jpmschweitzerandClaude Sonnet 5
a81b9a071b
Every round-trip test across test_book_migration.py, test_scribe_migration.py and test_tenants_migration.py re-upgraded to its own revision on the way back up rather than to head, so a full suite run left core_api_test one or two migrations behind (008, with 009/010 released since). Each now ends on `upgrade head`; a new TestSchemaEndsAtHeadAfterTheMigrationSuite class checks the final state directly with no upgrade of its own, as a regression guard. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
407 lines
18 KiB
Python
407 lines
18 KiB
Python
"""
|
|
Tests for the tenant registry migrations 007 (core-api T-8, workspace D-73)
|
|
and 008 (core-api T-11, the default-address column).
|
|
|
|
Runs against `core_api_test` only (see conftest.py — skip when
|
|
TEST_DATABASE_URL is unset, hard failure if it resolves to `core_api`).
|
|
Self-contained: starts and ends at head regardless of what ran before it,
|
|
so test order never matters for the schema the rest of the suite needs.
|
|
"""
|
|
import pytest
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from tests.conftest import run_alembic
|
|
|
|
|
|
async def _table_exists(url: str, table: str) -> bool:
|
|
asyncpg_url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
engine = create_async_engine(asyncpg_url, poolclass=NullPool)
|
|
try:
|
|
async with engine.connect() as conn:
|
|
result = await conn.execute(
|
|
text("SELECT to_regclass(:name) IS NOT NULL AS present"),
|
|
{"name": table},
|
|
)
|
|
return bool(result.scalar())
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
async def _scalar(url: str, query: str, **params) -> object:
|
|
asyncpg_url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
engine = create_async_engine(asyncpg_url, poolclass=NullPool)
|
|
try:
|
|
async with engine.connect() as conn:
|
|
result = await conn.execute(text(query), params)
|
|
return result.scalar()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
class TestMigration007RoundTrip:
|
|
"""Migration 007 must upgrade and downgrade cleanly (dossier verification #2)."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upgrade_creates_tenant_tables(self, test_database_url):
|
|
result = run_alembic(test_database_url, "upgrade", "head")
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
for table in ("tenants", "tenant_consumers"):
|
|
assert await _table_exists(test_database_url, table), f"{table} missing after upgrade"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_downgrade_then_upgrade_round_trips_cleanly(self, test_database_url):
|
|
baseline = run_alembic(test_database_url, "upgrade", "head")
|
|
assert baseline.returncode == 0, baseline.stderr
|
|
|
|
down = run_alembic(test_database_url, "downgrade", "006")
|
|
assert down.returncode == 0, down.stderr
|
|
assert not await _table_exists(test_database_url, "tenants")
|
|
assert not await _table_exists(test_database_url, "tenant_consumers")
|
|
# Book and Scribe tables, created by migrations below this one, must
|
|
# survive a downgrade of 007 — proves 007 only touches its own
|
|
# tables (and the FKs it added onto theirs).
|
|
assert await _table_exists(test_database_url, "book_records")
|
|
assert await _table_exists(test_database_url, "scribe_addresses")
|
|
|
|
# Leave the schema at head for the rest of the suite (core-api T-14).
|
|
up = run_alembic(test_database_url, "upgrade", "head")
|
|
assert up.returncode == 0, up.stderr
|
|
assert await _table_exists(test_database_url, "tenants")
|
|
assert await _table_exists(test_database_url, "tenant_consumers")
|
|
|
|
|
|
class TestMigration007Checks:
|
|
"""The CHECK constraints on `tenants.kind` and `tenants.status`."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_kind_check_constraint_exists(self, test_database_url):
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT conname FROM pg_constraint WHERE conname = 'ck_tenants_kind'",
|
|
)
|
|
assert found == "ck_tenants_kind"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_check_constraint_exists(self, test_database_url):
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT conname FROM pg_constraint WHERE conname = 'ck_tenants_status'",
|
|
)
|
|
assert found == "ck_tenants_status"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_kind_check_rejects_an_invalid_kind(self, test_database_url):
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
asyncpg_url = test_database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
engine = create_async_engine(asyncpg_url, poolclass=NullPool)
|
|
try:
|
|
async with engine.connect() as conn:
|
|
with pytest.raises(IntegrityError):
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO tenants (id, display_name, kind, status) "
|
|
"VALUES ('bad-kind-tenant', 'x', 'nonsense', 'active')"
|
|
)
|
|
)
|
|
await conn.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
class TestMigration007Seed:
|
|
"""core-api T-8: the registry seeds exactly the tenants/grants the
|
|
dossier specifies — not a third, and not a second grant."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seeds_jpmschweitzer_and_llm_tester(self, test_database_url):
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
for tenant_id, display_name, kind in (
|
|
("jpmschweitzer", "Jeroen", "household"),
|
|
("llm_tester", "Playground", "test"),
|
|
):
|
|
row_kind = await _scalar(
|
|
test_database_url,
|
|
"SELECT kind FROM tenants WHERE id = :id AND display_name = :name AND status = 'active'",
|
|
id=tenant_id,
|
|
name=display_name,
|
|
)
|
|
assert row_kind == kind, f"{tenant_id} seed row missing or wrong"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_both_tenants_grant_only_office_claude(self, test_database_url):
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
for tenant_id in ("jpmschweitzer", "llm_tester"):
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT consumer FROM tenant_consumers WHERE tenant_id = :id AND consumer = 'office-claude'",
|
|
id=tenant_id,
|
|
)
|
|
assert found == "office-claude", f"{tenant_id} is missing its office-claude grant"
|
|
|
|
|
|
class TestMigration007ForeignKeys:
|
|
"""Every owner_tenant column (five of them, per the dossier) now carries
|
|
an FK to tenants.id; an unknown owner_tenant is refused at the database,
|
|
not only in the service layer (dossier verification #5)."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"fk_name",
|
|
[
|
|
"fk_book_records_owner_tenant",
|
|
"fk_book_tenant_counters_owner_tenant",
|
|
"fk_scribe_addresses_owner_tenant",
|
|
"fk_scribe_messages_owner_tenant",
|
|
"fk_scribe_mailbox_entries_owner_tenant",
|
|
],
|
|
)
|
|
async def test_owner_tenant_fk_exists(self, test_database_url, fk_name):
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT conname FROM pg_constraint WHERE conname = :name AND contype = 'f'",
|
|
name=fk_name,
|
|
)
|
|
assert found == fk_name
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fk_refuses_a_scribe_address_with_an_unknown_owner_tenant(self, test_database_url):
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
asyncpg_url = test_database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
engine = create_async_engine(asyncpg_url, poolclass=NullPool)
|
|
try:
|
|
async with engine.connect() as conn:
|
|
with pytest.raises(IntegrityError):
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO scribe_addresses (owner_tenant, name, created_by) "
|
|
"VALUES ('no-such-tenant', 'fk-probe', 'test-harness')"
|
|
)
|
|
)
|
|
await conn.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
class TestMigration008RoundTrip:
|
|
"""Migration 008 (core-api T-11) must upgrade and downgrade cleanly
|
|
(dossier verification #2) — same pattern as TestMigration007RoundTrip,
|
|
one revision up."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upgrade_adds_default_address_column(self, test_database_url):
|
|
result = run_alembic(test_database_url, "upgrade", "head")
|
|
assert result.returncode == 0, result.stderr
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT column_name FROM information_schema.columns "
|
|
"WHERE table_name = 'tenant_consumers' AND column_name = 'default_address'",
|
|
)
|
|
assert found == "default_address"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_downgrade_then_upgrade_round_trips_cleanly(self, test_database_url):
|
|
baseline = run_alembic(test_database_url, "upgrade", "head")
|
|
assert baseline.returncode == 0, baseline.stderr
|
|
|
|
down = run_alembic(test_database_url, "downgrade", "007")
|
|
assert down.returncode == 0, down.stderr
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT column_name FROM information_schema.columns "
|
|
"WHERE table_name = 'tenant_consumers' AND column_name = 'default_address'",
|
|
)
|
|
assert found is None
|
|
# tenant_consumers itself, and the rest of the registry, must
|
|
# survive a downgrade of 008 — proves 008 only touches its own
|
|
# column and FK.
|
|
assert await _table_exists(test_database_url, "tenant_consumers")
|
|
assert await _table_exists(test_database_url, "tenants")
|
|
|
|
# Leave the schema at head for the rest of the suite (core-api T-14).
|
|
up = run_alembic(test_database_url, "upgrade", "head")
|
|
assert up.returncode == 0, up.stderr
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT column_name FROM information_schema.columns "
|
|
"WHERE table_name = 'tenant_consumers' AND column_name = 'default_address'",
|
|
)
|
|
assert found == "default_address"
|
|
|
|
|
|
class TestMigration008ForeignKey:
|
|
"""The composite FK `(tenant_id, default_address)` ->
|
|
`scribe_addresses(owner_tenant, name)` (dossier verification #2/#3): a
|
|
default always names a real address in the same tenant."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fk_exists(self, test_database_url):
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT conname FROM pg_constraint "
|
|
"WHERE conname = 'fk_tenant_consumers_default_address' AND contype = 'f'",
|
|
)
|
|
assert found == "fk_tenant_consumers_default_address"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fk_refuses_a_default_naming_an_address_that_does_not_exist(
|
|
self, test_database_url
|
|
):
|
|
"""dossier verification #3: "A test inserts a default naming an
|
|
address that does not exist and sees the FK refuse it.\""""
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
asyncpg_url = test_database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
engine = create_async_engine(asyncpg_url, poolclass=NullPool)
|
|
try:
|
|
async with engine.connect() as conn:
|
|
with pytest.raises(IntegrityError):
|
|
await conn.execute(
|
|
text(
|
|
"UPDATE tenant_consumers SET default_address = 'no-such-address' "
|
|
"WHERE tenant_id = 'llm_tester' AND consumer = 'office-claude'"
|
|
)
|
|
)
|
|
await conn.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fk_allows_a_null_default_regardless_of_tenant(self, test_database_url):
|
|
"""A NULL default_address never triggers the composite check
|
|
(Postgres's default MATCH SIMPLE) — explicit, since this is the
|
|
state every grant but jpmschweitzer/office-claude is left in."""
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
asyncpg_url = test_database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
engine = create_async_engine(asyncpg_url, poolclass=NullPool)
|
|
try:
|
|
async with engine.begin() as conn:
|
|
await conn.execute(
|
|
text(
|
|
"UPDATE tenant_consumers SET default_address = NULL "
|
|
"WHERE tenant_id = 'llm_tester' AND consumer = 'office-claude'"
|
|
)
|
|
)
|
|
finally:
|
|
await engine.dispose()
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT default_address FROM tenant_consumers "
|
|
"WHERE tenant_id = 'llm_tester' AND consumer = 'office-claude'",
|
|
)
|
|
assert found is None
|
|
|
|
|
|
class TestMigration008DataStep:
|
|
"""jpmschweitzer/office-claude defaults to `claude-ai` only if that
|
|
address exists at upgrade time (dossier: "an UPDATE ... WHERE EXISTS
|
|
does this"); `llm_tester`/office-claude is never touched."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_a_noop_when_claude_ai_is_absent(self, test_database_url):
|
|
"""core_api_test seeds no scribe_addresses rows, so the data step
|
|
must run without error and leave the default null — this is the
|
|
state production's own test database is in today."""
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
down = run_alembic(test_database_url, "downgrade", "007")
|
|
assert down.returncode == 0, down.stderr
|
|
|
|
# Upgrade through to head (core-api T-14) — migrations 009/010 touch
|
|
# neither this table nor this column, so the assertion below still
|
|
# proves 008's data step alone.
|
|
up = run_alembic(test_database_url, "upgrade", "head")
|
|
assert up.returncode == 0, up.stderr
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT default_address FROM tenant_consumers "
|
|
"WHERE tenant_id = 'jpmschweitzer' AND consumer = 'office-claude'",
|
|
)
|
|
assert found is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_llm_tester_office_claude_stays_null(self, test_database_url):
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT default_address FROM tenant_consumers "
|
|
"WHERE tenant_id = 'llm_tester' AND consumer = 'office-claude'",
|
|
)
|
|
assert found is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sets_claude_ai_when_the_address_exists_at_upgrade_time(
|
|
self, test_database_url
|
|
):
|
|
"""Register claude-ai in jpmschweitzer *before* 008 runs (the
|
|
production shape), downgrade to 007 so the data step runs again on
|
|
the way back up, then prove the UPDATE actually sets the default
|
|
rather than leaving it null regardless — the mutation check's
|
|
positive case. Cleans up its own address and resets the default so
|
|
the rest of the suite sees the null state it otherwise expects."""
|
|
run_alembic(test_database_url, "upgrade", "head")
|
|
down = run_alembic(test_database_url, "downgrade", "007")
|
|
assert down.returncode == 0, down.stderr
|
|
|
|
asyncpg_url = test_database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
|
|
engine = create_async_engine(asyncpg_url, poolclass=NullPool)
|
|
try:
|
|
async with engine.begin() as conn:
|
|
await conn.execute(
|
|
text(
|
|
"INSERT INTO scribe_addresses (owner_tenant, name, created_by) "
|
|
"VALUES ('jpmschweitzer', 'claude-ai', 'test-harness')"
|
|
)
|
|
)
|
|
|
|
# Upgrade through to head (core-api T-14) — same reasoning as
|
|
# test_is_a_noop_when_claude_ai_is_absent above.
|
|
up = run_alembic(test_database_url, "upgrade", "head")
|
|
assert up.returncode == 0, up.stderr
|
|
|
|
found = await _scalar(
|
|
test_database_url,
|
|
"SELECT default_address FROM tenant_consumers "
|
|
"WHERE tenant_id = 'jpmschweitzer' AND consumer = 'office-claude'",
|
|
)
|
|
assert found == "claude-ai"
|
|
finally:
|
|
async with engine.begin() as conn:
|
|
await conn.execute(
|
|
text(
|
|
"UPDATE tenant_consumers SET default_address = NULL "
|
|
"WHERE tenant_id = 'jpmschweitzer' AND consumer = 'office-claude'"
|
|
)
|
|
)
|
|
await conn.execute(
|
|
text(
|
|
"DELETE FROM scribe_addresses "
|
|
"WHERE owner_tenant = 'jpmschweitzer' AND name = 'claude-ai'"
|
|
)
|
|
)
|
|
await engine.dispose()
|
|
|
|
|
|
class TestSchemaEndsAtHeadAfterTheMigrationSuite:
|
|
"""core-api T-14 regression guard: every round-trip test above (and in
|
|
test_book_migration.py / test_scribe_migration.py, which this file's
|
|
alphabetically-later collection runs after) must leave `core_api_test`
|
|
at head, not on its own revision. This class is the last one defined in
|
|
the last-collected migration test file, and its check is a bare
|
|
`current` — no `upgrade head` of its own — so a round-trip test that
|
|
regresses to ending on its own revision again shows up here as "not
|
|
head" rather than being silently repaired by this very check."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_is_at_head(self, test_database_url):
|
|
result = run_alembic(test_database_url, "current")
|
|
assert result.returncode == 0, result.stderr
|
|
assert "(head)" in result.stdout, (
|
|
f"core_api_test is not at head after the migration suite: {result.stdout!r}"
|
|
)
|