80 lines
4.1 KiB
Python
80 lines
4.1 KiB
Python
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()
|