135 lines
5.2 KiB
Python
135 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import sys
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
import unittest
|
|
from uuid import UUID
|
|
|
|
BACKEND_ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(BACKEND_ROOT))
|
|
|
|
|
|
@contextmanager
|
|
def repository_dsn() -> Iterator[str]:
|
|
configured = os.environ.get("PIPELINE_TEST_DATABASE_DSN")
|
|
if configured:
|
|
yield configured
|
|
return
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
yield f"sqlite:///{Path(directory) / 'pipeline-audit.db'}"
|
|
|
|
|
|
class ScriptConfigAuditTest(unittest.TestCase):
|
|
def test_rollback_writes_audit_event(self) -> None:
|
|
with repository_dsn() as dsn:
|
|
from src.application.auth import get_current_user_by_email
|
|
from src.application.seed_data import seed_reference_data
|
|
from src.application.site_config import (
|
|
create_script_config_version,
|
|
create_target_site,
|
|
rollback_script_config_version,
|
|
)
|
|
from src.domain.contracts import (
|
|
PublishingRules,
|
|
ScriptConfigVersionCreateRequest,
|
|
TargetSiteConfigCreateRequest,
|
|
)
|
|
from src.infrastructure.repositories import open_backend_repository
|
|
|
|
repository = open_backend_repository(dsn)
|
|
repository.setup()
|
|
seed_reference_data(repository)
|
|
admin = get_current_user_by_email(repository, "admin@example.com")
|
|
site = create_target_site(
|
|
repository,
|
|
TargetSiteConfigCreateRequest(
|
|
name="Audit Test Site",
|
|
slug="audit-site",
|
|
publishing_type="git_next",
|
|
default_language="en",
|
|
brand_voice="Audit-focused publishing profile.",
|
|
audience="QA operations team.",
|
|
seo_rules={"primary_keyword_required": True},
|
|
visual_rules={"hero_style": "technical"},
|
|
source_rules={"minimum_sources": 1},
|
|
publishing_rules=PublishingRules(
|
|
repository_url="git@github.com:example/audit-site.git",
|
|
production_branch="main",
|
|
content_format="mdx",
|
|
content_path_template="content/articles/{slug}.mdx",
|
|
asset_path_template="public/articles/{slug}/{filename}",
|
|
frontmatter_mapping={"title": "title"},
|
|
dry_run_renderer="next-mdx",
|
|
),
|
|
active_script_config_version_id=None,
|
|
),
|
|
).site
|
|
|
|
initial = create_script_config_version(
|
|
repository,
|
|
site_id=site.id,
|
|
current_user=admin,
|
|
request=self._script_version_payload(summary="initial", activate=True),
|
|
).version
|
|
candidate = create_script_config_version(
|
|
repository,
|
|
site_id=site.id,
|
|
current_user=admin,
|
|
request=self._script_version_payload(
|
|
summary="candidate",
|
|
activate=True,
|
|
rollback_target_version_id=initial.id,
|
|
),
|
|
).version
|
|
|
|
rolled_back = rollback_script_config_version(
|
|
repository,
|
|
site_id=site.id,
|
|
version_id=initial.id,
|
|
current_user=admin,
|
|
)
|
|
self.assertEqual(str(initial.id), str(rolled_back.version.id))
|
|
|
|
rollback_events = [
|
|
event
|
|
for event in repository.script_config_version_events.list_for_site(site.id)
|
|
if event["event_type"] == "SCRIPT_CONFIG_VERSION_ROLLBACK"
|
|
]
|
|
self.assertEqual(1, len(rollback_events))
|
|
rollback_event = rollback_events[0]
|
|
self.assertEqual(str(site.id), str(rollback_event["target_site_id"]))
|
|
self.assertEqual(str(initial.id), str(rollback_event["version_id"]))
|
|
self.assertEqual(str(admin.id), str(rollback_event["actor_user_id"]))
|
|
self.assertEqual(
|
|
str(candidate.id),
|
|
rollback_event["payload"]["rollback_target_version_id"],
|
|
)
|
|
|
|
active_site = repository.target_sites.get_by_id(site.id)
|
|
self.assertEqual(initial.id, active_site.active_script_config_version_id)
|
|
|
|
def _script_version_payload(
|
|
self,
|
|
summary: str,
|
|
activate: bool = False,
|
|
rollback_target_version_id: UUID | None = None,
|
|
) -> ScriptConfigVersionCreateRequest:
|
|
payload = ScriptConfigVersionCreateRequest(
|
|
publishing_yaml="target:\n repo: audit\n",
|
|
transform_script="export function transformArticle(article) { return article; }",
|
|
diff={"summary": summary},
|
|
activate=activate,
|
|
)
|
|
if rollback_target_version_id is not None:
|
|
payload.rollback_target_version_id = rollback_target_version_id
|
|
return payload
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|