# HG changeset patch # User Manuel Jacob # Date 2020-07-09 10:52:04 # Node ID eb26a9cf78214e76c4cc23f82c8a9121300f264e # Parent a5fa2761a6cd8b328f060fa804dfe92e5d5d9baa procutil: avoid use of deprecated tempfile.mktemp() In the previous version, I used tempfile.mktemp() because it seemed to be the only way to open a file from two processes (the Python documentation says the file backing NamedTemporaryFile can’t be opened a second time on Windows). However, it’s possible when passing the O_TEMPORARY flag to the second open. Source: https://stackoverflow.com/a/15235559/6366251 diff --git a/tests/test-stdio.py b/tests/test-stdio.py --- a/tests/test-stdio.py +++ b/tests/test-stdio.py @@ -34,6 +34,7 @@ FULLY_BUFFERED = b'[written aaa][written TEST_LARGE_WRITE_CHILD_SCRIPT = r''' +import os import signal import sys @@ -43,7 +44,10 @@ from mercurial.utils import procutil signal.signal(signal.SIGINT, lambda *x: None) dispatch.initstdio() write_result = procutil.{stream}.write(b'x' * 1048576) -with open({write_result_fn}, 'w') as write_result_f: +with os.fdopen( + os.open({write_result_fn!r}, os.O_WRONLY | getattr(os, 'O_TEMPORARY', 0)), + 'w', +) as write_result_f: write_result_f.write(str(write_result)) ''' @@ -201,8 +205,7 @@ class TestStdio(unittest.TestCase): ) def post_child_check(): - with open(write_result_fn, 'r') as write_result_f: - write_result_str = write_result_f.read() + write_result_str = write_result_f.read() if pycompat.ispy3: # On Python 3, we test that the correct number of bytes is # claimed to have been written. @@ -213,14 +216,10 @@ class TestStdio(unittest.TestCase): expected_write_result_str = 'None' self.assertEqual(write_result_str, expected_write_result_str) - try: - # tempfile.mktemp() is unsafe in general, as a malicious process - # could create the file before we do. But in tests, we're running - # in a controlled environment. - write_result_fn = tempfile.mktemp() + with tempfile.NamedTemporaryFile('r') as write_result_f: self._test( TEST_LARGE_WRITE_CHILD_SCRIPT.format( - stream=stream, write_result_fn=repr(write_result_fn) + stream=stream, write_result_fn=write_result_f.name ), stream, rwpair_generator, @@ -228,11 +227,6 @@ class TestStdio(unittest.TestCase): python_args, post_child_check=post_child_check, ) - finally: - try: - os.unlink(write_result_fn) - except OSError: - pass def test_large_write_stdout_devnull(self): self._test_large_write('stdout', _devnull)