运维：固定虚拟PCI设备并补充过夜观察与电源策略

tools/tests/test_fixed_pci_policy.py

@@ -0,0 +1,37 @@
+import importlib.util
+from pathlib import Path
+import shlex
+import unittest
+from unittest.mock import patch
+
+spec = importlib.util.spec_from_file_location("fixed_pci", Path(__file__).parents[1] / "set-fixed-pci.py")
+policy = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(policy)
+
+
+class FixedPciPolicyTests(unittest.TestCase):
+    def test_keeps_gpu_identity_and_quoted_arguments(self):
+        old = '-set device.hostpci0.x-pci-sub-vendor-id=0x1558 -set device.hostpci0.x-pci-sub-device-id=0x6557 -name "desktop test"'
+        updated = policy.fixed_args(old)
+        self.assertEqual(shlex.split(updated)[:len(shlex.split(old))], shlex.split(old))
+        self.assertTrue(updated.startswith(old))
+        self.assertEqual(policy.fixed_args(updated), updated)
+        self.assertFalse(any("usb" in token.lower() for token in shlex.split(updated)))
+
+    def test_conflicting_explicit_hotplug_setting_requires_review(self):
+        with self.assertRaisesRegex(ValueError, "Conflicting"):
+            policy.fixed_args('-global pcie-root-port.hotplug=on')
+
+    def test_online_vm_cannot_be_changed(self):
+        with patch.object(policy, "command", return_value="status: running") as call:
+            with self.assertRaisesRegex(RuntimeError, "normally shut down"):
+                policy.require_stopped(100)
+            call.assert_called_once_with(["qm", "status", "100"])
+
+    def test_stopped_vm_can_proceed(self):
+        with patch.object(policy, "command", return_value="status: stopped"):
+            policy.require_stopped(100)
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/tests/test_overnight_observer.py

@@ -0,0 +1,76 @@
+"""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()