feat: consolidate shared UI and harden browser authority for release

2026-09-08 01:35:05 +02:00
parent ac40774785
commit b75ca34295
143 changed files with 8664 additions and 752 deletions
@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from alembic import command
+from sqlalchemy import create_engine, inspect, select, text
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.ownership import OwnershipTransfer
+from govoplan_core.db.migrations import alembic_config
+
+
+class OwnershipHistoryMigrationTests(unittest.TestCase):
+    def test_upgrade_existing_databases_without_rewriting_ownership(self) -> None:
+        for track in ("release", "dev"):
+            for legacy in (True, False):
+                with self.subTest(track=track, legacy=legacy):
+                    self._verify_upgrade(track, legacy=legacy)
+
+    def _verify_upgrade(self, track: str, *, legacy: bool) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-ownership-upgrade-") as directory:
+            url = f"sqlite:///{Path(directory) / 'upgrade.db'}"
+            config = alembic_config(database_url=url, enabled_modules=(), migration_track=track)
+            command.upgrade(config, "b47e6f809a13")
+            engine = create_engine(url)
+            try:
+                evidence = [{"sequence": 1, "action": "requested"}]
+                now = datetime.now(timezone.utc)
+                with Session(engine) as session:
+                    session.add(OwnershipTransfer(
+                        id="transfer-1", tenant_id="tenant-1", resource_module="campaigns",
+                        resource_type="campaign", resource_id="campaign-1", kind="owner_initiated",
+                        status="awaiting_target_acceptance", current_owner_type="user",
+                        current_owner_id="owner-1", target_owner_type="user", target_owner_id="owner-2",
+                        initiated_by_type="user", initiated_by_id="owner-1", reason="Existing request",
+                        approvals=[{"actor_id": "owner-1"}], decisions=evidence,
+                        idempotency_key="request-1", canonical_request_hash="a" * 64,
+                        expires_at=now + timedelta(days=7), revision=3, metadata_={"retained": True},
+                        created_at=now, updated_at=now,
+                    ))
+                    session.commit()
+                with engine.begin() as connection:
+                    if legacy:
+                        connection.execute(text("ALTER TABLE core_ownership_transfers DROP COLUMN decisions"))
+                    before = dict(connection.execute(text(
+                        "SELECT * FROM core_ownership_transfers WHERE id = 'transfer-1'"
+                    )).mappings().one())
+                command.upgrade(config, "c58a2d7e9f10")
+                command.upgrade(config, "c58a2d7e9f10")
+                with engine.connect() as connection:
+                    columns = {column["name"]: column for column in inspect(connection).get_columns(
+                        "core_ownership_transfers"
+                    )}
+                    self.assertFalse(columns["decisions"]["nullable"])
+                    after = dict(connection.execute(text(
+                        "SELECT * FROM core_ownership_transfers WHERE id = 'transfer-1'"
+                    )).mappings().one())
+                    self.assertEqual({key: after[key] for key in before}, before)
+                with Session(engine) as session:
+                    transfer = session.scalars(select(OwnershipTransfer)).one()
+                    self.assertEqual(transfer.decisions, [] if legacy else evidence)
+                    self.assertEqual(transfer.approvals, [{"actor_id": "owner-1"}])
+                    transfer.decisions = [*transfer.decisions, {"action": "accepted"}]
+                    session.commit()
+                command.downgrade(config, "b47e6f809a13")
+                command.upgrade(config, "c58a2d7e9f10")
+                with Session(engine) as session:
+                    self.assertEqual(session.get(OwnershipTransfer, "transfer-1").decisions[-1], {
+                        "action": "accepted"
+                    })
+            finally:
+                engine.dispose()
+
+
+if __name__ == "__main__":
+    unittest.main()