from __future__ import annotations from types import SimpleNamespace import pytest from ariadne.services import wolf_api as wolf_api_module from ariadne.services import wolf_gatekeeper as wolf_gatekeeper_module from ariadne.services.wolf_api import WolfApiService from ariadne.services.wolf_gatekeeper import WolfGatekeeperService class DummyResponse: def __init__(self, payload) -> None: self._payload = payload def raise_for_status(self) -> None: pass def json(self): return self._payload class DummyClient: calls: list[tuple[str, str, object | None]] = [] payload: object = {"success": True} def __init__(self, timeout: float) -> None: self.timeout = timeout def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def request(self, method: str, url: str, json=None): self.calls.append((method, url, json)) return DummyResponse(self.payload) def _settings(**overrides): defaults = { "wolf_api_url": "http://wolf-api", "wolf_api_timeout_sec": 2.0, "wolf_gatekeeper_url": "http://wolf-gatekeeper", "wolf_gatekeeper_timeout_sec": 3.0, } defaults.update(overrides) return SimpleNamespace(**defaults) def test_wolf_api_service_requests(monkeypatch) -> None: DummyClient.calls = [] DummyClient.payload = {"success": True, "clients": []} monkeypatch.setattr(wolf_api_module, "settings", _settings()) monkeypatch.setattr(wolf_api_module.httpx, "Client", DummyClient) svc = WolfApiService() assert svc.enabled() is True assert svc.clients()["success"] is True assert svc.pending_pair_requests()["success"] is True assert svc.sessions()["success"] is True assert svc.apps()["success"] is True assert svc.pair_client("secret", "1234")["success"] is True assert DummyClient.calls[-1] == ("POST", "http://wolf-api/api/v1/pair/client", {"pair_secret": "secret", "pin": "1234"}) def test_wolf_api_service_errors(monkeypatch) -> None: monkeypatch.setattr(wolf_api_module, "settings", _settings(wolf_api_url="")) svc = WolfApiService() assert svc.enabled() is False with pytest.raises(RuntimeError, match="not configured"): svc.clients() DummyClient.payload = [] monkeypatch.setattr(wolf_api_module, "settings", _settings()) monkeypatch.setattr(wolf_api_module.httpx, "Client", DummyClient) with pytest.raises(RuntimeError, match="unexpected"): svc.clients() def test_wolf_gatekeeper_service_requests(monkeypatch) -> None: DummyClient.calls = [] DummyClient.payload = {"success": True, "active_unlocks": []} monkeypatch.setattr(wolf_gatekeeper_module, "settings", _settings()) monkeypatch.setattr(wolf_gatekeeper_module.httpx, "Client", DummyClient) svc = WolfGatekeeperService() assert svc.enabled() is True assert svc.status()["success"] is True assert svc.unlock("1.2.3.4", 300, "olya", "olya")["success"] is True assert svc.revoke("1.2.3.4")["success"] is True assert DummyClient.calls[1] == ( "POST", "http://wolf-gatekeeper/unlock", {"ip": "1.2.3.4", "ttl_seconds": 300, "actor": "olya", "target_user": "olya"}, ) assert DummyClient.calls[2] == ("POST", "http://wolf-gatekeeper/revoke", {"ip": "1.2.3.4"}) def test_wolf_gatekeeper_service_errors(monkeypatch) -> None: monkeypatch.setattr(wolf_gatekeeper_module, "settings", _settings(wolf_gatekeeper_url="")) svc = WolfGatekeeperService() assert svc.enabled() is False with pytest.raises(RuntimeError, match="not configured"): svc.status() DummyClient.payload = [] monkeypatch.setattr(wolf_gatekeeper_module, "settings", _settings()) monkeypatch.setattr(wolf_gatekeeper_module.httpx, "Client", DummyClient) with pytest.raises(RuntimeError, match="unexpected"): svc.status()