From 37c9f7d67d1fa0c13c5a8b439a5a891849ac6386 Mon Sep 17 00:00:00 2001 From: leetcrypt Date: Tue, 21 Jul 2026 11:52:42 -0700 Subject: [PATCH] test: add offline unit tests for BinRAW->RAW identification path Covers the two offline (no-server) identification behaviors: - BinRAW->RAW route (4396c74): parser reconstructs a signed pulse train from the demodulated Data_RAW bit stream so has_raw_data becomes true and a BinRAW capture identifies through the same path as a RAW file. - Per-match catalog-category surfacing (41850b5): every engine match carries its own catalog category in match_details['device_category'], cross-checked against the protocol database and the DeviceMatch source value. Fixtures are embedded and written to tmp_path (the repo ignores *.sub), so a bare pytest runs green with no docker, no bound server, and no network. 11 tests, all passing. Co-Authored-By: Claude Opus 4.6 --- tests/unit/test_binraw_identification.py | 188 +++++++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 tests/unit/test_binraw_identification.py diff --git a/tests/unit/test_binraw_identification.py b/tests/unit/test_binraw_identification.py new file mode 100644 index 0000000..d6dedc3 --- /dev/null +++ b/tests/unit/test_binraw_identification.py @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 +""" +Offline unit tests for the parser's device-identification path. + +Covers two things the offline (no-server) pipeline is responsible for: + + 1. BinRAW -> RAW identification route (commit 4396c74): a BinRAW capture has + no RAW_Data array, so the parser must reconstruct a signed pulse train from + the demodulated Data_RAW bit stream. That reconstruction is what lets a + BinRAW file flow through the SAME identification path as a RAW file + (``has_raw_data`` becomes true), yielding device matches. + + 2. Per-match catalog-category surfacing (commit 41850b5): every device match + the engine returns must carry its OWN catalog category in + ``match_details['device_category']`` (e.g. an Acurite sensor stays a + "Weather Sensor"), not just the single ML overall call. + +Everything here is author-time / offline: bare ``pytest`` runs it with no +Docker, no bound server, and no network. ``.sub`` fixtures are written to the +per-test ``tmp_path`` because ``*.sub`` is git-ignored repo-wide, so embedding +the content keeps the suite green on a fresh clone. +""" + +import pytest + +from src.parser.sub_parser import SubFileParser +from src.matcher.engine import SignatureMatcher +from src.matcher.pattern_decoder import PatternDecoder +from src.matcher.protocol_database import get_protocol_database + + +# --- minimal, self-contained fixtures (written to tmp_path at runtime) ------- + +# BinRAW: alternating bit stream (0xAA == 0b10101010) x 6 bytes = 48 bits. +# Alternating bits mean every run has length 1, so the reconstructed pulse +# train is exactly 48 pulses, each of magnitude TE (495 us), sign-alternating. +BINRAW_SUB = "\n".join([ + "Filetype: Flipper SubGhz Key File", + "Version: 1", + "Frequency: 868350000", + "Preset: FuriHalSubGhzPresetOok270Async", + "Protocol: BinRAW", + "Bit: 48", + "TE: 495", + "Bit_RAW: 48", + "Data_RAW: AA AA AA AA AA AA", + "", +]) + +# RAW: a small but real timing array (315 MHz OOK). +RAW_SUB = "\n".join([ + "Filetype: Flipper SubGhz RAW File", + "Version: 1", + "Frequency: 315000000", + "Preset: FuriHalSubGhzPresetOok650Async", + "Protocol: RAW", + "RAW_Data: 2980 -240 520 -980 520 -980 1000 -500 480 -1020 520 -980 1000 -500", + "", +]) + + +def _write(tmp_path, name, content): + p = tmp_path / name + p.write_text(content) + return str(p) + + +def _catalog_categories(): + """Authoritative {protocol_name: catalog_category} from the protocol DB.""" + return {proto.name: proto.category for proto in get_protocol_database().get_all()} + + +def _valid_categories(): + """The set of every catalog category the protocol DB knows about.""" + return {proto.category for proto in get_protocol_database().get_all()} + + +# --- BinRAW parsing ---------------------------------------------------------- + +class TestBinRawParsing: + def test_binraw_fields_extracted(self, tmp_path): + meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB)) + assert meta.file_format == "BinRAW" + assert meta.frequency == 868350000 + assert meta.modulation == "OOK" + assert meta.bin_raw_te == 495 + assert meta.bin_raw_bit == 48 + assert meta.bin_raw_data == bytes.fromhex("AAAAAAAAAAAA") + + def test_binraw_reconstructs_raw_pulse_train(self, tmp_path): + """The BinRAW->RAW route: Data_RAW must expand into raw_data pulses.""" + meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB)) + assert meta.has_raw_data is True + # 48 alternating bits -> 48 unit-length runs -> 48 pulses. + assert meta.pulse_count == 48 + assert len(meta.raw_data) == 48 + # Every pulse is exactly one TE, alternating sign. + assert all(abs(t) == 495 for t in meta.raw_data) + assert meta.raw_data[0] > 0 and meta.raw_data[1] < 0 + + +class TestBinRawToPulses: + """The run-length expansion is deterministic; test it directly.""" + + def test_run_length_encoding(self): + # 0xF0 == 11110000 -> run of 4 ones then 4 zeros. + pulses = SubFileParser._binraw_to_pulses(bytes([0xF0]), 8, 100) + assert pulses == [400, -400] + + def test_alternating_bits(self): + # 0xAA == 10101010 -> 8 unit runs, alternating sign starting high. + pulses = SubFileParser._binraw_to_pulses(bytes([0xAA]), 8, 100) + assert pulses == [100, -100, 100, -100, 100, -100, 100, -100] + + def test_truncates_to_n_bits(self): + # Only the first 4 bits (1111) of 0xF0 are used. + pulses = SubFileParser._binraw_to_pulses(bytes([0xF0]), 4, 50) + assert pulses == [200] + + +# --- RAW parsing ------------------------------------------------------------- + +class TestRawParsing: + def test_raw_fields_extracted(self, tmp_path): + meta = SubFileParser().parse(_write(tmp_path, "r.sub", RAW_SUB)) + assert meta.file_format == "RAW" + assert meta.frequency == 315000000 + assert meta.modulation == "OOK" + assert meta.has_raw_data is True + assert meta.pulse_count == 14 + assert meta.raw_data[0] == 2980 + assert meta.avg_pulse_width > 0 + + +# --- identification: BinRAW flows through RAW, categories surface ------------- + +class TestBinRawIdentificationRoute: + def test_binraw_yields_matches_via_raw_path(self, tmp_path): + """A BinRAW capture identifies at all only because of the RAW route.""" + meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB)) + matches = SignatureMatcher().match(meta, max_results=5) + assert len(matches) > 0 + for m in matches: + assert m.device_name # non-empty device name + cat = m.match_details.get("device_category") + assert isinstance(cat, str) and cat, \ + "every match must surface a per-device catalog category" + + def test_binraw_category_is_valid_catalog_category(self, tmp_path): + """Every surfaced device_category is a real category from the catalog.""" + meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB)) + valid = _valid_categories() + matches = SignatureMatcher().match(meta, max_results=5) + assert len(matches) > 0 + for m in matches: + assert m.match_details.get("device_category") in valid + + def test_binraw_decoder_category_is_protocol_catalog_category(self, tmp_path): + """At the source, DeviceMatch.category (surfaced by the engine) == protocol.category.""" + meta = SubFileParser().parse(_write(tmp_path, "b.sub", BINRAW_SUB)) + decoded = PatternDecoder().decode(meta) + assert len(decoded) > 0 + for dm in decoded: + assert dm.category == dm.protocol.category + + +class TestRawIdentificationCategory: + def test_raw_match_surfaces_catalog_category(self, tmp_path): + meta = SubFileParser().parse(_write(tmp_path, "r.sub", RAW_SUB)) + catalog = _catalog_categories() + matches = SignatureMatcher().match(meta, max_results=5) + assert len(matches) > 0 + checked = 0 + for m in matches: + cat = m.match_details.get("device_category") + assert isinstance(cat, str) and cat + if m.device_name in catalog: + assert cat == catalog[m.device_name] + checked += 1 + assert checked > 0 + + def test_decoder_category_is_protocol_catalog_category(self, tmp_path): + """DeviceMatch.category (the source of the surfaced value) == protocol.category.""" + meta = SubFileParser().parse(_write(tmp_path, "r.sub", RAW_SUB)) + decoded = PatternDecoder().decode(meta) + assert len(decoded) > 0 + for dm in decoded: + assert dm.category == dm.protocol.category