From 74f7d7f13206b3f7705921f066ddc18c6daa8456 Mon Sep 17 00:00:00 2001 From: Gianfranco Mauro <95578129+gianfrancomauro@users.noreply.github.com> Date: Sat, 30 Nov 2024 15:57:01 +0100 Subject: [PATCH] Add tests for multiprocessing forkserver and dependencies * **Lib/test/test_multiprocessing_forkserver.py** - Add tests for shared memory access, process return code, and signal handling in forkserver. * **Lib/test/test_dependencies.py** - Add tests to check for the existence of `pyvenv.cfg` and availability of `multiprocessing.shared_memory`, `ssl`, `pdb`, and `warnings` modules. --- Lib/test/test_dependencies.py | 36 ++++++++++++++++ Lib/test/test_multiprocessing_forkserver.py | 48 +++++++++++++++++++++ 2 files changed, 84 insertions(+) create mode 100644 Lib/test/test_dependencies.py create mode 100644 Lib/test/test_multiprocessing_forkserver.py diff --git a/Lib/test/test_dependencies.py b/Lib/test/test_dependencies.py new file mode 100644 index 00000000000000..4a22cf236bbd3b --- /dev/null +++ b/Lib/test/test_dependencies.py @@ -0,0 +1,36 @@ +import unittest +import os +import sys + +class TestDependencies(unittest.TestCase): + + def test_pyvenv_cfg_exists(self): + venv_path = os.path.join(sys.prefix, 'pyvenv.cfg') + self.assertTrue(os.path.exists(venv_path), f"{venv_path} does not exist") + + def test_shared_memory_access(self): + try: + import multiprocessing.shared_memory + except ImportError: + self.fail("multiprocessing.shared_memory module is not available") + + def test_ssl_module(self): + try: + import ssl + except ImportError: + self.fail("ssl module is not available") + + def test_pdb_module(self): + try: + import pdb + except ImportError: + self.fail("pdb module is not available") + + def test_warnings_module(self): + try: + import warnings + except ImportError: + self.fail("warnings module is not available") + +if __name__ == '__main__': + unittest.main() diff --git a/Lib/test/test_multiprocessing_forkserver.py b/Lib/test/test_multiprocessing_forkserver.py new file mode 100644 index 00000000000000..16afa54091c18a --- /dev/null +++ b/Lib/test/test_multiprocessing_forkserver.py @@ -0,0 +1,48 @@ +import unittest +import multiprocessing +import os +import signal +import time + +class TestForkserver(unittest.TestCase): + + def setUp(self): + self.manager = multiprocessing.Manager() + self.shared_dict = self.manager.dict() + + def tearDown(self): + self.manager.shutdown() + + def test_shared_memory_access(self): + def worker(shared_dict): + shared_dict['key'] = 'value' + + process = multiprocessing.Process(target=worker, args=(self.shared_dict,)) + process.start() + process.join() + + self.assertEqual(self.shared_dict['key'], 'value') + + def test_forkserver_process_return_code(self): + def worker(): + os._exit(1) + + process = multiprocessing.Process(target=worker) + process.start() + process.join() + + self.assertEqual(process.exitcode, 1) + + def test_forkserver_signal_handling(self): + def worker(): + time.sleep(5) + + process = multiprocessing.Process(target=worker) + process.start() + os.kill(process.pid, signal.SIGTERM) + process.join() + + self.assertEqual(process.exitcode, -signal.SIGTERM) + +if __name__ == '__main__': + unittest.main()