Files
TJWaterServerBinary/tests/api/test_agent_auth_endpoints.py
T
jiang 76cf6c32bc
Server CI/CD / docker-image (push) Successful in 16s
Server CI/CD / deploy-fallback-log (push) Has been skipped
fix(agent): expose network context
2026-07-08 18:41:56 +08:00

83 lines
2.8 KiB
Python

from types import SimpleNamespace
from uuid import uuid4
from fastapi import HTTPException, status
from fastapi.testclient import TestClient
from app.api.v1.endpoints import agent_auth as agent_auth_endpoint
from app.auth.keycloak_dependencies import get_current_keycloak_payload
from app.auth.metadata_dependencies import get_current_metadata_user
from app.auth.project_dependencies import ProjectContext, get_project_context
from tests.conftest import build_test_app
def _build_client(*, project_context=None, current_user=None) -> TestClient:
app = build_test_app(agent_auth_endpoint.router, "/api/v1")
if project_context is not None:
app.dependency_overrides[get_project_context] = lambda: project_context
if current_user is not None:
app.dependency_overrides[get_current_metadata_user] = lambda: current_user
app.dependency_overrides[get_current_keycloak_payload] = lambda: {"exp": 1781183400}
return TestClient(app)
def test_agent_auth_context_returns_metadata_user_and_project_context():
user_id = uuid4()
keycloak_sub = uuid4()
project_id = uuid4()
client = _build_client(
project_context=ProjectContext(
project_id=project_id,
project_code="fengyang",
user_id=user_id,
project_role="editor",
),
current_user=SimpleNamespace(
id=user_id,
keycloak_id=keycloak_sub,
username="alice",
role="user",
is_superuser=False,
),
)
response = client.get("/api/v1/agent/auth/context")
assert response.status_code == 200
assert response.json() == {
"user_id": str(user_id),
"keycloak_sub": str(keycloak_sub),
"username": "alice",
"role": "user",
"is_superuser": False,
"project_id": str(project_id),
"network": "fengyang",
"project_role": "editor",
"token_expires_at": "2026-06-11T13:10:00+00:00",
}
def test_agent_auth_context_propagates_project_auth_failures():
def reject_project():
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="No access to project",
)
app = build_test_app(agent_auth_endpoint.router, "/api/v1")
app.dependency_overrides[get_project_context] = reject_project
app.dependency_overrides[get_current_metadata_user] = lambda: SimpleNamespace(
id=uuid4(),
keycloak_id=uuid4(),
username="alice",
role="user",
is_superuser=False,
)
app.dependency_overrides[get_current_keycloak_payload] = lambda: {"exp": 1781183400}
client = TestClient(app)
response = client.get("/api/v1/agent/auth/context")
assert response.status_code == 403
assert response.json()["detail"] == "No access to project"