AlexsandrSnytkin
/
Triangulation
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
main
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'main'
${ noResults }
Triangulation/test_service_integration.py

738 lines
27 KiB
Python
Raw Permalink Blame History

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\\(s\\) rejected payload"):
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_parse_source_payload_accepts_compact_short_keys():
payload = {
"receiver_id": "r0",
"samples": [
{"f_mhz": 868.1, "rssi": -60.0},
{"f_hz": 433_920_000.0, "dbm": -62.5},
],
}
parsed = service.parse_source_payload(
payload=payload,
source_label="source_url=test",
expected_receiver_id="r0",
)
assert parsed[0][0] == pytest.approx(868_100_000.0)
assert parsed[1][0] == pytest.approx(433_920_000.0)
assert parsed[1][1] == pytest.approx(-62.5)
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)
current_service = http_server.RequestHandlerClass.service_holder["current"] # type: ignore[attr-defined]
current_service.stop()
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_config_applies_without_manual_restart(monkeypatch: pytest.MonkeyPatch):
config = _base_config()
monkeypatch.setattr(
service,
"_fetch_measurements",
lambda *_, **__: [(915e6, -60.0)],
)
svc = service.AutoService(config)
http_server, thread, base_url = _start_api_server_for_test(svc)
try:
new_config = _base_config()
new_config["runtime"]["poll_interval_s"] = 0.25 # type: ignore[index]
raw = json.dumps(new_config).encode("utf-8")
req = urllib_request.Request(
url=f"{base_url}/config",
method="POST",
data=raw,
headers={"Content-Type": "application/json"},
)
with urllib_request.urlopen(req) as response:
payload = json.loads(response.read().decode("utf-8"))
assert payload["status"] == "ok"
assert payload["applied"] is True
assert payload["restart_required"] is False
with urllib_request.urlopen(f"{base_url}/config") as response:
payload = json.loads(response.read().decode("utf-8"))
assert payload["config"]["runtime"]["poll_interval_s"] == 0.25
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"]
assert set(sent_payload.keys()) == {"x", "y", "z"}
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_multiple_output_servers_with_names(monkeypatch: pytest.MonkeyPatch):
config = _base_config()
config["runtime"]["output_servers"] = [ # type: ignore[index]
{
"name": "sink_a",
"enabled": True,
"ip": "127.0.0.1",
"port": 8080,
"path": "/triangulation",
"timeout_s": 1.0,
},
{
"name": "sink_b",
"enabled": True,
"ip": "127.0.0.2",
"port": 8081,
"path": "/triangulation",
"timeout_s": 1.0,
},
]
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)
calls: List[Dict[str, object]] = []
def _fake_send_payload_to_server(**kwargs):
calls.append(kwargs)
if kwargs["server_ip"] == "127.0.0.2":
return 500, "fail"
return 200, "ok"
monkeypatch.setattr(service, "send_payload_to_server", _fake_send_payload_to_server)
svc = service.AutoService(config)
with pytest.raises(RuntimeError, match="sink_b"):
svc.refresh_once()
snapshot = svc.snapshot()
assert snapshot["output_delivery"]["status"] in ("partial", "error")
servers = snapshot["output_delivery"]["servers"]
assert isinstance(servers, list)
assert {row["name"] for row in servers} == {"sink_a", "sink_b"}
assert len(calls) == 2
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 configured filters"):
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)
def test_refresh_once_supports_more_than_three_receivers_and_chooses_best_triplet(
monkeypatch: pytest.MonkeyPatch,
):
config = _base_config()
config["input"]["receivers"].append( # type: ignore[index]
{
"receiver_id": "r3",
"center": {"x": 50.0, "y": 50.0, "z": 0.0},
"source_url": "http://r3.local/measurements",
}
)
responses = {
"http://r0.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -60.0}]},
"http://r1.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -61.0}]},
"http://r2.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -62.0}]},
"http://r3.local/measurements": {"measurements": [{"frequency_hz": 915e6, "rssi_dbm": -120.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
row = payload["frequency_table"][0]
assert row["used_receivers_count"] == 3
assert row["available_receivers_count"] == 4
assert len(row["receivers"]) == 3
def test_receiver_access_url_and_default_input_filter(monkeypatch: pytest.MonkeyPatch):
config = _base_config()
config["input"]["default_input_filter"] = { # type: ignore[index]
"enabled": True,
"min_frequency_mhz": 900.0,
"max_frequency_mhz": 920.0,
"min_rssi_dbm": -90.0,
"max_rssi_dbm": -50.0,
}
for receiver in config["input"]["receivers"]: # type: ignore[index]
receiver["access"] = {"url": receiver["source_url"]}
del receiver["source_url"]
receiver.pop("input_filter", None)
responses = {
"http://r0.local/measurements": {
"measurements": [
{"frequency_hz": 915e6, "rssi_dbm": -60.0},
{"frequency_hz": 433e6, "rssi_dbm": -60.0},
]
},
"http://r1.local/measurements": {
"measurements": [
{"frequency_hz": 915e6, "rssi_dbm": -61.0},
{"frequency_hz": 433e6, "rssi_dbm": -61.0},
]
},
"http://r2.local/measurements": {
"measurements": [
{"frequency_hz": 915e6, "rssi_dbm": -62.0},
{"frequency_hz": 433e6, "rssi_dbm": -62.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_hz"] == pytest.approx(915e6)
def test_receiver_configured_frequencies_limit_trilateration(monkeypatch: pytest.MonkeyPatch):
config = _base_config()
for receiver in config["input"]["receivers"]: # type: ignore[index]
receiver["frequencies_mhz"] = [915.0]
responses = {
"http://r0.local/measurements": {
"measurements": [
{"frequency_hz": 915e6, "rssi_dbm": -60.0},
{"frequency_hz": 433_920_000.0, "rssi_dbm": -61.0},
]
},
"http://r1.local/measurements": {
"measurements": [
{"frequency_hz": 915e6, "rssi_dbm": -61.0},
{"frequency_hz": 433_920_000.0, "rssi_dbm": -62.0},
]
},
"http://r2.local/measurements": {
"measurements": [
{"frequency_hz": 915e6, "rssi_dbm": -62.0},
{"frequency_hz": 433_920_000.0, "rssi_dbm": -63.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_hz"] == pytest.approx(915e6)
Reference in New Issue View Git Blame Copy Permalink