116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
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()
|