kaledonya-clips/intake/tests/test_server.py

import shutil
import subprocess
from urllib.parse import quote, unquote

import httpx
import pytest
from fastapi.testclient import TestClient

from app.server import Settings, create_app


class FakeZipline:
    def __init__(self):
        self.names = set()      # names Zipline already has
        self.race = set()       # free at HEAD time, taken by the time we upload
        self.fail_with = None   # force this status on /api/upload
        self.raw_urls = False   # return the url unencoded, like a careless server would
        self.head_fails = False  # HEAD raises, as if Zipline went away mid-request
        self.uploads = []

    def handler(self, request: httpx.Request) -> httpx.Response:
        if request.method == "HEAD" and self.head_fails:
            raise httpx.ConnectError("down")
        if request.method == "HEAD":
            name = unquote(request.url.path[1:])
            if name in self.names:
                # Zipline redirects text files (and password-protected ones) to /view/<name>
                return httpx.Response(302 if name.endswith(".txt") else 200)
            return httpx.Response(404)
        if request.url.path == "/api/upload":
            if self.fail_with:
                return httpx.Response(self.fail_with, json={"error": "boom"})
            stem = unquote(request.headers["x-zipline-filename"])
            name = stem + request.headers.get("x-zipline-file-extension", "")
            if name in self.names or name in self.race:
                self.race.discard(name)
                self.names.add(name)
                return httpx.Response(400, json={"error": "file with the same name already exists"})
            self.names.add(name)
            self.uploads.append({"name": name, "headers": dict(request.headers), "body": request.read()})
            return httpx.Response(200, json={"files": [{
                "id": "1", "name": name, "type": "video/mp4",
                "url": f"https://clips.example.com/{name if self.raw_urls else quote(name)}",
            }]})
        return httpx.Response(404)


@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)))


AUTH = {"Authorization": "tok"}


def up(client, name, data=b"x" * 10, headers=AUTH):
    return client.post("/upload", files={"file": (name, data, "application/octet-stream")}, headers=headers)


def test_health(client):
    assert client.get("/health").json() == {"ok": True}


def test_missing_token_is_401(client, zl):
    r = up(client, "a.mp4", headers={})
    assert r.status_code == 401
    assert zl.uploads == []


def test_missing_file_field_is_400(client):
    r = client.post("/upload", data={"nope": "1"}, headers=AUTH)
    assert r.status_code == 400


def test_spaces_become_dashes_and_headers(client, zl):
    r = up(client, "my clip 1.mp4")
    assert r.status_code == 200
    assert r.json() == {"url": "https://clips.example.com/my-clip-1.mp4", "name": "my-clip-1.mp4", "remuxed": False}
    h = zl.uploads[0]["headers"]
    assert h["authorization"] == "tok"
    assert h["x-zipline-format"] == "name"
    assert h["x-zipline-filename"] == "my-clip-1"
    assert h["x-zipline-file-extension"] == ".mp4"
    assert h["x-zipline-deletes-at"] == "90d"
    assert h["x-zipline-original-name"] == "true"
    assert zl.uploads[0]["body"].count(b"x" * 10) == 1


def test_duplicate_gets_suffix(client, zl):
    zl.names |= {"clip.mp4", "clip-2.mp4"}
    assert up(client, "clip.mp4").json()["name"] == "clip-3.mp4"


def test_taken_text_file_redirect_counts_as_taken(client, zl):
    zl.names |= {"n.txt", "n-2.txt", "n-3.txt"}
    r = up(client, "n.txt")
    assert r.status_code == 200
    assert r.json()["name"] == "n-4.txt"


def test_race_retries_next_name(client, zl):
    zl.race.add("clip.mp4")
    r = up(client, "clip.mp4")
    assert r.status_code == 200
    assert r.json()["name"] == "clip-2.mp4"


def test_unicode_and_special_chars(client, zl):
    r = up(client, "çılgın an #1.mp4")
    assert r.status_code == 200
    assert r.json()["name"] == "çılgın-an-#1.mp4"
    assert zl.uploads[0]["headers"]["x-zipline-filename"] == quote("çılgın-an-#1")


def test_no_extension(client, zl):
    r = up(client, "README")
    assert r.json()["name"] == "README"
    assert "x-zipline-file-extension" not in zl.uploads[0]["headers"]


def test_exact_limit_ok(client):
    assert up(client, "a.bin", data=b"x" * 1000).status_code == 200


def test_one_byte_over_is_413(client, zl):
    r = up(client, "a.bin", data=b"x" * 1001)
    assert r.status_code == 413
    assert r.json()["error"] == f"file is larger than the {1000 / 1024**3:g} GB limit"
    assert zl.uploads == []


def test_huge_content_length_rejected_early(client, zl):
    r = client.post("/upload", content=b"x", headers={**AUTH, "content-length": str(10**12),
                                                       "content-type": "multipart/form-data; boundary=b"})
    assert r.status_code == 413


def test_zipline_error_is_502_and_cleans_up(client, zl, tmp_path):
    zl.fail_with = 500
    r = up(client, "a b.mkv", data=b"garbage")
    assert r.status_code == 502
    assert list(tmp_path.iterdir()) == []


def test_zipline_4xx_passed_through(client, zl):
    zl.fail_with = 401
    assert up(client, "a.mp4").status_code == 401


def test_bad_mkv_kept_as_mkv(client, zl):
    r = up(client, "a b.mkv", data=b"not a video")
    assert r.status_code == 200
    assert r.json() == {"url": "https://clips.example.com/a-b.mkv", "name": "a-b.mkv", "remuxed": False}


@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not installed")
def test_good_mkv_becomes_mp4(zl, tmp_path):
    src = tmp_path / "src.mkv"
    subprocess.run(["ffmpeg", "-v", "error", "-y", "-f", "lavfi", "-i", "testsrc=size=320x240:rate=30",
                    "-t", "1", "-c:v", "libx264", str(src)], check=True)
    work = tmp_path / "work"
    work.mkdir()
    settings = Settings(zipline_url="http://zipline:3000", site_origin="https://clips.example.com",
                        upload_url="https://upload.example.com", max_bytes=10**8, deletes_at="90d", work_dir=work)
    c = TestClient(create_app(settings, transport=httpx.MockTransport(zl.handler)))
    r = up(c, "my clip.mkv", data=src.read_bytes())
    assert r.json()["name"] == "my-clip.mp4"
    assert r.json()["remuxed"] is True
    assert zl.uploads[0]["headers"]["x-zipline-file-extension"] == ".mp4"
    assert list(work.iterdir()) == []


def test_generic_content_type_is_guessed_from_extension(client, zl):
    # curl/ShareX send application/octet-stream; Discord only plays video/mp4 inline
    assert up(client, "clip.mp4").status_code == 200
    assert b"Content-Type: video/mp4" in zl.uploads[0]["body"]


def test_unknown_extension_stays_octet_stream(client, zl):
    assert up(client, "data.zzz9").status_code == 200
    assert b"Content-Type: application/octet-stream" in zl.uploads[0]["body"]


def test_upload_with_ticket(client, zl):
    ticket = client.app.state.tickets.issue("tok-from-ticket")
    r = up(client, "t.mp4", headers={"x-upload-ticket": ticket})
    assert r.status_code == 200
    assert zl.uploads[0]["headers"]["authorization"] == "tok-from-ticket"


def test_ticket_single_use(client, zl):
    ticket = client.app.state.tickets.issue("tok")
    assert up(client, "a.mp4", headers={"x-upload-ticket": ticket}).status_code == 200
    r = up(client, "b.mp4", headers={"x-upload-ticket": ticket})
    assert r.status_code == 401
    assert r.json() == {"error": "upload ticket expired, try again"}
    assert len(zl.uploads) == 1


def test_bad_ticket_is_401(client, zl):
    assert up(client, "a.mp4", headers={"x-upload-ticket": "made-up"}).status_code == 401
    assert zl.uploads == []


def test_cors_preflight_for_site_origin(client):
    r = client.options("/upload", headers={
        "origin": "https://clips.example.com",
        "access-control-request-method": "POST",
        "access-control-request-headers": "x-upload-ticket",
    })
    assert r.status_code == 200
    assert r.headers["access-control-allow-origin"] == "https://clips.example.com"
    assert "x-upload-ticket" in r.headers["access-control-allow-headers"].lower()


def test_cors_refuses_other_origin(client):
    r = client.options("/upload", headers={
        "origin": "https://evil.example",
        "access-control-request-method": "POST",
    })
    assert "access-control-allow-origin" not in r.headers


def test_cors_headers_on_error_responses(client):
    r = up(client, "a.bin", data=b"x" * 1001, headers={**AUTH, "origin": "https://clips.example.com"})
    assert r.status_code == 413
    assert r.headers["access-control-allow-origin"] == "https://clips.example.com"


def test_link_is_built_from_the_name_not_ziplines_url(client, zl):
    # a '#' left unencoded in Zipline's url would cut the shared link off
    zl.raw_urls = True
    r = up(client, "çılgın an #1.mp4")
    assert r.json()["url"] == "https://clips.example.com/" + quote("çılgın-an-#1.mp4")


def test_zipline_down_during_name_check_is_502(client, zl):
    zl.head_fails = True
    r = up(client, "a.mp4")
    assert r.status_code == 502


def test_no_free_name_is_409(client, zl, monkeypatch):
    from app import server
    monkeypatch.setattr(server, "next_free_name", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("full")))
    assert up(client, "a.mp4").status_code == 409