77 lines
3.7 KiB
Python
77 lines
3.7 KiB
Python
"""Reboot evidence/state tests without restarting or changing a real box."""
|
||
import importlib.util
|
||
from importlib.machinery import SourceFileLoader
|
||
import json
|
||
from pathlib import Path
|
||
import tempfile
|
||
import unittest
|
||
from unittest.mock import patch
|
||
|
||
script = Path(__file__).resolve().parents[2]/"terminal/kiosk-shell/scripts/vdi-overnight-observer"
|
||
spec = importlib.util.spec_from_loader("observer", SourceFileLoader("observer", str(script)))
|
||
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
|
||
|
||
|
||
class RebootEvidence(unittest.TestCase):
|
||
def baseline(self): return m.transition(None, "boot-a", "start")[0]
|
||
|
||
def test_baseline_zero_reboots(self):
|
||
state,event=m.transition(None,"boot-a","start")
|
||
self.assertEqual(event,"boot_baseline"); self.assertEqual(state["reboots"],0)
|
||
|
||
def test_service_restart_is_not_box_reboot(self):
|
||
before=self.baseline(); after,event=m.transition(before,"boot-a","later")
|
||
self.assertEqual(event,"observer_restart_same_boot"); self.assertEqual(after["reboots"],0)
|
||
self.assertEqual(after["started_at"],"start")
|
||
|
||
def test_boot_change_preserves_last_heartbeat(self):
|
||
before=self.baseline(); before.update(last_seen="last",uptime_seconds=100)
|
||
after,event=m.transition(before,"boot-b","next")
|
||
self.assertEqual(event,"boot_change"); self.assertEqual(after["reboots"],1)
|
||
self.assertEqual(after["history"][-1]["previous_last_seen"],"last")
|
||
self.assertEqual(before["reboots"],0)
|
||
|
||
def test_unknown_reason_is_not_reported_as_power_loss(self):
|
||
text=m.classify({})
|
||
self.assertIn("未确认",text); self.assertIn("不能区分",text)
|
||
|
||
def test_orderly_shutdown_requires_real_markers(self):
|
||
self.assertIn("正常关机",m.classify({"tail":{"text":"systemd-shutdown[1]: Syncing filesystems"}}))
|
||
self.assertIn("未确认",m.classify({"tail":{"text":"USB power management enabled; watchdog capability present"}}))
|
||
|
||
def test_oom_is_a_clue_not_a_proven_reboot_cause(self):
|
||
text=m.classify({"warnings":{"text":"Out of memory: Killed process 123"}})
|
||
self.assertIn("不能单独认定",text)
|
||
|
||
def test_only_changed_pstore_is_a_new_clue(self):
|
||
archive={"text":"Kernel panic - not syncing: example"}
|
||
self.assertIn("未确认",m.classify({"old_pstore":archive}))
|
||
self.assertIn("不能单独认定",m.classify({"pstore_new":archive}))
|
||
|
||
def test_invalid_state_preserved(self):
|
||
for value in ({"version":2,"reboots":0},{"version":1,"reboots":True}):
|
||
with self.assertRaises(ValueError): m.transition(value,"new","now")
|
||
|
||
def test_multiple_boots_and_history_bound(self):
|
||
state=self.baseline()
|
||
for i in range(40): state,event=m.transition(state,"next-"+str(i),"now")
|
||
self.assertEqual(state["reboots"],40); self.assertEqual(len(state["history"]),32)
|
||
|
||
def test_durable_state_and_chinese_report(self):
|
||
with tempfile.TemporaryDirectory() as d:
|
||
o=m.Observer(Path(d)/"logs",Path(d)/"state"); o.state=self.baseline(); o.save()
|
||
self.assertEqual(json.loads(o.state_file.read_text())["boot_id"],"boot-a")
|
||
self.assertIn("重启次数：0",(o.logs/"overnight-summary.txt").read_text(encoding="utf-8"))
|
||
|
||
def test_daily_rollover_uses_each_event_date(self):
|
||
from datetime import datetime,timezone
|
||
with tempfile.TemporaryDirectory() as d:
|
||
o=m.Observer(Path(d)/"logs",Path(d)/"state")
|
||
for day in (4,5):
|
||
with patch.object(m,"datetime") as clock:
|
||
clock.now.return_value=datetime(2026,11,day,0,0,tzinfo=timezone.utc)
|
||
o.log("sample")
|
||
self.assertEqual(len(list(o.logs.glob("*.log"))),2)
|
||
|
||
if __name__=="__main__": unittest.main()
|