From 7b94813549aa0b85d8c0a5e97da4f9c7f3efa00a Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 18:16:36 +1100 Subject: [PATCH] feat(auth): control schema, local accounts and sign-in sessions (M2a, #92) Migration 0004 adds a control schema (name from CONTROL_SCHEMA) with tenants, users, memberships (the four ADR-0003 roles) and sessions. Passwords use stdlib scrypt; session tokens are random, returned once and stored as SHA-256 hashes. POST /api/auth/login, POST /api/auth/logout and GET /api/auth/me; require_session is ready for the data routers (#93, #94) but not applied to them yet. scripts/create_tenant_admin.py creates a tenant and its first admin. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019YPoXzBSj89ZHgw7nfTmbB --- alembic/versions/0004_control_schema.py | 62 +++++++++++ api/auth.py | 25 ++++- api/config.py | 4 + api/main.py | 5 +- api/routers/auth_routes.py | 48 +++++++++ api/services/accounts.py | 114 ++++++++++++++++++++ docs/ENV_VARIABLES.md | 2 + scripts/create_tenant_admin.py | 50 +++++++++ scripts/lint.sh | 6 ++ tests/test_accounts.py | 135 ++++++++++++++++++++++++ tests/test_migrations.py | 22 +++- 11 files changed, 466 insertions(+), 7 deletions(-) create mode 100644 alembic/versions/0004_control_schema.py create mode 100644 api/routers/auth_routes.py create mode 100644 api/services/accounts.py create mode 100644 scripts/create_tenant_admin.py create mode 100644 tests/test_accounts.py diff --git a/alembic/versions/0004_control_schema.py b/alembic/versions/0004_control_schema.py new file mode 100644 index 0000000..78b5a3c --- /dev/null +++ b/alembic/versions/0004_control_schema.py @@ -0,0 +1,62 @@ +"""Control schema: tenants, users, memberships and sessions (M2a, #92, BR-36). + +Shared by every tenant. Roles are the four from ADR-0003. Session tokens are +stored only as SHA-256 hashes, so a database read does not yield usable tokens. + +Revision ID: 0004 +Revises: 0003 +Create Date: 2026-10-26 +""" + +from alembic import op + +from api.config import settings + +revision = "0004" +down_revision = "0003" +branch_labels = None +depends_on = None + + +def _schema() -> str: + return op.get_bind().dialect.identifier_preparer.quote(settings.CONTROL_SCHEMA) + + +def upgrade() -> None: + s = _schema() + op.execute(f"CREATE SCHEMA IF NOT EXISTS {s}") + op.execute(f""" + CREATE TABLE {s}.tenants ( + id BIGSERIAL PRIMARY KEY, + slug TEXT NOT NULL UNIQUE CHECK (slug ~ '^[a-z][a-z0-9_]{{1,40}}$'), + name TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + )""") + op.execute(f""" + CREATE TABLE {s}.users ( + id BIGSERIAL PRIMARY KEY, + username TEXT NOT NULL UNIQUE, + password_hash TEXT NOT NULL, + is_active BOOLEAN NOT NULL DEFAULT true, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() + )""") + op.execute(f""" + CREATE TABLE {s}.memberships ( + user_id BIGINT NOT NULL REFERENCES {s}.users(id) ON DELETE CASCADE, + tenant_id BIGINT NOT NULL REFERENCES {s}.tenants(id) ON DELETE CASCADE, + role TEXT NOT NULL CHECK (role IN ('uploader', 'reviewer', 'tenant_admin', 'auditor')), + PRIMARY KEY (user_id, tenant_id) + )""") + op.execute(f""" + CREATE TABLE {s}.sessions ( + token_hash TEXT PRIMARY KEY, + user_id BIGINT NOT NULL REFERENCES {s}.users(id) ON DELETE CASCADE, + tenant_id BIGINT NOT NULL REFERENCES {s}.tenants(id) ON DELETE CASCADE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + expires_at TIMESTAMPTZ NOT NULL, + revoked_at TIMESTAMPTZ + )""") + + +def downgrade() -> None: + op.execute(f"DROP SCHEMA IF EXISTS {_schema()} CASCADE") diff --git a/api/auth.py b/api/auth.py index 71dccef..d164bf1 100644 --- a/api/auth.py +++ b/api/auth.py @@ -13,9 +13,11 @@ import secrets -from fastapi import Header, HTTPException +from fastapi import Depends, Header, HTTPException from api.config import settings +from api.db import Conn +from api.services.accounts import session_for def require_api_key(x_api_key: str = Header(default="")) -> None: @@ -24,3 +26,24 @@ def require_api_key(x_api_key: str = Header(default="")) -> None: return if not secrets.compare_digest(x_api_key, settings.API_KEY): raise HTTPException(status_code=401, detail="Missing or invalid X-API-Key header") + + +def bearer_token(authorization: str = Header(default="")) -> str: + """The token from an ``Authorization: Bearer `` header, or 401.""" + scheme, _, token = authorization.partition(" ") + if scheme.lower() != "bearer" or not token.strip(): + raise HTTPException(status_code=401, detail="Sign in required") + return token.strip() + + +def require_session(token: str = Depends(bearer_token)) -> dict: + """The caller's live session (user, tenant, role), or 401 (M2a, #92). + + Not yet applied to the data routers; tenant routing (#93) and the role + matrix (#94) switch them over. + """ + with Conn() as conn, conn.cursor() as cur: + session = session_for(cur, token) + if session is None: + raise HTTPException(status_code=401, detail="Session expired or invalid; sign in again") + return session diff --git a/api/config.py b/api/config.py index 6e54a0c..b169c7f 100644 --- a/api/config.py +++ b/api/config.py @@ -12,6 +12,10 @@ class Settings: # Schema where dynamically created per-CSV tables + the registry live. UPLOADS_SCHEMA: str = os.environ.get("CSV_UPLOADS_SCHEMA", "csv_uploads") + # M2 (#92): tenants, users, memberships and sessions, shared by all tenants. + CONTROL_SCHEMA: str = os.environ.get("CONTROL_SCHEMA", "control") + # Lifetime of a sign-in session, in hours. + SESSION_TTL_HOURS: int = int(os.environ.get("SESSION_TTL_HOURS", "12")) # Schema holding the fixed T&E tables (deploy_all.sh dev creates te_dev). TE_SCHEMA: str = os.environ.get("TE_SCHEMA", "te_dev") diff --git a/api/main.py b/api/main.py index 56a3301..af18112 100644 --- a/api/main.py +++ b/api/main.py @@ -18,7 +18,7 @@ from api import db from api.config import settings from api.ratelimit import rate_limit_middleware -from api.routers import audit_routes, csv_routes, te_routes +from api.routers import audit_routes, auth_routes, csv_routes, te_routes logger = logging.getLogger(__name__) @@ -86,7 +86,7 @@ async def lifespan(app: FastAPI): CORSMiddleware, allow_origins=settings.CORS_ORIGINS, allow_methods=["GET", "POST", "DELETE"], - allow_headers=["Content-Type", "X-API-Key"], + allow_headers=["Content-Type", "X-API-Key", "Authorization"], ) # O-03: rate limiting. Registered after CORSMiddleware so it sits outside it in @@ -94,6 +94,7 @@ async def lifespan(app: FastAPI): # which the browser needs in order to read the status at all. app.middleware("http")(rate_limit_middleware) +app.include_router(auth_routes.router) app.include_router(csv_routes.router) app.include_router(te_routes.router) app.include_router(audit_routes.router) diff --git a/api/routers/auth_routes.py b/api/routers/auth_routes.py new file mode 100644 index 0000000..280cdb2 --- /dev/null +++ b/api/routers/auth_routes.py @@ -0,0 +1,48 @@ +"""Sign-in with local accounts (M2a, #92, BR-36). + +Sessions are bearer tokens: ``Authorization: Bearer ``. Login failures +all return the same 401 so a caller cannot probe for valid usernames or tenants. +The global rate limiter's write budget applies to login attempts. +""" + +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, Field + +from api.auth import bearer_token, require_session +from api.db import Conn +from api.services import accounts + +router = APIRouter(prefix="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/api/auth", tags=["auth"]) + + +class LoginRequest(BaseModel): + tenant: str = Field(min_length=1, max_length=64) + username: str = Field(min_length=1, max_length=128) + password: str = Field(min_length=1, max_length=1024) + + +@router.post("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/login") +def login(req: LoginRequest) -> dict: + with Conn() as conn: + with conn.cursor() as cur: + session = accounts.login(cur, req.tenant, req.username, req.password) + conn.commit() + if session is None: + raise HTTPException(status_code=401, detail="Invalid tenant, username or password") + return session + + +@router.post("/logout") +def logout(token: str = Depends(bearer_token)) -> dict: + with Conn() as conn: + with conn.cursor() as cur: + accounts.revoke(cur, token) + conn.commit() + return {"status": "ok"} + + +@router.get("/me") +def me(session: dict = Depends(require_session)) -> dict: + return session diff --git a/api/services/accounts.py b/api/services/accounts.py new file mode 100644 index 0000000..e9e28b2 --- /dev/null +++ b/api/services/accounts.py @@ -0,0 +1,114 @@ +"""Local accounts and sign-in sessions (M2a, #92, BR-36). + +Passwords use stdlib scrypt, so air-gapped installs need no extra package. +Session tokens are random, returned once, and stored only as SHA-256 hashes. +""" + +from __future__ import annotations + +import hashlib +import hmac +import secrets + +from psycopg2 import sql + +from api.config import settings + +ROLES = ("uploader", "reviewer", "tenant_admin", "auditor") +_N, _R, _P = 2**14, 8, 1 + + +def hash_password(password: str) -> str: + salt = secrets.token_bytes(16) + dk = hashlib.scrypt(password.encode(), salt=salt, n=_N, r=_R, p=_P) + return f"scrypt${_N}${_R}${_P}${salt.hex()}${dk.hex()}" + + +def verify_password(password: str, stored: str) -> bool: + try: + algo, n, r, p, salt, want = stored.split("$") + if algo != "scrypt": + return False + dk = hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt), n=int(n), r=int(r), p=int(p)) + except (ValueError, TypeError): + return False + return hmac.compare_digest(dk.hex(), want) + + +# Checked when the username is unknown, so a miss costs the same time as a hit. +_DUMMY_HASH = hash_password(secrets.token_hex(8)) + + +def _token_hash(token: str) -> str: + return hashlib.sha256(token.encode()).hexdigest() + + +def _q(query: str) -> sql.Composed: + return sql.SQL(query).format(c=sql.Identifier(settings.CONTROL_SCHEMA)) + + +def create_tenant(cur, slug: str, name: str) -> int: + cur.execute(_q("INSERT INTO {c}.tenants (slug, name) VALUES (%s, %s) RETURNING id"), (slug, name)) + return cur.fetchone()[0] + + +def create_user(cur, username: str, password: str) -> int: + cur.execute(_q("INSERT INTO {c}.users (username, password_hash) VALUES (%s, %s) RETURNING id"), + (username, hash_password(password))) + return cur.fetchone()[0] + + +def add_membership(cur, user_id: int, tenant_id: int, role: str) -> None: + if role not in ROLES: + raise ValueError(f"Unknown role {role!r}; expected one of {ROLES}") + cur.execute(_q("INSERT INTO {c}.memberships (user_id, tenant_id, role) VALUES (%s, %s, %s)"), + (user_id, tenant_id, role)) + + +def login(cur, tenant_slug: str, username: str, password: str) -> dict | None: + """Return a new session for an active member of the tenant, else None. + + The caller commits. Wrong tenant, unknown user, wrong password and inactive + user all return None, so a caller cannot tell which one failed. + """ + cur.execute(_q( + "SELECT u.id, u.password_hash, u.is_active, t.id, m.role " + "FROM {c}.users u " + "LEFT JOIN {c}.tenants t ON t.slug = %s " + "LEFT JOIN {c}.memberships m ON m.user_id = u.id AND m.tenant_id = t.id " + "WHERE u.username = %s"), (tenant_slug, username)) + row = cur.fetchone() + ok = verify_password(password, row[1] if row else _DUMMY_HASH) + if not (row and ok and row[2] and row[4]): + return None + user_id, tenant_id, role = row[0], row[3], row[4] + token = secrets.token_urlsafe(32) + cur.execute(_q( + "INSERT INTO {c}.sessions (token_hash, user_id, tenant_id, expires_at) " + "VALUES (%s, %s, %s, now() + make_interval(hours => %s)) RETURNING expires_at"), + (_token_hash(token), user_id, tenant_id, settings.SESSION_TTL_HOURS)) + expires_at = cur.fetchone()[0] + return {"token": token, "expiresAt": expires_at.isoformat(), "username": username, + "tenant": tenant_slug, "role": role} + + +def session_for(cur, token: str) -> dict | None: + """The live session for a token, with its user, tenant and role, else None.""" + cur.execute(_q( + "SELECT u.id, u.username, t.id, t.slug, m.role, s.expires_at " + "FROM {c}.sessions s " + "JOIN {c}.users u ON u.id = s.user_id AND u.is_active " + "JOIN {c}.tenants t ON t.id = s.tenant_id " + "JOIN {c}.memberships m ON m.user_id = u.id AND m.tenant_id = t.id " + "WHERE s.token_hash = %s AND s.revoked_at IS NULL AND s.expires_at > now()"), + (_token_hash(token),)) + row = cur.fetchone() + if not row: + return None + return {"userId": row[0], "username": row[1], "tenantId": row[2], "tenant": row[3], + "role": row[4], "expiresAt": row[5].isoformat()} + + +def revoke(cur, token: str) -> None: + cur.execute(_q("UPDATE {c}.sessions SET revoked_at = now() " + "WHERE token_hash = %s AND revoked_at IS NULL"), (_token_hash(token),)) diff --git a/docs/ENV_VARIABLES.md b/docs/ENV_VARIABLES.md index 9061a52..64100d1 100644 --- a/docs/ENV_VARIABLES.md +++ b/docs/ENV_VARIABLES.md @@ -39,6 +39,8 @@ Read by `api/config.py`. Set these in the shell before running | `API_POOL_GETCONN_TIMEOUT` | `5.0` | No | Seconds a request waits to borrow a database connection from the pool. Exceeded = 503 response. | | `CORS_ORIGINS` | `http://localhost:8080,http://localhost:5173,http://localhost:3000` | Yes | Comma-separated list of allowed CORS origins. Set to your frontend URL in production. | | `CSV_UPLOADS_SCHEMA` | `csv_uploads` | No | Schema where per-CSV tables and the file registry live. | +| `CONTROL_SCHEMA` | `control` | No | Schema for tenants, users, memberships and sign-in sessions (M2, #92). | +| `SESSION_TTL_HOURS` | `12` | No | How long a sign-in session lasts before the user must sign in again. | | `TE_SCHEMA` | `te_dev` | No | Schema holding the 12 fixed T&E tables. Switch to `te_test`, `te_staging`, or `te_prod` per environment. | | `MAX_UPLOAD_BYTES` | `52428800` (50 MB) | No | Maximum file size accepted by the upload endpoint. | diff --git a/scripts/create_tenant_admin.py b/scripts/create_tenant_admin.py new file mode 100644 index 0000000..7945620 --- /dev/null +++ b/scripts/create_tenant_admin.py @@ -0,0 +1,50 @@ +"""Create a tenant and its first tenant admin (M2a, #92). + +Run from the repo root with the API's PG* environment set: + python scripts/create_tenant_admin.py --tenant acme --name "Acme Pty Ltd" --username alice + +The password is read from TE_ADMIN_PASSWORD if set, otherwise prompted for. +Brings the database to the current migration first. +""" + +from __future__ import annotations + +import argparse +import getpass +import os +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from api import db # noqa: E402 +from api.services import accounts # noqa: E402 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--tenant", required=True, help="tenant slug: lowercase letters, digits, underscore") + ap.add_argument("--name", required=True, help="tenant display name") + ap.add_argument("--username", required=True) + args = ap.parse_args() + password = os.environ.get("TE_ADMIN_PASSWORD") or getpass.getpass("Password for the new admin: ") + if len(password) < 12: + print("Password must be at least 12 characters.", file=sys.stderr) + return 2 + db.init_pool() + db.bootstrap() + try: + with db.Conn() as conn: + with conn.cursor() as cur: + tenant_id = accounts.create_tenant(cur, args.tenant, args.name) + user_id = accounts.create_user(cur, args.username, password) + accounts.add_membership(cur, user_id, tenant_id, "tenant_admin") + conn.commit() + finally: + db.close_pool() + print(f"Created tenant '{args.tenant}' with tenant admin '{args.username}'.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/lint.sh b/scripts/lint.sh index a83313c..7efab81 100644 --- a/scripts/lint.sh +++ b/scripts/lint.sh @@ -17,6 +17,11 @@ PY_SRC=( alembic/versions/0001_baseline.py alembic/versions/0002_registry_unique.py alembic/versions/0003_import_counts.py + alembic/versions/0004_control_schema.py + api/auth.py + api/services/accounts.py + api/routers/auth_routes.py + scripts/create_tenant_admin.py ) PY_TESTS=( @@ -30,6 +35,7 @@ PY_TESTS=( tests/test_evals_tier_e_x.py tests/test_migrations.py tests/test_import_counts.py + tests/test_accounts.py ) # Collect scripts/*.py dynamically diff --git a/tests/test_accounts.py b/tests/test_accounts.py new file mode 100644 index 0000000..6024673 --- /dev/null +++ b/tests/test_accounts.py @@ -0,0 +1,135 @@ +"""Local accounts and sign-in sessions (M2a, #92, BR-36). + +Unit tests for password hashing; integration tests for /api/auth over a live +PostgreSQL (they fail, never skip, without one). +""" + +from __future__ import annotations + +import os +import unittest +import uuid + +import pytest +from fastapi.testclient import TestClient +from psycopg2 import sql + +from api.config import settings +from api.db import Conn +from api.main import app +from api.services import accounts + +_HELP = ("Start PostgreSQL and ensure api/ can reach it (PGHOST/PGPORT/PGUSER/" + "PGPASSWORD/PGDATABASE). See scripts/start-api.ps1.") +_PW = "correct horse battery staple" + + +@pytest.mark.unit +class Passwords(unittest.TestCase): + def test_round_trip(self): + self.assertTrue(accounts.verify_password(_PW, accounts.hash_password(_PW))) + + def test_wrong_password_fails(self): + self.assertFalse(accounts.verify_password("nope", accounts.hash_password(_PW))) + + def test_same_password_hashes_differently(self): + self.assertNotEqual(accounts.hash_password(_PW), accounts.hash_password(_PW)) + + def test_malformed_hash_fails_closed(self): + for bad in ("", "plain", "md5$1$2$3$aa$bb", "scrypt$x$8$1$aa$bb"): + self.assertFalse(accounts.verify_password(_PW, bad), bad) + + +@pytest.mark.integration +class SignIn(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.ctx = TestClient(app) + try: + cls.client = cls.ctx.__enter__() + except Exception as exc: # noqa: BLE001 + raise AssertionError(f"Could not start the API: {type(exc).__name__}: {exc}\n{_HELP}") from None + + @classmethod + def tearDownClass(cls): + cls.ctx.__exit__(None, None, None) + + def setUp(self): + tag = f"{os.getpid()}_{uuid.uuid4().hex[:8]}" + self.tenant, self.other, self.user = f"t_{tag}", f"o_{tag}", f"user_{tag}" + with Conn() as conn: + with conn.cursor() as cur: + tid = accounts.create_tenant(cur, self.tenant, "Test tenant") + accounts.create_tenant(cur, self.other, "Other tenant") + self.user_id = accounts.create_user(cur, self.user, _PW) + accounts.add_membership(cur, self.user_id, tid, "uploader") + conn.commit() + + def _exec(self, query: str, *params) -> None: + with Conn() as conn: + with conn.cursor() as cur: + cur.execute(sql.SQL(query).format(c=sql.Identifier(settings.CONTROL_SCHEMA)), params) + conn.commit() + + def _login(self, tenant=None, username=None, password=_PW): + return self.client.post("/api/auth/login", json={ + "tenant": tenant or self.tenant, "username": username or self.user, "password": password}) + + def _me(self, token: str): + return self.client.get("/api/auth/me", headers={"Authorization": f"Bearer {token}"}) + + def test_login_returns_a_session_for_the_tenant_and_role(self): + res = self._login() + self.assertEqual(res.status_code, 200, res.text) + body = res.json() + self.assertEqual((body["tenant"], body["username"], body["role"]), (self.tenant, self.user, "uploader")) + me = self._me(body["token"]).json() + self.assertEqual((me["tenant"], me["username"], me["role"]), (self.tenant, self.user, "uploader")) + + def test_every_login_failure_is_the_same_401(self): + cases = { + "wrong password": self._login(password="wrong"), + "unknown user": self._login(username="nobody_" + uuid.uuid4().hex), + "not a member of that tenant": self._login(tenant=self.other), + "unknown tenant": self._login(tenant="no_such_tenant"), + } + for name, res in cases.items(): + self.assertEqual((res.status_code, res.json()["detail"]), + (401, "Invalid tenant, username or password"), name) + + def test_inactive_user_cannot_sign_in_and_loses_live_sessions(self): + token = self._login().json()["token"] + self._exec("UPDATE {c}.users SET is_active = false WHERE id = %s", self.user_id) + self.assertEqual(self._login().status_code, 401) + self.assertEqual(self._me(token).status_code, 401) + + def test_logout_revokes_the_session(self): + token = self._login().json()["token"] + headers = {"Authorization": f"Bearer {token}"} + self.assertEqual(self.client.post("/api/auth/logout", headers=headers).status_code, 200) + self.assertEqual(self._me(token).status_code, 401) + + def test_expired_session_is_rejected(self): + token = self._login().json()["token"] + self._exec("UPDATE {c}.sessions SET expires_at = now() - interval '1 second' WHERE user_id = %s", + self.user_id) + self.assertEqual(self._me(token).status_code, 401) + + def test_removing_the_membership_ends_the_session(self): + token = self._login().json()["token"] + self._exec("DELETE FROM {c}.memberships WHERE user_id = %s", self.user_id) + self.assertEqual(self._me(token).status_code, 401) + + def test_missing_or_malformed_token_is_401(self): + for headers in ({}, {"Authorization": "Bearer "}, {"Authorization": "Basic abc"}, + {"Authorization": "Bearer not-a-real-token"}): + self.assertEqual(self.client.get("/api/auth/me", headers=headers).status_code, 401, headers) + + def test_tokens_are_stored_only_as_hashes(self): + token = self._login().json()["token"] + with Conn() as conn, conn.cursor() as cur: + cur.execute(sql.SQL("SELECT token_hash FROM {c}.sessions WHERE user_id = %s").format( + c=sql.Identifier(settings.CONTROL_SCHEMA)), (self.user_id,)) + stored = [r[0] for r in cur.fetchall()] + self.assertEqual(len(stored), 1) + self.assertNotIn(token, stored) diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 87bcb6c..fcd4e5f 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -50,8 +50,10 @@ class MigrationsOnScratchSchema(unittest.TestCase): """Upgrade and downgrade run against a throwaway schema, never the live one.""" def setUp(self): - self.original = settings.UPLOADS_SCHEMA - settings.UPLOADS_SCHEMA = f"mig_test_{uuid.uuid4().hex[:10]}" + self.original = (settings.UPLOADS_SCHEMA, settings.CONTROL_SCHEMA) + tag = uuid.uuid4().hex[:10] + settings.UPLOADS_SCHEMA = f"mig_test_{tag}" + settings.CONTROL_SCHEMA = f"mig_ctl_{tag}" # never touch the live control schema self.cfg = Config(_ALEMBIC_INI) self.conn = _connect() self.conn.autocommit = True @@ -59,8 +61,9 @@ def setUp(self): def tearDown(self): with self.conn.cursor() as cur: cur.execute(f'DROP SCHEMA IF EXISTS "{settings.UPLOADS_SCHEMA}" CASCADE') + cur.execute(f'DROP SCHEMA IF EXISTS "{settings.CONTROL_SCHEMA}" CASCADE') self.conn.close() - settings.UPLOADS_SCHEMA = self.original + settings.UPLOADS_SCHEMA, settings.CONTROL_SCHEMA = self.original def _constraints(self) -> set[str]: with self.conn.cursor() as cur: @@ -80,6 +83,17 @@ def test_upgrade_head_creates_tables_and_unique_constraints(self): self.assertTrue({"csv_files", "audit_log", "alembic_version"} <= self._tables()) self.assertEqual(self._constraints(), {"csv_files_file_hash_key", "csv_files_file_name_key"}) + def test_control_schema_is_created_and_removed(self): + command.upgrade(self.cfg, "head") + with self.conn.cursor() as cur: + cur.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = %s", + (settings.CONTROL_SCHEMA,)) + self.assertEqual({r[0] for r in cur.fetchall()}, {"tenants", "users", "memberships", "sessions"}) + command.downgrade(self.cfg, "0003") + with self.conn.cursor() as cur: + cur.execute("SELECT 1 FROM pg_namespace WHERE nspname = %s", (settings.CONTROL_SCHEMA,)) + self.assertIsNone(cur.fetchone()) + def test_downgrade_reverses_each_step(self): command.upgrade(self.cfg, "head") command.downgrade(self.cfg, "0001") @@ -134,7 +148,7 @@ def setUp(self): def test_bootstrap_brings_the_schema_to_head(self): with Conn() as conn, conn.cursor() as cur: cur.execute(f'SELECT version_num FROM "{settings.UPLOADS_SCHEMA}".alembic_version') - self.assertEqual(cur.fetchone()[0], "0003") + self.assertEqual(cur.fetchone()[0], "0004") def test_duplicate_hash_is_rejected_by_the_database(self): insert = (f'INSERT INTO "{settings.UPLOADS_SCHEMA}".csv_files (file_name, file_hash, table_name, mode) '