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 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user