Merge: R2 S2b — backend pytest harness (pure-logic tests + isolated test image)
This commit is contained in:
@@ -0,0 +1,12 @@
|
|||||||
|
# Test-only image — the app's Python deps + pytest/ruff, and NOTHING else. Code is bind-mounted at
|
||||||
|
# run time (not baked), so pytest always sees the working tree, and this file never touches the prod
|
||||||
|
# Dockerfile, so the published image can never carry test tooling. Same python + same pinned
|
||||||
|
# requirements.txt as the runtime stage, so tests run against the prod dependency set. Rebuilt only
|
||||||
|
# when a requirements file changes (the pip layer is cache-keyed on them); otherwise a no-op.
|
||||||
|
#
|
||||||
|
# docker build -f backend/Dockerfile.test -t siftlode-test:local backend
|
||||||
|
# docker run --rm -v "<repo>/backend:/app" -w /app siftlode-test:local python -m pytest
|
||||||
|
FROM python:3.12-slim
|
||||||
|
WORKDIR /app
|
||||||
|
COPY requirements.txt requirements-dev.txt ./
|
||||||
|
RUN pip install --no-cache-dir -r requirements.txt -r requirements-dev.txt
|
||||||
@@ -0,0 +1,6 @@
|
|||||||
|
[pytest]
|
||||||
|
# Run from backend/ (the image's /app). `pythonpath = .` puts the app package on sys.path so tests
|
||||||
|
# import it as `app.*`, matching runtime. The suite is pure-logic only (no DB/network) — see tests/.
|
||||||
|
testpaths = tests
|
||||||
|
pythonpath = .
|
||||||
|
addopts = -q
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
# Test + lint tooling — NEVER in the prod image. Installed only by backend/Dockerfile.test (the
|
||||||
|
# throwaway test image), which the prod build (root Dockerfile) never references. Pinned so the
|
||||||
|
# suite is reproducible: pytest for the pure-logic tests, and ruff at the SAME version the gate's
|
||||||
|
# backend lint lane currently runs on the host (see the run_lint note in siftlode.sh). httpx is
|
||||||
|
# already an app dependency (requirements.txt), so its TestClient needs nothing extra here.
|
||||||
|
pytest==8.4.2
|
||||||
|
ruff==0.15.21
|
||||||
@@ -0,0 +1,53 @@
|
|||||||
|
"""Signed-grant HMAC for password-protected share links: a grant proves the viewer unlocked a
|
||||||
|
token, rides in the file URL (a <video src> can't send a header), and expires."""
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
from types import SimpleNamespace
|
||||||
|
|
||||||
|
from app.downloads.links import _GRANT_TTL, check_grant, is_expired, make_grant, new_token
|
||||||
|
|
||||||
|
|
||||||
|
def test_new_token_is_unguessable_and_unique():
|
||||||
|
a, b = new_token(), new_token()
|
||||||
|
assert a != b
|
||||||
|
assert len(a) >= 30 # token_urlsafe(24) → ~32 chars
|
||||||
|
|
||||||
|
|
||||||
|
class TestGrant:
|
||||||
|
def test_a_fresh_grant_verifies_for_its_own_token(self):
|
||||||
|
grant = make_grant("tok")
|
||||||
|
assert check_grant("tok", grant) is True
|
||||||
|
|
||||||
|
def test_a_grant_does_not_verify_for_a_different_token(self):
|
||||||
|
# The signature binds the token, so a grant minted for one link can't unlock another.
|
||||||
|
assert check_grant("other", make_grant("tok")) is False
|
||||||
|
|
||||||
|
def test_a_tampered_signature_is_rejected(self):
|
||||||
|
exp, _sig = make_grant("tok").split(".", 1)
|
||||||
|
assert check_grant("tok", f"{exp}.deadbeef") is False
|
||||||
|
|
||||||
|
def test_a_tampered_expiry_is_rejected(self):
|
||||||
|
# Extending the exp field invalidates the signature (which covers token.exp).
|
||||||
|
_exp, sig = make_grant("tok").split(".", 1)
|
||||||
|
future = int(datetime.now(timezone.utc).timestamp()) + 10 * _GRANT_TTL
|
||||||
|
assert check_grant("tok", f"{future}.{sig}") is False
|
||||||
|
|
||||||
|
def test_an_expired_grant_is_rejected(self):
|
||||||
|
past = datetime.now(timezone.utc).timestamp() - 2 * _GRANT_TTL
|
||||||
|
assert check_grant("tok", make_grant("tok", now=past)) is False
|
||||||
|
|
||||||
|
def test_malformed_grants_are_rejected_not_crashed(self):
|
||||||
|
for bad in [None, "", "no-dot", "notanumber.sig"]:
|
||||||
|
assert check_grant("tok", bad) is False
|
||||||
|
|
||||||
|
|
||||||
|
class TestIsExpired:
|
||||||
|
def test_none_expiry_never_expires(self):
|
||||||
|
assert is_expired(SimpleNamespace(expires_at=None)) is False
|
||||||
|
|
||||||
|
def test_a_future_expiry_is_live(self):
|
||||||
|
future = datetime.now(timezone.utc) + timedelta(hours=1)
|
||||||
|
assert is_expired(SimpleNamespace(expires_at=future)) is False
|
||||||
|
|
||||||
|
def test_a_past_expiry_is_expired(self):
|
||||||
|
past = datetime.now(timezone.utc) - timedelta(hours=1)
|
||||||
|
assert is_expired(SimpleNamespace(expires_at=past)) is True
|
||||||
@@ -0,0 +1,26 @@
|
|||||||
|
"""App-assembly smoke test — DB-free.
|
||||||
|
|
||||||
|
Uses TestClient WITHOUT its context-manager form on purpose: that skips the lifespan (which opens a
|
||||||
|
DB session + starts the scheduler), so the app object can be built and its OpenAPI schema generated
|
||||||
|
with no database. That alone proves every included router imported and wired cleanly — a broken
|
||||||
|
route registration or import fails right here.
|
||||||
|
|
||||||
|
Endpoints that actually answer (/healthz, /api/version, the auth/feed/downloads routers) all touch
|
||||||
|
the database, so exercising THEM belongs in a DB-backed integration lane (needs a test Postgres +
|
||||||
|
migrations) — deferred; see siftlode-ops/ROADMAP.md R2 S2b.
|
||||||
|
"""
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from app.main import app
|
||||||
|
|
||||||
|
client = TestClient(app) # no `with` → lifespan/startup does not run → no DB required
|
||||||
|
|
||||||
|
|
||||||
|
def test_app_assembles_and_serves_openapi():
|
||||||
|
schema = client.get("/openapi.json")
|
||||||
|
assert schema.status_code == 200
|
||||||
|
paths = schema.json()["paths"]
|
||||||
|
# A handful of endpoints that must always exist — a rename/removal is a real contract change,
|
||||||
|
# and their absence here would mean a router failed to register.
|
||||||
|
for path in ["/healthz", "/api/version", "/api/feed", "/api/me"]:
|
||||||
|
assert path in paths, f"missing route {path} — a router failed to wire up"
|
||||||
@@ -0,0 +1,107 @@
|
|||||||
|
"""On-disk layout helpers: name sanitation, the Plex/flat path shapes, and — most importantly —
|
||||||
|
the safe_abs_path traversal guard that every file-serving endpoint depends on."""
|
||||||
|
from datetime import date
|
||||||
|
|
||||||
|
from app.downloads.storage import (
|
||||||
|
MediaMeta,
|
||||||
|
download_filename,
|
||||||
|
rel_path,
|
||||||
|
safe_abs_path,
|
||||||
|
sanitize,
|
||||||
|
)
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
class TestSanitize:
|
||||||
|
def test_spaces_become_underscores_and_result_has_no_separators(self):
|
||||||
|
out = sanitize("My Great Video")
|
||||||
|
assert out == "My_Great_Video"
|
||||||
|
assert "/" not in out and "\\" not in out and " " not in out
|
||||||
|
|
||||||
|
def test_illegal_and_emoji_chars_are_removed(self):
|
||||||
|
out = sanitize('a/b:c*?"<>| 🔥')
|
||||||
|
for bad in '/\\:*?"<>|🔥':
|
||||||
|
assert bad not in out
|
||||||
|
|
||||||
|
def test_repeated_punctuation_collapses(self):
|
||||||
|
assert sanitize("wow!!!___---end") == "wow!_-end"
|
||||||
|
|
||||||
|
def test_length_is_capped(self):
|
||||||
|
assert len(sanitize("x" * 500, limit=10)) <= 10
|
||||||
|
|
||||||
|
def test_empty_or_all_junk_falls_back_to_untitled(self):
|
||||||
|
assert sanitize("") == "untitled"
|
||||||
|
assert sanitize("🔥🔥") == "untitled"
|
||||||
|
|
||||||
|
|
||||||
|
class TestRelPath:
|
||||||
|
meta = MediaMeta(
|
||||||
|
video_id="abc123",
|
||||||
|
title="Cool Clip",
|
||||||
|
uploader="Some Channel",
|
||||||
|
upload_date=date(2024, 5, 2),
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_flat_layout_is_a_single_file(self):
|
||||||
|
assert rel_path(self.meta, "mp4", layout="flat") == "Cool_Clip_[abc123].mp4"
|
||||||
|
|
||||||
|
def test_plex_layout_nests_channel_and_season(self):
|
||||||
|
out = rel_path(self.meta, "mp4", layout="plex")
|
||||||
|
assert out == "Some_Channel/Season_2024/Some_Channel_-_2024-05-02_-_Cool_Clip_[abc123].mp4"
|
||||||
|
|
||||||
|
def test_missing_date_uses_year_zero_and_placeholder_date(self):
|
||||||
|
meta = MediaMeta(video_id="x", title="t", uploader="c", upload_date=None)
|
||||||
|
out = rel_path(meta, "mp4")
|
||||||
|
assert "Season_0/" in out and "0000-00-00" in out
|
||||||
|
|
||||||
|
|
||||||
|
class TestDownloadFilename:
|
||||||
|
def test_appends_the_container_extension(self):
|
||||||
|
assert download_filename("My Video", "mp4", Path("/x/f.webm")) == "My Video.mp4"
|
||||||
|
|
||||||
|
def test_does_not_double_an_extension_already_present(self):
|
||||||
|
assert download_filename("clip.mp4", "mp4", Path("/x/f.mp4")) == "clip.mp4"
|
||||||
|
|
||||||
|
def test_falls_back_to_the_file_suffix_when_no_container(self):
|
||||||
|
assert download_filename("song", None, Path("/x/f.m4a")) == "song.m4a"
|
||||||
|
|
||||||
|
|
||||||
|
class TestSafeAbsPath:
|
||||||
|
"""The path-traversal guard: only paths that resolve INSIDE download_root AND exist are returned."""
|
||||||
|
|
||||||
|
def test_returns_the_path_for_a_file_inside_the_root(self, tmp_path: Path):
|
||||||
|
(tmp_path / "sub").mkdir()
|
||||||
|
f = tmp_path / "sub" / "video.mp4"
|
||||||
|
f.write_text("data")
|
||||||
|
got = safe_abs_path(str(tmp_path), "sub/video.mp4")
|
||||||
|
assert got is not None and got == f.resolve()
|
||||||
|
|
||||||
|
def test_rejects_dot_dot_traversal_out_of_the_root(self, tmp_path: Path):
|
||||||
|
root = tmp_path / "root"
|
||||||
|
root.mkdir()
|
||||||
|
secret = tmp_path / "secret.txt"
|
||||||
|
secret.write_text("nope")
|
||||||
|
# ../secret.txt escapes root even though the file exists — must be refused.
|
||||||
|
assert safe_abs_path(str(root), "../secret.txt") is None
|
||||||
|
|
||||||
|
def test_rejects_an_absolute_path_escape(self, tmp_path: Path):
|
||||||
|
root = tmp_path / "root"
|
||||||
|
root.mkdir()
|
||||||
|
outside = tmp_path / "outside.txt"
|
||||||
|
outside.write_text("nope")
|
||||||
|
assert safe_abs_path(str(root), str(outside)) is None
|
||||||
|
|
||||||
|
def test_rejects_a_sibling_dir_whose_name_prefixes_the_root(self, tmp_path: Path):
|
||||||
|
# The one input that distinguishes the correct `root not in path.parents` guard from the
|
||||||
|
# classic buggy `str(path).startswith(str(root))`: /root vs a real sibling /rootx. A file in
|
||||||
|
# rootx is OUTSIDE root, but its path DOES start with the root string — a prefix check would
|
||||||
|
# wrongly serve it. This pins the guard against that regression.
|
||||||
|
root = tmp_path / "root"
|
||||||
|
root.mkdir()
|
||||||
|
sibling = tmp_path / "rootx"
|
||||||
|
sibling.mkdir()
|
||||||
|
(sibling / "steal.mp4").write_text("nope")
|
||||||
|
assert safe_abs_path(str(root), "../rootx/steal.mp4") is None
|
||||||
|
|
||||||
|
def test_returns_none_for_a_nonexistent_file_inside_the_root(self, tmp_path: Path):
|
||||||
|
assert safe_abs_path(str(tmp_path), "does/not/exist.mp4") is None
|
||||||
@@ -0,0 +1,41 @@
|
|||||||
|
"""normalize_title: de-shout / de-clutter video titles without mangling non-English text."""
|
||||||
|
from app.titles import normalize_title
|
||||||
|
|
||||||
|
|
||||||
|
def test_empty_and_whitespace_pass_through_unchanged():
|
||||||
|
assert normalize_title(None) is None
|
||||||
|
assert normalize_title("") == ""
|
||||||
|
assert normalize_title(" ") == " "
|
||||||
|
|
||||||
|
|
||||||
|
def test_plain_title_is_left_alone():
|
||||||
|
assert normalize_title("How I built a boat") == "How I built a boat"
|
||||||
|
|
||||||
|
|
||||||
|
def test_shouting_title_is_de_capsed_and_first_letter_kept_upper():
|
||||||
|
# >=2 all-caps runs and >=50% caps → treated as shouting: title-case the long words.
|
||||||
|
out = normalize_title("THIS IS A HUGE ANNOUNCEMENT")
|
||||||
|
assert out is not None
|
||||||
|
assert out == out[0].upper() + out[1:] # first alpha char stays capitalized
|
||||||
|
assert "HUGE" not in out and "ANNOUNCEMENT" not in out
|
||||||
|
|
||||||
|
|
||||||
|
def test_trailing_hashtags_are_stripped():
|
||||||
|
assert normalize_title("Great cooking video #food #yummy") == "Great cooking video"
|
||||||
|
|
||||||
|
|
||||||
|
def test_emoji_and_symbols_are_dropped():
|
||||||
|
assert "🔥" not in (normalize_title("Fire recipe 🔥🔥") or "")
|
||||||
|
|
||||||
|
|
||||||
|
def test_repeated_punctuation_is_collapsed():
|
||||||
|
assert normalize_title("Wait what!!!!") == "Wait what!"
|
||||||
|
|
||||||
|
|
||||||
|
def test_accented_non_english_letters_survive():
|
||||||
|
# Siftlode is trilingual — Hungarian/German letters must not be ASCII-folded away.
|
||||||
|
for title in ["Gulyásleves főzés", "Über die Straße"]:
|
||||||
|
out = normalize_title(title) or ""
|
||||||
|
for ch in "áöüőűé":
|
||||||
|
if ch in title:
|
||||||
|
assert ch in out
|
||||||
Reference in New Issue
Block a user