mivanchenko
b5c0fc8a5a
Adds the plumbing every other booking ticket builds on: resources, services, users, and password_reset_tokens tables, plus the clients columns (slug, timezone, auto_confirm, ics_token) and the bookings resource_id/EXCLUDE-constraint double-booking protection described in #14. booking_db.py is the only place raw SQL runs against these tables -- every function takes client_id and injects the tenant filter itself, and create/update_booking additionally verify the resource_id belongs to that client before writing, closing a guessed-ID cross-tenant hole. Tests spin up a real throwaway Postgres 16 container (matching prod) and exercise the EXCLUDE constraint, tenancy isolation, and password-reset single-use semantics end to end, per #14's "real Postgres, no mocking" testing decision. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
76 lines
2.2 KiB
Python
76 lines
2.2 KiB
Python
"""Spins up a throwaway Postgres 16 container (matching production) and
|
|
applies backoffice/db/init.sql against it, per the module's testing decision
|
|
(#14): real Postgres, no mocking, so the EXCLUDE constraint and tenancy
|
|
filters are exercised for real, not asserted by inspection.
|
|
"""
|
|
import atexit
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
|
|
import psycopg
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
_CONTAINER = f"smb-booking-test-db-{uuid.uuid4().hex[:8]}"
|
|
|
|
|
|
def _free_port():
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
return s.getsockname()[1]
|
|
|
|
|
|
def _wait_ready(url, timeout=30):
|
|
deadline = time.time() + timeout
|
|
last_err = None
|
|
while time.time() < deadline:
|
|
try:
|
|
with psycopg.connect(url, connect_timeout=2):
|
|
return
|
|
except psycopg.OperationalError as e:
|
|
last_err = e
|
|
time.sleep(0.5)
|
|
raise RuntimeError(f"test db never became ready: {last_err}")
|
|
|
|
|
|
def _start_db():
|
|
port = _free_port()
|
|
subprocess.run(
|
|
["docker", "run", "-d", "--rm", "--name", _CONTAINER,
|
|
"-e", "POSTGRES_DB=smbcrm_test",
|
|
"-e", "POSTGRES_USER=smbcrm",
|
|
"-e", "POSTGRES_PASSWORD=test",
|
|
"-p", f"127.0.0.1:{port}:5432",
|
|
"postgres:16-alpine"],
|
|
check=True, capture_output=True)
|
|
atexit.register(
|
|
lambda: subprocess.run(["docker", "stop", _CONTAINER], capture_output=True))
|
|
url = f"postgresql://smbcrm:test@127.0.0.1:{port}/smbcrm_test"
|
|
_wait_ready(url)
|
|
schema_path = os.path.join(os.path.dirname(__file__), "..", "..", "db", "init.sql")
|
|
with open(schema_path) as f:
|
|
schema = f.read()
|
|
with psycopg.connect(url) as conn, conn.cursor() as cur:
|
|
cur.execute(schema)
|
|
conn.commit()
|
|
return url
|
|
|
|
|
|
DATABASE_URL = _start_db()
|
|
os.environ["DATABASE_URL"] = DATABASE_URL
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_tables():
|
|
with psycopg.connect(DATABASE_URL) as conn, conn.cursor() as cur:
|
|
cur.execute(
|
|
"TRUNCATE resources, services, bookings, users, "
|
|
"password_reset_tokens RESTART IDENTITY CASCADE")
|
|
conn.commit()
|
|
yield
|