149 lines
6.2 KiB
Python
149 lines
6.2 KiB
Python
from tests.unit.app.app_route_helpers import *
|
|
|
|
|
|
def test_health_ok(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"ok": True}
|
|
|
|
def test_startup_and_shutdown(monkeypatch) -> None:
|
|
monkeypatch.setattr(app_module.provisioning, "start", lambda: None)
|
|
monkeypatch.setattr(app_module.scheduler, "add_task", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(app_module.scheduler, "start", lambda: None)
|
|
monkeypatch.setattr(app_module.scheduler, "stop", lambda: None)
|
|
monkeypatch.setattr(app_module.provisioning, "stop", lambda: None)
|
|
monkeypatch.setattr(app_module.portal_db, "close", lambda: None)
|
|
monkeypatch.setattr(app_module.ariadne_db, "close", lambda: None)
|
|
|
|
app_module._startup()
|
|
app_module._shutdown()
|
|
|
|
def test_startup_registers_metis_watch(monkeypatch) -> None:
|
|
tasks = []
|
|
|
|
monkeypatch.setattr(app_module.provisioning, "start", lambda: None)
|
|
monkeypatch.setattr(app_module.scheduler, "start", lambda: None)
|
|
monkeypatch.setattr(app_module.scheduler, "stop", lambda: None)
|
|
monkeypatch.setattr(app_module.provisioning, "stop", lambda: None)
|
|
monkeypatch.setattr(app_module.portal_db, "close", lambda: None)
|
|
monkeypatch.setattr(app_module.ariadne_db, "close", lambda: None)
|
|
monkeypatch.setattr(
|
|
app_module.scheduler,
|
|
"add_task",
|
|
lambda name, cron_expr, runner: tasks.append((name, cron_expr)),
|
|
)
|
|
|
|
app_module._startup()
|
|
|
|
assert any(name == "schedule.metis_sentinel_watch" for name, _cron in tasks)
|
|
assert any(name == "schedule.metis_k3s_token_sync" for name, _cron in tasks)
|
|
assert any(name == "schedule.platform_quality_suite_probe" for name, _cron in tasks)
|
|
assert any(name == "schedule.jenkins_build_weather" for name, _cron in tasks)
|
|
assert any(name == "schedule.jenkins_workspace_cleanup" for name, _cron in tasks)
|
|
|
|
def test_record_event_handles_exception(monkeypatch) -> None:
|
|
monkeypatch.setattr(app_module.storage, "record_event", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("fail")))
|
|
app_module._record_event("event", {"ok": True})
|
|
|
|
def test_parse_event_detail_variants() -> None:
|
|
assert app_module._parse_event_detail(None) == ""
|
|
assert app_module._parse_event_detail("not-json") == "not-json"
|
|
|
|
def test_missing_auth_header(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
|
|
resp = client.get("/api/admin/access/requests")
|
|
assert resp.status_code == 401
|
|
|
|
def test_invalid_token(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
monkeypatch.setattr(app_module.authenticator, "authenticate", lambda token: (_ for _ in ()).throw(ValueError("bad")))
|
|
|
|
resp = client.get(
|
|
"/api/admin/access/requests",
|
|
headers={"Authorization": "Bearer token"},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
def test_account_access_denied(monkeypatch) -> None:
|
|
ctx = AuthContext(username="alice", email="", groups=["guest"], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
|
|
resp = client.post(
|
|
"/api/account/firefly/reset",
|
|
headers={"Authorization": "Bearer token"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
def test_account_access_allows_missing_groups(monkeypatch) -> None:
|
|
ctx = AuthContext(username="alice", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
|
|
resp = client.post(
|
|
"/api/account/firefly/reset",
|
|
headers={"Authorization": "Bearer token"},
|
|
)
|
|
assert resp.status_code != 403
|
|
|
|
def test_retry_access_request_ok(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
executed = []
|
|
invoked = {}
|
|
|
|
monkeypatch.setattr(app_module.keycloak_admin, "ready", lambda: True)
|
|
monkeypatch.setattr(app_module.portal_db, "fetchone", lambda *_args, **_kwargs: {"status": "accounts_building"})
|
|
monkeypatch.setattr(app_module.portal_db, "execute", lambda query, params=None: executed.append((query, params)))
|
|
monkeypatch.setattr(app_module.provisioning, "provision_access_request", lambda code: invoked.setdefault("code", code))
|
|
monkeypatch.setattr(app_module, "_record_event", lambda *args, **kwargs: None)
|
|
|
|
resp = client.post("/api/access/requests/REQ123/retry")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["request_code"] == "REQ123"
|
|
assert invoked["code"] == "REQ123"
|
|
assert any("provision_attempted_at" in query for query, _params in executed)
|
|
|
|
def test_retry_access_request_not_found(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
monkeypatch.setattr(app_module.keycloak_admin, "ready", lambda: True)
|
|
monkeypatch.setattr(app_module.portal_db, "fetchone", lambda *_args, **_kwargs: None)
|
|
|
|
resp = client.post("/api/access/requests/REQ123/retry")
|
|
assert resp.status_code == 404
|
|
|
|
def test_retry_access_request_not_retryable(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
monkeypatch.setattr(app_module.keycloak_admin, "ready", lambda: True)
|
|
monkeypatch.setattr(app_module.portal_db, "fetchone", lambda *_args, **_kwargs: {"status": "ready"})
|
|
|
|
resp = client.post("/api/access/requests/REQ123/retry")
|
|
assert resp.status_code == 409
|
|
|
|
def test_metrics_endpoint(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
|
|
resp = client.get("/metrics")
|
|
assert resp.status_code == 200
|
|
|
|
def test_mailu_event_endpoint(monkeypatch) -> None:
|
|
ctx = AuthContext(username="", email="", groups=[], claims={})
|
|
client = _client(monkeypatch, ctx)
|
|
|
|
class DummyEvents:
|
|
def handle_event(self, payload):
|
|
assert payload == {"wait": False}
|
|
return 202, {"status": "accepted"}
|
|
|
|
monkeypatch.setattr(app_module, "mailu_events", DummyEvents())
|
|
|
|
resp = client.post("/events", json={"wait": False})
|
|
assert resp.status_code == 202
|
|
assert resp.json()["status"] == "accepted"
|