"""Offline tests for the corrected envelope/cipher/socket wire format. No radio/hardware needed. Vectors mirror the blutter disassembly of ``core/communication_object.dart``, ``core/encryption.dart`` and ``socket/`` (see ``docs/RE_DART_FINDINGS.md`` ยงยง3, 5, 8). """ import json import unittest from franklinwh_agate_client.protocol import ( DEFAULT_KEY_RADIX, CommunicationCmd, CommunicationObject, SequenceNumber, socket_cmd_type, SocketFrameDecoder, decrypt_data, decrypt_request, direct_connect_request, encrypt_object, frame_socket_payload, generate_secret_key, header_len_for, split_envelope, ) from franklinwh_agate_client.protocol.crypto import MARKER SN = "A03X2516013210060005" def _apower_json(snno=1, time_stamp=1792105196): env = direct_connect_request( CommunicationCmd.DEVICE_APOWER_INFO, SN, data={"update": 2}, snno=snno, ) env.time_stamp = time_stamp env.set_data({"update": 2}) return env.to_json_string() class TestSocketCmdSpace(unittest.TestCase): def test_socket_ids(self): self.assertEqual(socket_cmd_type(CommunicationCmd.QUERY_DEVICE_INFO), 1101) self.assertEqual(socket_cmd_type(CommunicationCmd.DEVICE_APOWER_INFO), 1307) self.assertEqual(socket_cmd_type(CommunicationCmd.BMS), 1705) self.assertEqual(socket_cmd_type(CommunicationCmd.EMS), 1707) def test_direct_request_uses_socket_id(self): env = direct_connect_request(CommunicationCmd.DEVICE_APOWER_INFO, SN) self.assertEqual(env.cmd_type, 1307) self.assertTrue(env.to_json_string().startswith('{"cmdType":1307,')) def test_discovery_shape_like_capture(self): env = direct_connect_request(CommunicationCmd.QUERY_DEVICE_INFO, "") body = env.to_json_string() self.assertTrue(body.startswith( '{"cmdType":1101,"equipNo":"00000000",')) 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_empty_sn_key(self): self.assertEqual(generate_secret_key(""), "0") class TestEnvelope(unittest.TestCase): def test_key_order_no_data_key(self): env = direct_connect_request( CommunicationCmd.DEVICE_APOWER_INFO, SN, data={"update": 2}) d = env.to_dict() self.assertEqual( list(d), ["cmdType", "equipNo", "type", "timeStamp", "snno", "len", "crc", "dataArea"], ) self.assertNotIn("data", d) # socket payload is always seeded with the operation type first self.assertEqual(d["dataArea"], {"opt": 0, "update": 2}) def test_no_data_still_carries_opt(self): env = direct_connect_request(CommunicationCmd.BMS, SN, data=None) d = env.to_dict() self.assertIn("dataArea", d) self.assertEqual(d["dataArea"], {"opt": 0}) self.assertEqual(env.length, len(b'{"opt":0}')) self.assertTrue(env.crc) def test_empty_map_still_carries_opt(self): env = direct_connect_request(CommunicationCmd.BMS, SN, data={}) self.assertEqual(env.to_dict()["dataArea"], {"opt": 0}) def test_len_crc_auto_uppercase(self): env = direct_connect_request( CommunicationCmd.DEVICE_APOWER_INFO, SN, data={"update": 2}) self.assertEqual(env.length, len(b'{"opt":0,"update":2}')) import zlib self.assertEqual( env.crc, format(zlib.crc32(b'{"opt":0,"update":2}'), "X")) self.assertEqual(env.crc, env.crc.upper()) def test_int_fields(self): env = direct_connect_request(CommunicationCmd.BMS, SN, snno=7) d = env.to_dict() self.assertIsInstance(d["type"], int) self.assertIsInstance(d["timeStamp"], int) self.assertIsInstance(d["snno"], int) self.assertEqual(d["snno"], 7) self.assertEqual(d["type"], 0) # query # whole seconds, 10 digits self.assertEqual(len(str(d["timeStamp"])), 10) def test_snno_sequence_wraps(self): seq = SequenceNumber() self.assertEqual([seq.next(), seq.next()], [1, 2]) seq = SequenceNumber(start=9999) self.assertEqual(seq.next(), 1) def test_empty_sn_placeholder(self): env = CommunicationObject(0, "") self.assertEqual(env.equip_no, "00000000") def test_from_dict_roundtrip(self): env = direct_connect_request( CommunicationCmd.DEVICE_APOWER_INFO, SN, data={"update": 2}, snno=3) clone = CommunicationObject.from_json_string(env.to_json_string()) self.assertEqual(clone.to_dict(), env.to_dict()) class TestCipher(unittest.TestCase): def test_header_plaintext_first(self): wire = encrypt_object(_apower_json(), generate_secret_key(SN)) a, b = split_envelope(_apower_json()) self.assertEqual(wire[:len(a)], a) self.assertTrue(a.endswith(b",")) self.assertEqual(len(wire), len(a) + len(b)) def test_header_len_matches(self): a, _ = split_envelope(_apower_json()) self.assertEqual(header_len_for(1307, SN), len(a)) def test_stream_is_absolute_index_plus_n(self): a, b = split_envelope(_apower_json()) n = int(generate_secret_key(SN), 32) wire = encrypt_object(_apower_json(), generate_secret_key(SN)) tail = wire[len(a):] for i, byte in enumerate(b): self.assertEqual(tail[i], (byte + i + len(a) + n) & 0xFF) def test_request_roundtrip(self): body = _apower_json() wire = encrypt_object(body, generate_secret_key(SN)) self.assertEqual( decrypt_request(wire, generate_secret_key(SN), cmd_type=1307, equip_no=SN), body, ) def test_decrypt_response_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) def test_decrypt_data_failures_are_empty(self): self.assertEqual(decrypt_data(b""), "") self.assertEqual(decrypt_data(b"no markers here"), "") # only one marker self.assertEqual(decrypt_data(b"abc\x58def"), "") # probe not JSON self.assertEqual(decrypt_data(b"\x58garbage\x58\x00\x01"), "") class TestSocketFraming(unittest.TestCase): def test_send_is_raw(self): payload = b"\x01\x02\x03" 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 dec = SocketFrameDecoder() # first half: incomplete -> nothing yet self.assertEqual(dec.feed(resp[:len(resp) // 2]), []) # rest: one complete JSON string out = dec.feed(resp[len(resp) // 2:]) self.assertEqual(len(out), 1) self.assertEqual(json.loads(out[0])["equipNo"], SN) if __name__ == "__main__": unittest.main()