117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
|
|
"""Offline test rig: a fake backend fed from JSON fixtures.
|
||
|
|
|
||
|
|
Same idea as scripts/ha/tests/test_import_api_offline.sh — no network, no HA
|
||
|
|
instance, no token. The fixtures under tests/fixtures/ are shaped exactly
|
||
|
|
like real `/api/states` and WebSocket registry payloads (trimmed down);
|
||
|
|
read_automation tests read the real repo automations, which is offline too.
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
SERVICE_DIR = Path(__file__).resolve().parents[1]
|
||
|
|
REPO_ROOT = SERVICE_DIR.parents[1]
|
||
|
|
FIXTURES = Path(__file__).resolve().parent / "fixtures"
|
||
|
|
|
||
|
|
sys.path.insert(0, str(SERVICE_DIR / "src"))
|
||
|
|
|
||
|
|
from ha_mcp import backend as backend_mod # noqa: E402
|
||
|
|
|
||
|
|
|
||
|
|
def load_fixture(name):
|
||
|
|
with open(FIXTURES / name, encoding="utf-8") as f:
|
||
|
|
return json.load(f)
|
||
|
|
|
||
|
|
|
||
|
|
class FakeBackend:
|
||
|
|
"""Stands in for LiveBackend: canned states/config/registries, real repo files."""
|
||
|
|
|
||
|
|
def __init__(self, instance="ken", states=None, registries=None, config=None,
|
||
|
|
area_source="websocket", automations_instance=None):
|
||
|
|
self.instance = instance
|
||
|
|
self.cfg = {
|
||
|
|
"adapter": "api",
|
||
|
|
"base_url": "http://ha.example:8123",
|
||
|
|
"status": "active",
|
||
|
|
"site": "ken",
|
||
|
|
"token_path": "/nonexistent",
|
||
|
|
}
|
||
|
|
self._states = load_fixture("states.json") if states is None else states
|
||
|
|
reg = load_fixture("registries.json") if registries is None else registries
|
||
|
|
self._config = config or {
|
||
|
|
"version": "2026.7.2",
|
||
|
|
"location_name": "KEN",
|
||
|
|
"time_zone": "Europe/Warsaw",
|
||
|
|
}
|
||
|
|
self._index = backend_mod._index_from_registries(
|
||
|
|
reg.get("areas"), reg.get("entities"), reg.get("devices"), area_source
|
||
|
|
)
|
||
|
|
self._automations_instance = automations_instance or instance
|
||
|
|
|
||
|
|
@property
|
||
|
|
def status(self):
|
||
|
|
return self.cfg.get("status") or "unknown"
|
||
|
|
|
||
|
|
def states(self):
|
||
|
|
return self._states
|
||
|
|
|
||
|
|
def ha_config(self):
|
||
|
|
return self._config
|
||
|
|
|
||
|
|
def area_index(self):
|
||
|
|
return self._index
|
||
|
|
|
||
|
|
def automation_files(self):
|
||
|
|
return backend_mod.automation_files(self._automations_instance, REPO_ROOT)
|
||
|
|
|
||
|
|
|
||
|
|
class UnreachableBackend(FakeBackend):
|
||
|
|
"""Live calls fail; repo-backed calls still work (read_automation offline path)."""
|
||
|
|
|
||
|
|
def states(self):
|
||
|
|
raise backend_mod.BackendError("instance 'ken' unreachable at http://ha.example:8123")
|
||
|
|
|
||
|
|
def ha_config(self):
|
||
|
|
raise backend_mod.BackendError("instance 'ken' unreachable at http://ha.example:8123")
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def fake_backend():
|
||
|
|
return FakeBackend()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def unreachable_backend():
|
||
|
|
return UnreachableBackend()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def repo_root():
|
||
|
|
return REPO_ROOT
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(scope="session")
|
||
|
|
def real_ken_backend():
|
||
|
|
"""FakeBackend over the newest committed /api/states snapshot of `ken`.
|
||
|
|
|
||
|
|
Still offline (the fixture is a file in the repo), but ~1650 real
|
||
|
|
entities instead of ten — ranking behaves differently at that scale.
|
||
|
|
Areas come from the storage-export fallback, same as a live run with the
|
||
|
|
WebSocket unavailable.
|
||
|
|
"""
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
from ha_mcp import backend as bm
|
||
|
|
|
||
|
|
snapshots = sorted((REPO_ROOT / "services" / "home-assistant" / "fixtures").glob("ken-states-*.yaml"))
|
||
|
|
if not snapshots:
|
||
|
|
pytest.skip("no ken-states-*.yaml fixture in the repo")
|
||
|
|
with open(snapshots[-1], encoding="utf-8") as f:
|
||
|
|
states = yaml.safe_load(f)
|
||
|
|
|
||
|
|
backend = FakeBackend(states=states)
|
||
|
|
backend._index = bm.load_offline_area_index("ken", REPO_ROOT)
|
||
|
|
return backend
|