From 93eaaec920f5a3919645124c0cc4164f1d6973ea Mon Sep 17 00:00:00 2001 From: Ralph Lange Date: Sun, 21 Jan 2024 15:16:53 +0100 Subject: [PATCH] test: improve RegularNamedTemporaryFile accept keyword args like tempfile.NamedTemporaryFile and split them properly to low-level calls --- src/p4p/test/test_gw.py | 10 +++++----- src/p4p/test/utils.py | 10 ++++++++-- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/src/p4p/test/test_gw.py b/src/p4p/test/test_gw.py index f740fc5..bf04f9d 100644 --- a/src/p4p/test/test_gw.py +++ b/src/p4p/test/test_gw.py @@ -225,7 +225,7 @@ def test_monitor(self): Q2 = Queue(maxsize=4) with self._ds_client.monitor('pv:ro', Q1.put, notify_disconnect=True): - + self.assertIsInstance(Q1.get(timeout=self.timeout), Disconnected) self.assertEqual(42, Q1.get(timeout=self.timeout)) @@ -296,7 +296,7 @@ def put(pv, op): _log.debug("Exit setUp") def setUpGW(self, usconfig): - cfile = self._cfile = NamedTemporaryFile('w+') + cfile = self._cfile = NamedTemporaryFile(mode='w+') json.dump({ 'version':2, 'clients':[{ @@ -474,7 +474,7 @@ class TestHighLevelChained(TestHighLevel): def setUpGW(self, usconfig): # First GW, connected to upstream server - cfile = self._cfile = NamedTemporaryFile('w+') + cfile = self._cfile = NamedTemporaryFile(mode='w+') json.dump({ 'version':2, 'clients':[{ @@ -510,7 +510,7 @@ def setUpGW(self, usconfig): gw1config = self._app1.servers[u'server1_0'].conf() # Second GW, connected to first - cfile = self._cfile = NamedTemporaryFile('w+') + cfile = self._cfile = NamedTemporaryFile(mode='w+') json.dump({ 'version':2, 'clients':[{ @@ -635,7 +635,7 @@ def log(self): return self._log.read() def write(self, content): - F = NamedTemporaryFile('w+') + F = NamedTemporaryFile(mode='w+') self._files.append(F) F.write(content) F.flush() diff --git a/src/p4p/test/utils.py b/src/p4p/test/utils.py index 4b8d150..3b3aa78 100644 --- a/src/p4p/test/utils.py +++ b/src/p4p/test/utils.py @@ -170,9 +170,15 @@ class RegularNamedTemporaryFile(object): """Like tempfile.NamedTemporaryFile which doesn't use O_TEMPORARY on windows """ def __init__(self, *args, **kws): - fd, self.name = tempfile.mkstemp(**kws) + mkstemp_kws = {k: v for k, v in kws.items() + if k in ['suffix', 'prefix', 'dir']} + if 'mode' in kws: + mode = kws['mode'] + else: + mode = None + fd, self.name = tempfile.mkstemp(**mkstemp_kws) try: - self.file = os.fdopen(fd, *args, **kws) + self.file = os.fdopen(fd, mode) self.read = self.file.read self.write = self.file.write self.flush = self.file.flush