import json import threading from typing import Any, Dict, List from urllib import error, request as urllib_request import pytest import service class _FakeResponse: def __init__(self, payload: object, status: int = 200): self._raw = json.dumps(payload).encode("utf-8") self.status = status def read(self) -> bytes: return self._raw def __enter__(self): return self def __exit__(self, exc_type, exc, tb) -> None: return None def _base_config() -> Dict[str, object]: return { "model": { "tx_power_dbm": 20.0, "tx_gain_dbi": 0.0, "rx_gain_dbi": 0.0, "path_loss_exponent": 2.0, "reference_distance_m": 1.0, "min_distance_m": 0.001, }, "solver": {"tolerance": 0.001, "z_preference": "positive"}, "runtime": { "poll_interval_s": 1.0, "output_server": { "enabled": False, "ip": "192.168.1.10", "port": 8080, "path": "/triangulation", "timeout_s": 3.0, }, }, "input": { "mode": "http_sources", "aggregation": "median", "source_timeout_s": 1.0, "receivers": [ { "receiver_id": "r0", "center": {"x": 0.0, "y": 0.0, "z": 0.0}, "source_url": "http://r0.local/measurements", }, { "receiver_id": "r1", "center": {"x": 10.0, "y": 0.0, "z": 0.0}, "source_url": "http://r1.local/measurements", }, { "receiver_id": "r2", "center": {"x": 0.0, "y": 8.0, "z": 0.0}, "source_url": "http://r2.local/measurements", }, ], }, } def _install_urlopen(monkeypatch: pytest.MonkeyPatch, responses: Dict[str, object]) -> None: def _fake_urlopen(req: Any, timeout: float = 0.0): url = getattr(req, "full_url", str(req)) payload_or_exc = responses[url] if isinstance(payload_or_exc, Exception): raise payload_or_exc return _FakeResponse(payload_or_exc) monkeypatch.setattr(service.request, "urlopen", _fake_urlopen) def _start_api_server_for_test(svc: service.AutoService): http_server = service.ThreadingHTTPServer(("127.0.0.1", 0), service._make_handler(svc)) thread = threading.Thread(target=http_server.serve_forever, daemon=True) thread.start() host, port = http_server.server_address return http_server, thread, f"http://{host}:{port}" def test_refresh_once_builds_frequency_table_for_common_frequencies( monkeypatch: pytest.MonkeyPatch, ): config = _base_config() freq_a = 433_920_000.0 freq_b = 868_100_000.0 responses = { "http://r0.local/measurements": { "receiver_id": "r0", "measurements": [ {"frequency_hz": freq_a, "rssi_dbm": -61.0}, {"frequency_hz": freq_b, "rssi_dbm": -68.0}, ], }, "http://r1.local/measurements": { "receiver_id": "r1", "measurements": [ {"frequency_hz": freq_a, "rssi_dbm": -64.0}, {"frequency_hz": freq_b, "rssi_dbm": -70.0}, ], }, "http://r2.local/measurements": { "receiver_id": "r2", "measurements": [ {"frequency_hz": freq_a, "rssi_dbm": -63.0}, {"frequency_hz": freq_b, "rssi_dbm": -69.0}, ], }, } _install_urlopen(monkeypatch, responses) svc = service.AutoService(config) svc.refresh_once() snapshot = svc.snapshot() payload = snapshot["payload"] assert snapshot["last_error"] == "" assert payload is not None assert payload["selected_frequency_hz"] in (freq_a, freq_b) table = payload["frequency_table"] assert isinstance(table, list) assert len(table) == 2 for row in table: assert row["frequency_hz"] in (freq_a, freq_b) assert "position" in row assert len(row["receivers"]) == 3 def test_refresh_once_fails_when_no_common_frequencies(monkeypatch: pytest.MonkeyPatch): config = _base_config() responses = { "http://r0.local/measurements": {"measurements": [{"frequency_hz": 100.0, "rssi_dbm": -60.0}]}, "http://r1.local/measurements": {"measurements": [{"frequency_hz": 200.0, "rssi_dbm": -60.0}]}, "http://r2.local/measurements": {"measurements": [{"frequency_hz": 300.0, "rssi_dbm": -60.0}]}, } _install_urlopen(monkeypatch, responses) svc = service.AutoService(config) with pytest.raises(RuntimeError, match="No common frequencies across all 3 receivers"): svc.refresh_once() def test_refresh_once_reports_row_validation_error_with_source_context( monkeypatch: pytest.MonkeyPatch, ): config = _base_config() responses = { "http://r0.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -60.0}]}, "http://r1.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": "bad"}]}, "http://r2.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -58.0}]}, } _install_urlopen(monkeypatch, responses) svc = service.AutoService(config) with pytest.raises(RuntimeError, match=r"source_url=http://r1\.local/measurements: row #1 field 'rssi_dbm' must be numeric"): svc.refresh_once() def test_refresh_once_validates_receiver_id_mismatch(monkeypatch: pytest.MonkeyPatch): config = _base_config() responses = { "http://r0.local/measurements": {"receiver_id": "r0", "measurements": [{"frequency_hz": 915e6, "rssi_dbm": -60.0}]}, "http://r1.local/measurements": {"receiver_id": "WRONG", "measurements": [{"frequency_hz": 915e6, "rssi_dbm": -59.0}]}, "http://r2.local/measurements": {"receiver_id": "r2", "measurements": [{"frequency_hz": 915e6, "rssi_dbm": -58.0}]}, } _install_urlopen(monkeypatch, responses) svc = service.AutoService(config) with pytest.raises(RuntimeError, match="does not match expected 'r1'"): svc.refresh_once() def test_refresh_once_raises_when_output_server_rejects_payload( monkeypatch: pytest.MonkeyPatch, ): config = _base_config() config["runtime"]["output_server"]["enabled"] = True # type: ignore[index] responses = { "http://r0.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -60.0}]}, "http://r1.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -59.0}]}, "http://r2.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -58.0}]}, } _install_urlopen(monkeypatch, responses) monkeypatch.setattr( service, "send_payload_to_server", lambda **_: (500, "internal error"), ) svc = service.AutoService(config) with pytest.raises(RuntimeError, match="Output server rejected payload: HTTP 500"): svc.refresh_once() def test_refresh_once_propagates_source_http_error(monkeypatch: pytest.MonkeyPatch): config = _base_config() responses = { "http://r0.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -60.0}]}, "http://r1.local/measurements": error.URLError("connection refused"), "http://r2.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -58.0}]}, } _install_urlopen(monkeypatch, responses) svc = service.AutoService(config) with pytest.raises(RuntimeError, match="Cannot reach 'http://r1.local/measurements': connection refused"): svc.refresh_once() def test_output_delivery_is_disabled_when_output_server_off(monkeypatch: pytest.MonkeyPatch): config = _base_config() responses = { "http://r0.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -60.0}]}, "http://r1.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -59.0}]}, "http://r2.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -58.0}]}, } _install_urlopen(monkeypatch, responses) svc = service.AutoService(config) svc.refresh_once() snapshot = svc.snapshot() assert snapshot["output_delivery"]["enabled"] is False assert snapshot["output_delivery"]["status"] == "disabled" def test_parse_source_payload_rejects_non_finite_values(): payload = {"measurements": [{"frequency_hz": float("inf"), "rssi_dbm": -60.0}]} with pytest.raises(ValueError, match="must be finite"): service.parse_source_payload(payload, source_label="source_url=test") def test_parse_source_payload_accepts_frequency_mhz(): payload = {"measurements": [{"frequency_mhz": 868.1, "rssi_dbm": -60.0}]} parsed = service.parse_source_payload(payload, source_label="source_url=test") assert parsed[0][0] == pytest.approx(868_100_000.0) def test_parse_source_payload_treats_generic_frequency_as_mhz(): payload = {"measurements": [{"frequency": 433.92, "rssi_dbm": -60.0}]} parsed = service.parse_source_payload(payload, source_label="source_url=test") assert parsed[0][0] == pytest.approx(433_920_000.0) def test_http_blocks_static_path_traversal(monkeypatch: pytest.MonkeyPatch): config = _base_config() svc = service.AutoService(config) http_server, thread, base_url = _start_api_server_for_test(svc) try: with pytest.raises(error.HTTPError) as exc_info: urllib_request.urlopen(f"{base_url}/static/../service.py") assert exc_info.value.code == 404 finally: http_server.shutdown() http_server.server_close() thread.join(timeout=1.0) def test_http_config_rejects_empty_body(monkeypatch: pytest.MonkeyPatch): config = _base_config() svc = service.AutoService(config) http_server, thread, base_url = _start_api_server_for_test(svc) try: req = urllib_request.Request( url=f"{base_url}/config", method="POST", data=b"", headers={"Content-Type": "application/json"}, ) with pytest.raises(error.HTTPError) as exc_info: urllib_request.urlopen(req) body = exc_info.value.read().decode("utf-8") assert exc_info.value.code == 400 assert "Empty request body" in body finally: http_server.shutdown() http_server.server_close() thread.join(timeout=1.0) def test_http_config_rejects_too_large_payload(monkeypatch: pytest.MonkeyPatch): config = _base_config() svc = service.AutoService(config) http_server, thread, base_url = _start_api_server_for_test(svc) try: huge_payload = b"{" + b" " * (service.MAX_CONFIG_BODY_BYTES + 10) + b"}" req = urllib_request.Request( url=f"{base_url}/config", method="POST", data=huge_payload, headers={"Content-Type": "application/json"}, ) try: urllib_request.urlopen(req) raise AssertionError("Expected request rejection for oversized payload") except error.HTTPError as exc: assert exc.code == 413 except ConnectionAbortedError: # On some Windows stacks, server closes connection before status line is read. pass except OSError as exc: if getattr(exc, "winerror", None) == 10053: pass else: raise finally: http_server.shutdown() http_server.server_close() thread.join(timeout=1.0) def test_http_refresh_requires_write_token_when_configured(monkeypatch: pytest.MonkeyPatch): config = _base_config() config["runtime"]["write_api_token"] = "secret" # type: ignore[index] svc = service.AutoService(config) svc.refresh_once = lambda: None # type: ignore[method-assign] http_server, thread, base_url = _start_api_server_for_test(svc) try: unauthorized_req = urllib_request.Request( url=f"{base_url}/refresh", method="POST", data=b"{}", headers={"Content-Type": "application/json"}, ) with pytest.raises(error.HTTPError) as exc_info: urllib_request.urlopen(unauthorized_req) assert exc_info.value.code == 401 authorized_req = urllib_request.Request( url=f"{base_url}/refresh", method="POST", data=b"{}", headers={"Content-Type": "application/json", "X-API-Token": "secret"}, ) with urllib_request.urlopen(authorized_req) as response: assert response.status == 200 finally: http_server.shutdown() http_server.server_close() thread.join(timeout=1.0) def test_http_config_get_redacts_write_token(monkeypatch: pytest.MonkeyPatch): config = _base_config() config["runtime"]["write_api_token"] = "secret" # type: ignore[index] svc = service.AutoService(config) http_server, thread, base_url = _start_api_server_for_test(svc) try: with urllib_request.urlopen(f"{base_url}/config") as response: body = response.read().decode("utf-8") payload = json.loads(body) runtime = payload["config"]["runtime"] assert runtime["write_api_token"] == "" assert runtime["write_api_token_set"] is True finally: http_server.shutdown() http_server.server_close() thread.join(timeout=1.0) def test_output_payload_is_filtered_by_frequency_range(monkeypatch: pytest.MonkeyPatch): config = _base_config() config["runtime"]["output_server"]["enabled"] = True # type: ignore[index] config["runtime"]["output_server"]["frequency_filter_enabled"] = True # type: ignore[index] config["runtime"]["output_server"]["min_frequency_mhz"] = 800.0 # type: ignore[index] config["runtime"]["output_server"]["max_frequency_mhz"] = 900.0 # type: ignore[index] responses = { "http://r0.local/measurements": { "measurements": [ {"frequency_hz": 433_920_000.0, "rssi_dbm": -60.0}, {"frequency_hz": 868_100_000.0, "rssi_dbm": -64.0}, ] }, "http://r1.local/measurements": { "measurements": [ {"frequency_hz": 433_920_000.0, "rssi_dbm": -62.0}, {"frequency_hz": 868_100_000.0, "rssi_dbm": -66.0}, ] }, "http://r2.local/measurements": { "measurements": [ {"frequency_hz": 433_920_000.0, "rssi_dbm": -61.0}, {"frequency_hz": 868_100_000.0, "rssi_dbm": -65.0}, ] }, } _install_urlopen(monkeypatch, responses) captured = {} def _fake_send_payload_to_server(**kwargs): captured["payload"] = kwargs["payload"] return 200, "ok" monkeypatch.setattr(service, "send_payload_to_server", _fake_send_payload_to_server) svc = service.AutoService(config) svc.refresh_once() sent_payload = captured["payload"] freq_rows = sent_payload["frequency_table"] assert len(freq_rows) == 1 assert freq_rows[0]["frequency_hz"] == 868_100_000.0 assert sent_payload["selected_frequency_hz"] == 868_100_000.0 def test_output_delivery_skipped_when_range_has_no_frequencies(monkeypatch: pytest.MonkeyPatch): config = _base_config() config["runtime"]["output_server"]["enabled"] = True # type: ignore[index] config["runtime"]["output_server"]["frequency_filter_enabled"] = True # type: ignore[index] config["runtime"]["output_server"]["min_frequency_mhz"] = 2_000.0 # type: ignore[index] config["runtime"]["output_server"]["max_frequency_mhz"] = 3_000.0 # type: ignore[index] responses = { "http://r0.local/measurements": {"measurements": [{"frequency_hz": 433_920_000.0, "rssi_dbm": -60.0}]}, "http://r1.local/measurements": {"measurements": [{"frequency_hz": 433_920_000.0, "rssi_dbm": -62.0}]}, "http://r2.local/measurements": {"measurements": [{"frequency_hz": 433_920_000.0, "rssi_dbm": -61.0}]}, } _install_urlopen(monkeypatch, responses) monkeypatch.setattr( service, "send_payload_to_server", lambda **_: (_ for _ in ()).throw(AssertionError("send_payload_to_server must not be called")), ) svc = service.AutoService(config) svc.refresh_once() snapshot = svc.snapshot() assert snapshot["output_delivery"]["status"] == "skipped" def test_config_validation_rejects_invalid_frequency_filter_range(): config = _base_config() config["runtime"]["output_server"]["frequency_filter_enabled"] = True # type: ignore[index] config["runtime"]["output_server"]["min_frequency_mhz"] = 900.0 # type: ignore[index] config["runtime"]["output_server"]["max_frequency_mhz"] = 800.0 # type: ignore[index] with pytest.raises(ValueError, match="max_frequency_mhz must be >= min_frequency_mhz"): service.AutoService(config) def test_receiver_input_filter_applies_per_server(monkeypatch: pytest.MonkeyPatch): config = _base_config() # Keep only ~433.92 MHz and tighter RSSI range for r0. config["input"]["receivers"][0]["input_filter"] = { # type: ignore[index] "enabled": True, "min_frequency_mhz": 430.0, "max_frequency_mhz": 440.0, "min_rssi_dbm": -61.0, "max_rssi_dbm": -59.0, } responses = { "http://r0.local/measurements": { "measurements": [ {"frequency_mhz": 433.92, "rssi_dbm": -60.0}, {"frequency_mhz": 868.1, "rssi_dbm": -60.0}, ] }, "http://r1.local/measurements": { "measurements": [ {"frequency_mhz": 433.92, "rssi_dbm": -63.0}, {"frequency_mhz": 868.1, "rssi_dbm": -66.0}, ] }, "http://r2.local/measurements": { "measurements": [ {"frequency_mhz": 433.92, "rssi_dbm": -62.0}, {"frequency_mhz": 868.1, "rssi_dbm": -65.0}, ] }, } _install_urlopen(monkeypatch, responses) svc = service.AutoService(config) svc.refresh_once() payload = svc.snapshot()["payload"] assert payload is not None assert len(payload["frequency_table"]) == 1 assert payload["frequency_table"][0]["frequency_mhz"] == pytest.approx(433.92, abs=1e-6) assert payload["receivers"][0]["raw_samples_count"] == 2 assert payload["receivers"][0]["filtered_samples_count"] == 1 def test_receiver_input_filter_empty_result_raises(monkeypatch: pytest.MonkeyPatch): config = _base_config() config["input"]["receivers"][0]["input_filter"] = { # type: ignore[index] "enabled": True, "min_frequency_mhz": 433.0, "max_frequency_mhz": 434.0, "min_rssi_dbm": -10.0, "max_rssi_dbm": -5.0, } responses = { "http://r0.local/measurements": {"measurements": [{"frequency_mhz": 433.92, "rssi_dbm": -60.0}]}, "http://r1.local/measurements": {"measurements": [{"frequency_mhz": 433.92, "rssi_dbm": -62.0}]}, "http://r2.local/measurements": {"measurements": [{"frequency_mhz": 433.92, "rssi_dbm": -61.0}]}, } _install_urlopen(monkeypatch, responses) svc = service.AutoService(config) with pytest.raises(RuntimeError, match="no measurements left after input_filter"): svc.refresh_once() def test_receiver_input_filter_validation_rejects_invalid_rssi_range(): config = _base_config() config["input"]["receivers"][1]["input_filter"] = { # type: ignore[index] "enabled": True, "min_frequency_mhz": 430.0, "max_frequency_mhz": 440.0, "min_rssi_dbm": -30.0, "max_rssi_dbm": -80.0, } with pytest.raises(ValueError, match="max_rssi_dbm must be >= min_rssi_dbm"): service.AutoService(config)