import json from typing import Any, Dict, List from urllib import error 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 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()