112 lines
4.5 KiB
Python
112 lines
4.5 KiB
Python
"""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])
|