Use ProcessPoolExecutor in tests instead of multiprocessing

Makes it easier to run functions in a separate processes because
`.result` returns the values directly instead of using Pipes.

Author: igo95862

test/test_namespaces.py

@@ -2,46 +2,44 @@
 # SPDX-FileCopyrightText: 2024 igo95862
 from __future__ import annotations
 
-from multiprocessing import Pipe
+from concurrent.futures import ProcessPoolExecutor
 from os import getuid
 from unittest import TestCase
 
 from lxns.namespaces import UserNamespace, unshare_namespaces
 
-from .test_os import run_in_subprocess
-
 
 class TestNamespaces(TestCase):
+    @staticmethod
+    def unshare_namespaces_test() -> tuple[int, int]:
+        user_namespace_id_before = UserNamespace.get_current_ns_id()
+        unshare_namespaces(user=True)
+        return user_namespace_id_before, UserNamespace.get_current_ns_id()
+
     def test_unshare_namespaces(self) -> None:
         current_user_ns_id = UserNamespace.get_current_ns_id()
 
-        read_pipe, write_pipe = Pipe()
-
-        def unshare_test() -> None:
-            write_pipe.send(UserNamespace.get_current_ns_id())
-            unshare_namespaces(user=True)
-            write_pipe.send(UserNamespace.get_current_ns_id())
+        with ProcessPoolExecutor() as executor:
+            u_ns_id_before, u_ns_id_after = executor.submit(
+                self.unshare_namespaces_test
+            ).result(3)
 
-        run_in_subprocess(unshare_test)
-        write_pipe.close()
-
-        current_user_ns_id = UserNamespace.get_current_ns_id()
+        self.assertEqual(current_user_ns_id, u_ns_id_before)
+        self.assertNotEqual(current_user_ns_id, u_ns_id_after)
 
-        self.assertEqual(current_user_ns_id, read_pipe.recv())
-        self.assertNotEqual(current_user_ns_id, read_pipe.recv())
-
-    def test_unshare_from_class(self) -> None:
+    @staticmethod
+    def unshare_from_class_test() -> tuple[int, int]:
         uid_before = getuid()
+        UserNamespace.unshare()
+        return uid_before, getuid()
 
-        read_pipe, write_pipe = Pipe()
-
-        def unshare_test() -> None:
-            write_pipe.send(getuid())
-            UserNamespace.unshare()
-            write_pipe.send(getuid())
+    def test_unshare_from_class(self) -> None:
+        uid_now = getuid()
 
-        run_in_subprocess(unshare_test)
-        write_pipe.close()
+        with ProcessPoolExecutor() as executor:
+            uid_before, uid_after = executor.submit(
+                self.unshare_from_class_test
+            ).result(3)
 
-        self.assertEqual(uid_before, read_pipe.recv())
-        self.assertNotEqual(uid_before, read_pipe.recv())
+        self.assertEqual(uid_now, uid_before)
+        self.assertNotEqual(uid_now, uid_after)

test/test_os.py

@@ -2,36 +2,31 @@
 # SPDX-FileCopyrightText: 2024 igo95862
 from __future__ import annotations
 
-from multiprocessing import Pipe, Process
+from concurrent.futures import ProcessPoolExecutor
 from os import getuid
-from typing import TYPE_CHECKING
 from unittest import TestCase
 
 from lxns.os import CLONE_NEWUSER, unshare
 
-if TYPE_CHECKING:
-    from collections.abc import Callable
 
+class TestLxnsOs(TestCase):
 
-def run_in_subprocess(func: Callable[[], None]) -> None:
-    subproc = Process(target=func)
-    subproc.start()
-    subproc.join(3)
+    @staticmethod
+    def unshare_test() -> tuple[int, int]:
+        uid_before_unshare = getuid()
+        unshare(CLONE_NEWUSER)
+        uid_after_unshare = getuid()
 
+        return uid_before_unshare, uid_after_unshare
 
-class TestLxnsOs(TestCase):
     def test_unshare_user_namespace(self) -> None:
-        uid_before = getuid()
-
-        read_pipe, write_pipe = Pipe()
+        uid_before_test = getuid()
 
-        def unshare_test() -> None:
-            write_pipe.send(getuid())
-            unshare(CLONE_NEWUSER)
-            write_pipe.send(getuid())
+        with ProcessPoolExecutor() as executor:
+            uid_before, uid_after = executor.submit(self.unshare_test).result(3)
 
-        run_in_subprocess(unshare_test)
-        write_pipe.close()
+        # Unit test process should not unshare
+        self.assertEqual(uid_before_test, getuid())
 
-        self.assertEqual(uid_before, read_pipe.recv())
-        self.assertNotEqual(uid_before, read_pipe.recv())
+        self.assertEqual(uid_before_test, uid_before)
+        self.assertNotEqual(uid_before_test, uid_after)