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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
"""Tests for app/logging_kit.py — log naming, teeing, and retention."""
import io
import os
import tempfile
import time
import unittest
from pathlib import Path
import logging_kit
class _TempDir(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.dir = Path(self.tmp.name)
class DayStreamTests(_TempDir):
def test_creates_dated_stream_and_appends(self):
handle = logging_kit.day_stream("tui", log_dir=self.dir)
self.assertIsNotNone(handle)
logging_kit.write_line(handle, "first")
logging_kit.write_line(handle, "second")
handle.close()
files = list(self.dir.glob("tui_*.log"))
self.assertEqual(len(files), 1)
self.assertRegex(files[0].name, r"^tui_\d{8}\.log$")
self.assertEqual(files[0].read_text(encoding="utf-8"),
"first\nsecond\n")
def test_unwritable_dir_returns_none(self):
blocker = self.dir / "file" # a file where the directory would be
blocker.write_text("x", encoding="utf-8")
self.assertIsNone(
logging_kit.day_stream("tui", log_dir=blocker))
class RunArtifactTests(_TempDir):
def test_creates_timestamped_artifact(self):
path, handle = logging_kit.run_artifact("audiocpp_build",
log_dir=self.dir)
self.assertIsNotNone(handle)
self.assertRegex(path.name, r"^audiocpp_build_\d{8}_\d{6}\.log$")
logging_kit.write_line(handle, "out")
handle.close()
self.assertEqual(path.read_text(encoding="utf-8"), "out\n")
def test_returns_path_even_when_unwritable(self):
blocker = self.dir / "file"
blocker.write_text("x", encoding="utf-8")
path, handle = logging_kit.run_artifact("x", log_dir=blocker)
self.assertIsNone(handle)
self.assertEqual(path.parent, blocker)
class WriteLineTests(unittest.TestCase):
def test_none_handle_is_a_noop(self):
logging_kit.write_line(None, "x") # must not raise
def test_broken_handle_is_swallowed(self):
class Broken:
def write(self, _):
raise OSError("nope")
def flush(self):
raise OSError("nope")
logging_kit.write_line(Broken(), "x") # must not raise
class TeeWriterTests(unittest.TestCase):
def test_mirrors_to_both_sides(self):
logf, inner = io.StringIO(), io.StringIO()
writer = logging_kit.TeeWriter(logf, inner)
self.assertEqual(writer.write("hello"), 5)
writer.flush()
self.assertEqual(logf.getvalue(), "hello")
self.assertEqual(inner.getvalue(), "hello")
def test_empty_write_returns_zero(self):
self.assertEqual(logging_kit.TeeWriter(io.StringIO()).write(""), 0)
def test_none_sides_and_errors_are_swallowed(self):
writer = logging_kit.TeeWriter(None, None)
self.assertEqual(writer.write("x"), 1) # must not raise
writer.flush()
self.assertFalse(writer.isatty())
class PruneTests(_TempDir):
def test_deletes_old_logs_keeps_new_server_logs_and_pid_files(self):
old = self.dir / "tui_20200101.log"
old.write_text("x", encoding="utf-8")
new = self.dir / "audiobook_20990101.log"
new.write_text("x", encoding="utf-8")
server = self.dir / "audiocpp-server.log"
server.write_text("x", encoding="utf-8")
pid = self.dir / "audiocpp-server.pid"
pid.write_text("1", encoding="utf-8")
ancient = time.time() - 100 * 86400
os.utime(old, (ancient, ancient))
os.utime(server, (ancient, ancient))
logging_kit.prune_logs(log_dir=self.dir)
self.assertFalse(old.exists())
self.assertTrue(new.exists())
self.assertTrue(server.exists()) # owned by a server process
self.assertTrue(pid.exists()) # not a log
def test_missing_dir_is_a_noop(self):
logging_kit.prune_logs(log_dir=self.dir / "nope") # must not raise
if __name__ == "__main__":
unittest.main()
|
