477 lines
20 KiB
Python
477 lines
20 KiB
Python
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_tiff_and_invalid_color_profile_are_accepted(service):
|
|
_, client, data = service
|
|
invalid_profile = io.BytesIO()
|
|
Image.new("RGB", (40, 20), "purple").save(
|
|
invalid_profile, "JPEG", icc_profile=b"not an ICC profile")
|
|
|
|
for raw in (photo(format="TIFF"), invalid_profile.getvalue()):
|
|
response = upload(client, [raw])
|
|
assert response.status_code == 202
|
|
output = data / "submissions" / response.json["id"] / "images/00.jpg"
|
|
with Image.open(output) as image:
|
|
assert image.format == "JPEG"
|
|
|
|
|
|
def test_heic_format_label_is_accepted(service, monkeypatch):
|
|
_, client, _ = service
|
|
actual_open = Image.open
|
|
|
|
class HeicLabel:
|
|
def __init__(self, image):
|
|
self.image = image
|
|
self.format = "HEIC"
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
self.image.close()
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self.image, name)
|
|
|
|
monkeypatch.setattr(Image, "open", lambda source: HeicLabel(actual_open(source)))
|
|
assert upload(client, [photo()]).status_code == 202
|
|
|
|
|
|
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&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"
|