feat(tenancy): give each tenant its own T&E tables (#99)

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)

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_<slug>``. 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_<slug>``. 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)

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()