Skip to content

R
re6stnet
Commit 1b0048c3 authored by Tom Niget's avatar Tom Niget
Browse files
Options

more python2 to python3

parent 03f248d3
Show whitespace changes
Inline Side-by-side
Showing with 119 additions and 92 deletions
re6st/registry.py
View file @ 1b0048c3
......@@ -91,7 +91,7 @@ class RegistryServer(object):
"name TEXT PRIMARY KEY NOT NULL",
"value")
self.prefix = self.getConfig("prefix", None)
self.version = str(self.getConfig("version", b'\x00')) # BBB: blob
self.version = self.getConfig("version", b'\x00') # BBB: blob
utils.sqliteCreateTable(self.db, "token",
"token TEXT PRIMARY KEY NOT NULL",
"email TEXT NOT NULL",
......@@ -189,15 +189,15 @@ class RegistryServer(object):
self.sendto(self.prefix, 0)
# The following entry lists values that are base64-encoded.
kw[''] = 'version',
kw['version'] = base64.b64encode(self.version)
kw['version'] = base64.b64encode(self.version).decode("ascii")
self.network_config = kw
def increaseVersion(self):
x = utils.packInteger(1 + utils.unpackInteger(self.version)[0:1])
x = utils.packInteger(1 + utils.unpackInteger(self.version)[0])
self.version = x + self.cert.sign(x)
def sendto(self, prefix, code):
self.sock.sendto("%s\0%c" % (prefix, code), ('::1', tunnel.PORT))
self.sock.sendto(("%s\0%c" % (prefix, code)).encode("utf-8"), ('::1', tunnel.PORT))
def recv(self, code):
try:
......@@ -314,9 +314,11 @@ class RegistryServer(object):
except HTTPError as e:
return request.send_error(*e.args)
except:
logging.warning(request.requestline, exc_info=1)
logging.warning(request.requestline, exc_info=True)
return request.send_error(http.client.INTERNAL_SERVER_ERROR)
if result:
if type(result) is str:
result = result.encode("utf-8")
request.send_response(http.client.OK)
request.send_header("Content-Length", str(len(result)))
else:
......@@ -593,8 +595,8 @@ class RegistryServer(object):
hmac = [self.getConfig(k, None) for k in BABEL_HMAC]
for i, v in enumerate(v for v in hmac if v is not None):
config[('babel_hmac_sign', 'babel_hmac_accept')[i]] = \
v and base64.b64encode(x509.encrypt(cert, v))
return zlib.compress(json.dumps(config))
v and base64.b64encode(x509.encrypt(cert, v)).decode("ascii")
return zlib.compress(json.dumps(config).encode("utf-8"))
def _queryAddress(self, peer):
self.sendto(peer, 1)
......@@ -800,7 +802,7 @@ class RegistryClient(object):
_hmac = None
user_agent = "re6stnet/%s, %s" % (version.version, platform.platform())
def __init__(self, url, cert=None, auto_close=True):
def __init__(self, url, cert: x509.Cert=None, auto_close=True):
self.cert = cert
self.auto_close = auto_close
url_parsed = urlparse(url)
......@@ -862,7 +864,7 @@ class RegistryClient(object):
except HTTPError:
raise
except Exception:
logging.info(url, exc_info=1)
logging.info(url, exc_info=True)
else:
logging.info('%s\nUnexpected response %s %s',
url, response.status, response.reason)
......
re6st/tests/__init__.py
View file @ 1b0048c3
from pathlib2 import Path
from pathlib import Path
DEMO_PATH = Path(__file__).resolve().parent.parent.parent / "demo"
re6st/tests/test_end2end/test_registry_client.py
View file @ 1b0048c3
......@@ -60,7 +60,7 @@ class TestRegistryClientInteract(unittest.TestCase):
# read token from db
db = sqlite3.connect(str(self.server.db), isolation_level=None)
token = None
for _ in xrange(100):
for _ in range(100):
time.sleep(.1)
token = db.execute("SELECT token FROM token WHERE email=?",
(email,)).fetchone()
......
re6st/tests/test_network/network_build.py
View file @ 1b0048c3
......@@ -4,7 +4,7 @@ import nemu
import time
import weakref
from subprocess import PIPE
from pathlib2 import Path
from pathlib import Path
from re6st.tests import DEMO_PATH
......@@ -60,7 +60,7 @@ class NetManager(object):
Raise:
AssertionError
"""
for reg, nodes in self.registries.iteritems():
for reg, nodes in self.registries.items():
for node in nodes:
app0 = node.Popen(["ping", "-c", "1", reg.ip], stdout=PIPE)
ret = app0.wait()
......
re6st/tests/test_network/re6st_wrap.py
View file @ 1b0048c3
......@@ -8,11 +8,12 @@ import logging
import re
import shutil
import sqlite3
import sys
import tempfile
import time
import weakref
from subprocess import PIPE
from pathlib2 import Path
from pathlib import Path
from re6st.tests import tools
from re6st.tests import DEMO_PATH
......@@ -20,9 +21,9 @@ from re6st.tests import DEMO_PATH
WORK_DIR = Path(__file__).parent / "temp_net_test"
DH_FILE = DEMO_PATH / "dh2048.pem"
RE6STNET = "python -m re6st.cli.node"
RE6ST_REGISTRY = "python -m re6st.cli.registry"
RE6ST_CONF = "python -m re6st.cli.conf"
RE6STNET = f"{sys.executable} -m re6st.cli.node"
RE6ST_REGISTRY = f"{sys.executable} -m re6st.cli.registry"
RE6ST_CONF = f"{sys.executable} -m re6st.cli.conf"
def initial():
"""create the workplace"""
......@@ -72,7 +73,7 @@ class Re6stRegistry(object):
self.run()
# wait the servcice started
p = self.node.Popen(['python', '-c', """if 1:
p = self.node.Popen([sys.executable, '-c', """if 1:
import socket, time
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
while True:
......@@ -115,7 +116,7 @@ class Re6stRegistry(object):
'--client-count', (self.client_number+1)//2, '--port', self.port]
#PY3: convert PosixPath to str, can be remove in Python 3
cmd = map(str, cmd)
cmd = list(map(str, cmd))
cmd[:0] = RE6ST_REGISTRY.split()
......@@ -210,7 +211,7 @@ class Re6stNode(object):
# read token
db = sqlite3.connect(str(self.registry.db), isolation_level=None)
token = None
for _ in xrange(100):
for _ in range(100):
time.sleep(.1)
token = db.execute("SELECT token FROM token WHERE email=?",
(self.email,)).fetchone()
......@@ -223,7 +224,7 @@ class Re6stNode(object):
out, _ = p.communicate(str(token[0]))
# logging.debug("re6st-conf output: {}".format(out))
# find the ipv6 subnet of node
self.ip6 = re.search('(?<=subnet: )[0-9:a-z]+', out).group(0)
self.ip6 = re.search('(?<=subnet: )[0-9:a-z]+', out.decode("utf-8")).group(0)
data = {'ip6': self.ip6, 'hash': self.registry.ident}
with open(str(self.data_file), 'w') as f:
json.dump(data, f)
......@@ -236,7 +237,7 @@ class Re6stNode(object):
'--key', self.key, '-v4', '--registry', self.registry.url,
'--console', self.console]
#PY3: same as for Re6stRegistry.run
cmd = map(str, cmd)
cmd = list(map(str, cmd))
cmd[:0] = RE6STNET.split()
cmd += args
......
re6st/tests/test_network/test_net.py
View file @ 1b0048c3
"""contain ping-test for re6set net"""
import os
import sys
import unittest
import time
import psutil
import logging
import random
from pathlib2 import Path
from pathlib import Path
import network_build
import re6st_wrap
......@@ -47,12 +48,12 @@ def wait_stable(nodes, timeout=240):
for node in nodes:
sub_ips = set(ips) - {node.ip6}
node.ping_proc = node.node.Popen(
["python", PING_PATH, '--retry', '-a'] + list(sub_ips))
[sys.executable, PING_PATH, '--retry', '-a'] + list(sub_ips), env=os.environ)
# check all the node network can ping each other, in order reverse
unfinished = list(nodes)
while unfinished:
for i in xrange(len(unfinished)-1, -1, -1):
for i in range(len(unfinished)-1, -1, -1):
node = unfinished[i]
if node.ping_proc.poll() is not None:
logging.debug("%s 's network is stable", node.name)
......
re6st/tests/test_unit/test_conf.py
View file @ 1b0048c3
#!/usr/bin/python2
#!/usr/bin/env python3
""" unit test for re6st-conf
"""
......@@ -36,7 +36,7 @@ class TestConf(unittest.TestCase):
# mocked server cert and pkey
cls.pkey, cls.cert = create_ca_file(os.devnull, os.devnull)
cls.fingerprint = "".join( cls.cert.digest("sha1").split(":"))
cls.fingerprint = "".join( cls.cert.digest("sha1").decode("ascii").split(":"))
# client.getCa should return a string form cert
cls.cert = crypto.dump_certificate(crypto.FILETYPE_PEM, cls.cert)
......
re6st/tests/test_unit/test_registry.py
View file @ 1b0048c3
......@@ -13,12 +13,13 @@ import tempfile
from argparse import Namespace
from OpenSSL import crypto
from mock import Mock, patch
from pathlib2 import Path
from pathlib import Path
from re6st import registry
from re6st.tests.tools import *
from re6st.tests import DEMO_PATH
# TODO test for request_dump, requestToken, getNetworkConfig, getBoostrapPeer
# getIPV4Information, versions
......@@ -49,6 +50,7 @@ def insert_cert(cur, ca, prefix, not_after=None, email=None):
insert_cert.serial += 1
return key, cert
insert_cert.serial = 0
......@@ -80,14 +82,23 @@ class TestRegistryServer(unittest.TestCase):
+ "@mail.com"
def test_recv(self):
recv = self.server.sock.recv = Mock()
recv.side_effect = [
side_effect = iter([
"0001001001001a_msg",
"0001001001002\0001dqdq",
"0001001001001\000a_msg",
"0001001001001\000\4a_msg",
"0000000000000\0" # ERROR, IndexError: msg is null
]
])
class SocketProxy:
def __init__(self, wrappee):
self.wrappee = wrappee
self.recv = lambda _: next(side_effect)
def __getattr__(self, attr):
return getattr(self.wrappee, attr)
self.server.sock = SocketProxy(self.server.sock)
try:
res1 = self.server.recv(4)
......@@ -115,7 +126,7 @@ class TestRegistryServer(unittest.TestCase):
now = int(time.time()) - self.config.grace_period + 20
# makeup data
insert_cert(cur, self.server.cert, prefix_old, 1)
insert_cert(cur, self.server.cert, prefix, now -1)
insert_cert(cur, self.server.cert, prefix, now - 1)
cur.execute("INSERT INTO token VALUES (?,?,?,?)",
(token_old, self.email, 4, 2))
cur.execute("INSERT INTO token VALUES (?,?,?,?)",
......@@ -143,7 +154,7 @@ class TestRegistryServer(unittest.TestCase):
prefix = "0000000011111111"
method = "func"
protocol = 7
params = {"cn" : prefix, "a" : 1, "b" : 2}
params = {"cn": prefix, "a": 1, "b": 2}
func.getcallargs.return_value = params
del func._private
func.return_value = result = b"this_is_a_result"
......@@ -176,12 +187,12 @@ class TestRegistryServer(unittest.TestCase):
def test_handle_request_private(self, func):
"""case request with _private attr"""
method = "func"
params = {"a" : 1, "b" : 2}
params = {"a": 1, "b": 2}
func.getcallargs.return_value = params
func.return_value = None
request_good = Mock()
request_good.client_address = self.config.authorized_origin
request_good.headers = {'X-Forwarded-For':self.config.authorized_origin[0]}
request_good.headers = {'X-Forwarded-For': self.config.authorized_origin[0]}
request_bad = Mock()
request_bad.client_address = ["wrong_address"]
......@@ -282,7 +293,7 @@ class TestRegistryServer(unittest.TestCase):
nb_less = 0
for cert in self.server.iterCert():
s = cert[0].get_subject().serialNumber
if(s and int(s) <= serial):
if (s and int(s) <= serial):
nb_less += 1
self.assertEqual(nb_less, serial)
......@@ -378,7 +389,7 @@ class TestRegistryServer(unittest.TestCase):
hmacs = get_hmac()
key_1 = hmacs[1]
self.assertEqual(hmacs, [None, key_1, ''])
self.assertEqual(hmacs, [None, key_1, b''])
# step 2
self.server.updateHMAC()
......@@ -397,12 +408,11 @@ class TestRegistryServer(unittest.TestCase):
self.assertEqual(get_hmac(), [None, key_2, key_1])
#setp 5
# setp 5
self.server.updateHMAC()
self.assertEqual(get_hmac(), [key_2, None, None])
def test_getNodePrefix(self):
# prefix in short format
prefix = "0000000101"
......@@ -426,19 +436,33 @@ class TestRegistryServer(unittest.TestCase):
('0000000000000001', '2 0/16 6/16')
]
recv.side_effect = recv_case
def side_effct(rlist, wlist, elist, timeout):
# rlist is true until the len(recv_case)th call
side_effct.i -= side_effct.i > 0
return [side_effct.i, wlist, None]
side_effct.i = len(recv_case) + 1
select.side_effect = side_effct
res = self.server.topology()
expect_res = '{"36893488147419103232/80": ["0/16", "7/16"], ' \
'"": ["36893488147419103232/80", "3/16", "1/16", "0/16", "7/16"], ' \
'"4/16": ["0/16"], "3/16": ["0/16", "7/16"], "0/16": ["6/16", "7/16"], '\
'"1/16": ["6/16", "0/16"], "7/16": ["6/16", "4/16"]}'''
class CustomDecoder(json.JSONDecoder):
def __init__(self, **kwargs):
json.JSONDecoder.__init__(self, **kwargs)
self.parse_array = self.JSONArray
self.scan_once = json.scanner.py_make_scanner(self)
def JSONArray(self, s_and_end, scan_once, **kwargs):
values, end = json.decoder.JSONArray(s_and_end, scan_once, **kwargs)
return set(values), end
res = json.loads(res, cls=CustomDecoder)
expect_res = {"36893488147419103232/80": {"0/16", "7/16"},
"": {"36893488147419103232/80", "3/16", "1/16", "0/16", "7/16"}, "4/16": {"0/16"},
"3/16": {"0/16", "7/16"}, "0/16": {"6/16", "7/16"}, "1/16": {"6/16", "0/16"},
"7/16": {"6/16", "4/16"}}
self.assertEqual(res, expect_res)
......
re6st/tests/test_unit/test_registry_client.py
View file @ 1b0048c3
......@@ -52,9 +52,9 @@ class TestRegistryClient(unittest.TestCase):
self.client._hmac = None
self.client.hello = Mock(return_value = "aaabbb")
self.client.cert = Mock()
key = "this_is_a_key"
key = b"this_is_a_key"
self.client.cert.decrypt.return_value = key
h = hmac.HMAC(key, query, hashlib.sha1).digest()
h = hmac.HMAC(key, query.encode("ascii"), hashlib.sha1).digest()
key = hashlib.sha1(key).digest()
# response part
body = b'this is a body'
......
re6st/tests/test_unit/test_tunnel/test_base_tunnel_manager.py
View file @ 1b0048c3
#!/usr/bin/python2
#!/usr/bin/env python3
import os
import sys
import unittest
......@@ -67,7 +67,7 @@ class testBaseTunnelManager(unittest.TestCase):
# @patch("re6st.tunnel.BaseTunnelManager._makeTunnel", create=True)
# def test_processPacket_address_with_msg_peer(self, makeTunnel):
# """code is 1, peer and msg not none """
# c = chr(1)
# c = b"\x01"
# msg = "address"
# peer = x509.Peer("000001")
# self.tunnel._connecting = {peer}
......@@ -81,7 +81,7 @@ class testBaseTunnelManager(unittest.TestCase):
def test_processPacket_address(self):
"""code is 1, for address. And peer or msg are none"""
c = chr(1)
c = b"\x01"
self.tunnel._address = {1: "1,1", 2: "2,2"}
res = self.tunnel._processPacket(c)
......@@ -95,7 +95,7 @@ class testBaseTunnelManager(unittest.TestCase):
and each address join by ;
it will truncate address which has more than 3 element
"""
c = chr(1)
c = b"\x01"
peer = x509.Peer("000001")
peer.protocol = 1
self.tunnel._peers.append(peer)
......@@ -111,11 +111,11 @@ class testBaseTunnelManager(unittest.TestCase):
"""code is 0, for network version, peer is not none
2 case, one modify the version, one not
"""
c = chr(0)
c = b"\x00"
peer = x509.Peer("000001")
version1 = "00003"
version2 = "00007"
self.tunnel._version = version3 = "00005"
version1 = b"00003"
version2 = b"00007"
self.tunnel._version = version3 = b"00005"
self.tunnel._peers.append(peer)
res = self.tunnel._processPacket(c + version1, peer)
......
re6st/tests/test_unit/test_tunnel/test_multi_gateway_manager.py
View file @ 1b0048c3
#!/usr/bin/python2
#!/usr/bin/env python3
import os
import sys
import unittest
......
re6st/tests/tools.py
View file @ 1b0048c3
......@@ -30,9 +30,9 @@ def generate_cert(ca, ca_key, csr, prefix, serial, not_after=None):
return
crypto.X509Cert in pem format
"""
if type(ca) is str:
if type(ca) is bytes:
ca = crypto.load_certificate(crypto.FILETYPE_PEM, ca)
if type(ca_key) is str:
if type(ca_key) is bytes:
ca_key = crypto.load_privatekey(crypto.FILETYPE_PEM, ca_key)
req = crypto.load_certificate_request(crypto.FILETYPE_PEM, csr)
......@@ -84,9 +84,9 @@ def create_ca_file(pkey_file, cert_file, serial=0x120010db80042):
cert.set_pubkey(key)
cert.sign(key, "sha512")
with open(pkey_file, 'w') as pkey_file:
with open(pkey_file, 'wb') as pkey_file:
pkey_file.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, key))
with open(cert_file, 'w') as cert_file:
with open(cert_file, 'wb') as cert_file:
cert_file.write(crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
return key, cert
......
re6st/tunnel.py
View file @ 1b0048c3
......@@ -329,7 +329,7 @@ class BaseTunnelManager(object):
def _getPeer(self, prefix):
return self._peers[bisect(self._peers, prefix) - 1]
def sendto(self, prefix, msg):
def sendto(self, prefix: str, msg):
to = utils.ipFromBin(self._network + prefix), PORT
peer = self._getPeer(prefix)
if peer.prefix != prefix:
......@@ -360,6 +360,7 @@ class BaseTunnelManager(object):
if address[0] == '::1':
try:
prefix, msg = msg.split(b'\0', 1)
prefix = prefix.decode()
int(prefix, 2)
except ValueError:
return
......@@ -371,7 +372,7 @@ class BaseTunnelManager(object):
if msg:
self._sendto(to, '%s\0%c%s' % (prefix, code, msg))
else:
self.sendto(prefix, chr(code | 0x80) + msg[1:])
self.sendto(prefix, bytes([code | 0x80]) + msg[1:])
return
try:
sender = utils.binFromIp(address[0])
......@@ -384,7 +385,7 @@ class BaseTunnelManager(object):
msg = peer.decode(msg)
if type(msg) is tuple:
seqno, msg, protocol = msg
def handleHello(peer, seqno, msg, retry):
def handleHello(peer, seqno, msg: bytes, retry):
if seqno == 2:
i = len(msg) // 2
h = msg[:i]
......@@ -394,7 +395,7 @@ class BaseTunnelManager(object):
except (AttributeError, crypto.Error, x509.NewSessionError,
subprocess.CalledProcessError):
logging.debug('ignored new session key from %r',
address, exc_info=1)
address, exc_info=True)
return
peer.version = self._version \
if self._sendto(to, b'\0' + self._version, peer) else b''
......
re6st/upnpigd.py
View file @ 1b0048c3
......@@ -69,7 +69,7 @@ class Forwarder(object):
try:
return self._refresh()
except UPnPException as e:
logging.debug("UPnP failure", exc_info=1)
logging.debug("UPnP failure", exc_info=True)
self.clear()
try:
self.discover()
......
re6st/utils.py
View file @ 1b0048c3
import argparse, errno, fcntl, hashlib, logging, os, select as _select
import shlex, signal, socket, sqlite3, struct, subprocess
import sys, textwrap, threading, time, traceback
from typing import Optional
# PY3: It will be even better to use Popen(pass_fds=...),
# and then socket.SOCK_CLOEXEC will be useless.
......@@ -37,12 +38,12 @@ class FileHandler(logging.FileHandler):
finally:
self.lock.release()
# In the rare case _reopen is set just before the lock was released
if self._reopen and self.lock.acquire(0):
if self._reopen and self.lock.acquire(False):
self.release()
def async_reopen(self, *_):
self._reopen = True
if self.lock.acquire(0):
if self.lock.acquire(False):
self.release()
def setupLog(log_level, filename=None, **kw):
......@@ -150,7 +151,7 @@ class exit(object):
def handler(*args):
if self.status is None:
self.status = status
if self.acquire(0):
if self.acquire(False):
self.release()
for sig in sigs:
signal.signal(sig, handler)
......@@ -263,7 +264,7 @@ newHmacSecret = newHmacSecret()
# - there's always a unique way to encode a value
# - the 3 first bits code the number of bytes
def packInteger(i):
def packInteger(i: int) -> bytes:
for n in range(8):
x = 32 << 8 * n
if i < x:
......@@ -271,7 +272,7 @@ def packInteger(i):
i -= x
raise OverflowError
def unpackInteger(x):
def unpackInteger(x: bytes) -> Optional[tuple[int, int]]:
n = x[0] >> 5
try:
i, = struct.unpack("!Q", b'\0' * (7 - n) + x[:n+1])
......
re6st/x509.py
View file @ 1b0048c3
......@@ -52,7 +52,7 @@ def maybe_renew(path, cert, info, renew, force=False):
if time.time() < next_renew:
return cert, next_renew
try:
pem = renew()
pem: bytes = renew()
if not pem or pem == crypto.dump_certificate(
crypto.FILETYPE_PEM, cert):
exc_info = 0
......@@ -62,7 +62,7 @@ def maybe_renew(path, cert, info, renew, force=False):
exc_info = 1
break
new_path = path + '.new'
with open(new_path, 'w') as f:
with open(new_path, 'wb') as f:
f.write(pem)
try:
s = os.stat(path)
......@@ -90,9 +90,9 @@ class Cert(object):
self.ca_path = ca
self.cert_path = cert
self.key_path = key
with open(ca) as f:
with open(ca, "rb") as f:
self.ca = crypto.load_certificate(crypto.FILETYPE_PEM, f.read())
with open(key) as f:
with open(key, "rb") as f:
self.key = crypto.load_privatekey(crypto.FILETYPE_PEM, f.read())
if cert:
with open(cert) as f:
......@@ -152,13 +152,13 @@ class Cert(object):
raise VerifyError(int(code), int(depth), msg.strip())
return r
def verify(self, sign, data):
def verify(self, sign: bytes, data):
crypto.verify(self.ca, sign, data, 'sha512')
def sign(self, data):
def sign(self, data) -> bytes:
return crypto.sign(self.key, data, 'sha512')
def decrypt(self, data):
def decrypt(self, data: bytes) -> bytes:
p = openssl('rsautl', '-decrypt', '-inkey', self.key_path)
out, err = p.communicate(data)
if p.returncode:
......@@ -209,7 +209,7 @@ class Peer(object):
stop_date = float('inf')
version = b''
def __init__(self, prefix):
def __init__(self, prefix: str):
self.prefix = prefix
@property
......@@ -253,7 +253,7 @@ class Peer(object):
def _hmac(self, msg):
return hmac.HMAC(self._key, msg, hashlib.sha1).digest()
def newSession(self, key, protocol):
def newSession(self, key: bytes, protocol):
if key <= self._key:
raise NewSessionError(self._key, key)
self._key = key
......
setup.py
View file @ 1b0048c3
......@@ -15,7 +15,7 @@ def copy_file(self, infile, outfile, *args, **kw):
if infile == version["__file__"]:
if not self.dry_run:
log.info("generating %s -> %s", infile, outfile)
with open(outfile, "wb") as f:
with open(outfile, "w", encoding="utf-8") as f:
for x in sorted(version.items()):
if not x[0].startswith("_"):
f.write("%s = %r\n" % x)
......@@ -33,11 +33,8 @@ def copy_file(self, infile, outfile, *args, **kw):
executable = self.distribution.command_obj['build'].executable
patched = "#!%s%s\n" % (executable, m.group(1) or '')
patched += src.read()
dst = os.open(outfile, os.O_CREAT | os.O_WRONLY | os.O_TRUNC)
try:
os.write(dst, patched)
finally:
os.close(dst)
with open(outfile, "w") as dst:
dst.write(patched)
return outfile, 1
cls, = self.__class__.__bases__
return cls.copy_file(self, infile, outfile, *args, **kw)
......@@ -97,7 +94,7 @@ setup(
extras_require = {
'geoip': ['geoip2'],
'multicast': ['PyYAML'],
'test': ['mock', 'pathlib2', 'nemu', 'python-unshare', 'python-passfd', 'multiping']
'test': ['mock', 'nemu3', 'unshare', 'multiping']
},
#dependency_links = [
# "http://miniupnp.free.fr/files/download.php?file=miniupnpc-1.7.20120714.tar.gz#egg=miniupnpc-1.7",
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7