zwitschi
ef22e217c7
CI / lint-test-build (push) Failing after 11s
- Added PG_PASSWORD to .env.example for database connection. - Removed unnecessary imports and streamlined code in various modules. - Enhanced error handling in ConfigSettingRepository and ConfigPairingRepository. - Updated test files to remove unused imports and improve clarity.
52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from arbitrade.config.settings import get_settings
|
|
from arbitrade.storage.pg_store import PgStore
|
|
from arbitrade.storage.repositories import AuditRecord, AuditRepository
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _pg() -> AsyncIterator[PgStore]:
|
|
s = get_settings()
|
|
store = PgStore(s)
|
|
try:
|
|
await store.start()
|
|
await store.migrate()
|
|
yield store
|
|
finally:
|
|
await store.stop()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_repository_inserts_and_lists_recent() -> None:
|
|
async with _pg() as store:
|
|
repository = AuditRepository(store)
|
|
|
|
await repository.insert(
|
|
AuditRecord(
|
|
occurred_at=datetime.now(UTC),
|
|
actor="dashboard_user",
|
|
event_type="dashboard.control.start",
|
|
decision="approved",
|
|
payload={"execution_status": "running"},
|
|
correlation_id="req-1",
|
|
)
|
|
)
|
|
|
|
recent = await repository.list_recent(limit=5)
|
|
|
|
assert len(recent) == 1
|
|
assert recent[0].actor == "dashboard_user"
|
|
assert recent[0].event_type == "dashboard.control.start"
|
|
assert recent[0].decision == "approved"
|
|
assert recent[0].payload == {"execution_status": "running"}
|
|
assert recent[0].correlation_id == "req-1"
|