python3: clean up system resources (pipes, processes)

Python3 added automatic detection of leaked resources, which causes lots of warnings in Re6st, mostly because of unclosed subprocess streams.

demo/demo

@@ -74,7 +74,8 @@ if args.duration:
     signal.signal(signal.SIGALRM, handler)
     signal.alarm(args.duration)
 
-exec(compile(open("fixnemu.py", "rb").read(), "fixnemu.py", 'exec'))
+with open('fixnemu.py') as f:
+    exec(compile(f.read(), 'fixnemu.py', 'exec'))
 
 class Re6stNode(nemu.Node):
     name: str

re6st/cli/conf.py

@@ -76,9 +76,10 @@ def main():
         print("WARNING: it is strongly recommended to use --fingerprint option.")
     network = x509.networkFromCa(ca)
     if config.is_needed:
-        route, err = subprocess.Popen(('ip', '-6', '-o', 'route', 'get',
+        with subprocess.Popen(('ip', '-6', '-o', 'route', 'get',
                                utils.ipFromBin(network)),
-                                      stdout=subprocess.PIPE).communicate()
+                              stdout=subprocess.PIPE) as proc:
+            route, err = proc.communicate()
         sys.exit(err or route and
             utils.binFromIp(route.split()[8]).startswith(network))
 
@@ -89,7 +90,7 @@ def main():
     reserved = 'CN', 'serial'
     req = crypto.X509Req()
     try:
-        with open(cert_path) as f:
+        with open(cert_path, "rb") as f:
             cert = loadCert(f.read())
         components = {k.decode(): v for k, v in cert.get_subject().get_components()}
         for k in reserved:

re6st/cli/node.py

@@ -266,14 +266,8 @@ def main():
 
     def call(cmd):
         logging.debug('%r', cmd)
-        p = subprocess.Popen(cmd, stdout=subprocess.PIPE,
-                                  stderr=subprocess.PIPE)
-        stdout, stderr = p.communicate()
-        if p.returncode:
-            raise EnvironmentError("%r failed with error %u\n%s"
-                                   % (' '.join(cmd), p.returncode, stderr))
-        return stdout
-    def ip4(object, *args):
+        return subprocess.run(cmd, capture_output=True, check=True).stdout
+    def ip4(object: str, *args):
         args = ['ip', '-4', object, 'add'] + list(args)
         call(args)
         args[3] = 'del'

re6st/registry.py

@@ -155,6 +155,11 @@ class RegistryServer:
         else:
             self.newHMAC(0)
 
+    def close(self):
+        self.sock.close()
+        self.db.close()
+        self.ctl.close()
+
     def getConfig(self, name, *default):
         r, = next(self.db.execute(
             "SELECT value FROM config WHERE name=?", (name,)), default)

re6st/tests/test_end2end/test_registry_client.py

@@ -41,7 +41,8 @@ class TestRegistryClientInteract(unittest.TestCase):
         self.max_clients = 10
 
     def tearDown(self):
-        self.server.proc.terminate()
+        with self.server.proc as p:
+            p.terminate()
 
     def test_1_main(self):
         """ a client interact a server, no re6stnet node test basic function"""

re6st/tests/test_network/network_build.py

@@ -3,7 +3,7 @@ import logging
 import nemu
 import time
 import weakref
-from subprocess import PIPE
+from subprocess import DEVNULL, PIPE
 from pathlib import Path
 
 from re6st.tests import DEMO_PATH
@@ -62,7 +62,7 @@ class NetManager:
         """
         for reg, nodes in self.registries.items():
             for node in nodes:
-                app0 = node.Popen(["ping", "-c", "1", reg.ip], stdout=PIPE)
+                with node.Popen(["ping", "-c", "1", reg.ip], stdout=DEVNULL) as app0:
                     ret = app0.wait()
                 if ret:
                     raise ConnectableError(

re6st/tests/test_network/re6st_wrap.py

@@ -134,13 +134,17 @@ class Re6stRegistry:
             if e.errno != errno.ENOENT:
                 raise
 
-    def __del__(self):
+    def terminate(self):
         try:
             logging.debug("teminate process %s", self.proc.pid)
-            self.proc.destroy()
+            with self.proc as p:
+                p.destroy()
         except:
             pass
 
+    def __del__(self):
+        self.terminate()
+
 
 class Re6stNode:
     """class run a re6stnet service on a namespace"""
@@ -263,7 +267,8 @@ class Re6stNode:
     def stop(self):
         """stop running re6stnet process"""
         logging.debug("%s teminate process %s", self.name, self.proc.pid)
-        self.proc.destroy()
+        with self.proc as p:
+            p.destroy()
 
     def __del__(self):
         """teminate process and rm temp dir"""

re6st/tests/test_network/test_net.py

@@ -12,25 +12,6 @@ from . import network_build, re6st_wrap
 
 PING_PATH = str(Path(__file__).parent.resolve() / "ping.py")
 
-def deploy_re6st(nm, recreate=False):
-    net = nm.registries
-    nodes = []
-    registries = []
-    re6st_wrap.Re6stRegistry.registry_seq = 0
-    re6st_wrap.Re6stNode.node_seq = 0
-    for registry in net:
-        reg = re6st_wrap.Re6stRegistry(registry, "2001:db8:42::", len(net[registry]),
-                                       recreate=recreate)
-        reg_node = re6st_wrap.Re6stNode(registry, reg, name=reg.name)
-        registries.append(reg)
-        reg_node.run("--gateway", "--disable-proto", "none", "--ip", registry.ip)
-        nodes.append(reg_node)
-        for m in net[registry]:
-            node = re6st_wrap.Re6stNode(m, reg)
-            node.run("-i" + m.iface.name)
-            nodes.append(node)
-    return nodes, registries
-
 def wait_stable(nodes, timeout=240):
     """try use ping6 from each node to the other until ping success to all the
     other nodes
@@ -77,6 +58,36 @@ class TestNet(unittest.TestCase):
         logging.basicConfig(level=logging.INFO)
         re6st_wrap.initial()
 
+    def deploy_re6st(self, nm, recreate=False):
+        net = nm.registries
+        nodes = []
+        registries = []
+        re6st_wrap.Re6stRegistry.registry_seq = 0
+        re6st_wrap.Re6stNode.node_seq = 0
+        for registry in net:
+            reg = re6st_wrap.Re6stRegistry(registry, "2001:db8:42::", len(net[registry]),
+                                           recreate=recreate)
+            reg_node = re6st_wrap.Re6stNode(registry, reg, name=reg.name)
+            registries.append(reg)
+            reg_node.run("--gateway", "--disable-proto", "none", "--ip", registry.ip)
+            nodes.append(reg_node)
+            for m in net[registry]:
+                node = re6st_wrap.Re6stNode(m, reg)
+                node.run("-i" + m.iface.name)
+                nodes.append(node)
+
+        def clean_re6st():
+            for node in nodes:
+                node.node.destroy()
+                node.stop()
+
+            for reg in registries:
+                reg.terminate()
+
+        self.addCleanup(clean_re6st)
+
+        return nodes, registries
+
     @classmethod
     def tearDownClass(cls):
         """watch any process leaked after tests"""
@@ -94,7 +105,7 @@ class TestNet(unittest.TestCase):
         """create a network in a net segment, test the connectivity by ping
         """
         nm = network_build.net_route()
-        nodes, _ = deploy_re6st(nm)
+        nodes, registries = self.deploy_re6st(nm)
 
         wait_stable(nodes, 40)
         time.sleep(10)
@@ -107,7 +118,7 @@ class TestNet(unittest.TestCase):
         then test if network recover, this test seems always failed
         """
         nm = network_build.net_demo()
-        nodes, _ = deploy_re6st(nm)
+        nodes, registries = self.deploy_re6st(nm)
 
         wait_stable(nodes, 100)
 
@@ -126,7 +137,7 @@ class TestNet(unittest.TestCase):
         then test if network recover,
         """
         nm = network_build.net_route()
-        nodes, _ = deploy_re6st(nm)
+        nodes, registries = self.deploy_re6st(nm)
 
         wait_stable(nodes, 40)
 

re6st/tests/test_unit/test_registry.py

@@ -70,6 +70,7 @@ class TestRegistryServer(unittest.TestCase):
 
     @classmethod
     def tearDownClass(cls):
+        cls.server.close()
         # remove database
         for file in [cls.config.db, cls.config.ca, cls.config.key]:
             try:

re6st/tests/tools.py

@@ -103,7 +103,7 @@ def decrypt(pkey, incontent):
     with open("node.key", 'w') as f:
         f.write(pkey.decode())
     args = "openssl rsautl -decrypt -inkey node.key".split()
-    p = subprocess.Popen(
-        args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    with subprocess.Popen(
+        args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as p:
         outcontent, err = p.communicate(incontent)
     return outcontent