118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
import pytest
|
|
|
|
SERVER_DIR = Path(__file__).resolve().parents[1]
|
|
if str(SERVER_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SERVER_DIR))
|
|
|
|
ADMIN_TOKEN = "admin-secret-token"
|
|
INVITE_CODE = "invite-123"
|
|
ADMIN_AUTH = {"Authorization": f"Bearer {ADMIN_TOKEN}"}
|
|
|
|
# The instance this repo ships with (start-server.bat / server-worker.cmd point
|
|
# at it) — used only to derive/verify the environment, never to store test data.
|
|
LOCAL_DSN = "postgresql://dsh@127.0.0.1:15433/postgres"
|
|
|
|
|
|
def _db_name(dsn: str) -> str:
|
|
return (urlsplit(dsn).path or "/").lstrip("/")
|
|
|
|
|
|
def _ensure_test_database(test_dsn: str) -> None:
|
|
"""Create the test database if it does not exist yet.
|
|
|
|
db.reset_schema() drops the whole `public` schema, so the suite needs a
|
|
database of its own; connecting to the instance's default database to
|
|
CREATE one is the only write this module performs on a non-test database.
|
|
"""
|
|
import psycopg
|
|
|
|
try:
|
|
psycopg.connect(test_dsn, connect_timeout=3).close()
|
|
return
|
|
except psycopg.OperationalError:
|
|
pass
|
|
|
|
admin_dsn = os.environ.get("DSH_DATABASE_URL", "").strip() or LOCAL_DSN
|
|
name = _db_name(test_dsn)
|
|
conn = psycopg.connect(admin_dsn, connect_timeout=3, autocommit=True)
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (name,))
|
|
if cur.fetchone() is None:
|
|
cur.execute(f'CREATE DATABASE "{name}"')
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _resolve_test_dsn() -> str:
|
|
"""Test DSN, guaranteed to be a different database from production.
|
|
|
|
Default is the production instance with `_test` appended to the database
|
|
name. Pointing the suite at the production database wipes live data on the
|
|
first test (reset_schema drops the schema), so that combination is refused
|
|
outright rather than trusted to a comment.
|
|
"""
|
|
prod_dsn = os.environ.get("DSH_DATABASE_URL", "").strip() or LOCAL_DSN
|
|
explicit = os.environ.get("DSH_TEST_DATABASE_URL", "").strip()
|
|
test_dsn = explicit or urlsplit(prod_dsn)._replace(
|
|
path="/" + _db_name(prod_dsn) + "_test"
|
|
).geturl()
|
|
|
|
same_instance = urlsplit(prod_dsn).netloc == urlsplit(test_dsn).netloc
|
|
if same_instance and _db_name(prod_dsn) == _db_name(test_dsn):
|
|
pytest.exit(
|
|
f"refusing to run: test database {test_dsn!r} is the production "
|
|
f"database. Set DSH_TEST_DATABASE_URL to a disposable database "
|
|
f"(the suite drops the `public` schema on every test).",
|
|
returncode=4,
|
|
)
|
|
_ensure_test_database(test_dsn)
|
|
return test_dsn
|
|
|
|
|
|
TEST_DATABASE_URL = _resolve_test_dsn()
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(monkeypatch):
|
|
psycopg = pytest.importorskip("psycopg")
|
|
|
|
try:
|
|
probe = psycopg.connect(TEST_DATABASE_URL, connect_timeout=3)
|
|
probe.close()
|
|
except Exception as e: # noqa: BLE001
|
|
pytest.skip(f"test PostgreSQL not reachable at {TEST_DATABASE_URL}: {e}")
|
|
|
|
tmp = tempfile.mkdtemp(prefix="dsh-test-")
|
|
monkeypatch.setenv("DSH_DATA_DIR", tmp)
|
|
monkeypatch.setenv("DSH_DATABASE_URL", TEST_DATABASE_URL)
|
|
monkeypatch.setenv("DSH_BOOTSTRAP_ADMIN_TOKEN", ADMIN_TOKEN)
|
|
monkeypatch.setenv("DSH_INVITE_CODE", INVITE_CODE)
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from dsh_sync.config import Config
|
|
from dsh_sync.db import reset_schema
|
|
from dsh_sync.main import create_app
|
|
|
|
cfg = Config.from_env()
|
|
reset_schema(cfg) # fresh, empty database per test
|
|
app = create_app(cfg)
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture()
|
|
def member_auth(client):
|
|
r = client.post("/v1/tokens", json={"username": "alice", "invite_code": INVITE_CODE})
|
|
assert r.status_code == 200, r.text
|
|
return {"Authorization": f"Bearer {r.json()['token']}"}
|