Add a real pytest suite, gating the CI build
64 tests covering: auth/setup and the CSRF gate, the "owner adds their own data, anyone linked can mute it" permission pattern shared across calendar-select/tasks-source/whiteboard-source, migration correctness (fresh install, idempotent re-run, expected columns), battery estimate outlier rejection, calendar_feed's fetch/merge/partial-failure handling, webdav_client's fetch/list-directory, the whiteboard force-refresh throttle bypass and browse endpoint, and render-size invariants across calendar views/orientations. No DB/HTTP fixtures need Docker, Node, or a real Immich/CalDAV/WebDAV server -- a fresh temp SQLite file plus a couple of small local HTTP servers as test doubles cover it all. Table data is wiped and reseeded between tests rather than relying on SQLAlchemy's transaction-rollback isolation pattern, which needs a pysqlite event-listener workaround app/db.py's engine doesn't have and has no reason to gain just for tests. Wired into .gitea/workflows/server-docker-build.yml as its own job that build-and-push now depends on, so a failing suite blocks the image push rather than just running alongside it for show.
This commit is contained in:
@@ -0,0 +1,140 @@
|
||||
"""Shared pytest fixtures for the server test suite.
|
||||
|
||||
DATABASE_URL must be set before app.db (and anything importing it,
|
||||
transitively including app.main) is first imported -- app.db builds its
|
||||
engine/SessionLocal at module import time, not lazily -- so this file
|
||||
sets it as the very first thing it does, ahead of any `from app...`
|
||||
import below. Importing app.main also runs migration.run_migrations()
|
||||
as a side effect of that import (see main.py), which is what actually
|
||||
creates the schema in the fresh temp database this points at.
|
||||
|
||||
Each test shares one migrated schema (re-migrating per test would be
|
||||
needless I/O), but gets a clean slate of *data*: every table is wiped
|
||||
after each test rather than relying on SQLAlchemy's transaction-rollback
|
||||
test-isolation pattern (Session bound to a connection-level transaction
|
||||
via join_transaction_mode="create_savepoint") -- that pattern needs the
|
||||
"pysqlite serializable" event-listener workaround (see SQLAlchemy's own
|
||||
docs on pysqlite's implicit-transaction quirks) that app/db.py's engine
|
||||
doesn't set up, and this suite has no reason to add production-affecting
|
||||
engine config just to make tests work. Deleting tables in reverse
|
||||
dependency order (children before parents) satisfies foreign keys
|
||||
without needing that workaround at all.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
_tmp_dir = tempfile.mkdtemp(prefix="espresso_frame_tests_")
|
||||
os.environ["DATABASE_URL"] = f"sqlite:///{Path(_tmp_dir) / 'test.db'}"
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app import db as db_module
|
||||
from app import migration
|
||||
from app.auth import hash_password
|
||||
from app.main import app
|
||||
from app.models import Base, Frame, User, UserFrame
|
||||
|
||||
|
||||
def _reset_db() -> None:
|
||||
"""Wipes every ORM-mapped table, then reseeds the same baseline a
|
||||
real fresh install gets (frame #1 + the server-settings singleton --
|
||||
see migration.py's _ensure_frame_one/_ensure_server_settings, both
|
||||
idempotent and both already called by run_migrations). The raw
|
||||
schema_version table isn't part of Base.metadata (see migration.py),
|
||||
so it survives the wipe untouched and run_migrations() skips
|
||||
straight to that reseed step instead of re-running every ALTER."""
|
||||
with db_module.engine.begin() as conn:
|
||||
for table in reversed(Base.metadata.sorted_tables):
|
||||
conn.execute(table.delete())
|
||||
migration.run_migrations()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_session():
|
||||
session = db_module.SessionLocal()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
_reset_db()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(db_session):
|
||||
"""A TestClient whose every request shares this test's own db_session
|
||||
-- so anything the test asserts against db_session sees exactly what
|
||||
the app just did. Table data from this test is wiped once db_session
|
||||
tears down (see _wipe_all_tables above)."""
|
||||
|
||||
def _override_get_db():
|
||||
yield db_session
|
||||
|
||||
app.dependency_overrides[db_module.get_db] = _override_get_db
|
||||
# follow_redirects=False: nearly every POST route in this app answers
|
||||
# success with a 303 (POST/redirect/GET) -- tests assert against that
|
||||
# 303 directly, matching what a real browser's network tab would show
|
||||
# before it follows the redirect itself.
|
||||
with TestClient(app, follow_redirects=False) as c:
|
||||
yield c
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
def make_user(db: Session, username: str, password: str = "testpass123", **kwargs) -> User:
|
||||
"""A user row directly via the ORM -- bypasses the HTTP signup/claim
|
||||
flow for tests that only care about what happens once a user already
|
||||
exists (permission boundaries, source ownership, etc.)."""
|
||||
import time as _time
|
||||
|
||||
user = User(
|
||||
username=username,
|
||||
display_name=kwargs.pop("display_name", username.capitalize()),
|
||||
password_hash=hash_password(password),
|
||||
created_at=_time.time(),
|
||||
**kwargs,
|
||||
)
|
||||
db.add(user)
|
||||
db.flush()
|
||||
return user
|
||||
|
||||
|
||||
def link_user(db: Session, user: User, frame: Frame) -> None:
|
||||
db.add(UserFrame(user_id=user.id, frame_id=frame.id))
|
||||
db.flush()
|
||||
|
||||
|
||||
def login(client: TestClient, username: str, password: str = "testpass123") -> None:
|
||||
resp = client.post("/login", data={"username": username, "password": password})
|
||||
assert resp.status_code == 303, resp.text
|
||||
|
||||
|
||||
def get_csrf_token(client: TestClient, page_url: str) -> str:
|
||||
"""Scrapes the csrf_token hidden field out of a rendered page -- the
|
||||
same value a real browser's form submit would carry, see auth.py's
|
||||
_csrf_ok."""
|
||||
import re
|
||||
|
||||
resp = client.get(page_url)
|
||||
assert resp.status_code == 200, resp.text
|
||||
m = re.search(r'name="csrf_token" value="([^"]+)"', resp.text)
|
||||
assert m, f"no csrf_token found on {page_url}"
|
||||
return m.group(1)
|
||||
|
||||
|
||||
def csrf_headers(client: TestClient, page_url: str = "/settings") -> dict:
|
||||
"""X-CSRF-Token header for JSON API POSTs (require_user_api's
|
||||
_csrf_ok checks this header, not a form field -- see common.js'
|
||||
fetchJson, which reads the same <meta name="csrf-token"> tag this
|
||||
scrapes)."""
|
||||
import re
|
||||
|
||||
resp = client.get(page_url)
|
||||
assert resp.status_code == 200, resp.text
|
||||
m = re.search(r'name="csrf-token" content="([^"]+)"', resp.text)
|
||||
assert m, f"no csrf-token meta tag found on {page_url}"
|
||||
return {"X-CSRF-Token": m.group(1)}
|
||||
Reference in New Issue
Block a user