203 lines
8.5 KiB
Python
203 lines
8.5 KiB
Python
import io
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import os
|
|
|
|
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):
|
|
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_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
|
|
entry = (folder / "entry.md").read_text()
|
|
assert "<script>" not in entry and "{%" not in entry and "{{" not in entry
|
|
assert "<br>second line" in entry
|
|
assert entry.index("00.jpg") < entry.index("01.jpg")
|
|
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"
|
|
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) <= 2560
|
|
if raw[:2] == b"\xff\xd8":
|
|
assert image.height > image.width
|
|
|
|
|
|
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
|
|
entry = (folder / "entry.md").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 body.index("00.jpg") < body.index("01.jpg")
|
|
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")
|
|
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_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"
|