huncode/content-factory
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement task 006 site config versioning and audit events

Browse Source
This commit is contained in:
huncode
2026-05-21 19:13:19 +03:00
parent cf84400118
commit ff1a2f78c0
23 changed files with 1878 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/src/application/site_config.py
+93
View File
@@ -16,6 +16,11 @@ from src.domain.contracts import (
)
_VERSION_CREATED_EVENT_TYPE = "SCRIPT_CONFIG_VERSION_CREATED"
_VERSION_ACTIVATED_EVENT_TYPE = "SCRIPT_CONFIG_VERSION_ACTIVATED"
_VERSION_ROLLBACK_EVENT_TYPE = "SCRIPT_CONFIG_VERSION_ROLLBACK"
def create_target_site(
repository: object,
request: TargetSiteConfigCreateRequest,
@@ -40,6 +45,14 @@ def create_target_site(
return TargetSiteConfigResponse(site=site)
def get_target_site(
repository: object,
site_id: UUID,
) -> TargetSiteConfigResponse:
site = repository.target_sites.get_by_id(site_id)
return TargetSiteConfigResponse(site=site)
def update_target_site(
repository: object,
site_id: UUID,
@@ -82,6 +95,16 @@ def create_script_config_version(
request: ScriptConfigVersionCreateRequest,
) -> ScriptConfigVersionResponse:
repository.target_sites.get_by_id(site_id)
if request.rollback_target_version_id is not None:
rollback_target = repository.script_config_versions.get_by_id(
request.rollback_target_version_id
)
if str(rollback_target["target_site_id"]) != str(site_id):
raise LookupError(
f"Rollback target version is not for site: {request.rollback_target_version_id}"
)
existing_versions = repository.script_config_versions.list_for_site(site_id)
version = request.version or _next_version(existing_versions)
now = _now()
@@ -112,6 +135,24 @@ def create_script_config_version(
target_site_id=site_id,
version_id=version_id,
activated_at=now,
rollback_target_version_id=request.rollback_target_version_id,
)
repository.script_config_version_events.create(
target_site_id=site_id,
version_id=version_id,
event_type=_VERSION_ACTIVATED_EVENT_TYPE,
actor_user_id=current_user.id,
payload={"action": "activate", "version": version},
created_at=now,
)
else:
repository.script_config_version_events.create(
target_site_id=site_id,
version_id=version_id,
event_type=_VERSION_CREATED_EVENT_TYPE,
actor_user_id=current_user.id,
payload={"action": "create", "version": version},
created_at=now,
)
return ScriptConfigVersionResponse(version=_script_version_summary(row))
@@ -122,6 +163,7 @@ def activate_script_config_version(
*,
site_id: UUID,
version_id: UUID,
current_user: CurrentUser,
) -> ScriptConfigVersionResponse:
repository.target_sites.get_by_id(site_id)
row = repository.script_config_versions.activate(
@@ -129,6 +171,54 @@ def activate_script_config_version(
version_id=version_id,
activated_at=_now(),
)
repository.script_config_version_events.create(
target_site_id=site_id,
version_id=version_id,
event_type=_VERSION_ACTIVATED_EVENT_TYPE,
actor_user_id=current_user.id,
payload={"action": "activate", "version_id": str(version_id)},
created_at=_now(),
)
return ScriptConfigVersionResponse(version=_script_version_summary(row))
def rollback_script_config_version(
repository: object,
*,
site_id: UUID,
version_id: UUID,
current_user: CurrentUser,
) -> ScriptConfigVersionResponse:
target_site = repository.target_sites.get_by_id(site_id)
version_row = repository.script_config_versions.get_by_id(version_id)
if str(version_row["target_site_id"]) != str(site_id):
raise LookupError(
f"Script config version not found for site: {site_id} {version_id}"
)
rollback_target_id = target_site.active_script_config_version_id
if rollback_target_id == version_id:
rollback_target_id = None
row = repository.script_config_versions.activate(
target_site_id=site_id,
version_id=version_id,
activated_at=_now(),
rollback_target_version_id=rollback_target_id,
)
repository.script_config_version_events.create(
target_site_id=site_id,
version_id=version_id,
event_type=_VERSION_ROLLBACK_EVENT_TYPE,
actor_user_id=current_user.id,
payload={
"action": "rollback",
"rollback_target_version_id": (
str(rollback_target_id) if rollback_target_id is not None else None
),
},
created_at=_now(),
)
return ScriptConfigVersionResponse(version=_script_version_summary(row))
@@ -147,10 +237,13 @@ def _script_version_summary(row: dict[str, object]) -> ScriptConfigVersionSummar
target_site_id=row["target_site_id"],
version=row["version"],
status=row["status"],
diff=row["diff"],
rollback_target_version_id=row["rollback_target_version_id"],
publishing_yaml_hash=row["publishing_yaml_hash"],
transform_script_hash=row["transform_script_hash"],
created_by=row["created_by"],
created_at=row["created_at"],
activated_at=row["activated_at"],
)
apps/backend/src/domain/contracts/models.py
+3
View File
@@ -113,10 +113,13 @@ class ScriptConfigVersionSummary(ContractModel):
target_site_id: UUID
version: int = Field(ge=1)
status: ScriptConfigVersionStatus
diff: JsonObject = Field(default_factory=dict)
publishing_yaml_hash: str = Field(min_length=1)
transform_script_hash: str = Field(min_length=1)
rollback_target_version_id: UUID | None = None
created_by: UUID
created_at: datetime
activated_at: datetime | None = None
class ScriptConfigVersionCreateRequest(ContractModel):
apps/backend/src/infrastructure/repositories.py
+108 -1
View File
@@ -37,6 +37,7 @@ class BackendRepository:
self.target_sites = TargetSitesRepository(self)
self.articles = ArticlesRepository(self)
self.script_config_versions = ScriptConfigVersionsRepository(self)
self.script_config_version_events = ScriptConfigVersionAuditEventsRepository(self)
self.schema = SchemaRepository(self)
def setup(self) -> None:
@@ -719,6 +720,7 @@ class ScriptConfigVersionsRepository:
target_site_id: UUID,
version_id: UUID,
activated_at: datetime,
rollback_target_version_id: UUID | None = None,
) -> dict[str, Any]:
placeholder = self._repository.placeholder()
with self._repository.connection() as connection:
@@ -756,13 +758,15 @@ class ScriptConfigVersionsRepository:
SET
status = {placeholder},
activated_at = {placeholder},
updated_at = {placeholder}
updated_at = {placeholder},
rollback_target_version_id = COALESCE({placeholder}, rollback_target_version_id)
WHERE id = {placeholder}
""",
(
ScriptConfigVersionStatus.ACTIVE.value,
_datetime_value(activated_at),
_datetime_value(activated_at),
_uuid_value(rollback_target_version_id),
str(version_id),
),
)
@@ -795,6 +799,97 @@ class ScriptConfigVersionsRepository:
return [_plain_row(row) for row in rows]
class ScriptConfigVersionAuditEventsRepository:
def __init__(self, repository: BackendRepository) -> None:
self._repository = repository
def create(
self,
*,
target_site_id: UUID,
version_id: UUID,
event_type: str,
actor_user_id: UUID | None,
payload: JsonObject,
created_at: datetime,
) -> dict[str, Any]:
placeholder = self._repository.placeholder()
sql = f"""
INSERT INTO script_config_version_events (
id,
target_site_id,
version_id,
event_type,
actor_user_id,
payload,
created_at
)
VALUES (
{placeholder},
{placeholder},
{placeholder},
{placeholder},
{placeholder},
{placeholder},
{placeholder}
)
"""
params = (
str(uuid4()),
str(target_site_id),
str(version_id),
event_type,
_uuid_value(actor_user_id),
_json_value(payload),
_datetime_value(created_at),
)
with self._repository.connection() as connection:
connection.execute(sql, params)
return self.get(event_type=event_type, target_site_id=target_site_id, version_id=version_id)
def list_for_site(self, target_site_id: UUID) -> list[dict[str, Any]]:
placeholder = self._repository.placeholder()
with self._repository.connection() as connection:
rows = connection.execute(
f"""
SELECT id, target_site_id, version_id, event_type, actor_user_id, payload, created_at
FROM script_config_version_events
WHERE target_site_id = {placeholder}
ORDER BY created_at
""",
(str(target_site_id),),
).fetchall()
return [_script_config_version_event_from_row(row) for row in rows]
def get(
self,
*,
target_site_id: UUID,
version_id: UUID,
event_type: str,
) -> dict[str, Any]:
placeholder = self._repository.placeholder()
with self._repository.connection() as connection:
row = connection.execute(
f"""
SELECT id, target_site_id, version_id, event_type, actor_user_id, payload, created_at
FROM script_config_version_events
WHERE target_site_id = {placeholder}
AND version_id = {placeholder}
AND event_type = {placeholder}
""",
(str(target_site_id), str(version_id), event_type),
).fetchone()
if row is None:
raise LookupError(
f"Script config version event not found: {target_site_id} {version_id} {event_type}"
)
return _script_config_version_event_from_row(row)
class SchemaRepository:
def __init__(self, repository: BackendRepository) -> None:
self._repository = repository
@@ -920,6 +1015,18 @@ def _workflow_event_from_row(row: Any) -> WorkflowEventSummary:
)
def _script_config_version_event_from_row(row: Any) -> dict[str, Any]:
return {
"id": _row_value(row, "id"),
"target_site_id": _row_value(row, "target_site_id"),
"version_id": _row_value(row, "version_id"),
"event_type": _row_value(row, "event_type"),
"actor_user_id": _row_value(row, "actor_user_id"),
"payload": _json_from_row(row, "payload"),
"created_at": _row_value(row, "created_at"),
}
def _plain_row(row: Any) -> dict[str, Any]:
if isinstance(row, sqlite3.Row):
result = dict(row)
apps/backend/src/infrastructure/schema.py
+30
View File
@@ -285,6 +285,17 @@ POSTGRES_SCHEMA_STATEMENTS: tuple[str, ...] = (
UNIQUE (prompt_key, version)
)
""",
f"""
CREATE TABLE IF NOT EXISTS script_config_version_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
target_site_id UUID NOT NULL REFERENCES target_sites(id) ON DELETE CASCADE,
version_id UUID NOT NULL REFERENCES script_config_versions(id) ON DELETE CASCADE,
event_type TEXT NOT NULL,
actor_user_id UUID REFERENCES users(id),
payload JSONB NOT NULL DEFAULT '{{}}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
)
""",
"CREATE INDEX IF NOT EXISTS idx_target_sites_slug ON target_sites (slug)",
"""
CREATE INDEX IF NOT EXISTS idx_articles_dashboard
@@ -314,6 +325,10 @@ POSTGRES_SCHEMA_STATEMENTS: tuple[str, ...] = (
CREATE INDEX IF NOT EXISTS idx_publish_commits_article
ON publish_commits (article_id, created_at)
""",
"""
CREATE INDEX IF NOT EXISTS idx_script_config_version_events_site
ON script_config_version_events (target_site_id, created_at)
""",
)
@@ -550,6 +565,17 @@ SQLITE_SCHEMA_STATEMENTS: tuple[str, ...] = (
UNIQUE (prompt_key, version)
)
""",
"""
CREATE TABLE IF NOT EXISTS script_config_version_events (
id TEXT PRIMARY KEY,
target_site_id TEXT NOT NULL,
version_id TEXT NOT NULL,
event_type TEXT NOT NULL,
actor_user_id TEXT,
payload TEXT NOT NULL DEFAULT '{}',
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""",
"CREATE INDEX IF NOT EXISTS idx_target_sites_slug ON target_sites (slug)",
"""
CREATE INDEX IF NOT EXISTS idx_articles_dashboard
@@ -579,6 +605,10 @@ SQLITE_SCHEMA_STATEMENTS: tuple[str, ...] = (
CREATE INDEX IF NOT EXISTS idx_publish_commits_article
ON publish_commits (article_id, created_at)
""",
"""
CREATE INDEX IF NOT EXISTS idx_script_config_version_events_site
ON script_config_version_events (target_site_id, created_at)
""",
)
apps/backend/src/presentation/routes/sites.py
+37 -1
View File
@@ -8,7 +8,9 @@ from src.application.site_config import (
activate_script_config_version,
create_script_config_version,
create_target_site,
get_target_site,
list_script_config_versions,
rollback_script_config_version,
update_target_site,
)
from src.domain.auth import ADMIN_ROLES, EDITOR_OR_ADMIN_ROLES
@@ -36,6 +38,18 @@ def list_sites(
return [TargetSiteConfigResponse(site=site) for site in repository.target_sites.list()]
@router.get("/sites/{site_id}", response_model=TargetSiteConfigResponse)
def get_site(
site_id: UUID,
_: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
repository: BackendRepository = Depends(get_repository),
) -> TargetSiteConfigResponse:
try:
return get_target_site(repository, site_id)
except LookupError as error:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not Found") from error
@router.post(
"/sites",
response_model=TargetSiteConfigResponse,
@@ -107,7 +121,7 @@ def post_script_config_version(
def post_activate_script_config_version(
site_id: UUID,
version_id: UUID,
_: CurrentUser = Depends(require_roles(ADMIN_ROLES)),
current_user: CurrentUser = Depends(require_roles(ADMIN_ROLES)),
repository: BackendRepository = Depends(get_repository),
) -> ScriptConfigVersionResponse:
try:
@@ -115,6 +129,28 @@ def post_activate_script_config_version(
repository,
site_id=site_id,
version_id=version_id,
current_user=current_user,
)
except LookupError as error:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not Found") from error
@router.post(
"/sites/{site_id}/publishing-config/versions/{version_id}/rollback",
response_model=ScriptConfigVersionResponse,
)
def post_rollback_script_config_version(
site_id: UUID,
version_id: UUID,
current_user: CurrentUser = Depends(require_roles(ADMIN_ROLES)),
repository: BackendRepository = Depends(get_repository),
) -> ScriptConfigVersionResponse:
try:
return rollback_script_config_version(
repository,
site_id=site_id,
version_id=version_id,
current_user=current_user,
)
except LookupError as error:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not Found") from error
apps/backend/tests/integration/test_auth_authorization_public_api.py
+157 -3
View File
@@ -121,6 +121,35 @@ class AuthAuthorizationPublicApiTest(unittest.TestCase):
version_activate_response.text,
)
def test_editor_cannot_rollback_script_config_version(self) -> None:
site_response = self.client.post(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._site_payload("rollback_denied_blog"),
)
self.assertEqual(201, site_response.status_code, site_response.text)
site_id = site_response.json()["site"]["id"]
first_version = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._script_version_payload(),
)
self.assertEqual(201, first_version.status_code, first_version.text)
version_id = first_version.json()["version"]["id"]
activate_response = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions/{version_id}/activate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
rollback_response = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions/{version_id}/rollback",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(403, activate_response.status_code, activate_response.text)
self.assertEqual(403, rollback_response.status_code, rollback_response.text)
def test_admin_can_create_and_activate_script_config_version(self) -> None:
site_response = self.client.post(
"/api/sites",
@@ -147,6 +176,121 @@ class AuthAuthorizationPublicApiTest(unittest.TestCase):
self.assertEqual(200, activate_response.status_code, activate_response.text)
self.assertEqual("ACTIVE", activate_response.json()["version"]["status"])
def test_admin_can_get_site_by_id(self) -> None:
site_response = self.client.post(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._site_payload("admin_get_site_by_id_blog"),
)
self.assertEqual(201, site_response.status_code, site_response.text)
site_id = site_response.json()["site"]["id"]
get_response = self.client.get(
f"/api/sites/{site_id}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, get_response.status_code, get_response.text)
self.assertEqual(site_id, get_response.json()["site"]["id"])
def test_admin_can_create_version_with_rollback_target(self) -> None:
site_response = self.client.post(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._site_payload("rollback_target_blog"),
)
self.assertEqual(201, site_response.status_code, site_response.text)
site_id = site_response.json()["site"]["id"]
baseline = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._script_version_payload(diff={"summary": "baseline"}),
)
self.assertEqual(201, baseline.status_code, baseline.text)
baseline_id = baseline.json()["version"]["id"]
create_response = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._script_version_payload(
diff={"summary": "rollback-aware"},
rollback_target_version_id=baseline_id,
activate=True,
),
)
self.assertEqual(201, create_response.status_code, create_response.text)
created = create_response.json()["version"]
self.assertEqual("ACTIVE", created["status"])
self.assertEqual("rollback-aware", created["diff"].get("summary"))
self.assertEqual(baseline_id, created["rollback_target_version_id"])
versions_response = self.client.get(
f"/api/sites/{site_id}/publishing-config/versions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, versions_response.status_code, versions_response.text)
self.assertEqual(2, len(versions_response.json()["versions"]))
def test_admin_can_rollback_script_config_version(self) -> None:
site_response = self.client.post(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._site_payload("rollback_action_blog"),
)
self.assertEqual(201, site_response.status_code, site_response.text)
site_id = site_response.json()["site"]["id"]
active = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._script_version_payload(
diff={"summary": "initial active"},
activate=True,
),
)
self.assertEqual(201, active.status_code, active.text)
active_id = active.json()["version"]["id"]
next_version = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._script_version_payload(diff={"summary": "next version"}),
)
self.assertEqual(201, next_version.status_code, next_version.text)
next_id = next_version.json()["version"]["id"]
self.client.post(
f"/api/sites/{site_id}/publishing-config/versions/{next_id}/activate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
)
rollback_response = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions/{active_id}/rollback",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
)
self.assertEqual(200, rollback_response.status_code, rollback_response.text)
self.assertEqual(active_id, rollback_response.json()["version"]["id"])
self.assertEqual(
"ACTIVE",
rollback_response.json()["version"]["status"],
)
self.assertEqual(
next_id,
rollback_response.json()["version"]["rollback_target_version_id"],
)
get_response = self.client.get(
f"/api/sites/{site_id}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, get_response.status_code, get_response.text)
self.assertEqual(
active_id,
get_response.json()["site"]["active_script_config_version_id"],
)
def test_editor_can_create_articles_and_approve_plan_review(self) -> None:
sites_response = self.client.get(
"/api/sites",
@@ -215,12 +359,22 @@ class AuthAuthorizationPublicApiTest(unittest.TestCase):
},
}
def _script_version_payload(self) -> dict[str, object]:
return {
def _script_version_payload(
self,
diff: dict[str, object] | None = None,
rollback_target_version_id: str | None = None,
activate: bool = False,
) -> dict[str, object]:
payload = {
"publishing_yaml": "target: editorial_ops\nrepository:\n branch: main\n",
"transform_script": "export function transformArticle(article) { return article; }\n",
"diff": {"summary": "Task 004 authorization fixture."},
"diff": diff or {"summary": "Task 004 authorization fixture."},
}
if rollback_target_version_id is not None:
payload["rollback_target_version_id"] = rollback_target_version_id
if activate:
payload["activate"] = True
return payload
if __name__ == "__main__":
apps/backend/tests/integration/test_script_config_audit.py
+134
View File
@@ -0,0 +1,134 @@
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()