import json
import math
import re
from urllib.parse import quote, unquote
import httpx
import pytest
from fastapi.testclient import TestClient
from app.server import Settings, create_app
def mkfile(fid, name, size=100, thumb=True):
return {
"id": fid, "name": name, "originalName": name, "size": size, "type": "video/mp4", "views": 3,
"createdAt": "2026-10-01T16:46:15.378Z", "deletesAt": "2026-12-30T16:46:15.375Z",
"thumbnail": {"path": f".thumbnail.{fid}.jpg"} if thumb else None, "tags": [], "url": "/" + name,
}
class FakeZipline:
def __init__(self):
self.users = {"alice": ("pw", "tok-alice"), "bob": ("pw2", "tok-bob")}
self.files = {
"tok-alice": [mkfile("e1", "clutch.mp4"), mkfile("e2", "çılgın-#1.mp4", thumb=False)],
"tok-bob": [mkfile("c1", "friend.mp4")],
}
self.down = False
self.calls = []
def owner(self, tok):
return next((name for name, (_, t) in self.users.items() if t == tok), None)
def handler(self, request: httpx.Request) -> httpx.Response:
if self.down:
raise httpx.ConnectError("down")
path, method = request.url.path, request.method
self.calls.append((method, path, request.headers.get("authorization"), request.headers.get("cookie")))
if path == "/api/auth/login":
body = json.loads(request.content)
user = self.users.get(body["username"])
if not user or user[0] != body["password"]:
return httpx.Response(400, json={"error": "E1044: Invalid username or password"})
return httpx.Response(
200, json={"user": {"id": "u-" + body["username"], "username": body["username"]}},
headers={"set-cookie": f"zipline_session=sess-{body['username']}; Max-Age=1209600; Path=/; HttpOnly; Secure; SameSite=Lax"},
)
if path == "/api/user/token":
cookie = request.headers.get("cookie") or ""
for name, (_, tok) in self.users.items():
if f"zipline_session=sess-{name}" in cookie:
return httpx.Response(200, json={"token": tok})
return httpx.Response(401, json={"error": "E2001: Invalid token"})
if method == "HEAD":
name = unquote(path[1:])
f = next((f for files in self.files.values() for f in files if f["name"] == name), None)
if f is None:
return httpx.Response(404)
return httpx.Response(200, headers={"content-type": f["type"], "content-length": str(f["size"])})
tok = request.headers.get("authorization")
who = self.owner(tok)
if who is None:
return httpx.Response(401, json={"error": "E2001: Invalid token"})
mine = self.files.setdefault(tok, [])
if path == "/api/user":
return httpx.Response(200, json={"user": {"id": "u-" + who, "username": who, "role": "USER"}})
if path == "/api/user/stats":
return httpx.Response(200, json={"storageUsed": sum(f["size"] for f in mine)})
if path == "/api/user/files":
page = int(request.url.params.get("page", "1"))
per = int(request.url.params.get("perpage", "15"))
chunk = mine[(page - 1) * per: page * per]
return httpx.Response(200, json={"page": chunk, "total": len(mine), "pages": max(1, math.ceil(len(mine) / per))})
if path.startswith("/api/user/files/"):
ident = unquote(path[len("/api/user/files/"):])
f = next((f for f in mine if ident in (f["id"], f["name"])), None)
if f is None:
return httpx.Response(404, json={"error": "E4000: File not found"})
if method == "GET":
return httpx.Response(200, json=f)
if method == "PATCH":
f["name"] = json.loads(request.content)["name"]
return httpx.Response(200, json=f)
if method == "DELETE":
mine.remove(f)
return httpx.Response(200, json=f)
return httpx.Response(404, json={"error": "not found"})
@pytest.fixture
def zl():
return FakeZipline()
@pytest.fixture
def client(zl, tmp_path):
settings = Settings(zipline_url="http://zipline:3000", site_origin="https://clips.example.com",
upload_url="https://upload.example.com", max_bytes=1000, deletes_at="90d", work_dir=tmp_path)
return TestClient(create_app(settings, transport=httpx.MockTransport(zl.handler)), base_url="https://testserver")
def login(client, user="alice", pw="pw", ip="1.2.3.4"):
return client.post("/_site/api/login", json={"username": user, "password": pw}, headers={"cf-connecting-ip": ip})
def test_pages_serve_the_site(client):
for path in ("/", "/v/clutch.mp4", "/v/%C3%A7%C4%B1lg%C4%B1n-%231.mp4"):
r = client.get(path)
assert r.status_code == 200
assert "text/html" in r.headers["content-type"]
assert "/_site/static/app.js" in r.text
assert client.get("/_site/static/style.css").status_code == 200
@pytest.fixture
def fake_media(monkeypatch):
seen = []
monkeypatch.setattr("app.site_api.video_size", lambda url, length: seen.append(url) or (1920, 1080))
monkeypatch.setattr("app.site_api.still_jpeg", lambda url, length: seen.append(url) or b"jpeg-bytes")
return seen
def test_watch_page_has_discord_player_tags(client, fake_media):
# Discord gives up on big raw links; the /v/ page hands it the player URL, size and a still instead
r = client.get("/v/%C3%A7%C4%B1lg%C4%B1n-%231.mp4")
clip = "https://clips.example.com/%C3%A7%C4%B1lg%C4%B1n-%231.mp4"
assert f'' in r.text
assert '' in r.text
assert '' in r.text
assert '' in r.text
assert ('') in r.text
assert fake_media == ["http://zipline:3000/%C3%A7%C4%B1lg%C4%B1n-%231.mp4"]
def test_shared_clip_gets_a_random_stripe_color(client, fake_media):
colors = {re.search(r'name="theme-color" content="(#[0-9a-f]{6})"', client.get("/v/clutch.mp4").text).group(1)
for _ in range(5)}
assert len(colors) > 1 and "#3e4637" not in colors
assert 'name="theme-color" content="#3e4637"' in client.get("/").text
def test_watch_page_without_clip_has_no_player_tags(client, fake_media):
for path in ("/v/nope.mp4", '/v/">'):
r = client.get(path)
assert r.status_code == 200
assert "og:video" not in r.text and "