96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import delete, select
|
|
|
|
from app.infrastructure.db import SessionFactory
|
|
from app.model.audit import InteractionAudit
|
|
from app.model.configuration import ConfigRelease
|
|
from app.model.platform import DomainEventOutbox
|
|
from app.service.config_release_service import ConfigReleaseError, ConfigReleaseService
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
async def test_config_release_lifecycle_is_versioned_and_audited() -> None:
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
release_no = f"it-{uuid4().hex}"
|
|
release_id = 0
|
|
rollback_id = 0
|
|
try:
|
|
async with SessionFactory() as session:
|
|
release = ConfigRelease(
|
|
release_no=release_no,
|
|
title="integration",
|
|
change_summary="integration",
|
|
status="draft",
|
|
created_by=101,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(release)
|
|
await session.flush()
|
|
release_id = release.id
|
|
service = ConfigReleaseService(session)
|
|
await service.submit_for_review(release_id, 101)
|
|
with pytest.raises(ConfigReleaseError, match="creator cannot"):
|
|
await service.approve(release_id, 101)
|
|
await service.approve(release_id, 202)
|
|
await service.activate(release_id, 202)
|
|
await session.commit()
|
|
|
|
async with SessionFactory() as session:
|
|
service = ConfigReleaseService(session)
|
|
rollback = await service.rollback(release_id, 202)
|
|
rollback_id = rollback.id
|
|
await session.commit()
|
|
|
|
async with SessionFactory() as session:
|
|
release = await session.get(ConfigRelease, release_id)
|
|
rollback = await session.get(ConfigRelease, rollback_id)
|
|
events = list(
|
|
await session.scalars(
|
|
select(DomainEventOutbox).where(
|
|
DomainEventOutbox.event_type == "config.cache_invalidate_requested",
|
|
DomainEventOutbox.aggregate_id.in_([str(release_id), str(rollback_id)]),
|
|
)
|
|
)
|
|
)
|
|
audits = list(
|
|
await session.scalars(
|
|
select(InteractionAudit).where(
|
|
InteractionAudit.action_type.in_(
|
|
[
|
|
"config.submit_review",
|
|
"config.approve",
|
|
"config.activate",
|
|
"config.rollback",
|
|
]
|
|
),
|
|
InteractionAudit.detail["release_id"].in_([release_id, rollback_id]),
|
|
)
|
|
)
|
|
)
|
|
assert release is not None and release.status == "superseded"
|
|
assert rollback is not None and rollback.status == "active"
|
|
assert rollback.rollback_of_release_id == release_id
|
|
assert len(events) == 2
|
|
assert len(audits) == 4
|
|
finally:
|
|
async with SessionFactory() as session:
|
|
await session.execute(
|
|
delete(InteractionAudit).where(
|
|
InteractionAudit.detail["release_id"].in_([release_id, rollback_id])
|
|
)
|
|
)
|
|
await session.execute(
|
|
delete(DomainEventOutbox).where(
|
|
DomainEventOutbox.aggregate_id.in_([str(release_id), str(rollback_id)])
|
|
)
|
|
)
|
|
await session.execute(
|
|
delete(ConfigRelease).where(ConfigRelease.release_no.like("it-%"))
|
|
)
|
|
await session.commit()
|