feat: DSH编排基建首版（server/tests 88用例/P2插件/8020端口/PORT-NOTE/MOBILE_APP）

2026-09-12 14:20:02 +08:00
commit f6acd20325
128 changed files with 24471 additions and 0 deletions
@@ -0,0 +1,117 @@
+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']}"}