diff --git a/.gitea/workflows/server-docker-build.yml b/.gitea/workflows/server-docker-build.yml index 28fb4a2..86cfd85 100644 --- a/.gitea/workflows/server-docker-build.yml +++ b/.gitea/workflows/server-docker-build.yml @@ -8,7 +8,27 @@ on: - ".gitea/workflows/server-docker-build.yml" jobs: + test: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install dependencies + working-directory: server + run: pip install -r requirements-dev.txt + + - name: Run tests + working-directory: server + run: pytest + build-and-push: + needs: test runs-on: ubuntu-latest steps: - name: Checkout diff --git a/.gitignore b/.gitignore index 6e00562..894b91d 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,7 @@ server/**/__pycache__/ server/.venv/ server/*.egg-info/ server/data/ +server/.pytest_cache/ # render-service/ (whiteboard mode's Node sidecar) -- installed fresh # inside the Docker image, never committed. No package-lock.json exists # yet either (no Node/npm available in this project's dev environment to diff --git a/server/README.md b/server/README.md index d76df26..4cec5ca 100644 --- a/server/README.md +++ b/server/README.md @@ -303,3 +303,19 @@ CONFIG_PATH=./data/config.json uvicorn app.main:app --reload --host 0.0.0.0 --po `--host 0.0.0.0` matters here: without it, uvicorn defaults to `127.0.0.1` (localhost-only), which the ESP32 can't reach over the LAN. The Docker image already binds `0.0.0.0` by default. + +## Running tests + +``` +pip install -r requirements-dev.txt +pytest +``` + +Runs against a fresh temp SQLite database (`tests/conftest.py` sets +`DATABASE_URL` before anything imports `app.db`), with every table wiped +and reseeded (frame #1 + server settings, same as a real fresh install) +between tests -- no Docker, Node, or a real Immich/CalDAV/WebDAV server +needed; a few tests spin up small local HTTP servers as fixtures to +stand in for those. Also runs as its own job in +`.gitea/workflows/server-docker-build.yml`, gating the image build/push +-- a failing test suite blocks the push, not just decorates it. diff --git a/server/pytest.ini b/server/pytest.ini new file mode 100644 index 0000000..3f45d54 --- /dev/null +++ b/server/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +pythonpath = . +testpaths = tests +filterwarnings = + ignore::DeprecationWarning diff --git a/server/requirements-dev.txt b/server/requirements-dev.txt new file mode 100644 index 0000000..1441c92 --- /dev/null +++ b/server/requirements-dev.txt @@ -0,0 +1,2 @@ +-r requirements.txt +pytest==9.1.1 diff --git a/server/tests/__init__.py b/server/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/server/tests/conftest.py b/server/tests/conftest.py new file mode 100644 index 0000000..be952f1 --- /dev/null +++ b/server/tests/conftest.py @@ -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 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)} diff --git a/server/tests/test_auth_and_setup.py b/server/tests/test_auth_and_setup.py new file mode 100644 index 0000000..905570e --- /dev/null +++ b/server/tests/test_auth_and_setup.py @@ -0,0 +1,106 @@ +"""First-run setup, login, and the basic frame-visibility permission +gate (require_frame_view/can_view_frame) -- the things every other +endpoint's own permission test implicitly depends on already working.""" + +from __future__ import annotations + +from app.models import Frame + +from .conftest import get_csrf_token, link_user, login, make_user + + +def test_setup_creates_admin_and_claims_migrated_frame(client, db_session): + resp = client.post("/setup", data={ + "username": "alice", "password": "hunter22", "display_name": "Alice", + }) + assert resp.status_code == 303 + assert resp.headers["location"] == "/" + + frame = db_session.get(Frame, 1) + assert frame is not None + assert frame.owner_user_id is not None + assert frame.controlled_by_user_id is not None + + +def test_setup_only_works_once(client): + resp = client.post("/setup", data={"username": "alice", "password": "hunter22"}) + assert resp.status_code == 303 + + resp = client.post("/setup", data={"username": "mallory", "password": "hunter22"}) + assert resp.status_code == 403 + + +def test_login_requires_correct_password(client): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + client.cookies.clear() + + resp = client.post("/login", data={"username": "alice", "password": "wrong"}) + assert resp.status_code == 401 + + resp = client.post("/login", data={"username": "alice", "password": "hunter22"}) + assert resp.status_code == 303 + + +def test_root_redirects_to_setup_before_any_user_exists(client): + resp = client.get("/") + assert resp.status_code == 303 + assert resp.headers["location"] == "/setup" + + +def test_unauthenticated_request_redirects_to_login_once_a_user_exists(client): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + client.cookies.clear() + + resp = client.get("/") + assert resp.status_code == 303 + assert resp.headers["location"] == "/login" + + +def test_user_not_linked_to_a_frame_cannot_view_it(client, db_session): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + bob = make_user(db_session, "bob") + db_session.flush() + + client.cookies.clear() + login(client, "bob") + + resp = client.get("/frames/1") + assert resp.status_code == 404 + + +def test_linked_user_can_view_but_not_configure_by_default(client, db_session): + """Being linked grants view access; whether they can also *control* + (change settings/take the wheel) is a separate, narrower gate -- + require_frame_control, exercised via the permission-boundary tests + for individual features rather than here.""" + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + bob = make_user(db_session, "bob") + frame = db_session.get(Frame, 1) + link_user(db_session, bob, frame) + + client.cookies.clear() + login(client, "bob") + + resp = client.get("/frames/1") + assert resp.status_code == 200 + + +def test_csrf_token_required_for_settings_save(client): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + resp = client.post("/settings", data={ + "display_name": "Alice", "email": "alice@example.com", "csrf_token": "bogus", + }) + assert resp.status_code == 403 + + +def test_settings_save_round_trips_with_real_csrf_token(client, db_session): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + csrf = get_csrf_token(client, "/settings") + resp = client.post("/settings", data={ + "display_name": "Alice Smith", "email": "alice@example.com", "csrf_token": csrf, + }) + assert resp.status_code in (200, 303), resp.text + + from app.models import User + user = db_session.query(User).filter_by(username="alice").one() + assert user.display_name == "Alice Smith" diff --git a/server/tests/test_battery_estimate.py b/server/tests/test_battery_estimate.py new file mode 100644 index 0000000..3781944 --- /dev/null +++ b/server/tests/test_battery_estimate.py @@ -0,0 +1,71 @@ +"""_reject_outlier_drops -- the outlier-rejection pass in the battery +remaining-time estimate (see routers/common.py's battery_estimate_s). +Pure function, no DB/HTTP -- (recency_weight, drop_pct) pairs in, +filtered pairs out.""" + +from __future__ import annotations + +from app.routers.common import _reject_outlier_drops + + +def _steps(drops: list[float]) -> list[tuple[int, float]]: + return [(i + 1, d) for i, d in enumerate(drops)] + + +def _avg(steps: list[tuple[int, float]]) -> float: + total_weight = sum(w for w, _ in steps) + return sum(w * d for w, d in steps) / total_weight + + +def test_no_outlier_keeps_every_step(): + steps = _steps([1, 1, 2, 1, 1, 2, 1]) + assert _reject_outlier_drops(steps) == steps + + +def test_single_glitch_dip_is_rejected(): + """The exact shape reported in production: 18 ordinary 1%-per-wake + steps and one spliced-in 26% glitch -- the naive median-based MAD + degenerates to 0 here (more than half the steps tie at the median), + which used to let the glitch sail straight through untouched.""" + normal = [1] * 18 + glitchy = normal[:9] + [26] + normal[9:] + + kept = _reject_outlier_drops(_steps(glitchy)) + kept_drops = [d for _, d in kept] + assert 26 not in kept_drops + assert len(kept) == 18 + + # the whole point: the estimate should come out the same as if the + # glitch had never been recorded at all + baseline_avg = _avg(_steps(normal)) + filtered_avg = _avg(kept) + assert abs(filtered_avg - baseline_avg) < 1e-9 + + +def test_single_glitch_spike_is_rejected(): + normal = [2] * 18 + glitchy = normal[:5] + [40] + normal[5:] + + kept = _reject_outlier_drops(_steps(glitchy)) + kept_drops = [d for _, d in kept] + assert 40 not in kept_drops + assert len(kept) == 18 + + +def test_identical_steps_reject_nothing(): + """Every step tied at the exact same value -- both the median MAD + and the mean-absolute-deviation fallback are 0 here, which is the + one case _reject_outlier_drops explicitly bails out of rather than + filtering down to nothing.""" + steps = _steps([1] * 10) + assert _reject_outlier_drops(steps) == steps + + +def test_never_filters_down_to_nothing(): + """Even a genuinely bimodal series (half the wakes cheap, half + expensive -- not a single-glitch shape at all) shouldn't empty the + list; a battery_estimate_s caller treats an empty result as + "insufficient data," which a merely-noisy history isn't.""" + steps = _steps([1, 1, 1, 1, 10, 10, 10, 10]) + kept = _reject_outlier_drops(steps) + assert len(kept) > 0 diff --git a/server/tests/test_calendar_feed.py b/server/tests/test_calendar_feed.py new file mode 100644 index 0000000..f6645f2 --- /dev/null +++ b/server/tests/test_calendar_feed.py @@ -0,0 +1,157 @@ +"""calendar_feed.py's fetch/parse/merge against a local HTTP server +serving fixture .ics text -- pure functions, no ORM/FastAPI.""" + +from __future__ import annotations + +import threading +from datetime import date +from http.server import BaseHTTPRequestHandler, HTTPServer + +import pytest + +from app.calendar_feed import CalendarSource, merge_events + +_PLAIN_ICS = b"""BEGIN:VCALENDAR +VERSION:2.0 +BEGIN:VEVENT +UID:plain-1@example.com +SUMMARY:Dentist +DTSTART:20260801T140000Z +DTEND:20260801T150000Z +END:VEVENT +END:VCALENDAR +""" + +_RECURRING_ICS = b"""BEGIN:VCALENDAR +VERSION:2.0 +BEGIN:VEVENT +UID:weekly-1@example.com +SUMMARY:Standup +DTSTART:20260803T090000Z +DTEND:20260803T091500Z +RRULE:FREQ=WEEKLY;COUNT=4 +END:VEVENT +END:VCALENDAR +""" + +_SHARED_EVENT_ICS_A = b"""BEGIN:VCALENDAR +VERSION:2.0 +BEGIN:VEVENT +UID:shared-a@example.com +SUMMARY:Family Dinner +DTSTART:20260805T230000Z +DTEND:20260806T010000Z +END:VEVENT +END:VCALENDAR +""" + +_SHARED_EVENT_ICS_B = b"""BEGIN:VCALENDAR +VERSION:2.0 +BEGIN:VEVENT +UID:shared-b@example.com +SUMMARY:Family Dinner +DTSTART:20260805T230000Z +DTEND:20260806T010000Z +END:VEVENT +END:VCALENDAR +""" + + +class _Handler(BaseHTTPRequestHandler): + def do_GET(self): + body = self.server.feeds.get(self.path) # type: ignore[attr-defined] + if body is None: + self.send_response(404) + self.end_headers() + return + self.send_response(200) + self.send_header("Content-Type", "text/calendar") + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + + +@pytest.fixture +def ics_server(): + server = HTTPServer(("127.0.0.1", 0), _Handler) + server.feeds = { + "/plain.ics": _PLAIN_ICS, + "/recurring.ics": _RECURRING_ICS, + "/shared-a.ics": _SHARED_EVENT_ICS_A, + "/shared-b.ics": _SHARED_EVENT_ICS_B, + } + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + yield f"http://127.0.0.1:{port}" + finally: + server.shutdown() + thread.join() + + +_WINDOW_START = date(2026, 7, 1) +_WINDOW_END = date(2026, 9, 1) + + +def test_single_source_fetches_its_event(ics_server): + sources = [CalendarSource("Alice", "ics", f"{ics_server}/plain.ics")] + events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END) + assert summary == "" + assert len(events) == 1 + assert events[0]["summary"] == "Dentist" + assert events[0]["sources"] == [{"owner_display_name": "Alice", "color_index": None}] + + +def test_recurring_event_expands_within_window(ics_server): + sources = [CalendarSource("Bob", "ics", f"{ics_server}/recurring.ics")] + events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END) + assert summary == "" + assert len(events) == 4 # COUNT=4 + assert all(e["summary"] == "Standup" for e in events) + # distinct occurrences, not the same one repeated + assert len({e["start"] for e in events}) == 4 + + +def test_unreachable_source_does_not_blank_others(ics_server): + sources = [ + CalendarSource("Alice", "ics", f"{ics_server}/plain.ics"), + CalendarSource("Bob", "ics", f"{ics_server}/does-not-exist.ics"), + ] + events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END) + assert len(events) == 1 + assert events[0]["summary"] == "Dentist" + assert summary == "1 of 2 calendars unavailable" + + +def test_all_sources_unreachable(ics_server): + sources = [ + CalendarSource("Alice", "ics", f"{ics_server}/nope1.ics"), + CalendarSource("Bob", "ics", f"{ics_server}/nope2.ics"), + ] + events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END) + assert events == [] + assert summary == "2 of 2 calendars unavailable" + + +def test_duplicate_event_across_calendars_collapses_with_both_sources(ics_server): + """A shared event synced onto two people's calendars (same summary/ + start/end/all_day) should show up once, but carry both owners in + its `sources` list -- see merge_events' own docstring.""" + sources = [ + CalendarSource("Alice", "ics", f"{ics_server}/shared-a.ics"), + CalendarSource("Bob", "ics", f"{ics_server}/shared-b.ics"), + ] + events, summary = merge_events(sources, _WINDOW_START, _WINDOW_END) + assert summary == "" + assert len(events) == 1 + owners = {s["owner_display_name"] for s in events[0]["sources"]} + assert owners == {"Alice", "Bob"} + + +def test_events_outside_window_are_excluded(ics_server): + sources = [CalendarSource("Alice", "ics", f"{ics_server}/plain.ics")] + events, _ = merge_events(sources, date(2020, 1, 1), date(2020, 2, 1)) + assert events == [] diff --git a/server/tests/test_migrations.py b/server/tests/test_migrations.py new file mode 100644 index 0000000..d2490cd --- /dev/null +++ b/server/tests/test_migrations.py @@ -0,0 +1,59 @@ +"""migration.py structural sanity: MIGRATIONS is well-formed, a fresh +install lands on the latest schema version with frame #1 + server +settings seeded, and re-running run_migrations() is a true no-op (it's +called unconditionally at every app.main import -- see main.py -- so it +has to tolerate being invoked against an already-current database).""" + +from __future__ import annotations + +from sqlalchemy import inspect, text + +from app import db as db_module +from app.migration import MIGRATIONS, run_migrations +from app.models import Frame, ServerSettings + + +def test_migrations_list_is_sequential_and_unique(): + versions = [v for v, _ in MIGRATIONS] + assert versions == sorted(versions) + assert len(versions) == len(set(versions)) + assert versions == list(range(1, len(versions) + 1)) + + +def test_fresh_install_lands_on_latest_version(db_session): + with db_module.engine.connect() as conn: + row = conn.execute(text("SELECT version FROM schema_version")).fetchone() + assert row is not None + assert row[0] == MIGRATIONS[-1][0] + + +def test_fresh_install_seeds_frame_one_and_server_settings(db_session): + frame = db_session.get(Frame, 1) + assert frame is not None + assert frame.name + + settings = db_session.get(ServerSettings, 1) + assert settings is not None + + +def test_rerunning_migrations_is_a_no_op(db_session): + frame_count_before = len(db_session.query(Frame).all()) + run_migrations() + run_migrations() + frame_count_after = len(db_session.query(Frame).all()) + assert frame_count_before == frame_count_after == 1 + + +def test_expected_columns_exist_on_current_schema(): + """A light spot-check, not exhaustive -- one column from a handful of + the more recent migrations, to catch an ALTER that silently didn't + apply (e.g. a typo'd table/column name in a migration function).""" + inspector = inspect(db_module.engine) + user_columns = {c["name"] for c in inspector.get_columns("users")} + frame_columns = {c["name"] for c in inspector.get_columns("frames")} + + assert "webdav_base_url" in user_columns # migration 15 + assert "webdav_username" in user_columns # migration 14 + assert "calendar_caldav_url" in user_columns + assert "whiteboard_cached_image" in frame_columns # migration 14 + assert "calendar_week_start_offset" in frame_columns diff --git a/server/tests/test_permission_boundaries.py b/server/tests/test_permission_boundaries.py new file mode 100644 index 0000000..80f1f93 --- /dev/null +++ b/server/tests/test_permission_boundaries.py @@ -0,0 +1,178 @@ +"""The "owner controls adding their own data; anyone linked can mute it" +permission pattern, repeated across calendar-select, tasks-source, and +whiteboard-source -- exercised at the HTTP layer (not just unit-level) +since the whole point is verifying the *endpoint's* authorization check, +not just a helper function's logic.""" + +from __future__ import annotations + +from app.models import Frame, FrameCalendar, User + +from .conftest import csrf_headers, link_user, login, make_user + + +def _setup_two_linked_users(client, db_session) -> Frame: + """alice is the frame's admin/owner (via /setup); bob is a second + user linked to the same frame #1 but neither owns nor controls it.""" + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + bob = make_user(db_session, "bob") + frame = db_session.get(Frame, 1) + link_user(db_session, bob, frame) + return frame + + +# --- whiteboard-source --- + +def test_whiteboard_source_owner_can_set_it(client, db_session): + _setup_two_linked_users(client, db_session) + # alice is still logged in from /setup + resp = client.post("/api/frames/1/whiteboard-source", json={ + "url": "https://cloud.example.com/dav/files/alice/board.whiteboard", + }, headers=csrf_headers(client)) + assert resp.status_code == 200, resp.text + + frame = db_session.get(Frame, 1) + assert frame.whiteboard_user_id is not None + assert frame.whiteboard_url == "https://cloud.example.com/dav/files/alice/board.whiteboard" + + +def test_whiteboard_source_set_always_targets_the_caller(client, db_session): + """bob has no way to point the frame at someone else's account -- + there's no target-user field in the request at all, so a "set" call + from bob always attaches to bob, even if he pastes alice's URL.""" + _setup_two_linked_users(client, db_session) + client.cookies.clear() + login(client, "bob") + + resp = client.post("/api/frames/1/whiteboard-source", json={ + "url": "https://cloud.example.com/dav/files/alice/board.whiteboard", + }, headers=csrf_headers(client)) + assert resp.status_code == 200, resp.text + + frame = db_session.get(Frame, 1) + bob_row = db_session.query(User).filter_by(username="bob").one() + assert frame.whiteboard_user_id == bob_row.id + + +def test_whiteboard_source_anyone_linked_can_clear(client, db_session): + _setup_two_linked_users(client, db_session) + client.post("/api/frames/1/whiteboard-source", json={"url": "https://cloud.example.com/board.whiteboard"}, + headers=csrf_headers(client)) + + client.cookies.clear() + login(client, "bob") + resp = client.post("/api/frames/1/whiteboard-source", json={"url": None}, headers=csrf_headers(client)) + assert resp.status_code == 200, resp.text + + frame = db_session.get(Frame, 1) + assert frame.whiteboard_url == "" + assert frame.whiteboard_user_id is None + + +def test_whiteboard_source_rejects_non_http_url(client, db_session): + _setup_two_linked_users(client, db_session) + resp = client.post("/api/frames/1/whiteboard-source", json={"url": "javascript:alert(1)"}, + headers=csrf_headers(client)) + assert resp.status_code == 400 + + +def test_whiteboard_source_unlinked_user_cannot_touch_it(client, db_session): + _setup_two_linked_users(client, db_session) + make_user(db_session, "mallory") # exists, but never linked to frame 1 + client.cookies.clear() + login(client, "mallory") + + resp = client.post("/api/frames/1/whiteboard-source", json={"url": "https://x.example.com/b.whiteboard"}, + headers=csrf_headers(client, "/settings")) + assert resp.status_code == 404 + + +# --- tasks-source --- + +def test_tasks_source_set_always_targets_the_caller(client, db_session): + _setup_two_linked_users(client, db_session) + client.cookies.clear() + login(client, "bob") + + resp = client.post("/api/frames/1/tasks-source", json={"calendar_key": "caldav:/some/tasks/"}, + headers=csrf_headers(client)) + assert resp.status_code == 200, resp.text + + bob_row = db_session.query(User).filter_by(username="bob").one() + frame = db_session.get(Frame, 1) + assert frame.calendar_tasks_user_id == bob_row.id + assert frame.calendar_tasks_calendar_key == "caldav:/some/tasks/" + + +def test_tasks_source_anyone_linked_can_clear(client, db_session): + _setup_two_linked_users(client, db_session) + client.post("/api/frames/1/tasks-source", json={"calendar_key": "caldav:/alice/tasks/"}, + headers=csrf_headers(client)) + + client.cookies.clear() + login(client, "bob") + resp = client.post("/api/frames/1/tasks-source", json={"calendar_key": None}, headers=csrf_headers(client)) + assert resp.status_code == 200, resp.text + + frame = db_session.get(Frame, 1) + assert frame.calendar_tasks_user_id is None + assert frame.calendar_tasks_calendar_key is None + + +# --- calendar-select --- + +def test_calendar_select_bob_cannot_add_alices_calendar(client, db_session): + _setup_two_linked_users(client, db_session) + alice_id = db_session.query(User).filter_by(username="alice").one().id + + client.cookies.clear() + login(client, "bob") + resp = client.post("/api/frames/1/calendar-select", json={ + "user_id": alice_id, "calendar_key": "ics", "included": True, + }, headers=csrf_headers(client)) + assert resp.status_code == 403 + + +def test_calendar_select_owner_can_add_their_own(client, db_session): + _setup_two_linked_users(client, db_session) + alice_id = db_session.query(User).filter_by(username="alice").one().id + + resp = client.post("/api/frames/1/calendar-select", json={ + "user_id": alice_id, "calendar_key": "ics", "included": True, + }, headers=csrf_headers(client)) + assert resp.status_code == 200, resp.text + + row = db_session.query(FrameCalendar).filter_by(frame_id=1, user_id=alice_id, calendar_key="ics").one() + assert row.included is True + + +def test_calendar_select_bob_can_mute_alices_calendar(client, db_session): + """Muting is a display-preference veto anyone linked gets, unlike + adding -- the one-sided half of this endpoint's permission split.""" + _setup_two_linked_users(client, db_session) + alice_id = db_session.query(User).filter_by(username="alice").one().id + client.post("/api/frames/1/calendar-select", json={ + "user_id": alice_id, "calendar_key": "ics", "included": True, + }, headers=csrf_headers(client)) + + client.cookies.clear() + login(client, "bob") + resp = client.post("/api/frames/1/calendar-select", json={ + "user_id": alice_id, "calendar_key": "ics", "included": False, + }, headers=csrf_headers(client)) + assert resp.status_code == 200, resp.text + + row = db_session.query(FrameCalendar).filter_by(frame_id=1, user_id=alice_id, calendar_key="ics").one() + assert row.included is False + + +def test_calendar_select_cannot_mute_a_calendar_that_was_never_added(client, db_session): + _setup_two_linked_users(client, db_session) + alice_id = db_session.query(User).filter_by(username="alice").one().id + + client.cookies.clear() + login(client, "bob") + resp = client.post("/api/frames/1/calendar-select", json={ + "user_id": alice_id, "calendar_key": "ics", "included": False, + }, headers=csrf_headers(client)) + assert resp.status_code == 404 diff --git a/server/tests/test_render_size_invariants.py b/server/tests/test_render_size_invariants.py new file mode 100644 index 0000000..5182700 --- /dev/null +++ b/server/tests/test_render_size_invariants.py @@ -0,0 +1,67 @@ +"""Every renderer that produces a device-facing frame must return +exactly EPD_WIDTH*EPD_HEIGHT/2 bytes (the panel's packed 2px/byte +format) -- firmware writes this straight to the display with no length +checking of its own, so a renderer that's off by even one byte is a +silent on-device corruption bug, not a clean error. This is a cheap, +high-value regression guard: pure PIL rendering, no DB/HTTP/Node.""" + +from __future__ import annotations + +import pytest + +from app.calendar_render import CALENDAR_VIEWS, render_calendar +from app.image_pipeline import EPD_HEIGHT, EPD_WIDTH, render_placeholder + +EXPECTED_BYTES = EPD_WIDTH * EPD_HEIGHT // 2 +ORIENTATIONS = ["landscape", "landscape_flipped", "portrait", "portrait_flipped"] + +_SAMPLE_EVENTS = [ + { + "summary": "Dentist", "start": "2026-08-01T14:00:00+00:00", "end": "2026-08-01T15:00:00+00:00", + "all_day": False, "sources": [{"owner_display_name": "Alice", "color_index": None}], + }, + { + "summary": "Team Offsite", "start": "2026-08-03T00:00:00", "end": "2026-08-04T00:00:00", + "all_day": True, "sources": [{"owner_display_name": "Bob", "color_index": 2}], + }, +] + + +@pytest.mark.parametrize("orientation", ORIENTATIONS) +def test_placeholder_render_size(orientation): + data = render_placeholder(["Not configured yet"], orientation=orientation) + assert len(data) == EXPECTED_BYTES + + +@pytest.mark.parametrize("view", CALENDAR_VIEWS) +def test_calendar_render_size_across_views(view): + data = render_calendar(_SAMPLE_EVENTS, view, browse_offset=0, orientation="landscape", + palette_rgb=None, timezone="UTC") + assert len(data) == EXPECTED_BYTES + + +@pytest.mark.parametrize("orientation", ORIENTATIONS) +def test_calendar_render_size_across_orientations(orientation): + data = render_calendar(_SAMPLE_EVENTS, "agenda", browse_offset=0, orientation=orientation, + palette_rgb=None, timezone="UTC") + assert len(data) == EXPECTED_BYTES + + +def test_calendar_render_size_empty_events(): + data = render_calendar([], "week", browse_offset=0, orientation="landscape", + palette_rgb=None, timezone="UTC") + assert len(data) == EXPECTED_BYTES + + +def test_calendar_render_size_with_fetch_summary_and_tasks(): + tasks = [{"summary": "Buy milk", "completed": False}, {"summary": "Walk the dog", "completed": True}] + data = render_calendar(_SAMPLE_EVENTS, "week", browse_offset=0, orientation="landscape", + palette_rgb=None, timezone="UTC", fetch_summary="1 of 2 calendars unavailable", + week_days=5, week_layout="vertical", tasks=tasks) + assert len(data) == EXPECTED_BYTES + + +def test_calendar_render_size_with_week_start_offset(): + data = render_calendar(_SAMPLE_EVENTS, "week", browse_offset=0, orientation="landscape", + palette_rgb=None, timezone="UTC", week_days=3, week_start_offset=2) + assert len(data) == EXPECTED_BYTES diff --git a/server/tests/test_webdav_client.py b/server/tests/test_webdav_client.py new file mode 100644 index 0000000..25f94fa --- /dev/null +++ b/server/tests/test_webdav_client.py @@ -0,0 +1,138 @@ +"""webdav_client.py against a real local WebDAV-ish HTTP server (Basic +auth + PROPFIND) -- no ORM, no FastAPI, pure protocol-level fetch/list +functions used by whiteboard frame mode.""" + +from __future__ import annotations + +import base64 +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer + +import pytest + +from app.webdav_client import WebDavError, fetch_file, list_directory, parent_directory_url + +_USERNAME = "alice" +_PASSWORD = "secret" +_FILE_BYTES = b'{"elements": [], "appState": {}}' + +_PROPFIND_RESPONSE = b""" + + + /dav/files/alice/Boards/ + + + Boards + + + HTTP/1.1 200 OK + + + + /dav/files/alice/Boards/Family%20Board.whiteboard + + + Family Board.whiteboard + + + HTTP/1.1 200 OK + + + + /dav/files/alice/Boards/Archive/ + + + Archive + + + HTTP/1.1 200 OK + + + +""" + +_AUTH_HEADER = "Basic " + base64.b64encode(f"{_USERNAME}:{_PASSWORD}".encode()).decode() + + +class _Handler(BaseHTTPRequestHandler): + def _authed(self) -> bool: + return self.headers.get("Authorization", "") == _AUTH_HEADER + + def do_GET(self): + if self.path == "/dav/files/alice/Boards/Family%20Board.whiteboard": + if not self._authed(): + self.send_response(401) + self.end_headers() + return + self.send_response(200) + self.end_headers() + self.wfile.write(_FILE_BYTES) + return + self.send_response(404) + self.end_headers() + + def do_PROPFIND(self): + if not self._authed(): + self.send_response(401) + self.end_headers() + return + self.send_response(207) + self.send_header("Content-Type", "application/xml") + self.end_headers() + self.wfile.write(_PROPFIND_RESPONSE) + + def log_message(self, *args): + pass + + +@pytest.fixture(scope="module") +def webdav_server(): + server = HTTPServer(("127.0.0.1", 0), _Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + yield f"http://127.0.0.1:{port}" + finally: + server.shutdown() + thread.join() + + +def test_fetch_file_success(webdav_server): + data = fetch_file(f"{webdav_server}/dav/files/alice/Boards/Family%20Board.whiteboard", _USERNAME, _PASSWORD) + assert data == _FILE_BYTES + + +def test_fetch_file_wrong_password_raises(webdav_server): + with pytest.raises(WebDavError): + fetch_file(f"{webdav_server}/dav/files/alice/Boards/Family%20Board.whiteboard", _USERNAME, "wrong") + + +def test_fetch_file_missing_raises(webdav_server): + with pytest.raises(WebDavError): + fetch_file(f"{webdav_server}/dav/files/alice/Boards/nope.whiteboard", _USERNAME, _PASSWORD) + + +def test_list_directory_excludes_self_and_sorts_dirs_first(webdav_server): + entries = list_directory(f"{webdav_server}/dav/files/alice/Boards/", _USERNAME, _PASSWORD) + assert [e["name"] for e in entries] == ["Archive", "Family Board.whiteboard"] + assert entries[0]["is_dir"] is True + assert entries[1]["is_dir"] is False + + +def test_list_directory_wrong_password_raises(webdav_server): + with pytest.raises(WebDavError): + list_directory(f"{webdav_server}/dav/files/alice/Boards/", _USERNAME, "wrong") + + +def test_list_directory_urls_are_absolute(webdav_server): + entries = list_directory(f"{webdav_server}/dav/files/alice/Boards/", _USERNAME, _PASSWORD) + archive = next(e for e in entries if e["name"] == "Archive") + assert archive["url"] == f"{webdav_server}/dav/files/alice/Boards/Archive/" + + +def test_parent_directory_url_stops_at_base(): + base = "https://cloud.example.com/dav/files/alice/" + assert parent_directory_url(base, base) is None + assert parent_directory_url(base, base + "Boards/") == base + assert parent_directory_url(base, base + "Boards/Family/") == base + "Boards/" diff --git a/server/tests/test_whiteboard_refresh_and_browse.py b/server/tests/test_whiteboard_refresh_and_browse.py new file mode 100644 index 0000000..57f4dc7 --- /dev/null +++ b/server/tests/test_whiteboard_refresh_and_browse.py @@ -0,0 +1,150 @@ +"""get_or_refresh_whiteboard's fetch throttle (and force=True bypassing +it) plus the whiteboard-browse HTTP endpoint. whiteboard.fetch_and_render +is monkeypatched -- it talks to a real WebDAV server and the Node render +sidecar (see render-service/), neither of which this suite needs a real +copy of to verify the *throttle*/*permission* logic around it.""" + +from __future__ import annotations + +import time + +from app import whiteboard +from app.models import Frame, User +from app.routers.common import get_or_refresh_whiteboard + +from .conftest import csrf_headers, link_user, login, make_user + + +def _configure_whiteboard(db_session, frame: Frame, user: User) -> None: + frame.whiteboard_user_id = user.id + frame.whiteboard_url = "http://example.invalid/board.whiteboard" + frame.whiteboard_cached_image = b"OLD_CACHED_PNG" + frame.whiteboard_checked_at = time.time() # just refreshed -- well within the throttle + db_session.commit() + + +def test_unforced_call_within_throttle_uses_cache(client, db_session, monkeypatch): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + alice = db_session.query(User).filter_by(username="alice").one() + alice.webdav_username = "alice" + alice.webdav_password = "secret" + frame = db_session.get(Frame, 1) + _configure_whiteboard(db_session, frame, alice) + + calls = [] + monkeypatch.setattr(whiteboard, "fetch_and_render", + lambda url, u, p: calls.append(1) or b"NEW_PNG") + + result = get_or_refresh_whiteboard(db_session, frame) + assert result == b"OLD_CACHED_PNG" + assert calls == [] + + +def test_force_bypasses_throttle_and_persists(client, db_session, monkeypatch): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + alice = db_session.query(User).filter_by(username="alice").one() + alice.webdav_username = "alice" + alice.webdav_password = "secret" + frame = db_session.get(Frame, 1) + _configure_whiteboard(db_session, frame, alice) + + calls = [] + monkeypatch.setattr(whiteboard, "fetch_and_render", + lambda url, u, p: calls.append(1) or b"NEW_PNG") + + result = get_or_refresh_whiteboard(db_session, frame, force=True) + assert result == b"NEW_PNG" + assert len(calls) == 1 + + db_session.refresh(frame) + assert frame.whiteboard_cached_image == b"NEW_PNG" + + +def test_preview_endpoint_force_query_param_bypasses_throttle(client, db_session, monkeypatch): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + alice = db_session.query(User).filter_by(username="alice").one() + alice.webdav_username = "alice" + alice.webdav_password = "secret" + frame = db_session.get(Frame, 1) + + # The preview endpoint runs whatever get_or_refresh_whiteboard returns + # through PIL (Image.open) -- unlike the other tests in this file, + # the placeholder "cached" bytes need to be real, valid PNG data, not + # just an arbitrary marker string. + import io + + from PIL import Image + buf = io.BytesIO() + Image.new("RGB", (1, 1), (255, 255, 255)).save(buf, format="PNG") + tiny_png = buf.getvalue() + + frame.whiteboard_user_id = alice.id + frame.whiteboard_url = "http://example.invalid/board.whiteboard" + frame.whiteboard_cached_image = tiny_png + frame.whiteboard_checked_at = time.time() + db_session.commit() + + calls = [] + monkeypatch.setattr(whiteboard, "fetch_and_render", + lambda url, u, p: calls.append(1) or tiny_png) + + resp = client.get("/api/frames/1/preview/whiteboard") + assert resp.status_code == 200 + assert calls == [] # fresh cache, no force -- no refetch + + resp = client.get("/api/frames/1/preview/whiteboard?force=1") + assert resp.status_code == 200 + assert len(calls) == 1 # force=1 -- refetched despite fresh cache + + +# --- whiteboard-browse --- + +def test_browse_requires_webdav_creds(client, db_session): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + resp = client.get("/api/frames/1/whiteboard-browse") + assert resp.status_code == 400 + assert "credentials" in resp.json()["detail"].lower() + + +def test_browse_requires_a_base_url_when_none_supplied(client, db_session): + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + alice = db_session.query(User).filter_by(username="alice").one() + alice.webdav_username = "alice" + alice.webdav_password = "secret" + db_session.commit() + + resp = client.get("/api/frames/1/whiteboard-browse") + assert resp.status_code == 400 + assert "browse root" in resp.json()["detail"].lower() + + +def test_browse_uses_the_caller_own_creds_not_the_frames_owner(client, db_session, monkeypatch): + """bob, linked but not the frame's whiteboard owner, should still be + able to browse HIS OWN webdav account to pick a file for himself -- + browsing is a "help me find a file" lookup against the caller's own + credentials, unrelated to whose account the frame is currently + pointed at.""" + client.post("/setup", data={"username": "alice", "password": "hunter22"}) + bob = make_user(db_session, "bob") + frame = db_session.get(Frame, 1) + link_user(db_session, bob, frame) + bob.webdav_username = "bob" + bob.webdav_password = "bobsecret" + bob.webdav_base_url = "https://cloud.example.com/dav/files/bob/" + db_session.commit() + + seen_creds = [] + + def fake_list_directory(url, username, password): + seen_creds.append((url, username, password)) + return [{"name": "Board.whiteboard", "url": url + "Board.whiteboard", "is_dir": False}] + + from app import webdav_client + monkeypatch.setattr(webdav_client, "list_directory", fake_list_directory) + + client.cookies.clear() + login(client, "bob") + resp = client.get("/api/frames/1/whiteboard-browse") + assert resp.status_code == 200, resp.text + assert seen_creds == [("https://cloud.example.com/dav/files/bob/", "bob", "bobsecret")] + assert resp.json()["entries"][0]["name"] == "Board.whiteboard"