165 lines
5.2 KiB
Python
165 lines
5.2 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from ai_platform.platform.application.tenant_key_repository import (
|
|
TenantKeyConflict,
|
|
TenantKeyNotFound,
|
|
TenantKeyRecord,
|
|
)
|
|
from ai_platform.platform.application.tenant_keys import (
|
|
create_tenant_key,
|
|
revoke_tenant_key,
|
|
rotate_tenant_key,
|
|
)
|
|
from ai_platform.platform.security import digest
|
|
|
|
|
|
class FakeTenantKeyRepository:
|
|
def __init__(self, *, operation_error: Exception | None = None) -> None:
|
|
self.operation_error = operation_error
|
|
self.mutations: list[tuple[str, str, str | None]] = []
|
|
self.record = TenantKeyRecord(
|
|
tenant_id="tenant-1",
|
|
active=True,
|
|
created_at=datetime(2025, 1, 1, tzinfo=UTC),
|
|
updated_at=datetime(2025, 1, 1, tzinfo=UTC),
|
|
)
|
|
|
|
async def list(self) -> list[TenantKeyRecord]:
|
|
return [self.record]
|
|
|
|
async def create(self, tenant_id: str, verifier: str, actor: str) -> None:
|
|
self.mutations.append(("create", tenant_id, verifier))
|
|
if self.operation_error:
|
|
raise self.operation_error
|
|
assert actor == "admin@example.invalid"
|
|
|
|
async def rotate(self, tenant_id: str, verifier: str, actor: str) -> TenantKeyRecord:
|
|
self.mutations.append(("rotate", tenant_id, verifier))
|
|
if self.operation_error:
|
|
raise self.operation_error
|
|
assert actor == "admin@example.invalid"
|
|
return self.record
|
|
|
|
async def revoke(self, tenant_id: str, actor: str) -> TenantKeyRecord:
|
|
self.mutations.append(("revoke", tenant_id, None))
|
|
if self.operation_error:
|
|
raise self.operation_error
|
|
assert actor == "admin@example.invalid"
|
|
return self.record
|
|
|
|
|
|
class FakeTenantKeyGenerator:
|
|
def __init__(self, *values: str) -> None:
|
|
self._values = iter(values)
|
|
|
|
def __call__(self) -> str:
|
|
return next(self._values)
|
|
|
|
|
|
class FakeTenantKeyTransaction:
|
|
def __init__(self, *, commit_error: Exception | None = None) -> None:
|
|
self.commit_error = commit_error
|
|
self.commits = 0
|
|
self.rollbacks = 0
|
|
|
|
async def commit(self) -> None:
|
|
self.commits += 1
|
|
if self.commit_error:
|
|
raise self.commit_error
|
|
|
|
async def rollback(self) -> None:
|
|
self.rollbacks += 1
|
|
|
|
|
|
async def test_tenant_key_use_cases_commit_each_successful_mutation():
|
|
repository = FakeTenantKeyRepository()
|
|
transaction = FakeTenantKeyTransaction()
|
|
|
|
created_key = await create_tenant_key(
|
|
repository,
|
|
"tenant-1",
|
|
"admin@example.invalid",
|
|
key_generator=FakeTenantKeyGenerator("tnt-created"),
|
|
commit=transaction.commit,
|
|
rollback=transaction.rollback,
|
|
)
|
|
rotated_key = await rotate_tenant_key(
|
|
repository,
|
|
"tenant-1",
|
|
"admin@example.invalid",
|
|
key_generator=FakeTenantKeyGenerator("tnt-rotated"),
|
|
commit=transaction.commit,
|
|
rollback=transaction.rollback,
|
|
)
|
|
revoked_record = await revoke_tenant_key(
|
|
repository,
|
|
"tenant-1",
|
|
"admin@example.invalid",
|
|
commit=transaction.commit,
|
|
rollback=transaction.rollback,
|
|
)
|
|
|
|
assert created_key == "tnt-created"
|
|
assert rotated_key == "tnt-rotated"
|
|
assert digest(created_key) == repository.mutations[0][2]
|
|
assert digest(rotated_key) == repository.mutations[1][2]
|
|
assert [mutation[0] for mutation in repository.mutations] == ["create", "rotate", "revoke"]
|
|
assert transaction.commits == 3
|
|
assert transaction.rollbacks == 0
|
|
assert revoked_record is repository.record
|
|
|
|
|
|
async def test_tenant_key_use_case_rolls_back_repository_errors():
|
|
repository = FakeTenantKeyRepository(operation_error=TenantKeyConflict())
|
|
transaction = FakeTenantKeyTransaction()
|
|
|
|
with pytest.raises(TenantKeyConflict):
|
|
await create_tenant_key(
|
|
repository,
|
|
"tenant-1",
|
|
"admin@example.invalid",
|
|
key_generator=FakeTenantKeyGenerator("tnt-unused"),
|
|
commit=transaction.commit,
|
|
rollback=transaction.rollback,
|
|
)
|
|
|
|
assert transaction.commits == 0
|
|
assert transaction.rollbacks == 1
|
|
|
|
|
|
async def test_tenant_key_use_case_rolls_back_failed_commit():
|
|
repository = FakeTenantKeyRepository()
|
|
transaction = FakeTenantKeyTransaction(commit_error=RuntimeError("commit failed"))
|
|
|
|
with pytest.raises(RuntimeError, match="commit failed"):
|
|
await rotate_tenant_key(
|
|
repository,
|
|
"tenant-1",
|
|
"admin@example.invalid",
|
|
key_generator=FakeTenantKeyGenerator("tnt-unused"),
|
|
commit=transaction.commit,
|
|
rollback=transaction.rollback,
|
|
)
|
|
|
|
assert transaction.commits == 1
|
|
assert transaction.rollbacks == 1
|
|
|
|
|
|
async def test_tenant_key_use_case_rolls_back_missing_key():
|
|
repository = FakeTenantKeyRepository(operation_error=TenantKeyNotFound())
|
|
transaction = FakeTenantKeyTransaction()
|
|
|
|
with pytest.raises(TenantKeyNotFound):
|
|
await revoke_tenant_key(
|
|
repository,
|
|
"tenant-1",
|
|
"admin@example.invalid",
|
|
commit=transaction.commit,
|
|
rollback=transaction.rollback,
|
|
)
|
|
|
|
assert transaction.commits == 0
|
|
assert transaction.rollbacks == 1
|