From e70e4efea2ad247667b391536175b20e2eabdea6 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 21:03:41 +1100 Subject: [PATCH] feat(tenancy): give each tenant its own T&E tables (#99) Tenant provisioning (CLI and every startup) now clones the 12 T&E tables from the deployment's T&E schema into tenant_: columns, defaults, checks, indexes and comments via LIKE ... INCLUDING ALL, then foreign keys, the updated_at function and its triggers re-pointed at the tenant schema. Keys are UUIDs, so tenants share no sequence. Idempotent. T&E mode under a tenant session now loads into that tenant's tables, so the 409 added in #93 is gone. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019YPoXzBSj89ZHgw7nfTmbB --- api/routers/csv_routes.py | 5 +-- api/tenancy.py | 67 +++++++++++++++++++++++++++++++--- tests/test_tenant_isolation.py | 63 ++++++++++++++++++++++++++++++-- 3 files changed, 122 insertions(+), 13 deletions(-) diff --git a/api/routers/csv_routes.py b/api/routers/csv_routes.py index 377f633..55853b9 100644 --- a/api/routers/csv_routes.py +++ b/api/routers/csv_routes.py @@ -10,7 +10,7 @@ from api.auth import allow, tenant_scope from api.config import settings -from api.tenancy import current_schema, uploads_schema +from api.tenancy import uploads_schema from api.db import Conn from api.services.audit import coerce_file_id, record_audit, write_audit from api.services.csv_parse import build_preview @@ -72,9 +72,6 @@ def upload(req: UploadRequest) -> dict: if len(req.content.encode("utf-8")) > settings.MAX_UPLOAD_BYTES: raise HTTPException(413, "File too large") if req.mode == "te": - if current_schema() is not None: - raise HTTPException(409, "T&E mode isn't available for tenants yet; it arrives with " - "per-tenant T&E tables (#99). Use dynamic mode.") if not req.targetTable: raise HTTPException(422, "targetTable is required when mode is 'te'") result = upload_te(req.fileName, req.content, req.targetTable) diff --git a/api/tenancy.py b/api/tenancy.py index 0e43efb..e08f1f2 100644 --- a/api/tenancy.py +++ b/api/tenancy.py @@ -1,7 +1,7 @@ """Per-request tenant schema routing (M2b, #93, BR-36; decision 6: schema per tenant). -Each tenant's registry, audit log and uploaded tables live in schema -``tenant_``. A request signed in to a tenant runs with that schema set +Each tenant's registry, audit log, uploaded tables and T&E tables live in +schema ``tenant_``. A request signed in to a tenant runs with that schema set here; ``uploads_schema()`` and ``te_schema()`` read it. A request without a session (API-key or dev) keeps the deployment's default schemas. """ @@ -12,7 +12,9 @@ from contextlib import contextmanager from contextvars import ContextVar -from api.config import settings +from psycopg2 import sql + +from api.config import TE_TABLES, settings _current: ContextVar[str | None] = ContextVar("tenant_schema", default=None) _migrate_lock = threading.Lock() @@ -32,8 +34,6 @@ def uploads_schema() -> str: def te_schema() -> str: - # ponytail: per-tenant T&E tables arrive in #99; until then a tenant's - # T&E requests are refused before reaching this (csv_routes.upload). return _current.get() or settings.TE_SCHEMA @@ -66,3 +66,60 @@ def migrate_tenant(slug: str) -> None: command.upgrade(cfg, "head") finally: settings.UPLOADS_SCHEMA = original + from api.db import Conn + + with Conn() as conn: + with conn.cursor() as cur: + provision_te_tables(cur, schema_for(slug), settings.TE_SCHEMA) + conn.commit() + + +def provision_te_tables(cur, schema: str, template: str) -> int: + """Give a tenant its own T&E tables, cloned from the deployment's T&E schema (#99). + + ``template`` is the schema build/deploy_all.sh created. Copies columns, + defaults, checks, indexes and comments, then the foreign keys, the + updated_at function and its triggers, re-pointed at ``schema``. Keys are + UUIDs, so no sequence is shared between tenants. Idempotent: only missing + tables are created. Returns how many were. The caller commits. + """ + cur.execute("SET LOCAL search_path = pg_catalog") # makes catalog DDL schema-qualified + cur.execute("SELECT quote_ident(%s), quote_ident(%s)", (template, schema)) + qt, qs = cur.fetchone() + cur.execute("SELECT relname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace " + "WHERE n.nspname = %s AND c.relkind = 'r' AND c.relname = ANY(%s)", (template, TE_TABLES)) + if {r[0] for r in cur.fetchall()} != set(TE_TABLES): + raise RuntimeError(f"T&E template schema {template!r} is missing or incomplete; " + "deploy it with build/deploy_all.sh first") + cur.execute("SELECT relname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace " + "WHERE n.nspname = %s AND c.relname = ANY(%s)", (schema, TE_TABLES)) + present = {r[0] for r in cur.fetchall()} + missing = [t for t in TE_TABLES if t not in present] + if not missing: + return 0 + for table in missing: + cur.execute(sql.SQL("CREATE TABLE {}.{} (LIKE {}.{} INCLUDING ALL)").format( + sql.Identifier(schema), sql.Identifier(table), sql.Identifier(template), sql.Identifier(table))) + + def defs(query: str) -> list[tuple]: + cur.execute(query, (template, missing)) + return cur.fetchall() + + for table, name, ddl in defs( + "SELECT r.relname, c.conname, pg_get_constraintdef(c.oid) FROM pg_constraint c " + "JOIN pg_class r ON r.oid = c.conrelid JOIN pg_namespace n ON n.oid = r.relnamespace " + "WHERE n.nspname = %s AND c.contype = 'f' AND r.relname = ANY(%s) ORDER BY c.conname"): + cur.execute(sql.SQL("ALTER TABLE {}.{} ADD CONSTRAINT {} ").format( + sql.Identifier(schema), sql.Identifier(table), sql.Identifier(name)) + + sql.SQL(ddl.replace(f"REFERENCES {qt}.", f"REFERENCES {qs}."))) + cur.execute("SELECT pg_get_functiondef(p.oid) FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace " + "WHERE n.nspname = %s AND p.proname = 'set_updated_at'", (template,)) + for (ddl,) in cur.fetchall(): + cur.execute(sql.SQL(ddl.replace(f"FUNCTION {qt}.", f"FUNCTION {qs}.", 1))) + for (ddl,) in defs( + "SELECT pg_get_triggerdef(t.oid) FROM pg_trigger t JOIN pg_class r ON r.oid = t.tgrelid " + "JOIN pg_namespace n ON n.oid = r.relnamespace " + "WHERE n.nspname = %s AND NOT t.tgisinternal AND r.relname = ANY(%s)"): + cur.execute(sql.SQL(ddl.replace(f" ON {qt}.", f" ON {qs}.").replace(f"FUNCTION {qt}.", f"FUNCTION {qs}."))) + cur.execute("RESET search_path") + return len(missing) diff --git a/tests/test_tenant_isolation.py b/tests/test_tenant_isolation.py index d3962d7..c60e80f 100644 --- a/tests/test_tenant_isolation.py +++ b/tests/test_tenant_isolation.py @@ -11,6 +11,7 @@ import unittest import uuid +import psycopg2.errors import pytest from fastapi.testclient import TestClient from psycopg2 import sql @@ -136,11 +137,58 @@ def names(who): self.assertNotIn(name, names("b")) self.assertNotIn(name, names(None)) - def test_te_mode_is_refused_for_tenants_until_99(self): - res = self.client.post("/api/csv/upload", headers=self.headers["a"], json={ - "fileName": "t.csv", "content": "name,org_type,country\nX,government,AU\n", + def _te_upload(self, who: str, org: str) -> dict: + res = self.client.post("/api/csv/upload", headers=self.headers[who], json={ + "fileName": f"{org}.csv", "content": f"name,org_type,country\n{org},government,AU\n", "mode": "te", "targetTable": "organisations"}) - self.assertEqual(res.status_code, 409, res.text) + self.assertEqual(res.status_code, 200, res.text) + return res.json() + + def _orgs(self, schema: str) -> list[str]: + with Conn() as conn, conn.cursor() as cur: + cur.execute(sql.SQL("SELECT name FROM {}.organisations").format(sql.Identifier(schema))) + return [r[0] for r in cur.fetchall()] + + def test_te_mode_loads_into_the_tenants_own_te_tables(self): + org = f"ORG_{uuid.uuid4().hex[:10]}" + res = self._te_upload("a", org) + self.assertEqual((res["status"], res["insertedRows"]), ("ok", 1), res) + self.assertEqual(res["tableName"], f"{schema_for(self.slugs['a'])}.organisations") + self.assertIn(org, self._orgs(schema_for(self.slugs["a"]))) + self.assertNotIn(org, self._orgs(schema_for(self.slugs["b"]))) + self.assertNotIn(org, self._orgs(settings.TE_SCHEMA), "must not touch the shared T&E schema") + + def test_te_table_counts_are_per_tenant(self): + def count(who): + tables = self.client.get("/api/te/tables", headers=self.headers[who]).json() + self.assertTrue(all(t["exists"] for t in tables), tables) # all 12 provisioned + return next(t["rowCount"] for t in tables if t["table"] == "organisations") + + before_b = count("b") + self._te_upload("a", f"ORG_{uuid.uuid4().hex[:10]}") + self.assertEqual(count("b"), before_b) + + def test_tenant_te_tables_keep_their_foreign_keys_and_triggers(self): + schema = schema_for(self.slugs["b"]) + with Conn() as conn, conn.cursor() as cur: + cur.execute("SELECT count(*) FROM information_schema.table_constraints " + "WHERE table_schema = %s AND constraint_type = 'FOREIGN KEY'", (schema,)) + tenant_fks = cur.fetchone()[0] + cur.execute("SELECT count(*) FROM information_schema.table_constraints " + "WHERE table_schema = %s AND constraint_type = 'FOREIGN KEY'", (settings.TE_SCHEMA,)) + self.assertEqual(tenant_fks, cur.fetchone()[0]) + cur.execute("SELECT count(*) FROM information_schema.referential_constraints rc " + "JOIN information_schema.table_constraints tc USING (constraint_schema, constraint_name) " + "WHERE rc.constraint_schema = %s AND rc.unique_constraint_schema <> %s", (schema, schema)) + self.assertEqual(cur.fetchone()[0], 0, "foreign keys must point inside the tenant schema") + cur.execute("SELECT count(*) FROM information_schema.triggers WHERE trigger_schema = %s", (schema,)) + self.assertGreater(cur.fetchone()[0], 0) + with self.assertRaises(psycopg2.errors.ForeignKeyViolation): + cur.execute(sql.SQL( + "INSERT INTO {}.personnel (org_id, full_name, email, te_role, password_hash) " + "VALUES (gen_random_uuid(), 'x', %s, 'observer', 'x')").format(sql.Identifier(schema)), + (f"{uuid.uuid4().hex}@example.com",)) + conn.rollback() def test_invalid_session_is_401_not_a_fallback_to_default(self): res = self.client.get("/api/csv/files", headers={"Authorization": "Bearer not-a-token"}) @@ -171,3 +219,10 @@ def visible(who): results = list(pool.map(visible, ["a", "b", None] * 20)) for who, seen in results: self.assertEqual(seen, {marks[who]} if who else set(), who) + + def test_provisioning_is_idempotent(self): + from api.tenancy import provision_te_tables + + with Conn() as conn, conn.cursor() as cur: + self.assertEqual(provision_te_tables(cur, schema_for(self.slugs["a"]), settings.TE_SCHEMA), 0) + conn.rollback()