Mercurial > hg
changeset 45104:eb26a9cf7821
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
author | Manuel Jacob <me@manueljacob.de> |
---|---|
date | Thu, 09 Jul 2020 12:52:04 +0200 |
parents | a5fa2761a6cd |
children | 5d09a120b4be |
files | tests/test-stdio.py |
diffstat | 1 files changed, 8 insertions(+), 14 deletions(-) [+] |
line wrap: on
line diff
--- a/tests/test-stdio.py Wed Jul 15 15:09:21 2020 +0200 +++ b/tests/test-stdio.py Thu Jul 09 12:52:04 2020 +0200 @@ -34,6 +34,7 @@ TEST_LARGE_WRITE_CHILD_SCRIPT = r''' +import os import signal import sys @@ -43,7 +44,10 @@ 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 @@ ) 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 @@ 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 @@ 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)