Rotary Logger  1.0.2
The middleware rotary logger
Loading...
Searching...
No Matches
test_tee_stream_basic.py
Go to the documentation of this file.
1"""
2# +==== BEGIN rotary_logger =================+
3# LOGO:
4# ..........####...####..........
5# ......###.....#.#########......
6# ....##........#.###########....
7# ...#..........#.############...
8# ...#..........#.#####.######...
9# ..#.....##....#.###..#...####..
10# .#.....#.##...#.##..##########.
11# #.....##########....##...######
12# #.....#...##..#.##..####.######
13# .#...##....##.#.##..###..#####.
14# ..#.##......#.#.####...######..
15# ..#...........#.#############..
16# ..#...........#.#############..
17# ...##.........#.############...
18# ......#.......#.#########......
19# .......#......#.########.......
20# .........#####...#####.........
21# /STOP
22# PROJECT: rotary_logger
23# FILE: test_tee_stream_basic.py
24# CREATION DATE: 01-11-2025
25# LAST Modified: 7:42:0 01-11-2025
26# DESCRIPTION:
27# A module that provides a universal python light on iops way of logging to files your program execution.
28# /STOP
29# COPYRIGHT: (c) Asperguide
30# PURPOSE: This is the file in charge of testing the that TeeStream class works as expected.
31# // AR
32# +==== END rotary_logger =================+
33"""
34import io
35from pathlib import Path
36
37import pytest
38
39from rotary_logger.tee_stream import TeeStream
40
41
42def _find_log_file(root: Path):
43 logs = list(root.rglob('*.log'))
44 if not logs:
45 pytest.skip('No log file created')
46 return logs[0]
47
48
49def test_tee_stream_write_and_flush(tmp_path: Path) -> None:
50 # Use an explicit .log file path to avoid the library attempting to
51 # open a directory as a file (some FileInstance constructors accept
52 # a file path rather than a folder).
53 root = tmp_path / 'logs.log'
54 # use an in-memory stream to capture original_stream writes
55 orig = io.StringIO()
56 ts = TeeStream(root, orig, max_size_mb=1)
57 try:
58 ts.write('hello world\n')
59 ts.write('another line\n')
60 # flush should force buffer to disk
61 ts.flush()
62 # original stream should contain the messages
63 assert 'hello world' in orig.getvalue()
64 # file should exist with the contents
65 logfile = _find_log_file(tmp_path)
66 # use the FileInstance configured encoding
67 enc = ts.file_instance.get_encoding()
68 content = logfile.read_text(encoding=enc)
69 assert 'hello world' in content
70 assert logfile.stat().st_size > 0
71 finally:
72 # cleanup
73 try:
74 orig.close()
75 except Exception:
76 pass
77
78
79def test_tee_stream_encoding_byte_count(tmp_path: Path) -> None:
80 root = tmp_path / 'logs.log'
81 orig = io.StringIO()
82 # use utf-16 to ensure byte-counting uses configured encoding
83 # Create a FileInstance with the desired encoding first so the
84 # underlying descriptor is opened with utf-16.
85 from rotary_logger.file_instance import FileInstance
86 fi = FileInstance(root, max_size_mb=1, encoding='utf-16')
87 ts = TeeStream(fi, orig)
88 try:
89 s = 'é' * 10 + '\n'
90 ts.write(s)
91 ts.flush()
92 # _written_bytes should equal the encoded byte length
93 expected = len(s.encode(ts.file_instance.get_encoding()))
94 logfile = _find_log_file(tmp_path)
95 assert logfile.stat().st_size >= expected
96 finally:
97 try:
98 orig.close()
99 except Exception:
100 pass
None test_tee_stream_write_and_flush(Path tmp_path)
None test_tee_stream_encoding_byte_count(Path tmp_path)