43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
from types import SimpleNamespace
|
|
from typing import cast
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from ai_platform.platform.application.app_credentials import (
|
|
AppCredentialConflict,
|
|
AppCredentialNotFound,
|
|
AppCredentialRepository,
|
|
AppCredentialUseCases,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("operation", "error"),
|
|
[
|
|
("create", AppCredentialConflict()),
|
|
("rotate", AppCredentialNotFound()),
|
|
("revoke", RuntimeError("storage")),
|
|
],
|
|
)
|
|
async def test_repository_error_rolls_back(operation, error):
|
|
repo = cast(
|
|
AppCredentialRepository, SimpleNamespace(**{operation: AsyncMock(side_effect=error)})
|
|
)
|
|
commit, rollback = AsyncMock(), AsyncMock()
|
|
cases = AppCredentialUseCases(repo, commit, rollback, lambda: "key")
|
|
with pytest.raises(type(error)):
|
|
await getattr(cases, operation)("omni-local")
|
|
commit.assert_not_awaited()
|
|
rollback.assert_awaited_once()
|
|
|
|
|
|
async def test_failed_commit_rolls_back():
|
|
repo = cast(AppCredentialRepository, SimpleNamespace(rotate=AsyncMock()))
|
|
commit, rollback = AsyncMock(side_effect=RuntimeError("commit failed")), AsyncMock()
|
|
cases = AppCredentialUseCases(repo, commit, rollback, lambda: "key")
|
|
with pytest.raises(RuntimeError, match="commit failed"):
|
|
await cases.rotate("omni-local")
|
|
commit.assert_awaited_once()
|
|
rollback.assert_awaited_once()
|