Radix 16 (live-verified), universal reply splitter, golden capture vectors, host_probe6

tests/fixtures/discovery_req.bin

@@ -0,0 +1,2 @@
+{"cmdType":1101,"equipNo":"00000000",Ç ÌåÜÙÐ#(#'Úóëòõïïð÷úò÷ðç9568ìýùð;5?ô␍ù;K=ýÿ%+
+MK_M.`TQ,ndfk3*'jgmPsqwshus^n|-F/d@>ABADE8”•

tests/test_protocol.py

@@ -6,6 +6,7 @@ No radio/hardware needed.  Vectors mirror the blutter disassembly of
 """
 
 import json
+import os
 import unittest
 
 from franklinwh_agate_client.protocol import (
@@ -17,12 +18,14 @@ from franklinwh_agate_client.protocol import (
     SocketFrameDecoder,
     decrypt_data,
     decrypt_request,
+    decrypt_wire,
     direct_connect_request,
     encrypt_object,
     frame_socket_payload,
     generate_secret_key,
     header_len_for,
     split_envelope,
+    split_wire,
 )
 from franklinwh_agate_client.protocol.crypto import MARKER
 
@@ -62,9 +65,9 @@ class TestSecretKey(unittest.TestCase):
     def test_real_sn_key(self):
         self.assertEqual(generate_secret_key(SN), "1c")
 
-    def test_key_radix_is_32(self):
-        self.assertEqual(DEFAULT_KEY_RADIX, 32)
-        self.assertEqual(int(generate_secret_key(SN), 32), 44)
+    def test_key_radix_is_16(self):
+        self.assertEqual(DEFAULT_KEY_RADIX, 16)
+        self.assertEqual(int(generate_secret_key(SN), 16), 28)
 
     def test_empty_sn_key(self):
         self.assertEqual(generate_secret_key(""), "0")
@@ -148,7 +151,7 @@ class TestCipher(unittest.TestCase):
 
     def test_stream_is_absolute_index_plus_n(self):
         a, b = split_envelope(_apower_json())
-        n = int(generate_secret_key(SN), 32)
+        n = int(generate_secret_key(SN), 16)
         wire = encrypt_object(_apower_json(), generate_secret_key(SN))
         tail = wire[len(a):]
         for i, byte in enumerate(b):
@@ -163,16 +166,10 @@ class TestCipher(unittest.TestCase):
             body,
         )
 
-    def test_decrypt_response_roundtrip(self):
+    def test_wire_roundtrip(self):
         body = _apower_json()
-        cut = body.index('"type"')
-        s, tail = body[:cut - 1], body[cut:]
-        n = int(generate_secret_key(SN), 32)
-        prefix_len = len((s + ",").encode("utf-8"))
-        enc = bytes((byte + k + prefix_len + n) & 0xFF
-                    for k, byte in enumerate(tail.encode("utf-8")))
-        resp = s.encode("utf-8") + bytes((MARKER,)) + enc
-        self.assertEqual(decrypt_data(resp), body)
+        wire = encrypt_object(body, generate_secret_key(SN))
+        self.assertEqual(decrypt_wire(wire), body)
 
     def test_decrypt_data_failures_are_empty(self):
         self.assertEqual(decrypt_data(b""), "")
@@ -189,22 +186,75 @@ class TestSocketFraming(unittest.TestCase):
         self.assertEqual(frame_socket_payload(payload), payload)
 
     def test_decoder_accumulates_until_json(self):
-        body = _apower_json()
-        cut = body.index('"type"')
-        s, tail = body[:cut - 1], body[cut:]
-        n = int(generate_secret_key(SN), 32)
-        prefix_len = len((s + ",").encode("utf-8"))
-        enc = bytes((byte + k + prefix_len + n) & 0xFF
-                    for k, byte in enumerate(tail.encode("utf-8")))
-        resp = s.encode("utf-8") + bytes((MARKER,)) + enc
+        wire = open(os.path.join(FIXTURES, "discovery_req.bin"),
+                    "rb").read()
         dec = SocketFrameDecoder()
         # first half: incomplete -> nothing yet
-        self.assertEqual(dec.feed(resp[:len(resp) // 2]), [])
+        self.assertEqual(dec.feed(wire[:len(wire) // 2]), [])
         # rest: one complete JSON string
-        out = dec.feed(resp[len(resp) // 2:])
+        out = dec.feed(wire[len(wire) // 2:])
         self.assertEqual(len(out), 1)
-        self.assertEqual(json.loads(out[0])["equipNo"], SN)
+        self.assertEqual(json.loads(out[0])["equipNo"], "00000000")
 
 
 if __name__ == "__main__":
     unittest.main()
+
+
+FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures")
+
+
+class TestLiveCaptures(unittest.TestCase):
+    """Golden vectors: two real frames (app request + device reply)."""
+
+    def test_key_radix_is_16(self):
+        self.assertEqual(DEFAULT_KEY_RADIX, 16)
+        self.assertEqual(generate_secret_key("00000000"), "80")
+        self.assertEqual(int("80", 16), 128)
+        self.assertEqual(int(generate_secret_key(SN), 16), 28)
+
+    def test_discovery_request_vector(self):
+        wire = open(os.path.join(FIXTURES, "discovery_req.bin"), "rb").read()
+        self.assertEqual(len(wire), 153)
+        a, b = split_wire(wire)
+        self.assertEqual(a, b'{"cmdType":1101,"equipNo":"00000000",')
+        self.assertEqual(
+            decrypt_wire(wire),
+            '{"cmdType":1101,"equipNo":"00000000","type":0,'
+            '"timeStamp":1792117904,"snno":1,"len":37,"crc":"A474C16F",'
+            '"dataArea":{"opt":0,"minProtocolVer":"V1.00.00"}}',
+        )
+
+    def test_device_info_reply_vector(self):
+        wire = open(os.path.join(FIXTURES, "device_info_reply.bin"),
+                    "rb").read()
+        self.assertEqual(len(wire), 1154)
+        a, b = split_wire(wire)
+        self.assertEqual(
+            a, b'{"cmdType":1102,"equipNo":"10060005A03X25160132",')
+        reply = json.loads(decrypt_wire(wire))
+        self.assertEqual(reply["cmdType"], 1102)
+        self.assertEqual(reply["equipNo"], "10060005A03X25160132")
+        self.assertEqual(reply["len"], 1024)
+        data = reply["dataArea"]
+        self.assertEqual(data["opt"], 0)
+        self.assertEqual(data["result"], 0)
+        self.assertEqual(data["IBG_SN"], "10060005A03X25160132")
+        self.assertEqual(data["FHP_NUM"], 2)
+        self.assertEqual(len(data["FHP_SN"]), 2)
+        self.assertEqual(len(data["BMS_SN"]), 2)
+        # len/crc integrity, as the device checks them
+        import zlib
+        compact = json.dumps(data, separators=(",", ":"),
+                             ensure_ascii=False)
+        self.assertEqual(reply["len"], len(compact.encode("utf-8")))
+        self.assertEqual(
+            reply["crc"], format(zlib.crc32(compact.encode()), "X"))
+
+    def test_split_wire_errors(self):
+        with self.assertRaises(ValueError):
+            split_wire(b"no commas here")
+        with self.assertRaises(ValueError):
+            split_wire(b'{"cmdType":1101')
+        with self.assertRaises(ValueError):
+            decrypt_wire(b"not json at all")