feat(tests): add authentication and proxy tests with fixtures

2026-04-12 20:39:32 +02:00
parent c3baca215b
commit ccacab9e5b
6 changed files with 385 additions and 3 deletions
@@ -0,0 +1,112 @@
+import importlib
+from datetime import UTC, datetime, timedelta
+
+from core.security import AuthContext, get_auth_context
+from models.auth import AccessTokenPayload, AuthUserDocument
+
+auth_router_module = importlib.import_module("api.auth.router")
+
+
+def test_login_success_returns_bearer_token(client, monkeypatch: object):
+    user = AuthUserDocument(username="admin", password_hash="stored-hash")
+
+    async def fake_get_user_by_username(username: str) -> AuthUserDocument | None:
+        assert username == "admin"
+        return user
+
+    def fake_verify_password(plain_password: str, password_hash: str) -> bool:
+        return plain_password == "secret" and password_hash == "stored-hash"
+
+    def fake_create_access_token(*, subject: str) -> str:
+        assert subject == "admin"
+        return "token-123"
+
+    monkeypatch.setattr(
+        auth_router_module,
+        "get_user_by_username",
+        fake_get_user_by_username,
+    )
+    monkeypatch.setattr(auth_router_module, "verify_password", fake_verify_password)
+    monkeypatch.setattr(
+        auth_router_module,
+        "create_access_token",
+        fake_create_access_token,
+    )
+
+    response = client.post(
+        "/auth/login",
+        json={"username": "admin", "password": "secret"},
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {"access_token": "token-123", "token_type": "bearer"}
+
+
+def test_login_invalid_credentials_returns_401(client, monkeypatch: object):
+    async def fake_get_user_by_username(_: str) -> None:
+        return None
+
+    monkeypatch.setattr(
+        auth_router_module,
+        "get_user_by_username",
+        fake_get_user_by_username,
+    )
+
+    response = client.post(
+        "/auth/login",
+        json={"username": "admin", "password": "wrong"},
+    )
+
+    assert response.status_code == 401
+    assert response.json()["detail"] == "Invalid username or password"
+    assert response.headers["www-authenticate"] == "Bearer"
+
+
+def test_logout_revokes_active_token(client, monkeypatch: object):
+    captured: dict[str, object] = {}
+
+    async def fake_revoke_token(*, jti: str, expires_at: datetime) -> None:
+        captured["jti"] = jti
+        captured["expires_at"] = expires_at
+
+    monkeypatch.setattr(auth_router_module, "revoke_token", fake_revoke_token)
+
+    response = client.post("/auth/logout")
+
+    assert response.status_code == 204
+    assert captured["jti"] == "test-jti"
+    assert isinstance(captured["expires_at"], datetime)
+
+
+def test_logout_skips_revocation_when_token_is_expired(
+    client,
+    fake_user: AuthUserDocument,
+    monkeypatch: object,
+):
+    state = {"called": False}
+
+    async def fake_revoke_token(*, jti: str, expires_at: datetime) -> None:
+        state["called"] = True
+
+    monkeypatch.setattr(auth_router_module, "revoke_token", fake_revoke_token)
+
+    expired_payload = AccessTokenPayload(
+        sub=fake_user.username,
+        jti="expired-jti",
+        exp=int((datetime.now(UTC) - timedelta(minutes=1)).timestamp()),
+    )
+    expired_context = AuthContext(
+        token="expired-token",
+        user=fake_user,
+        payload=expired_payload,
+    )
+
+    async def override_auth_context() -> AuthContext:
+        return expired_context
+
+    client.app.dependency_overrides[get_auth_context] = override_auth_context
+
+    response = client.post("/auth/logout")
+
+    assert response.status_code == 204
+    assert state["called"] is False
@@ -0,0 +1,147 @@
+import importlib
+
+from core.security import get_auth_context, require_authenticated_user
+from models.proxy import ProxyDocument
+from models.service import ServiceEndpointDocument
+
+proxy_router_module = importlib.import_module("api.proxy.router")
+
+
+def make_proxy_document(
+    *,
+    proxy_port: int,
+    name: str = "svc",
+    code: str = "proxy-code",
+    address: str = "127.0.0.1",
+    port: int = 9000,
+) -> ProxyDocument:
+    return ProxyDocument(
+        proxy_port=proxy_port,
+        name=name,
+        code=code,
+        service=ServiceEndpointDocument(address=address, port=port),
+    )
+
+
+def test_proxy_requires_auth_without_dependency_overrides(client):
+    client.app.dependency_overrides.pop(require_authenticated_user, None)
+    client.app.dependency_overrides.pop(get_auth_context, None)
+
+    response = client.get("/proxy")
+
+    assert response.status_code == 401
+    assert response.json()["detail"] == "Could not validate credentials"
+
+
+def test_list_proxies_returns_documents(client, monkeypatch: object):
+    docs = [
+        make_proxy_document(proxy_port=8080, name="api"),
+        make_proxy_document(proxy_port=8081, name="worker"),
+    ]
+
+    async def fake_list_proxies() -> list[ProxyDocument]:
+        return docs
+
+    monkeypatch.setattr(proxy_router_module, "list_proxy_documents", fake_list_proxies)
+
+    response = client.get("/proxy")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert [item["proxy_port"] for item in body] == [8080, 8081]
+
+
+def test_get_proxy_returns_404_when_not_found(client, monkeypatch: object):
+    async def fake_get_proxy_by_port(proxy_port: int) -> None:
+        return None
+
+    monkeypatch.setattr(
+        proxy_router_module,
+        "get_proxy_by_port",
+        fake_get_proxy_by_port,
+    )
+
+    response = client.get("/proxy/9999")
+
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Proxy not found"
+
+
+def test_create_proxy_returns_409_on_port_conflict(client, monkeypatch: object):
+    existing = make_proxy_document(proxy_port=3671)
+
+    async def fake_get_proxy_by_port(proxy_port: int) -> ProxyDocument | None:
+        assert proxy_port == 3671
+        return existing
+
+    monkeypatch.setattr(
+        proxy_router_module,
+        "get_proxy_by_port",
+        fake_get_proxy_by_port,
+    )
+
+    response = client.post(
+        "/proxy",
+        json={
+            "proxy_port": 3671,
+            "name": "new",
+            "code": "abc",
+            "service": {"address": "localhost", "port": 8080},
+        },
+    )
+
+    assert response.status_code == 409
+    assert response.json()["detail"] == "Proxy already exists on port 3671"
+
+
+def test_create_proxy_returns_created_document(client, monkeypatch: object):
+    captured: dict[str, object] = {}
+
+    async def fake_get_proxy_by_port(proxy_port: int) -> ProxyDocument | None:
+        assert proxy_port == 4040
+        return None
+
+    async def fake_create_proxy(
+        *,
+        proxy_port: int,
+        name: str,
+        service: dict[str, object],
+        code: str,
+    ) -> ProxyDocument:
+        captured["proxy_port"] = proxy_port
+        captured["name"] = name
+        captured["service"] = service
+        captured["code"] = code
+        return make_proxy_document(
+            proxy_port=proxy_port,
+            name=name,
+            code=code,
+            address=str(service["address"]),
+            port=int(service["port"]),
+        )
+
+    monkeypatch.setattr(
+        proxy_router_module,
+        "get_proxy_by_port",
+        fake_get_proxy_by_port,
+    )
+    monkeypatch.setattr(proxy_router_module, "create_proxy", fake_create_proxy)
+
+    response = client.post(
+        "/proxy",
+        json={
+            "proxy_port": 4040,
+            "name": "edge",
+            "code": "edge-code",
+            "service": {"address": "service.local", "port": 8181},
+        },
+    )
+
+    assert response.status_code == 201
+    assert response.json()["proxy_port"] == 4040
+    assert captured == {
+        "proxy_port": 4040,
+        "name": "edge",
+        "service": {"address": "service.local", "port": 8181},
+        "code": "edge-code",
+    }