| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,7 +1,47 @@ | ||
| import os | ||
| import re | ||
| import unittest | ||
| from multiprocessing.util import _SUN_PATH_MAX | ||
| from test._test_multiprocessing import install_tests_in_module_dict | ||
| from test.support import os_helper, script_helper | ||
|
|
||
| install_tests_in_module_dict(globals(), 'forkserver', exclude_types=True) | ||
|
|
||
|
|
||
| class TestForkServerConfiguration(unittest.TestCase): | ||
| def test_respect_sun_path_max(self): | ||
| # Ensure that the calculation for temporary filepath lengths is correct. | ||
| # See https://github.com/python/cpython/issues/149527. | ||
|
|
||
| cmd = '''if 1: | ||
| from multiprocessing.connection import arbitrary_address | ||
| from multiprocessing.util import get_temp_dir | ||
| if __name__ == "__main__": | ||
| print(get_temp_dir()) | ||
| print(arbitrary_address("AF_UNIX")) | ||
| ''' | ||
| with os_helper.temp_dir() as root: | ||
| self.assertLess(len(root), _SUN_PATH_MAX) | ||
| _, out, _ = script_helper.assert_python_ok('-c', cmd, TMPDIR=root) | ||
| res = out.decode().strip().splitlines() | ||
| self.assertEqual(len(res), 2) | ||
|
|
||
| # get_temp_dir() with a TMPDIR that has a too large length ends up | ||
| # NOT being used as it is not possible to use it. In particular, | ||
| # the corresponding path will fallback to one of the "built-in" | ||
| # ones such as '/tmp', '/var/tmp', or '/usr/tmp'. | ||
| root_regex = "|".join(map(re.escape, [root, "/tmp", "/var/tmp", "/usr/tmp"])) | ||
|
|
||
| temp_pymp = res[0] | ||
| self.assertLess(len(temp_pymp), _SUN_PATH_MAX) | ||
| temp_pymp_regex = os.path.join(root_regex, r"pymp-\w{8}") | ||
| self.assertRegex(temp_pymp, temp_pymp_regex) | ||
|
|
||
| temp_sock = res[1] | ||
| self.assertLess(len(temp_sock), _SUN_PATH_MAX) | ||
| temp_sock_regex = os.path.join(temp_pymp_regex, r"sock-[0-9a-fA-F]{12}") | ||
| self.assertRegex(temp_sock, temp_sock_regex) | ||
|
|
||
|
|
||
| if __name__ == '__main__': | ||
| unittest.main() |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| :mod:`multiprocessing`: fix creation of temporary socket files following | ||
| :gh:`137335` (`PR#148578 <https://github.com/python/cpython/pull/148578>`__ | ||
| and corresponding backports). Patch by Bénédikt Tran. |
| Back | FazBrowse Home | New Git URL |
Uh oh!
There was an error while loading. Please reload this page.