theread.me/photo-upload/tests/test_upload.py

import io
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
import subprocess
import sys
import os
import random
import zipfile

import pytest
from PIL import Image

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from app import create_app

TOKEN = "t" * 48
AUTH = {"Authorization": "Bearer " + TOKEN}


@pytest.fixture
def service(tmp_path):
    app = create_app(tmp_path / "data", TOKEN)
    app.config["TESTING"] = True
    return app, app.test_client(), tmp_path / "data"


def photo(color="red", format="JPEG", size=(40, 20), exif=None):
    if exif is None:
        exif = Image.Exif()
        exif[36867] = "2025:01:02 03:04:05"
        exif[36881] = "+00:00"
    output = io.BytesIO()
    Image.new("RGB", size, color).save(output, format, **({"exif": exif} if exif else {}))
    return output.getvalue()


def upload(client, images=None, caption="", headers=AUTH, highlight=None):
    return client.post("/api/photos", headers=headers, data={
        "caption": caption,
        **({"highlight": highlight} if highlight is not None else {}),
        "photos": [(io.BytesIO(raw), "../../escape.jpg") for raw in (images or [photo()])],
    })


def test_authentication_and_request_limits(service):
    app, client, data = service
    assert upload(client, headers={}).status_code == 401
    assert upload(client, headers={"Authorization": "Bearer wrong"}).status_code == 401
    assert client.get("/api/photos/" + "a" * 64).status_code == 401
    assert client.get("/healthz").status_code == 200
    assert upload(client, caption="a" * 4001).status_code == 400
    assert client.post("/api/photos", headers=AUTH).status_code == 400
    assert upload(client, [photo()] * 21).status_code == 400
    app.config["MAX_CONTENT_LENGTH"] = 100
    assert upload(client).status_code == 413
    assert not list((data / "submissions").iterdir())


def test_validation_is_all_or_nothing(service):
    _, client, data = service
    assert upload(client, [photo(), b"not an image"]).status_code == 400
    assert upload(client, [photo(format="GIF")]).status_code == 400
    assert not list((data / "submissions").iterdir())
    assert not list((data / "temporary").iterdir())


def test_multi_megabyte_multipart_file(service):
    _, client, data = service
    output = io.BytesIO()
    exif = Image.Exif()
    exif[36867] = "2025:01:02 03:04:05"
    Image.frombytes("RGB", (1800, 1800), random.Random(2).randbytes(1800 * 1800 * 3)).save(
        output, "JPEG", quality=95, exif=exif)
    assert len(output.getvalue()) > 3 * 1024 * 1024
    response = upload(client, [output.getvalue()])
    assert response.status_code == 202, response.json
    path = data / "submissions" / response.json["id"] / "images/00.jpg"
    assert path.stat().st_size <= 300 * 1024
    with Image.open(path) as image:
        assert max(image.size) <= 1600
        assert image.info.get("progressive")


def test_zip_batch_preserves_all_images_order_and_deduplicates(service):
    _, client, data = service
    images = [photo("red"), photo("blue"), photo("green")]
    archive = io.BytesIO()
    with zipfile.ZipFile(archive, "w") as output:
        for i, raw in enumerate(images):
            output.writestr(f"selection/{i}.jpg", raw)
        output.writestr("__MACOSX/._0.jpg", b"metadata")
    response = client.post("/api/photos", headers=AUTH, data={
        "archive": (io.BytesIO(archive.getvalue()), "photos.zip"),
        "caption": "Batch", "highlight": "Trip",
    })
    assert response.status_code == 202, response.json
    assert response.json["count"] == 3
    folder = data / "submissions" / response.json["id"]
    for i, raw in enumerate(images):
        assert (folder / "originals" / str(i)).read_bytes() == raw
    assert upload(client, images, "Batch", highlight="Trip").json["id"] == response.json["id"]


def test_zip_invalid_or_oversized_batches_are_atomic(service):
    _, client, data = service
    for contents in ([b"bad"], [photo()] * 21):
        archive = io.BytesIO()
        with zipfile.ZipFile(archive, "w") as output:
            for i, raw in enumerate(contents):
                output.writestr(f"{i}.jpg", raw)
        response = client.post("/api/photos", headers=AUTH, data={"archive": (io.BytesIO(archive.getvalue()), "photos.zip")})
        assert response.status_code == 400
    assert not list((data / "submissions").iterdir())


def test_album_caption_order_and_retry_deduplication(service):
    _, client, data = service
    caption = '<script>alert(1)</script> {% include secret %} {{ site.email }}\nsecond line'
    images = [photo("red"), photo("blue")]
    response = upload(client, images, caption)
    assert response.status_code == 202
    job = response.json["id"]
    folder = data / "submissions" / job
    entries = sorted((folder / "entries").glob("*.md"))
    assert len(entries) == 2
    rendered = "\n".join(entry.read_text() for entry in entries)
    assert "<script>" not in rendered and "{%" not in rendered and "{{" not in rendered
    assert rendered.count("<br>second line") == 2
    assert "00.jpg" in entries[0].read_text() and "01.jpg" in entries[1].read_text()
    assert Image.open(folder / "images/00.jpg").getpixel((0, 0))[0] > 240
    assert Image.open(folder / "images/01.jpg").getpixel((0, 0))[2] > 240
    assert (folder / "originals/0").read_bytes() == images[0]
    assert upload(client, images, caption).status_code == 200
    assert len(list((data / "submissions").iterdir())) == 1
    assert client.get(response.json["status_url"], headers=AUTH).json["status"] == "queued"
    (data / "status" / f"{job}.json").write_text('{"status":"failed"}')
    assert client.post(f"/api/photos/{job}/retry", headers=AUTH).json["status"] == "queued"
    assert client.post("/api/photos/" + "a" * 64 + "/retry", headers=AUTH).status_code == 404


def test_orientation_metadata_and_heic(service):
    _, client, data = service
    exif = Image.Exif()
    exif[274] = 6
    exif[270] = "private metadata"
    exif[36867] = "2025:01:02 03:04:05"
    for raw in [photo(exif=exif, size=(3000, 1500)), photo(format="HEIF")]:
        response = upload(client, [raw])
        assert response.status_code == 202
        with Image.open(data / "submissions" / response.json["id"] / "images/00.jpg") as image:
            assert image.format == "JPEG"
            assert not image.getexif()
            assert max(image.size) <= 1600
            if raw[:2] == b"\xff\xd8":
                assert image.height > image.width


def test_photo_date_comes_from_original_metadata(service):
    _, client, data = service
    exif = Image.Exif()
    exif[36867] = "2025:04:18 14:23:11"
    exif[36881] = "+02:00"

    response = upload(client, [photo(exif=exif)])

    assert response.status_code == 202
    entry = (data / "submissions" / response.json["id"] / "entries/00.md").read_text()
    assert 'date: "2025-04-18T14:23:11+02:00"' in entry


def test_photo_dates_can_come_from_photos_library_metadata(service):
    _, client, data = service
    response = client.post("/api/photos", headers=AUTH, data={
        "photos": [
            (io.BytesIO(photo("red")), "first.jpg"),
            (io.BytesIO(photo("blue")), "second.jpg"),
        ],
        "photo_dates": "2025-04-18T14:23:11+02:00\n2025-04-19T09:10:05+02:00",
    })
    assert response.status_code == 202
    entries = sorted((data / "submissions" / response.json["id"] / "entries").glob("*.md"))
    assert 'date: "2025-04-18T14:23:11+02:00"' in entries[0].read_text()
    assert 'date: "2025-04-19T09:10:05+02:00"' in entries[1].read_text()

    invalid = client.post("/api/photos", headers=AUTH, data={
        "photos": [(io.BytesIO(photo()), "photo.jpg")],
        "photo_dates": "not-a-date",
    })
    assert invalid.status_code == 400
    assert "valid ISO dates" in invalid.json["error"]

    missing = upload(client, [photo(exif=Image.Exif())])
    assert missing.status_code == 202
    manifest = json.loads((data / "submissions" / missing.json["id"] / "manifest.json").read_text())
    assert datetime.fromisoformat(manifest["capture_dates"][0]).date() == datetime.now(timezone.utc).date()


def test_batch_date_taken_overrides_metadata_for_analog_photos(service):
    _, client, data = service
    response = client.post("/api/photos", headers=AUTH, data={
        "photos": [
            (io.BytesIO(photo(exif=Image.Exif())), "first.jpg"),
            (io.BytesIO(photo(exif=Image.Exif())), "second.jpg"),
        ],
        "photo_dates": "2026-01-02T03:04:05+00:00\n2026-01-03T03:04:05+00:00",
        "date_taken": "2025",
    })
    assert response.status_code == 202
    manifest = json.loads((data / "submissions" / response.json["id"] / "manifest.json").read_text())
    assert all(value.startswith("2025-01-01T") for value in manifest["capture_dates"])

    invalid = client.post("/api/photos", headers=AUTH, data={
        "photos": [(io.BytesIO(photo()), "photo.jpg")],
        "date_taken": "last summer",
    })
    assert invalid.status_code == 400
    assert "year, ISO date, or ISO timestamp" in invalid.json["error"]


def test_dated_retry_replaces_the_same_undated_upload(service):
    _, client, data = service
    raw = photo()
    original = upload(client, [raw], "Same caption", highlight="Trip").json
    replacement = client.post("/api/photos", headers=AUTH, data={
        "photos": [(io.BytesIO(raw), "photo.jpg")],
        "caption": "Same caption",
        "highlight": "Trip",
        "photo_dates": "2025-04-18T14:23:11+02:00",
    }).json

    assert replacement["id"] != original["id"]
    manifest = json.loads((data / "submissions" / replacement["id"] / "manifest.json").read_text())
    assert manifest["replaces"] == original["id"]


def test_optional_highlight_preserves_existing_upload_ids(service):
    _, client, data = service
    raw = photo()
    digest = hashlib.sha256(b"Caption")
    digest.update(hashlib.sha256(raw).digest())
    response = upload(client, [raw], "Caption")
    assert response.json["id"] == digest.hexdigest()
    assert upload(client, [raw], "Caption", highlight="  ").json["id"] == digest.hexdigest()
    entry = (data / "submissions" / digest.hexdigest() / "entry.md").read_text()
    assert "anchor:" not in entry
    assert entry.count(f'id="photo-{digest.hexdigest()}"') == 1


def test_highlight_album_and_safe_front_matter(service):
    _, client, data = service
    name = 'سفر: "Summer" & <friends>\n---\npublished: false'
    images = [photo("red"), photo("blue")]
    response = upload(client, images, "Album caption", highlight=name)
    assert response.status_code == 202
    job = response.json["id"]
    folder = data / "submissions" / job
    entries = sorted((folder / "entries").glob("*.md"))
    assert len(entries) == 2
    entry = entries[0].read_text()
    metadata, body = entry.split("---\n", 2)[1:]
    anchor = next(line for line in metadata.splitlines() if line.startswith("anchor: "))
    assert json.loads(anchor.removeprefix("anchor: ")) == name
    assert "\npublished:" not in metadata
    # The template emits the sole target at the start of this album, before 00.jpg.
    assert f'title: "photo-{job}"' in metadata
    assert f'id="photo-{job}"' not in body
    assert "00.jpg" in body and "01.jpg" in entries[1].read_text()
    assert "anchor:" not in entries[1].read_text()
    assert json.loads((folder / "manifest.json").read_text())["highlight"] == name
    assert upload(client, images, "Album caption", highlight=name).status_code == 200
    assert upload(client, images, "Album caption", highlight="Other trip").json["id"] != job
    assert upload(client, images, highlight="a" * 201).status_code == 400


def git(directory, *args):
    return subprocess.check_output(["git", "-C", str(directory), *args], text=True).strip()


@pytest.fixture
def publisher(tmp_path, service):
    _, client, data = service
    remote = tmp_path / "remote.git"
    source = tmp_path / "source"
    source.mkdir()
    subprocess.run(["git", "init", "--bare", str(remote)], check=True, capture_output=True)
    git(source, "init", "-b", "master")
    git(source, "config", "user.name", "Test")
    git(source, "config", "user.email", "test@example.com")
    (source / "art.html").write_text("Initial site")
    (source / "snippets.md").write_text("---\nlayout: post\n---\nIntroduction\n\n<!-- uploaded-snippets -->\n\nOld snippet\n")
    git(source, "add", ".")
    git(source, "commit", "-m", "Initial")
    git(source, "remote", "add", "origin", str(remote))
    git(source, "push", "origin", "master")
    env = dict(os.environ, PHOTO_DATA=str(data), PHOTO_WORK=str(tmp_path / "work"),
               PHOTO_REPOSITORY=str(remote), PHOTO_BRANCH="master")

    def run():
        return subprocess.run([sys.executable, str(Path(__file__).resolve().parents[1] / "publisher.py"), "--once"],
                              env=env, capture_output=True, text=True)
    return run, remote, client


def test_git_push_and_crash_recovery(publisher, service, monkeypatch, tmp_path):
    run, remote, client = publisher
    _, _, data = service
    response = upload(client)
    result = run()
    assert result.returncode == 0, result.stderr
    job = response.json["id"]
    paths = git(remote, "ls-tree", "-r", "--name-only", "master")
    assert f"img/arts/uploads/{job}/00.jpg" in paths
    assert "originals" not in paths
    assert client.get(response.json["status_url"], headers=AUTH).json["status"] == "pushed"
    head = git(remote, "rev-parse", "master")
    # Crash after push but before recording its success: retry must not create another commit.
    (data / "status" / f"{job}.json").write_text('{"status":"publishing"}')
    assert run().returncode == 0
    assert git(remote, "rev-parse", "master") == head
    receipt = tmp_path / "deployed.json"
    monkeypatch.setenv("PHOTO_DEPLOYMENT_FILE", str(receipt))
    receipt.write_text(json.dumps({"photos": [job], "commit": head}))
    assert client.get(response.json["status_url"], headers=AUTH).json["status"] == "published"


def test_publisher_names_photo_entry_from_capture_date(publisher, service):
    run, remote, client = publisher
    exif = Image.Exif()
    exif[36867] = "2025:04:18 14:23:11"
    response = upload(client, [photo(exif=exif)])

    result = run()

    assert result.returncode == 0, result.stderr
    paths = git(remote, "ls-tree", "-r", "--name-only", "master")
    assert f'_art/2025-04-18-photo-{response.json["id"]}-00.md' in paths


def test_publisher_replaces_an_undated_version_of_the_same_photos(publisher, service):
    run, remote, client = publisher
    raw = photo()
    original = upload(client, [raw], "Same caption", highlight="Trip").json
    assert run().returncode == 0
    replacement = client.post("/api/photos", headers=AUTH, data={
        "photos": [(io.BytesIO(raw), "photo.jpg")],
        "caption": "Same caption",
        "highlight": "Trip",
        "photo_dates": "2025-04-18T14:23:11+02:00",
    }).json

    result = run()

    assert result.returncode == 0, result.stderr
    paths = git(remote, "ls-tree", "-r", "--name-only", "master")
    assert f"photo-{original['id']}" not in paths
    assert f"img/arts/uploads/{original['id']}" not in paths
    assert f'_art/2025-04-18-photo-{replacement["id"]}-00.md' in paths


def test_rejected_push_can_retry(publisher):
    run, remote, client = publisher
    head = git(remote, "rev-parse", "master")
    response = upload(client)
    hook = remote / "hooks/pre-receive"
    hook.write_text("#!/bin/sh\nexit 1\n")
    hook.chmod(0o755)
    result = run()
    assert result.returncode != 0
    assert git(remote, "rev-parse", "master") == head
    assert client.get(response.json["status_url"], headers=AUTH).json["status"] == "failed"
    hook.unlink()
    client.post(response.json["status_url"] + "/retry", headers=AUTH)
    result = run()
    assert result.returncode == 0, result.stderr
    assert client.get(response.json["status_url"], headers=AUTH).json["status"] == "pushed"


def test_snippet_intake_validation_and_literal_text(service):
    _, client, data = service
    assert client.post("/api/snippets", data={"text": "hello"}).status_code == 401
    assert client.post("/api/snippets", headers=AUTH).status_code == 400
    assert client.post("/api/snippets", headers=AUTH, data={"text": "x" * 20001}).status_code == 400
    text = 'Book quote\nفارسی <script>bad</script> {% include secret %}\nhttps://example.com/?a=1&b=2'
    response = client.post("/api/snippets", headers=AUTH, data={"text": text})
    assert response.status_code == 202
    job = response.json["id"]
    assert response.json["url"] == f"/snippets/#snippet-{job}"
    assert client.get(f"/api/photos/{job}", headers=AUTH).status_code == 404
    entry = (data / "submissions" / job / "entry.md").read_text()
    assert '<script>' not in entry and '{%' not in entry
    assert '<p dir="auto">Book quote<br>\nفارسی' in entry
    assert '<a href="https://example.com/?a=1&amp;b=2">' in entry
    assert client.post("/api/snippets", headers=AUTH, data={"text": text}).status_code == 200
    assert client.post("/api/snippets", headers=AUTH, data={"photos": (io.BytesIO(b"bad"), "bad.jpg")}).status_code == 400


def test_snippet_single_photo_alias(service):
    _, client, data = service
    raw = photo()
    response = client.post("/api/snippets", headers=AUTH, data={
        "photo": (io.BytesIO(raw), "photo.jpg"), "text": "A note",
    })
    assert response.status_code == 202, response.json
    assert response.json["count"] == 1
    assert (data / "submissions" / response.json["id"] / "originals/0").read_bytes() == raw
    duplicate = client.post("/api/snippets", headers=AUTH, data={
        "photos": (io.BytesIO(raw), "photo.jpg"), "text": "A note",
    })
    assert duplicate.json["id"] == response.json["id"]
    assert duplicate.status_code == 200
    assert client.post("/api/snippets", headers=AUTH, data={
        "photo": [(io.BytesIO(raw), "one.jpg"), (io.BytesIO(raw), "two.jpg")],
    }).status_code == 400


def test_snippet_prepend_images_recovery_and_receipt(publisher, service, monkeypatch, tmp_path):
    run, remote, client = publisher
    _, _, data = service
    first = client.post("/api/snippets", headers=AUTH, data={"text": "First snippet"}).json
    second = client.post("/api/snippets", headers=AUTH, data={
        "caption": "Image note", "photos": (io.BytesIO(photo()), "photo.jpg"),
    }).json
    result = run()
    assert result.returncode == 0, result.stderr
    content = git(remote, "show", "master:snippets.md")
    assert content.startswith("---\nlayout: post\n---\nIntroduction")
    assert content.index(second["id"]) < content.index(first["id"]) < content.index("Old snippet")
    assert f'img/snippets/uploads/{second["id"]}/00.jpg' in git(remote, "ls-tree", "-r", "--name-only", "master")
    head = git(remote, "rev-parse", "master")
    for job in (first["id"], second["id"]):
        (data / "status" / f"{job}.json").write_text('{"status":"failed"}')
        assert client.post(f"/api/snippets/{job}/retry", headers=AUTH).json["status"] == "queued"
    assert run().returncode == 0
    assert git(remote, "rev-parse", "master") == head
    receipt = tmp_path / "deployed.json"
    monkeypatch.setenv("PHOTO_DEPLOYMENT_FILE", str(receipt))
    receipt.write_text(json.dumps({"snippets": [first["id"], second["id"]]}))
    assert client.get(first["status_url"], headers=AUTH).json["status"] == "published"