From e6d807ed3f3fc0b7804dece1c9eb6b935f5bafc6 Mon Sep 17 00:00:00 2001 From: codex Date: Tue, 21 Apr 2026 07:44:43 -0300 Subject: [PATCH] test(bstein-home): cover vaultwarden invite adapter --- backend/tests/test_vaultwarden.py | 260 ++++++++++++++++++++++++++++++ 1 file changed, 260 insertions(+) create mode 100644 backend/tests/test_vaultwarden.py diff --git a/backend/tests/test_vaultwarden.py b/backend/tests/test_vaultwarden.py new file mode 100644 index 0000000..8d2b5b1 --- /dev/null +++ b/backend/tests/test_vaultwarden.py @@ -0,0 +1,260 @@ +from __future__ import annotations + +"""Tests for Vaultwarden invite and Kubernetes discovery helpers.""" + +import base64 +from pathlib import Path + +import pytest + +from atlas_portal import vaultwarden + + +class DummyResponse: + """HTTP response double used by Vaultwarden tests.""" + + def __init__(self, payload=None, *, status_code: int = 200, text: str = "") -> None: + self._payload = payload if payload is not None else {} + self.status_code = status_code + self.text = text + + def json(self): + """Return configured JSON payload.""" + + return self._payload + + def raise_for_status(self) -> None: + """Raise for HTTP failure statuses.""" + + if self.status_code >= 400: + raise RuntimeError("bad status") + + +class DummyClient: + """httpx.Client replacement with queued responses.""" + + responses: list[DummyResponse] = [] + calls: list[tuple[str, str, dict]] = [] + closed = 0 + + def __init__(self, **kwargs): + self.kwargs = kwargs + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + @classmethod + def reset(cls, *responses: DummyResponse) -> None: + """Replace queued responses and clear captured calls.""" + + cls.responses = list(responses) + cls.calls = [] + cls.closed = 0 + + def close(self) -> None: + """Record cache eviction closing behavior.""" + + type(self).closed += 1 + + def get(self, url): + self.calls.append(("GET", url, {})) + return self.responses.pop(0) + + def post(self, url, **kwargs): + self.calls.append(("POST", url, kwargs)) + return self.responses.pop(0) + + +def _reset_admin_state(monkeypatch) -> None: + monkeypatch.setattr(vaultwarden, "_ADMIN_SESSION", None) + monkeypatch.setattr(vaultwarden, "_ADMIN_SESSION_EXPIRES_AT", 0.0) + monkeypatch.setattr(vaultwarden, "_ADMIN_SESSION_BASE_URL", "") + monkeypatch.setattr(vaultwarden, "_ADMIN_RATE_LIMITED_UNTIL", 0.0) + + +def test_service_account_and_k8s_json_paths(monkeypatch, tmp_path: Path) -> None: + sa_path = tmp_path / "sa" + sa_path.mkdir() + monkeypatch.setattr(vaultwarden, "_SA_PATH", sa_path) + + with pytest.raises(RuntimeError, match="token missing"): + vaultwarden._read_service_account() + + (sa_path / "token").write_text(" ") + (sa_path / "ca.crt").write_text("ca") + with pytest.raises(RuntimeError, match="token empty"): + vaultwarden._read_service_account() + + (sa_path / "token").write_text("token") + assert vaultwarden._read_service_account() == ("token", str(sa_path / "ca.crt")) + + monkeypatch.setattr(vaultwarden, "_read_service_account", lambda: ("token", "/ca.crt")) + monkeypatch.setattr(vaultwarden.httpx, "Client", DummyClient) + DummyClient.reset(DummyResponse({"items": []}), DummyResponse([])) + + assert vaultwarden._k8s_get_json("/api") == {"items": []} + with pytest.raises(RuntimeError, match="unexpected kubernetes response"): + vaultwarden._k8s_get_json("/api") + + +def test_k8s_pod_and_secret_helpers(monkeypatch) -> None: + encoded = base64.b64encode(b"admin-token").decode("ascii") + pods = { + "items": [ + {"status": {"phase": "Pending", "podIP": "10.0.0.1"}}, + { + "status": { + "phase": "Running", + "podIP": "10.0.0.2", + "conditions": [{"type": "Ready", "status": "True"}], + } + }, + ] + } + monkeypatch.setattr(vaultwarden, "_k8s_get_json", lambda path: pods if "pods" in path else {"data": {"token": encoded}}) + + assert vaultwarden._k8s_find_pod_ip("apps", "app=vaultwarden") == "10.0.0.2" + assert vaultwarden._k8s_get_secret_value("apps", "secret", "token") == "admin-token" + + no_condition_pod = {"items": [{"status": {"phase": "Running", "podIP": "10.0.0.3", "conditions": [None]}}]} + monkeypatch.setattr(vaultwarden, "_k8s_get_json", lambda path: no_condition_pod) + assert vaultwarden._k8s_find_pod_ip("apps", "app=vaultwarden") == "10.0.0.3" + + monkeypatch.setattr(vaultwarden, "_k8s_get_json", lambda path: {"items": []}) + with pytest.raises(RuntimeError, match="no vaultwarden pods"): + vaultwarden._k8s_find_pod_ip("apps", "app=vaultwarden") + + monkeypatch.setattr(vaultwarden, "_k8s_get_json", lambda path: {"items": [{"status": {"phase": "Running"}}]}) + with pytest.raises(RuntimeError, match="no IP"): + vaultwarden._k8s_find_pod_ip("apps", "app=vaultwarden") + + monkeypatch.setattr(vaultwarden, "_k8s_get_json", lambda path: {"data": {}}) + with pytest.raises(RuntimeError, match="secret key missing"): + vaultwarden._k8s_get_secret_value("apps", "secret", "token") + + monkeypatch.setattr(vaultwarden, "_k8s_get_json", lambda path: {"data": {"token": "bad-base64"}}) + with pytest.raises(RuntimeError, match="failed to decode"): + vaultwarden._k8s_get_secret_value("apps", "secret", "token") + + empty = base64.b64encode(b" ").decode("ascii") + monkeypatch.setattr(vaultwarden, "_k8s_get_json", lambda path: {"data": {"token": empty}}) + with pytest.raises(RuntimeError, match="value empty"): + vaultwarden._k8s_get_secret_value("apps", "secret", "token") + + +def test_admin_session_cache_and_rate_limit(monkeypatch) -> None: + _reset_admin_state(monkeypatch) + monkeypatch.setattr(vaultwarden, "_k8s_get_secret_value", lambda *args: "admin-token") + monkeypatch.setattr(vaultwarden.httpx, "Client", DummyClient) + monkeypatch.setattr(vaultwarden.time, "time", lambda: 100.0) + DummyClient.reset(DummyResponse({}, status_code=200)) + + session = vaultwarden._admin_session("http://vaultwarden") + assert vaultwarden._admin_session("http://vaultwarden") is session + + DummyClient.reset(DummyResponse({}, status_code=200)) + vaultwarden._admin_session("http://other") + assert DummyClient.closed == 1 + + class BadCloseSession: + def close(self) -> None: + raise RuntimeError("ignore") + + monkeypatch.setattr(vaultwarden, "_ADMIN_SESSION", BadCloseSession()) + monkeypatch.setattr(vaultwarden, "_ADMIN_SESSION_EXPIRES_AT", 999.0) + monkeypatch.setattr(vaultwarden, "_ADMIN_SESSION_BASE_URL", "http://old") + DummyClient.reset(DummyResponse({}, status_code=200)) + vaultwarden._admin_session("http://new") + + _reset_admin_state(monkeypatch) + DummyClient.reset(DummyResponse({}, status_code=429)) + with pytest.raises(RuntimeError, match="rate limited"): + vaultwarden._admin_session("http://vaultwarden") + with pytest.raises(RuntimeError, match="rate limited"): + vaultwarden._admin_session("http://vaultwarden") + + +def test_invite_user_success_and_idempotent_paths(monkeypatch) -> None: + _reset_admin_state(monkeypatch) + monkeypatch.setattr(vaultwarden.settings, "VAULTWARDEN_SERVICE_HOST", "vaultwarden.apps.svc:80") + monkeypatch.setattr(vaultwarden.settings, "VAULTWARDEN_NAMESPACE", "apps") + monkeypatch.setattr(vaultwarden.settings, "VAULTWARDEN_POD_LABEL", "app=vaultwarden") + monkeypatch.setattr(vaultwarden.settings, "VAULTWARDEN_POD_PORT", 8080) + monkeypatch.setattr(vaultwarden, "_k8s_find_pod_ip", lambda *args: "10.0.0.2") + + class InviteSession: + def __init__(self, response): + self.response = response + + def post(self, path, json=None): + return self.response + + monkeypatch.setattr(vaultwarden, "_admin_session", lambda base_url: InviteSession(DummyResponse(status_code=200))) + assert vaultwarden.invite_user("alice@example.dev").status == "invited" + + monkeypatch.setattr( + vaultwarden, + "_admin_session", + lambda base_url: InviteSession(DummyResponse(status_code=409, text="user already exists")), + ) + result = vaultwarden.invite_user("alice@example.dev") + assert result.ok and result.status == "already_present" + + monkeypatch.setattr(vaultwarden, "_ADMIN_RATE_LIMITED_UNTIL", 9999999999.0) + assert vaultwarden.invite_user("alice@example.dev").status == "rate_limited" + assert vaultwarden.invite_user("not-email").status == "invalid_email" + + monkeypatch.setattr(vaultwarden, "_ADMIN_RATE_LIMITED_UNTIL", 0.0) + monkeypatch.setattr(vaultwarden, "_admin_session", lambda base_url: InviteSession(DummyResponse(status_code=429))) + assert vaultwarden.invite_user("alice@example.dev").status == "rate_limited" + + +def test_invite_user_fallback_and_error_paths(monkeypatch) -> None: + _reset_admin_state(monkeypatch) + monkeypatch.setattr(vaultwarden.settings, "VAULTWARDEN_SERVICE_HOST", "vaultwarden.apps.svc:80") + monkeypatch.setattr(vaultwarden.settings, "VAULTWARDEN_NAMESPACE", "apps") + monkeypatch.setattr(vaultwarden.settings, "VAULTWARDEN_POD_LABEL", "app=vaultwarden") + monkeypatch.setattr(vaultwarden.settings, "VAULTWARDEN_POD_PORT", 8080) + monkeypatch.setattr(vaultwarden, "_k8s_find_pod_ip", lambda *args: "10.0.0.2") + + class SequenceSession: + def __init__(self): + self.calls = 0 + + def post(self, path, json=None): + self.calls += 1 + if self.calls == 1: + raise RuntimeError("service offline") + return DummyResponse(status_code=201) + + session = SequenceSession() + monkeypatch.setattr(vaultwarden, "_admin_session", lambda base_url: session) + assert vaultwarden.invite_user("alice@example.dev").status == "invited" + + class BadTextResponse: + status_code = 500 + + @property + def text(self): + raise RuntimeError("no body") + + class BadTextSession: + def post(self, path, json=None): + return BadTextResponse() + + monkeypatch.setattr(vaultwarden, "_admin_session", lambda base_url: BadTextSession()) + result = vaultwarden.invite_user("alice@example.dev") + assert result.status == "error" + assert "status 500" in result.detail + + monkeypatch.setattr(vaultwarden, "_k8s_find_pod_ip", lambda *args: (_ for _ in ()).throw(RuntimeError("no pod"))) + monkeypatch.setattr(vaultwarden, "_admin_session", lambda base_url: (_ for _ in ()).throw(RuntimeError("rate limited"))) + assert vaultwarden.invite_user("alice@example.dev").status == "rate_limited" + + monkeypatch.setattr(vaultwarden, "_admin_session", lambda base_url: (_ for _ in ()).throw(RuntimeError("offline"))) + result = vaultwarden.invite_user("alice@example.dev") + assert not result.ok + assert result.status == "error"