agent-platform/tests/test_auth_identity_service.py

from pathlib import Path

from tests.conftest import build_fastapi_test_client, prepare_known_service_import


def test_identity_bootstraps_demo_user_for_frontend_login(
    tmp_path: Path,
    monkeypatch,
) -> None:
    prepare_known_service_import("auth-service")

    from app.bootstrap.app import create_app
    from app.db.models import Base
    from sqlalchemy import create_engine

    database_url = f"sqlite:///{tmp_path / 'demo-auth.db'}"
    monkeypatch.setenv("AGENT_PLATFORM_DATABASE_URL", database_url)
    engine = create_engine(database_url, connect_args={"check_same_thread": False})
    Base.metadata.create_all(engine)

    app = create_app()
    client = build_fastapi_test_client(app)

    response = client.post(
        "/identity/auth/login",
        json={"username": "demo-user", "password": "demo-password"},
    )

    assert response.status_code == 200
    payload = response.json()
    assert payload["data"]["accessToken"]
    assert payload["data"]["user"]["username"] == "demo-user"


def test_identity_post_contract_supports_login_permissions_and_api_keys(
    tmp_path: Path,
    monkeypatch,
) -> None:
    prepare_known_service_import("auth-service")

    from app.bootstrap.app import create_app
    from app.db.models import Base
    from app.domain.repositories import RoleAssignmentRepository, RoleRepository, UserRepository
    from app.infrastructure.passwords import hash_password
    from core_db import create_session_factory
    from sqlalchemy import create_engine

    database_url = f"sqlite:///{tmp_path / 'auth.db'}"
    monkeypatch.setenv("AGENT_PLATFORM_DATABASE_URL", database_url)
    engine = create_engine(database_url, connect_args={"check_same_thread": False})
    Base.metadata.create_all(engine)

    app = create_app()
    app.state.session_factory = create_session_factory(engine)
    client = build_fastapi_test_client(app)

    session_factory = create_session_factory(engine)
    with session_factory() as db:
        user = UserRepository(db).create(
            username="demo",
            password_hash=hash_password("super-secret"),
            display_name="Demo User",
            email="demo@example.com",
            metadata_json={"source": "test"},
        )
        role = RoleRepository(db).create(
            code="admin",
            name="Admin",
            description="Platform admin",
            permissions_json=[],
        )
        RoleAssignmentRepository(db).create(
            user_id=user.id,
            role_id=role.id,
            scope_type=None,
            scope_id=None,
            expires_time=None,
        )
        user_id = user.id
        role_id = role.id

    binding_response = client.post(
        "/identity/rolePermissionBindings/add",
        json={
            "roleId": role_id,
            "permission": "agents:*",
        },
    )

    assert binding_response.status_code == 200
    binding_payload = binding_response.json()
    assert binding_payload["success"] is True
    assert binding_payload["data"]["roleId"] == role_id

    login_response = client.post(
        "/identity/auth/login",
        json={"username": "demo", "password": "super-secret"},
    )

    assert login_response.status_code == 200
    login_payload = login_response.json()
    assert login_payload["data"]["accessToken"]
    assert login_payload["data"]["user"]["displayName"] == "Demo User"
    assert "createdTime" in login_payload["data"]["user"]

    token = login_payload["data"]["accessToken"]
    me_response = client.post(
        "/identity/auth/me",
        headers={"Authorization": f"Bearer {token}"},
        json={},
    )

    assert me_response.status_code == 200
    me_payload = me_response.json()
    assert me_payload["data"]["user"]["id"] == user_id
    assert me_payload["data"]["permissions"] == ["agents:*"]

    permission_response = client.post(
        "/identity/permissions/check",
        json={
            "userId": user_id,
            "permission": "agents:create",
        },
    )

    assert permission_response.status_code == 200
    assert permission_response.json()["data"]["allowed"] is True

    api_key_response = client.post(
        "/identity/apiKeys/create",
        json={"name": "Gateway Test Key", "scopes": "agents:read"},
    )

    assert api_key_response.status_code == 200
    api_key_payload = api_key_response.json()["data"]
    assert api_key_payload["secret"].startswith("agp_")
    assert api_key_payload["apiKey"]["keyPrefix"]

    api_keys_response = client.post("/identity/apiKeys/list", json={"page": 1, "pageSize": 20})

    assert api_keys_response.status_code == 200
    assert api_keys_response.json()["data"]["total"] == 1