Skip to content

Commit

Permalink
test: improve RegularNamedTemporaryFile
Browse files Browse the repository at this point in the history
accept keyword args like tempfile.NamedTemporaryFile
and split them properly to low-level calls
  • Loading branch information
ralphlange committed Jan 21, 2024
1 parent ddb5819 commit 93eaaec
Show file tree
Hide file tree
Showing 2 changed files with 13 additions and 7 deletions.
10 changes: 5 additions & 5 deletions src/p4p/test/test_gw.py
Original file line number Diff line number Diff line change
Expand Up @@ -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))

Expand Down Expand Up @@ -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':[{
Expand Down Expand Up @@ -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':[{
Expand Down Expand Up @@ -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':[{
Expand Down Expand Up @@ -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()
Expand Down
10 changes: 8 additions & 2 deletions src/p4p/test/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down

0 comments on commit 93eaaec

Please sign in to comment.