2268b6a77a
Add app credentials, account-scoped idempotency and per-app terminal notifications. Consolidate the unreleased migration baseline, preserve rule descriptions as AI context, and regenerate provider contracts with regression coverage.
115 lines
5.4 KiB
Python
115 lines
5.4 KiB
Python
from importlib.util import module_from_spec, spec_from_file_location
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from alembic.autogenerate import compare_metadata
|
|
from alembic.config import Config
|
|
from alembic.migration import MigrationContext
|
|
from alembic.operations import Operations
|
|
from alembic.script import ScriptDirectory
|
|
from sqlalchemy import delete, insert, inspect, select, text
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from ai_platform.infrastructure.config.settings import Settings
|
|
from ai_platform.infrastructure.persistence.database import Base, make_session_factory
|
|
from ai_platform.infrastructure.persistence.models import (
|
|
AppCredential,
|
|
ScoringNotification,
|
|
ScoringRun,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_one_initial_migration_with_no_legacy_chain():
|
|
config = Config(str(ROOT / "alembic.ini"))
|
|
config.set_main_option("path_separator", "os")
|
|
config.set_main_option("script_location", str(ROOT / "migrations"))
|
|
scripts = ScriptDirectory.from_config(config)
|
|
revisions = list(scripts.walk_revisions())
|
|
assert len(revisions) == 1
|
|
assert revisions[0].revision == "0001_initial"
|
|
assert revisions[0].down_revision is None
|
|
assert scripts.get_heads() == ["0001_initial"]
|
|
|
|
|
|
async def test_initial_schema_matches_models_constraints_and_round_trip():
|
|
settings = Settings.load()
|
|
if urlsplit(settings.database_url).path.rstrip("/").rsplit("/", 1)[-1] != "ai_platform_test":
|
|
pytest.fail("Migration tests require isolated ai_platform_test")
|
|
path = ROOT / "migrations/versions/0001_initial.py"
|
|
spec = spec_from_file_location(path.stem, path)
|
|
assert spec is not None and spec.loader is not None
|
|
migration = module_from_spec(spec)
|
|
spec.loader.exec_module(migration)
|
|
factory = make_session_factory(settings.database_url)
|
|
schema = "migration_test_" + uuid4().hex
|
|
try:
|
|
async with factory.kw["bind"].connect() as connection:
|
|
transaction = await connection.begin()
|
|
try:
|
|
await connection.execute(text(f"CREATE SCHEMA {schema}"))
|
|
await connection.execute(text(f"SET LOCAL search_path TO {schema}"))
|
|
|
|
def apply(sync, direction):
|
|
with Operations.context(MigrationContext.configure(sync)):
|
|
getattr(migration, direction)()
|
|
|
|
def assert_schema(sync):
|
|
context = MigrationContext.configure(
|
|
sync, opts={"compare_server_default": True}
|
|
)
|
|
assert compare_metadata(context, Base.metadata) == []
|
|
assert set(inspect(sync).get_table_names()) == set(Base.metadata.tables)
|
|
|
|
await connection.run_sync(apply, "upgrade")
|
|
await connection.run_sync(assert_schema)
|
|
for app_id in ("app-a", "app-b"):
|
|
await connection.execute(
|
|
insert(AppCredential).values(app_id=app_id, verifier=app_id)
|
|
)
|
|
request_id, run_id = str(uuid4()), str(uuid4())
|
|
run = {
|
|
"app_id": "app-a",
|
|
"account_id": "3",
|
|
"request_id": request_id,
|
|
"fingerprint": "0" * 64,
|
|
"request": {},
|
|
"profile_version": "example",
|
|
}
|
|
await connection.execute(insert(ScoringRun).values(id=run_id, **run))
|
|
# The same request identity is valid for a different account or app.
|
|
await connection.execute(insert(ScoringRun).values(**{**run, "account_id": "4"}))
|
|
await connection.execute(insert(ScoringRun).values(**{**run, "app_id": "app-b"}))
|
|
for values in (run, {**run, "app_id": "unknown"}):
|
|
with pytest.raises(IntegrityError):
|
|
async with connection.begin_nested():
|
|
await connection.execute(insert(ScoringRun).values(**values))
|
|
event = {
|
|
"run_id": run_id,
|
|
"app_id": "app-a",
|
|
"account_id": "3",
|
|
"request_id": request_id,
|
|
"status": "completed",
|
|
"next_attempt_at": text("now()"),
|
|
"created_at": text("now()"),
|
|
}
|
|
await connection.execute(insert(ScoringNotification).values(**event))
|
|
for values in (event, {**event, "run_id": str(uuid4())}):
|
|
with pytest.raises(IntegrityError):
|
|
async with connection.begin_nested():
|
|
await connection.execute(insert(ScoringNotification).values(**values))
|
|
await connection.execute(delete(ScoringRun).where(ScoringRun.id == run_id))
|
|
assert (await connection.execute(select(ScoringNotification.id))).first() is None
|
|
await connection.run_sync(apply, "downgrade")
|
|
assert await connection.run_sync(lambda sync: inspect(sync).get_table_names()) == []
|
|
await connection.run_sync(apply, "upgrade")
|
|
await connection.run_sync(assert_schema)
|
|
finally:
|
|
# Only this test's schema/data are reverted; public data is untouched.
|
|
await transaction.rollback()
|
|
finally:
|
|
await factory.kw["bind"].dispose()
|