Skip to content

Commit c066518

Browse files
jsiirolajsiirola
committed
Improve test coverage
1 parent 15ccaef commit c066518

File tree

2 files changed

+205
-1
lines changed

2 files changed

+205
-1
lines changed

‎pyomo/common/tests/test_log.py

+6
Original file line numberDiff line numberDiff line change
@@ -526,6 +526,12 @@ def test_log_stream(self):
526526
ls.write("line 1\nline 2\n")
527527
self.assertEqual(OUT.getvalue(), "INFO: line 1\nINFO: line 2\n")
528528

529+
with LI as OUT:
530+
# empty writes do not generate log records
531+
ls.write("")
532+
ls.flush()
533+
self.assertEqual(OUT.getvalue(), "")
534+
529535
with LI as OUT:
530536
ls.write("line 1\nline 2")
531537
self.assertEqual(OUT.getvalue(), "INFO: line 1\n")

‎pyomo/common/tests/test_tee.py

+199-1
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020

2121
from io import StringIO, BytesIO
2222

23-
from pyomo.common.log import LoggingIntercept
23+
from pyomo.common.log import LoggingIntercept, LogStream
2424
import pyomo.common.unittest as unittest
2525
from pyomo.common.tempfiles import TempfileManager
2626
import pyomo.common.tee as tee
@@ -99,6 +99,23 @@ def test_err_and_out_are_different(self):
9999
self.assertIs(err, t.STDERR)
100100
self.assertIsNot(out, err)
101101

102+
def test_signal_flush(self):
103+
a = StringIO()
104+
with tee.TeeStream(a) as t:
105+
out = t.STDOUT
106+
self.assertIs(type(out), tee._SignalFlush)
107+
out.write("out1\n")
108+
out.writelines(["out2\n", "out3\n"])
109+
self.assertEqual(a.getvalue(), "out1\nout2\nout3\n")
110+
with tee.TeeStream(a) as t:
111+
err = t.STDERR
112+
self.assertIs(type(err), tee._AutoFlush)
113+
err.write("err1\n")
114+
err.writelines(["err2\n", "err3\n"])
115+
self.assertEqual(a.getvalue(), "out1\nout2\nout3\nerr1\nerr2\nerr3\n")
116+
with self.assertRaisesRegex(AttributeError, '.*is not writable'):
117+
tee.TeeStream().STDOUT.name = 'foo'
118+
102119
@unittest.skipIf(
103120
not tee._peek_available,
104121
"Requires the _mergedReader, but _peek_available==False",
@@ -259,6 +276,25 @@ def write(self, data):
259276
r"\nThe following was left in the output buffer:\n 'i\\n'\n$",
260277
)
261278

279+
280+
class TestCapture(unittest.TestCase):
281+
def setUp(self):
282+
self.streams = sys.stdout, sys.stderr
283+
self.reenable_gc = gc.isenabled()
284+
gc.disable()
285+
gc.collect()
286+
# Set a short switch interval so that the threading tests behave
287+
# as expected
288+
self.switchinterval = sys.getswitchinterval()
289+
sys.setswitchinterval(tee._poll_interval / 100)
290+
291+
def tearDown(self):
292+
sys.stdout, sys.stderr = self.streams
293+
sys.setswitchinterval(self.switchinterval)
294+
if self.reenable_gc:
295+
gc.enable()
296+
gc.collect()
297+
262298
def test_capture_output(self):
263299
out = StringIO()
264300
with tee.capture_output(out) as OUT:
@@ -287,6 +323,122 @@ def test_capture_output_logfile_string(self):
287323
result = f.read()
288324
self.assertEqual('HELLO WORLD\n', result)
289325

326+
logfile = os.path.join('path', 'to', 'nonexisting', 'file.txt')
327+
T = tee.capture_output(logfile)
328+
with self.assertRaisesRegex(FileNotFoundError, f".*{logfile}"):
329+
T.__enter__()
330+
self.assertEqual(T.context_stack, [])
331+
332+
def test_capture_to_logger(self):
333+
logger = logging.getLogger('_pyomo_no_logger')
334+
lstream = LogStream(logging.WARNING, logger)
335+
orig = logger.propagate, logger.handlers
336+
try:
337+
logger.propagate = False
338+
logger.handlers = []
339+
with LoggingIntercept(module='_pyomo_no_logger') as LOG:
340+
with tee.capture_output(lstream, capture_fd=False):
341+
sys.stderr.write("hi!\n")
342+
sys.stderr.flush()
343+
self.assertEqual(LOG.getvalue(), "hi!\n")
344+
345+
# test that we handle the lastResort logger correctly
346+
_lastResort = logging.lastResort
347+
with tee.capture_output() as OUT:
348+
with tee.capture_output(lstream, capture_fd=False):
349+
self.assertIsNot(_lastResort, logging.lastResort)
350+
sys.stderr.write("hi?\n")
351+
self.assertEqual(OUT.getvalue(), "hi?\n")
352+
353+
# test that we allow redirect-to-logger out
354+
with tee.capture_output() as OUT:
355+
logger.addHandler(logging.NullHandler())
356+
logger.addHandler(logging.StreamHandler(sys.stderr))
357+
with tee.capture_output(lstream, capture_fd=False):
358+
sys.stderr.write("hi.\n")
359+
self.assertEqual(OUT.getvalue(), "hi.\n")
360+
logger.handlers.clear()
361+
362+
# test a sub-logger
363+
lstream = LogStream(
364+
logging.WARNING, logging.getLogger('_pyomo_no_logger.foo')
365+
)
366+
with tee.capture_output() as OUT:
367+
logger.addHandler(logging.NullHandler())
368+
logger.addHandler(logging.StreamHandler(sys.stderr))
369+
with tee.capture_output(lstream, capture_fd=False):
370+
sys.stderr.write("hi,\n")
371+
self.assertEqual(OUT.getvalue(), "hi,\n")
372+
finally:
373+
logger.propagate, logger.handlers = orig
374+
375+
def test_capture_fd_to_logger(self):
376+
logger = logging.getLogger('_pyomo_no_logger')
377+
lstream = LogStream(logging.WARNING, logger)
378+
orig = logger.propagate, logger.handlers
379+
try:
380+
logger.propagate = False
381+
logger.handlers = []
382+
with LoggingIntercept(module='_pyomo_no_logger') as LOG:
383+
with tee.capture_output(lstream, capture_fd=True):
384+
sys.stderr.write("hi!\n")
385+
sys.stderr.flush()
386+
self.assertEqual(LOG.getvalue(), "hi!\n")
387+
388+
# test that we handle the lastResort logger correctly
389+
_lastResort = logging.lastResort
390+
with tee.capture_output() as OUT:
391+
with tee.capture_output(lstream, capture_fd=True):
392+
self.assertIsNot(_lastResort, logging.lastResort)
393+
sys.stderr.write("hi?\n")
394+
self.assertEqual(OUT.getvalue(), "hi?\n")
395+
396+
# test that we allow redirect-to-logger out
397+
with tee.capture_output() as OUT:
398+
logger.addHandler(logging.NullHandler())
399+
logger.addHandler(logging.StreamHandler(sys.stderr))
400+
with tee.capture_output(lstream, capture_fd=True):
401+
sys.stderr.write("hi.\n")
402+
self.assertEqual(OUT.getvalue(), "hi.\n")
403+
logger.handlers.clear()
404+
405+
# test a sub-logger
406+
lstream = LogStream(
407+
logging.WARNING, logging.getLogger('_pyomo_no_logger.foo')
408+
)
409+
with tee.capture_output() as OUT:
410+
logger.addHandler(logging.NullHandler())
411+
logger.addHandler(logging.StreamHandler(sys.stderr))
412+
with tee.capture_output(lstream, capture_fd=True):
413+
sys.stderr.write("hi,\n")
414+
self.assertEqual(OUT.getvalue(), "hi,\n")
415+
finally:
416+
logger.propagate, logger.handlers = orig
417+
418+
def test_no_fileno_stdout(self):
419+
T = tee.capture_output()
420+
with T:
421+
self.assertEqual(len(T.context_stack), 2)
422+
T = tee.capture_output(capture_fd=True)
423+
# out & err point to something other than fd 1 and 2
424+
sys.stdout = os.fdopen(os.dup(1), closefd=True)
425+
sys.stderr = os.fdopen(os.dup(2), closefd=True)
426+
with sys.stdout, sys.stderr:
427+
with T:
428+
self.assertEqual(len(T.context_stack), 7)
429+
# out & err point to fd 1 and 2
430+
sys.stdout = os.fdopen(1, closefd=False)
431+
sys.stderr = os.fdopen(2, closefd=False)
432+
with sys.stdout, sys.stderr:
433+
with T:
434+
self.assertEqual(len(T.context_stack), 5)
435+
# out & err have no fileno
436+
sys.stdout = StringIO()
437+
sys.stderr = StringIO()
438+
with sys.stdout, sys.stderr:
439+
with T:
440+
self.assertEqual(len(T.context_stack), 5)
441+
290442
def test_capture_output_stack_error(self):
291443
OUT1 = StringIO()
292444
OUT2 = StringIO()
@@ -334,6 +486,20 @@ def test_capture_output_invalid_ostream(self):
334486
"The following was left in the output buffer:\n 'hi\\n'\n",
335487
)
336488

489+
def test_exit_on_del(self):
490+
# THis is a weird "feature", but because things like the pyomo
491+
# script will create and "enter" a capture_output object without
492+
# using a context manager, it is possible that the object can be
493+
# deleted without calling __exit__. Check that the context
494+
# stack us correctly unwound
495+
T = tee.capture_output()
496+
T.__enter__()
497+
stack = T.context_stack
498+
self.assertGreater(len(stack), 0)
499+
del T
500+
gc.collect()
501+
self.assertEqual(len(stack), 0)
502+
337503
def test_deadlock(self):
338504
class MockStream(object):
339505
def write(self, data):
@@ -712,6 +878,38 @@ def test_capture_output_fd(self):
712878
os.close(w)
713879
self.assertEqual(FILE.read(), "to_stdout_2\nto_fd1_2\n")
714880

881+
def test_nested_capture_output(self):
882+
OUT2 = StringIO()
883+
r, w = os.pipe()
884+
os.dup2(w, 1)
885+
sys.stdout = stdout0 = os.fdopen(1, 'w', closefd=False)
886+
with tee.capture_output((sys.stdout, StringIO()), capture_fd=True) as (_, OUT1):
887+
stdout1 = sys.stdout
888+
self.assertIsNot(stdout0, stdout1)
889+
with tee.capture_output((sys.stdout, OUT2), capture_fd=True):
890+
stdout2 = sys.stdout
891+
self.assertIsNot(stdout1, stdout2)
892+
sys.stdout.write("to_stdout_1\n")
893+
sys.stdout.flush()
894+
with os.fdopen(1, 'w', closefd=False) as F:
895+
F.write("to_fd1_1\n")
896+
F.flush()
897+
898+
sys.stdout.write("to_stdout_2\n")
899+
sys.stdout.flush()
900+
with os.fdopen(1, 'w', closefd=False) as F:
901+
F.write("to_fd1_2\n")
902+
F.flush()
903+
904+
self.assertEqual(OUT1.getvalue(), "to_stdout_1\nto_fd1_1\n")
905+
self.assertEqual(OUT2.getvalue(), "to_stdout_1\nto_fd1_1\n")
906+
with os.fdopen(r, 'r') as FILE:
907+
os.close(1)
908+
os.close(w)
909+
self.assertEqual(
910+
FILE.read(), "to_stdout_1\nto_fd1_1\nto_stdout_2\nto_fd1_2\n"
911+
)
912+
715913

716914
if __name__ == '__main__':
717915
unittest.main()

0 commit comments

Comments
 (0)