102 lines
4.0 KiB
Python
102 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from govoplan_access.backend.db.models import Account, User
|
|
from govoplan_core.auth import ApiPrincipal, get_api_principal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_identity.backend.api.v1.routes import router as identity_router
|
|
from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
|
|
from tests.db_isolation import temporary_database
|
|
|
|
|
|
class IdentityApiTests(unittest.TestCase):
|
|
def _principal(self, scopes: set[str]) -> ApiPrincipal:
|
|
account = Account(
|
|
id="account-identity-api",
|
|
email="identity-admin@example.test",
|
|
normalized_email="identity-admin@example.test",
|
|
display_name="Identity Admin",
|
|
)
|
|
user = User(
|
|
id="user-identity-api",
|
|
tenant_id="tenant-identity-api",
|
|
account_id=account.id,
|
|
email=account.email,
|
|
display_name=account.display_name,
|
|
settings={},
|
|
mail_profile_policy={},
|
|
)
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id=account.id,
|
|
membership_id=user.id,
|
|
tenant_id=user.tenant_id,
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=account,
|
|
user=user,
|
|
)
|
|
|
|
def _app(self, scopes: set[str]) -> FastAPI:
|
|
app = FastAPI()
|
|
app.include_router(identity_router, prefix="/api/v1")
|
|
app.dependency_overrides[get_api_principal] = lambda: self._principal(scopes)
|
|
return app
|
|
|
|
def test_identity_search_returns_account_links(self) -> None:
|
|
root = Path(tempfile.mkdtemp(prefix="govoplan-identity-api-"))
|
|
try:
|
|
with temporary_database(f"sqlite:///{root / 'identity.db'}") as database:
|
|
Base.metadata.create_all(bind=database.engine)
|
|
with database.session() as session:
|
|
identity = Identity(
|
|
id="identity-alice",
|
|
display_name="Alice Registry",
|
|
external_subject="alice@example.test",
|
|
source="local",
|
|
)
|
|
session.add(identity)
|
|
session.add(
|
|
IdentityAccountLink(
|
|
identity_id=identity.id,
|
|
account_id="account-alice-primary",
|
|
is_primary=True,
|
|
source="local",
|
|
)
|
|
)
|
|
session.add(
|
|
IdentityAccountLink(
|
|
identity_id=identity.id,
|
|
account_id="account-alice-secondary",
|
|
is_primary=False,
|
|
source="local",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
app = self._app({"identity:identity:read"})
|
|
with TestClient(app) as client:
|
|
response = client.get("/api/v1/identity/identities", params={"query": "registry"})
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
payload = response.json()
|
|
self.assertEqual(1, len(payload["identities"]))
|
|
item = payload["identities"][0]
|
|
self.assertEqual("identity-alice", item["id"])
|
|
self.assertEqual("Alice Registry", item["display_name"])
|
|
self.assertEqual("account-alice-primary", item["primary_account_id"])
|
|
self.assertEqual(["account-alice-primary", "account-alice-secondary"], item["account_ids"])
|
|
finally:
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|