165 lines
9.0 KiB
Python
165 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from alembic import command
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, event, inspect, text
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_access.backend.api.v1.routes import router
|
|
from govoplan_access.backend.auth.dependencies import get_api_principal
|
|
from govoplan_access.backend.db.models import ExternalFunctionRoleAssignment, Role
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.db.migrations import alembic_config
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_core.tenancy.scope import Tenant
|
|
|
|
|
|
TABLE_NAME = "access_external_function_role_assignments"
|
|
REPAIR_REVISION = "d8f1b4e7a0c3"
|
|
|
|
|
|
class ExternalFunctionMappingMigrationTests(unittest.TestCase):
|
|
def test_release_missing_table_repair(self) -> None:
|
|
self._verify_upgrade("release", missing=True)
|
|
|
|
def test_release_existing_mappings_preserved(self) -> None:
|
|
self._verify_upgrade("release", missing=False)
|
|
|
|
def test_dev_missing_table_repair(self) -> None:
|
|
self._verify_upgrade("dev", missing=True)
|
|
|
|
def test_dev_existing_mappings_preserved(self) -> None:
|
|
self._verify_upgrade("dev", missing=False)
|
|
|
|
def _verify_upgrade(self, track: str, *, missing: bool) -> None:
|
|
previous = "c7e0a3d6f9b2" if track == "release" else "b6d9f2a5c8e1"
|
|
with tempfile.TemporaryDirectory(prefix="govoplan-function-mapping-upgrade-") as directory:
|
|
url = f"sqlite:///{Path(directory) / 'upgrade.db'}"
|
|
config = alembic_config(database_url=url, enabled_modules=("access",), migration_track=track)
|
|
command.upgrade(config, "4f2a9c8e7b6d")
|
|
command.upgrade(config, previous)
|
|
engine = create_engine(url, connect_args={"check_same_thread": False})
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def enforce_foreign_keys(connection, _record) -> None:
|
|
connection.execute("PRAGMA foreign_keys=ON")
|
|
|
|
try:
|
|
with Session(engine) as session:
|
|
session.add_all([
|
|
Tenant(id="tenant-1", slug="tenant-1", name="Existing tenant"),
|
|
Tenant(id="tenant-2", slug="tenant-2", name="Other tenant"),
|
|
])
|
|
session.flush()
|
|
session.add_all([
|
|
Role(id="role-1", tenant_id="tenant-1", slug="role-1", name="Existing role", permissions=["access:function:read"]),
|
|
Role(id="role-2", tenant_id="tenant-2", slug="role-2", name="Other role", permissions=["access:role:read"]),
|
|
])
|
|
session.commit()
|
|
if missing:
|
|
# Reproduce only in this isolated database: a recorded baseline
|
|
# with the exact missing table observed in the live 500 response.
|
|
with engine.begin() as connection:
|
|
connection.execute(text("DROP TABLE access_external_function_role_assignments"))
|
|
else:
|
|
self._insert_mapping(engine, "mapping-1", "tenant-1", "role-1")
|
|
self._insert_mapping(engine, "mapping-2", "tenant-2", "role-2")
|
|
with engine.connect() as connection:
|
|
tables_before = set(inspect(connection).get_table_names())
|
|
parents_before = self._parent_rows(connection)
|
|
mappings_before = [] if missing else self._mapping_rows(connection)
|
|
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/api/v1")
|
|
principal = ApiPrincipal(
|
|
principal=PrincipalRef(account_id="reader", membership_id="reader-1", tenant_id="tenant-1", scopes=frozenset({"access:function:read"})),
|
|
account=None,
|
|
user=None,
|
|
)
|
|
|
|
def test_session():
|
|
with Session(engine) as session:
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = test_session
|
|
app.dependency_overrides[get_api_principal] = lambda: principal
|
|
with TestClient(app, raise_server_exceptions=False) as client:
|
|
path = "/api/v1/admin/external-function-role-mappings"
|
|
if missing:
|
|
self.assertEqual(client.get(f"{path}/delta").status_code, 500)
|
|
command.upgrade(config, REPAIR_REVISION)
|
|
command.upgrade(config, REPAIR_REVISION)
|
|
for suffix in ("", "/delta"):
|
|
response = client.get(f"{path}{suffix}")
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertEqual([item["id"] for item in response.json()["mappings"]], [] if missing else ["mapping-1"])
|
|
self.assertEqual(response.json()["total"], 0 if missing else 1)
|
|
self.assertEqual(client.get(f"{path}{suffix}?tenant_id=tenant-2").status_code, 409)
|
|
principal.principal = PrincipalRef(account_id="reader", membership_id="reader-1", tenant_id="tenant-1", scopes=frozenset())
|
|
self.assertEqual(client.get(f"{path}/delta").status_code, 403)
|
|
|
|
with engine.connect() as connection:
|
|
inspector = inspect(connection)
|
|
self.assertEqual(set(inspector.get_table_names()), tables_before | {TABLE_NAME})
|
|
self.assertEqual(self._parent_rows(connection), parents_before)
|
|
self.assertEqual(self._mapping_rows(connection), mappings_before)
|
|
columns = inspector.get_columns(TABLE_NAME)
|
|
self.assertEqual({item["name"] for item in columns}, {"id", "tenant_id", "source_module", "function_id", "role_id", "settings", "created_at", "updated_at"})
|
|
self.assertTrue(all(not item["nullable"] for item in columns))
|
|
self.assertEqual(inspector.get_pk_constraint(TABLE_NAME)["constrained_columns"], ["id"])
|
|
self.assertIn(["tenant_id", "source_module", "function_id", "role_id"], [item["column_names"] for item in inspector.get_unique_constraints(TABLE_NAME)])
|
|
self.assertEqual({tuple(item["column_names"]) for item in inspector.get_indexes(TABLE_NAME)}, {("tenant_id",), ("role_id",), ("function_id",), ("source_module",)})
|
|
self.assertEqual({(tuple(item["constrained_columns"]), item["referred_table"], item["options"]["ondelete"]) for item in inspector.get_foreign_keys(TABLE_NAME)}, {(("role_id",), "access_roles", "CASCADE"), (("tenant_id",), "core_scopes", "CASCADE")})
|
|
|
|
self._insert_mapping(engine, "mapping-after-repair", "tenant-1", "role-1", function_id="new-function")
|
|
with self.assertRaises(IntegrityError):
|
|
self._insert_mapping(engine, "duplicate", "tenant-1", "role-1", function_id="new-function")
|
|
with self.assertRaises(IntegrityError):
|
|
self._insert_mapping(engine, "bad-role", "tenant-1", "missing-role")
|
|
with self.assertRaises(IntegrityError):
|
|
self._insert_mapping(engine, "bad-tenant", "missing-tenant", "role-1")
|
|
with engine.connect() as connection:
|
|
all_mappings = self._mapping_rows(connection)
|
|
command.downgrade(config, previous)
|
|
command.upgrade(config, REPAIR_REVISION)
|
|
with engine.connect() as connection:
|
|
self.assertEqual(self._mapping_rows(connection), all_mappings)
|
|
self.assertEqual(self._parent_rows(connection), parents_before)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
@staticmethod
|
|
def _insert_mapping(engine, mapping_id: str, tenant_id: str, role_id: str, *, function_id: str = "function-1") -> None:
|
|
now = datetime.now(timezone.utc)
|
|
with Session(engine) as session:
|
|
session.add(ExternalFunctionRoleAssignment(
|
|
id=mapping_id, tenant_id=tenant_id, role_id=role_id,
|
|
source_module="organizations", function_id=function_id,
|
|
settings={"meaning": "Existing mapping", "nested": {"retained": True}},
|
|
created_at=now, updated_at=now,
|
|
))
|
|
session.commit()
|
|
|
|
@staticmethod
|
|
def _mapping_rows(connection):
|
|
return [dict(row) for row in connection.execute(text("SELECT * FROM access_external_function_role_assignments ORDER BY id")).mappings()]
|
|
|
|
@staticmethod
|
|
def _parent_rows(connection):
|
|
return {
|
|
"roles": [dict(row) for row in connection.execute(text("SELECT * FROM access_roles ORDER BY id")).mappings()],
|
|
"tenants": [dict(row) for row in connection.execute(text("SELECT * FROM core_scopes ORDER BY id")).mappings()],
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|