import os from oms_tui.csvio import export_devices, import_rows from oms_tui.models import DeviceStore def _dll(device_id, mfct="KAM", media="Cold water meter"): return { "type": "dll", "id": device_id, "manufacturer": mfct, "version": "1b", "media": media, "driver": "kamwater", "c_field": "44", "rssi": 97, } def test_duplicate_dll_burst_counts_once(): # Two DLL lines for the SAME telegram (keyed meter + wildcard) arrive # microseconds apart -> counted once. store = DeviceStore() store.upsert_discovery(_dll("111"), now=1000.0) store.upsert_discovery(_dll("111"), now=1000.001) assert store.get("111").count == 1 def test_real_retransmissions_count_separately(): store = DeviceStore() store.upsert_discovery(_dll("111"), now=1000.0) store.upsert_discovery(_dll("111"), now=1005.0) # seconds later dev = store.get("111") assert dev.count == 2 assert dev.manufacturer_name == "Kamstrup" assert dev.status == "listening" # DLL only, encryption not yet known def test_encrypted_meter_with_plaintext_read_is_locked_not_open(): # The exact contradiction the user hit: an encrypted (mode 5) meter that # also exposes some plaintext records read by the wildcard ("scan"). store = DeviceStore() store.upsert_discovery(_dll("61471000")) store.apply_tpl( {"type": "tpl", "ci": "7a", "cfg": "8560", "nb": "6", "sec_mode": 5, "enc": "AES-CBC"}, "61471000", ) store.apply_json( {"type": "json", "id": "61471000", "name": "scan", "driver": "x", "media": "water", "fields": {"total_m3": 1.0}} ) dev = store.get("61471000") assert dev.encrypted is True assert dev.decrypted is True # we do have some values assert dev.key_decrypted is False assert dev.fields_source == "plain" assert dev.status == "locked" # NOT "open" assert dev.encryption == "mode 5 (AES-CBC)" def test_keyed_decrypt_marks_decrypted(): store = DeviceStore() store.upsert_discovery(_dll("61471000")) store.apply_tpl({"type": "tpl", "sec_mode": 5, "enc": "AES-CBC"}, "61471000") dev = store.get("61471000") dev.key = "00112233445566778899AABBCCDDEEFF" store.apply_json( {"type": "json", "id": "61471000", "name": "m_61471000", "driver": "x", "media": "water", "fields": {"total_m3": 2.0}} ) assert dev.key_decrypted is True assert dev.fields_source == "key" assert dev.status == "decrypted" def test_send_interval_estimate(): store = DeviceStore() # telegrams at t=0,16,32,64 -> gaps 16,16,32 (one missed) -> median 16 for t in (1000.0, 1016.0, 1032.0, 1064.0): store.upsert_discovery(_dll("777"), now=t) dev = store.get("777") assert dev.interval_samples == 3 assert dev.interval_estimate == 16 assert dev.interval_last == 32 assert dev.interval_min == 16 def test_json_only_still_counts_and_intervals(): # Regression: a device whose DLL line fails to parse still gets counted # (and its interval inferred) from the JSON telegrams alone. store = DeviceStore() for t in (2000.0, 2016.0, 2032.0): store.apply_json( {"type": "json", "id": "37287397", "name": "scan", "driver": "qheatv2", "media": "radio converter (meter side)", "fields": {"total_kwh": 0}}, now=t, ) dev = store.get("37287397") assert dev.count == 3 assert dev.interval_samples == 2 assert dev.interval_estimate == 16 def test_dll_and_its_json_count_once(): # The DLL and its own JSON (same telegram, ~same instant) must not double. store = DeviceStore() store.upsert_discovery(_dll("123"), now=3000.0) store.apply_json( {"type": "json", "id": "123", "name": "scan", "fields": {"x": 1}}, now=3000.05, ) assert store.get("123").count == 1 def test_single_telegram_has_no_interval(): store = DeviceStore() store.upsert_discovery(_dll("888"), now=1000.0) dev = store.get("888") assert dev.interval_samples == 0 assert dev.interval_estimate is None def test_duplicate_burst_does_not_pollute_interval(): store = DeviceStore() store.upsert_discovery(_dll("999"), now=1000.0) store.upsert_discovery(_dll("999"), now=1000.002) # duplicate DLL burst store.upsert_discovery(_dll("999"), now=1016.0) dev = store.get("999") assert dev.interval_samples == 1 assert dev.interval_estimate == 16 # not ~0 from the burst def test_unencrypted_device_is_open_not_locked(): store = DeviceStore() store.upsert_discovery(_dll("333")) store.apply_json( {"type": "json", "id": "333", "driver": "apator162", "media": "water", "fields": {"total_m3": 1.2}} ) dev = store.get("333") assert dev.decrypted is True assert dev.key is None assert dev.status == "open" # not "locked" assert dev.status_icon == "📖" assert dev.status_color == "green" def test_encryption_from_tpl_and_ell(): store = DeviceStore() store.upsert_discovery(_dll("444")) store.apply_tpl( {"type": "tpl", "ci": "7a", "cfg": "8560", "nb": "6", "sec_mode": 5, "enc": "AES-CBC"}, "444", ) dev = store.get("444") assert dev.sec_mode == 5 assert dev.encryption == "mode 5 (AES-CBC)" assert dev.enc_blocks == "6" store2 = DeviceStore() store2.upsert_discovery(_dll("555")) store2.apply_ell( {"type": "ell", "ci": "8d", "sn": "d37cac21", "session": "3", "enc": "AES-CTR"}, "555", ) assert store2.get("555").encryption == "AES-CTR (ELL)" def test_json_sets_decrypted_without_counting(): store = DeviceStore() store.upsert_discovery(_dll("222")) store.apply_json( {"type": "json", "id": "222", "driver": "kamwater", "media": "cold water", "fields": {"total_m3": 6.408}} ) dev = store.get("222") assert dev.count == 1 # JSON does not double-count assert dev.decrypted is True assert dev.fields["total_m3"] == 6.408 def test_csv_roundtrip_with_keys(tmp_path): store = DeviceStore() store.upsert_discovery(_dll("76348799")) store.get("76348799").key = "28F64A24988064A079AA2C807D6102AE" store.upsert_discovery(_dll("88888888", mfct="TCH", media="Heat")) path = os.path.join(tmp_path, "out.csv") assert export_devices(path, store.ordered()) == 2 rows = import_rows(path) by_id = {r["id"]: r for r in rows} assert by_id["76348799"]["aes_key"] == "28F64A24988064A079AA2C807D6102AE" assert by_id["76348799"]["manufacturer"] == "KAM" assert by_id["88888888"]["aes_key"] == "" def test_import_minimal_key_file(tmp_path): path = os.path.join(tmp_path, "keys.csv") with open(path, "w") as f: f.write("id,aes_key\n76348799,28F64A24988064A079AA2C807D6102AE\n") rows = import_rows(path) assert rows[0]["id"] == "76348799" assert rows[0]["aes_key"].startswith("28F64A")