mirror of
https://github.com/agentrhq/authsome.git
synced 2026-09-19 01:34:19 +08:00
Manoj Bajaj
57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from authsome.cli.identity import RuntimeIdentity
|
|
from authsome.server.identity_bootstrap import IdentityBootstrapService
|
|
from authsome.server.store import create_server_store
|
|
from authsome.server.ui_sessions import UiSessionStore
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bootstrap_requires_claim_until_identity_is_claimed(tmp_path: Path) -> None:
|
|
identity = RuntimeIdentity.create(tmp_path, "steady-wisely-boldly-0042")
|
|
store = await create_server_store(home=tmp_path)
|
|
ui_sessions = UiSessionStore("test-secret")
|
|
service = IdentityBootstrapService(
|
|
registry=store.identity_registry,
|
|
claims=store.identity_claims,
|
|
ui_sessions=ui_sessions,
|
|
server_base_url="http://127.0.0.1:7998",
|
|
)
|
|
|
|
try:
|
|
status = await service.register_identity(handle=identity.handle, did=identity.did)
|
|
|
|
assert status.registration_status == "claim_required"
|
|
assert status.claim_url.startswith("http://127.0.0.1:7998/claim?token=")
|
|
finally:
|
|
await store.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bootstrap_returns_claimed_status_after_claim(tmp_path: Path) -> None:
|
|
identity = RuntimeIdentity.create(tmp_path, "steady-wisely-boldly-0042")
|
|
store = await create_server_store(home=tmp_path)
|
|
ui_sessions = UiSessionStore("test-secret")
|
|
service = IdentityBootstrapService(
|
|
registry=store.identity_registry,
|
|
claims=store.identity_claims,
|
|
ui_sessions=ui_sessions,
|
|
server_base_url="http://127.0.0.1:7998",
|
|
)
|
|
|
|
try:
|
|
await store.identity_registry.register(handle=identity.handle, did=identity.did)
|
|
principal = await store.principals.create_by_email("dev@example.com")
|
|
await store.identity_claims.claim_identity(identity.handle, principal.principal_id)
|
|
await store.identity_claims.accept_claim(identity.handle)
|
|
|
|
status = await service.get_identity_status(handle=identity.handle)
|
|
|
|
assert status is not None
|
|
assert status.registration_status == "claimed"
|
|
assert status.principal_id == principal.principal_id
|
|
finally:
|
|
await store.close()
|