serversdown
aa341c7342
Two reliability fixes for the Series 3 tray app: 1. Auto-updater no longer dies on a Settings save. _restart_watcher set+replaced the shared stop_event that the updater loop also keyed off, so saving settings could kill the updater thread (race). Give the updater its own app-lifetime event (_app_stop). Same fix as thor-watcher 0.4.1. 2. Unsniffable .MLG files are now cached, so each is sniffed + logged once per session instead of every 5-min scan. A long-lived autocall folder with a few hundred unidentifiable files was flooding the log (~600K lines in 10 days) and re-reading every header each scan. Adds test_series3_tray.py (4 lifecycle tests) and test_scan_latest.py (2 scan tests). Full suite 50 passing. Bump to v1.6.1. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
"""
|
|
Tests for scan_latest's handling of unsniffable .MLG files.
|
|
|
|
Regression guard: a .MLG file the watcher can't extract a unit ID from must
|
|
be cached (as "no id") so it isn't re-sniffed and re-logged on every scan.
|
|
Previously the unsniffable result was never cached, so the
|
|
`[unsniffable-recent]` warning was re-emitted every scan cycle — ~200 files
|
|
x 288 scans/day flooded the log with hundreds of thousands of duplicate lines.
|
|
"""
|
|
import os
|
|
import time
|
|
import tempfile
|
|
import unittest
|
|
|
|
import series3_watcher
|
|
|
|
|
|
class UnsniffableCaching(unittest.TestCase):
|
|
def test_unsniffable_recent_file_logged_once_across_scans(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
# A .MLG file with no BE####/BA#### pattern → sniff returns None.
|
|
path = os.path.join(d, "junk001.MLG")
|
|
with open(path, "wb") as f:
|
|
f.write(b"\x00" * 512)
|
|
|
|
calls = []
|
|
cache = {}
|
|
recent_cutoff = time.time() - 86400.0 # file (mtime≈now) counts as recent
|
|
|
|
for _ in range(3):
|
|
series3_watcher.scan_latest(
|
|
watch=d,
|
|
header_bytes=2048,
|
|
cache=cache,
|
|
recent_cutoff=recent_cutoff,
|
|
max_age_days=365,
|
|
logger=calls.append,
|
|
)
|
|
|
|
unsniff = [c for c in calls if "[unsniffable-recent]" in c]
|
|
self.assertEqual(
|
|
len(unsniff), 1,
|
|
"unsniffable file should log once across scans, got {}: {}".format(
|
|
len(unsniff), unsniff),
|
|
)
|
|
|
|
def test_sniffable_file_still_returned(self):
|
|
# Sanity: a file WITH a unit id is still detected (happy path intact).
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = os.path.join(d, "evt001.MLG")
|
|
with open(path, "wb") as f:
|
|
f.write(b"hdr\x00BE12599\x00hdr") # contains a sniffable unit id
|
|
cache = {}
|
|
result = series3_watcher.scan_latest(
|
|
watch=d, header_bytes=2048, cache=cache,
|
|
recent_cutoff=time.time() - 86400.0, max_age_days=365, logger=None,
|
|
)
|
|
self.assertIn("BE12599", result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|