1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
#!/usr/bin/env python3
import os
import sys
import tempfile
import unittest
import logging_utils as lutils
import string_utils as sutils
import unittest_utils as uu
class TestLoggingUtils(unittest.TestCase):
def test_output_context(self):
unique_suffix = sutils.generate_uuid(True)
filename = f'/tmp/logging_utils_test.{unique_suffix}'
secret_message = f'This is a test, {unique_suffix}.'
with tempfile.SpooledTemporaryFile(mode='r+') as tmpfile1:
with uu.RecordStdout() as record:
with lutils.OutputMultiplexerContext(
lutils.OutputMultiplexer.Destination.FILENAMES
| lutils.OutputMultiplexer.Destination.FILEHANDLES
| lutils.OutputMultiplexer.Destination.LOG_INFO,
filenames=[filename, '/dev/null'],
handles=[tmpfile1, sys.stdout],
) as mplex:
mplex.print(secret_message, end='')
# Make sure it was written to the filename.
with open(filename, 'r') as rf:
self.assertEqual(rf.readline(), secret_message)
os.unlink(filename)
# Make sure it was written to stdout.
tmp = record().readline()
self.assertEqual(tmp, secret_message)
# Make sure it was written to the filehandle.
tmpfile1.seek(0)
tmp = tmpfile1.readline()
self.assertEqual(tmp, secret_message)
def test_record_streams(self):
with uu.RecordMultipleStreams(sys.stderr, sys.stdout) as record:
print("This is a test!")
print("This is one too.", file=sys.stderr)
self.assertEqual(record().readlines(), ["This is a test!\n", "This is one too.\n"])
if __name__ == '__main__':
unittest.main()
|
