feat(gas-logger): добавь непрерывный логгер и JSON-карту

c/gas-logger/tests/test_client.py

@@ -0,0 +1,111 @@
+"""Integration: Python client -> register callbacks -> compiled C PM35 logger."""
+import copy
+import ctypes
+import unittest
+from set_devices.gas_logger import GasLoggerClient, load_map, validate_map
+
+DEVICE = None  # set by run_tests.py before loading the suite
+
+
+class ClientTests(unittest.TestCase):
+    def setUp(self):
+        self.lib=DEVICE
+        self.assertEqual(self.lib.host_init(),0)
+        self.client=GasLoggerClient(self.read,self.write)
+        self.client.start()
+        for i in range(800):
+            self.assertEqual(self.lib.host_capture(i,i),0)
+
+    def read(self,address,count):
+        out=(ctypes.c_uint16*count)()
+        result=self.lib.host_read(address,out,count)
+        if result: raise RuntimeError('read '+str(result))
+        return list(out)
+
+    def write(self,address,value):
+        result=self.lib.host_write(address,value)
+        if result: raise RuntimeError('write '+str(result))
+
+    def test_download_keeps_recording_and_order(self):
+        def progress(done,total):
+            for i in range(20): self.assertEqual(self.lib.host_capture(900+done+i,99),0)
+        records=self.client.download(progress=progress)
+        self.assertEqual([r['time_ms'] for r in records],list(range(40,800)))
+        self.assertEqual([r['current'] for r in records],list(range(40,800)))
+        self.assertTrue(all(r['zpt2']==5678 for r in records))
+        self.assertEqual(self.client.status()[2],1)
+        self.assertGreater(self.client.status()[6],0)
+
+    def test_cancel_retains_snapshot_and_explicit_resume(self):
+        cancel=[False]
+        def progress(done,total): cancel[0]=True
+        with self.assertRaises(InterruptedError):
+            self.client.download(cancelled=lambda:cancel[0],progress=progress)
+        self.assertEqual(self.client.status()[2],3)
+        with self.assertRaises(RuntimeError): self.client.download()
+        self.assertEqual(len(self.client.download(resume=True)),760)
+
+    def test_partial_reply_leaves_pin(self):
+        original=self.client.read
+        self.client.read=lambda a,n: original(a,n)[:-1] if a>=0x6100 else original(a,n)
+        with self.assertRaises(ValueError): self.client.download()
+        self.client.read=original
+        self.assertEqual(self.client.status()[2],3)
+
+    def test_no_writes_for_wrong_schema(self):
+        m=load_map();m['channels'][0]['name']='Other'
+        writes=[]
+        client=GasLoggerClient(self.read,lambda *a:writes.append(a),m)
+        with self.assertRaises(ValueError): client.download()
+        self.assertEqual(writes,[])
+
+    def test_generation_mismatch_does_not_release(self):
+        self.write(0x6012,2)
+        with self.assertRaises(ValueError):self.client.release(999)
+        self.assertEqual(self.client.status()[2],3)
+
+    def test_classic_can_chunks_and_pcan_region(self):
+        def read(a,n):
+            words=[]
+            for i in range(n):
+                out=ctypes.c_uint16()
+                self.assertEqual(self.lib.host_pcan_read(a+i,ctypes.byref(out)),0)
+                words.append(out.value)
+            return words
+        client=GasLoggerClient(read,self.write,block_words=4)
+        self.assertEqual(len(client.download()),760)
+
+    def test_json_rejects_collisions_and_drift(self):
+        base=load_map()
+        for change in (
+            lambda m:m['channels'][0].update(gas=0x6000),
+            lambda m:m['channels'][1].update(gas=m['channels'][0]['gas']),
+            lambda m:m['service'].update(data_offset=512),
+            lambda m:m['service']['registers'][0].update(key='wrong'),
+            lambda m:m.update(capacity=65535),
+        ):
+            m=copy.deepcopy(base);change(m)
+            with self.assertRaises(ValueError):validate_map(m)
+
+    def test_replaced_snapshot_is_rejected_before_release(self):
+        changed=[False]
+        def progress(done,total):
+            if changed[0]:return
+            changed[0]=True
+            self.client.release(1)
+            self.assertEqual(self.lib.host_capture(2000,999),0)
+            self.write(0x6012,2)
+        # Simulates a second, unsupported controlling client. We must never
+        # accept a mixture of generations or release the newer snapshot.
+        with self.assertRaises((ValueError,RuntimeError)):
+            self.client.download(progress=progress)
+        status=self.client.status()
+        self.assertEqual(status[2],3)
+        self.assertEqual(status[8],2)
+
+    def test_start_does_not_reset_history(self):
+        before=self.client.status()
+        self.client.start()
+        after=self.client.status()
+        self.assertEqual(before[6],after[6])
+        self.assertEqual(before[12:14],after[12:14])