mirror of
https://github.com/openhop-dev/openhop_repeater.git
synced 2026-10-24 05:52:59 +00:00
246 lines
9.1 KiB
Python
246 lines
9.1 KiB
Python
"""Sensor manager discovery and secret round-trip contracts."""
|
|
|
|
import copy
|
|
|
|
import cherrypy
|
|
import yaml
|
|
|
|
from repeater.sensors import SensorRegistry
|
|
from repeater.web.api_endpoints import APIEndpoints
|
|
|
|
|
|
def _api(tmp_path):
|
|
config = {
|
|
"sensors": {
|
|
"enabled": True,
|
|
"definitions": [
|
|
{
|
|
"name": "modem",
|
|
"type": "openhop_modem",
|
|
"enabled": True,
|
|
"settings": {"host": "modem.local", "password": "private-value"},
|
|
},
|
|
{"name": "other", "type": "bme280", "settings": {"bus_number": 1}},
|
|
],
|
|
},
|
|
}
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text(yaml.safe_dump(config))
|
|
return APIEndpoints(config=copy.deepcopy(config), config_path=str(path)), path
|
|
|
|
|
|
def test_sensor_config_masks_password_without_mutating_runtime(tmp_path):
|
|
api, _ = _api(tmp_path)
|
|
result = api.sensors_config()
|
|
assert result["success"] is True
|
|
definitions = result["data"]["definitions"]
|
|
assert definitions[0]["settings"]["password"] == "*****"
|
|
assert definitions[1]["settings"]["bus_number"] == 1
|
|
assert api.config["sensors"]["definitions"][0]["settings"]["password"] == "private-value"
|
|
|
|
|
|
def test_masked_password_round_trip_preserves_secret_and_other_edits(tmp_path):
|
|
api, path = _api(tmp_path)
|
|
body = api.sensors_config()["data"]
|
|
body["definitions"][0]["settings"]["host"] = "new-modem.local"
|
|
cherrypy.request.method = "POST"
|
|
cherrypy.request.json = body
|
|
assert api.sensors_config_update()["success"] is True
|
|
saved = yaml.safe_load(path.read_text())["sensors"]["definitions"][0]["settings"]
|
|
assert saved["password"] == "private-value"
|
|
assert saved["host"] == "new-modem.local"
|
|
assert "private-value" not in str(body)
|
|
|
|
|
|
def test_renaming_unique_sensor_keeps_its_masked_password(tmp_path):
|
|
api, path = _api(tmp_path)
|
|
body = api.sensors_config()["data"]
|
|
body["definitions"][0]["name"] = "renamed-modem"
|
|
cherrypy.request.method = "POST"
|
|
cherrypy.request.json = body
|
|
assert api.sensors_config_update()["success"] is True
|
|
saved = yaml.safe_load(path.read_text())["sensors"]["definitions"][0]
|
|
assert saved["name"] == "renamed-modem"
|
|
assert saved["settings"]["password"] == "private-value"
|
|
|
|
|
|
def test_rename_existing_modem_while_adding_second_preserves_correct_password(tmp_path):
|
|
api, path = _api(tmp_path)
|
|
api.config["sensors"]["definitions"].reverse()
|
|
path.write_text(yaml.safe_dump(api.config))
|
|
body = api.sensors_config()["data"]
|
|
body["definitions"][1]["name"] = "renamed-modem"
|
|
body["definitions"].append(
|
|
{
|
|
"name": "new-modem",
|
|
"type": "openhop_modem",
|
|
"enabled": True,
|
|
"settings": {"host": "second.local", "password": "second-private-value"},
|
|
}
|
|
)
|
|
cherrypy.request.method = "POST"
|
|
cherrypy.request.json = body
|
|
result = api.sensors_config_update()
|
|
assert result["success"] is True, result.get("error")
|
|
saved = yaml.safe_load(path.read_text())["sensors"]["definitions"]
|
|
assert saved[1]["settings"]["password"] == "private-value"
|
|
assert saved[2]["settings"]["password"] == "second-private-value"
|
|
assert all("_original_name" not in d for d in saved)
|
|
|
|
|
|
def test_swapping_modem_names_does_not_swap_their_passwords(tmp_path):
|
|
api, path = _api(tmp_path)
|
|
first = api.config["sensors"]["definitions"][0]
|
|
api.config["sensors"]["definitions"] = [
|
|
first,
|
|
{
|
|
"name": "second",
|
|
"type": "openhop_modem",
|
|
"enabled": True,
|
|
"settings": {"host": "second.local", "password": "second-secret"},
|
|
},
|
|
]
|
|
path.write_text(yaml.safe_dump(api.config))
|
|
body = api.sensors_config()["data"]
|
|
body["definitions"][0]["name"] = "second"
|
|
body["definitions"][1]["name"] = "modem"
|
|
cherrypy.request.method = "POST"
|
|
cherrypy.request.json = body
|
|
assert api.sensors_config_update()["success"] is True
|
|
saved = yaml.safe_load(path.read_text())["sensors"]["definitions"]
|
|
assert [d["settings"]["password"] for d in saved] == ["private-value", "second-secret"]
|
|
|
|
|
|
def test_duplicate_origin_cannot_copy_another_modems_masked_password(tmp_path):
|
|
api, path = _api(tmp_path)
|
|
original = path.read_text()
|
|
body = api.sensors_config()["data"]
|
|
duplicate = copy.deepcopy(body["definitions"][0])
|
|
duplicate["name"] = "new-modem"
|
|
duplicate["settings"]["host"] = "new.local"
|
|
body["definitions"].append(duplicate)
|
|
cherrypy.request.method = "POST"
|
|
cherrypy.request.json = body
|
|
assert api.sensors_config_update()["success"] is False
|
|
assert path.read_text() == original
|
|
|
|
|
|
def test_new_password_and_explicit_clear(tmp_path):
|
|
api, path = _api(tmp_path)
|
|
cherrypy.request.method = "POST"
|
|
body = api.sensors_config()["data"]
|
|
body["definitions"][0]["settings"]["password"] = "replacement"
|
|
cherrypy.request.json = body
|
|
assert api.sensors_config_update()["success"] is True
|
|
assert (
|
|
yaml.safe_load(path.read_text())["sensors"]["definitions"][0]["settings"]["password"]
|
|
== "replacement"
|
|
)
|
|
body["definitions"][0]["settings"]["password"] = ""
|
|
cherrypy.request.json = body
|
|
assert api.sensors_config_update()["success"] is True
|
|
assert (
|
|
yaml.safe_load(path.read_text())["sensors"]["definitions"][0]["settings"]["password"] == ""
|
|
)
|
|
|
|
|
|
def test_masked_password_on_new_sensor_is_rejected_without_writing(tmp_path):
|
|
api, path = _api(tmp_path)
|
|
before = path.read_text()
|
|
body = api.sensors_config()["data"]
|
|
body["definitions"].append(
|
|
{"name": "new-modem", "type": "openhop_modem", "settings": {"password": "*****"}}
|
|
)
|
|
cherrypy.request.method = "POST"
|
|
cherrypy.request.json = body
|
|
assert api.sensors_config_update()["success"] is False
|
|
assert path.read_text() == before
|
|
|
|
|
|
def test_duplicate_sensor_names_are_rejected_without_writing(tmp_path):
|
|
api, path = _api(tmp_path)
|
|
original = path.read_text()
|
|
body = api.sensors_config()["data"]
|
|
body["definitions"].append(
|
|
{"name": "modem", "type": "openhop_modem", "settings": {"host": "other.local"}}
|
|
)
|
|
cherrypy.request.method = "POST"
|
|
cherrypy.request.json = body
|
|
result = api.sensors_config_update()
|
|
assert result["success"] is False
|
|
assert path.read_text() == original
|
|
|
|
|
|
def test_two_modems_and_legacy_alias_remain_loadable(tmp_path):
|
|
from repeater.sensors import SensorManager
|
|
|
|
api, path = _api(tmp_path)
|
|
section = api.config["sensors"]
|
|
section["definitions"] = [
|
|
{
|
|
"name": "north",
|
|
"type": "openhop_modem",
|
|
"enabled": True,
|
|
"settings": {"host": "north.local", "password": "north-secret"},
|
|
},
|
|
{
|
|
"name": "south",
|
|
"type": "openhop_modem",
|
|
"enabled": True,
|
|
"settings": {"host": "south.local", "password": "south-secret"},
|
|
},
|
|
{
|
|
"name": "legacy",
|
|
"type": "pymc_modem",
|
|
"enabled": True,
|
|
"settings": {"host": "legacy.local", "password": "legacy-secret"},
|
|
},
|
|
]
|
|
path.write_text(yaml.safe_dump(api.config))
|
|
masked = api.sensors_config()["data"]
|
|
assert [d["settings"]["password"] for d in masked["definitions"]] == ["*****"] * 3
|
|
masked["definitions"].reverse()
|
|
cherrypy.request.method = "POST"
|
|
cherrypy.request.json = masked
|
|
assert api.sensors_config_update()["success"] is True
|
|
saved = yaml.safe_load(path.read_text())["sensors"]["definitions"]
|
|
assert [(d["name"], d["type"]) for d in saved] == [
|
|
("legacy", "pymc_modem"),
|
|
("south", "openhop_modem"),
|
|
("north", "openhop_modem"),
|
|
]
|
|
assert [d["settings"]["password"] for d in saved] == [
|
|
"legacy-secret",
|
|
"south-secret",
|
|
"north-secret",
|
|
]
|
|
manager = SensorManager(api.config)
|
|
assert [sensor.name for sensor in manager.sensors] == ["legacy", "south", "north"]
|
|
assert manager.get_summary()["loaded"] == 3
|
|
|
|
|
|
def test_types_endpoint_discovers_new_installed_sensor_module(tmp_path, monkeypatch):
|
|
from repeater import sensors
|
|
|
|
folder = tmp_path / "extra-sensors"
|
|
folder.mkdir()
|
|
(folder / "future_sensor.py").write_text(
|
|
"from repeater.sensors.registry import SensorRegistry\n"
|
|
"from repeater.sensors.base import SensorBase\n"
|
|
"@SensorRegistry.register('future_sensor')\n"
|
|
"class FutureSensor(SensorBase):\n"
|
|
" sensor_type = 'future_sensor'\n"
|
|
" _settings_schema = [{'key': 'pin', 'type': 'integer', 'label': 'Pin', 'default': 1}]\n"
|
|
" def _read(self): return {'value': 1}\n"
|
|
)
|
|
monkeypatch.setattr(sensors, "__path__", list(sensors.__path__) + [str(folder)])
|
|
try:
|
|
api, _ = _api(tmp_path)
|
|
result = api.sensors_types()
|
|
assert result["success"] is True
|
|
assert "pymc_modem" not in {entry["type"] for entry in result["data"]["types"]}
|
|
future = next(t for t in result["data"]["types"] if t["type"] == "future_sensor")
|
|
assert future["settings"][0]["key"] == "pin"
|
|
finally:
|
|
SensorRegistry._factories.pop("future_sensor", None)
|