USB-AIO für automatische Geräteprüfung

2026-10-14 20:14:36 +02:00
commit 81f06cf6b4
40 changed files with 9956 additions and 0 deletions
--- a/tests/test_station.py
+++ b/tests/test_station.py
@@ -0,0 +1,41 @@
+from test_inventory_sync import bridge
+import unittest,tempfile,json
+from pathlib import Path
+from unittest.mock import patch
+
+class StationTests(unittest.TestCase):
+    def test_offline_inspection_is_durable_and_retry_is_idempotent(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root=Path(tmp);path=root/'inspection.json'
+            path.write_text(json.dumps({'asset':'HW-test','revision':2,'updated':20,'identity':{'uuid':'12345678-1234-1234-1234-123456789012','model':'Test'},'technician':'Tech','checks':{'wifi':{'outcome':'UNAVAILABLE','notes':'Tool unavailable'}},'notes':''}))
+            with self.assertRaises(FileNotFoundError):bridge.run_station(root,'progress',path)
+            files=list((root/'outbox/inspections').glob('*.json'));self.assertEqual(len(files),1)
+            first=json.loads(files[0].read_text());self.assertEqual(first['body']['results']['wifi']['outcome'],'UNAVAILABLE')
+            bridge.queue_inspection(root,path,'IN_PROGRESS');self.assertEqual(len(list((root/'outbox/inspections').glob('*.json'))),1)
+            with patch.object(bridge,'station_resolve',return_value={'number':99,'revision':2}),patch.object(bridge,'station_api',side_effect=OSError('offline')):
+                with self.assertRaises(OSError):bridge.flush_inspections(root,{})
+            self.assertTrue(files[0].exists())
+            with patch.object(bridge,'station_resolve',return_value={'number':99,'revision':3}),patch.object(bridge,'station_api',return_value={'number':99}) as api:
+                bridge.flush_inspections(root,{})
+            self.assertEqual(api.call_args.args[3]['request_id'],first['body']['request_id'])
+            self.assertFalse(files[0].exists());self.assertTrue((root/'outbox/sent'/files[0].name).exists())
+    def test_intake_does_not_overwrite_later_stage(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root=Path(tmp);p=root/'report.json';p.write_text(json.dumps({'system_uuid':'12345678-1234-1234-1234-123456789012','cpu':'CPU'}))
+            with patch.object(bridge,'station_config',return_value={}),patch.object(bridge,'station_resolve',return_value={'number':1,'stage':'STORAGE'}),patch.object(bridge,'station_api') as api:
+                bridge.run_station(root,'intake',p)
+                api.assert_not_called()
+
+    def test_boot_recovers_saved_inspection_before_network_access(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root=Path(tmp);(root/'inspections').mkdir()
+            saved={'asset':'HW-previous','revision':3,'updated':21,
+                   'identity':{'uuid':'12345678-1234-1234-1234-123456789012'},
+                   'sync_phase':'COMPLETE','technician':'Tech','checks':{'wifi':{'outcome':'PASS'}}}
+            (root/'inspections/previous.json').write_text(json.dumps(saved))
+            with self.assertRaises(FileNotFoundError):bridge.run_station(root,'boot',None)
+            queued=list((root/'outbox/inspections').glob('*.json'))
+            self.assertEqual(len(queued),1)
+            self.assertEqual(json.loads(queued[0].read_text())['body']['phase'],'COMPLETE')
+            with self.assertRaises(FileNotFoundError):bridge.run_station(root,'boot',None)
+            self.assertEqual(len(list((root/'outbox/inspections').glob('*.json'))),1)