diff --git a/.agents/skills/visual-acceptance/SKILL.md b/.agents/skills/visual-acceptance/SKILL.md index d5e4b43c..07cc7009 100644 --- a/.agents/skills/visual-acceptance/SKILL.md +++ b/.agents/skills/visual-acceptance/SKILL.md @@ -8,7 +8,8 @@ description: Capture, inspect, and attach feature-specific iOS simulator screens Make the requested behavior quick for Danis to assess in the PR description. Use the existing macOS CI/XCUI evidence pipeline; read [docs/remote-ios-verification.md](../../../docs/remote-ios-verification.md) -for retrieval, inspection, publication commands, and simulator limits. +for retrieval, inspection, publication commands, and simulator limits, including +the [compact PR media rules](../../../docs/remote-ios-verification.md#compact-pr-media). 1. Translate the request into observable acceptance criteria. Extend the relevant `GrowingUpUI` journey with assertions and named screenshot checkpoints that @@ -25,7 +26,7 @@ for retrieval, inspection, publication commands, and simulator limits. animation. Describe what you inspected, its outcome, and any device-only gaps in a temporary Markdown summary. Publish through `scripts/pr-visual-evidence.py`; it embeds native GitHub images/video in the PR body and preserves other sections. -4. Read back the PR description and confirm the media renders. After a new push or +4. Inspect the rendered PR attachments under the compact media rules. After a new push or CI rerun, replace evidence with reviewed media for the latest head/attempt before handoff. Media upload failures leave the description untouched; inspect a failed PR-update request before retrying. Remove diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md index 0ff2e279..31e9151f 100644 --- a/.github/pull_request_template.md +++ b/.github/pull_request_template.md @@ -11,6 +11,8 @@ Pending visual acceptance. TestFlight deployment requires Danis's explicit approval. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ed196bd6..821f48c3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,6 +43,8 @@ jobs: runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 + - name: Install media tools + run: sudo apt-get update && sudo apt-get install -y ffmpeg - name: Check evidence scripts run: | bash -n scripts/run-ui-journeys.sh diff --git a/AGENTS.md b/AGENTS.md index 356bec45..1fb5d875 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -5,7 +5,8 @@ - Never place DerivedData, `.xcresult` bundles, downloaded CI artifacts, or other reproducible validation output inside the repository. Use Xcode's default DerivedData location or a `mktemp` directory, and remove temporary validation artifacts when the run is complete. - Reuse an existing compatible simulator for local builds and tests. Never create a local simulator unless Danis explicitly asks for one; if temporary simulator creation is explicitly approved, delete it as soon as the run completes. - On Linux, validate iOS changes through the PR's macOS CI and `scripts/pr-evidence.sh PR_NUMBER`. Follow `docs/remote-ios-verification.md`: inspect relevant exported screenshots and recording frames for the latest head, fix concrete failures on the same PR, and remove downloaded evidence after review. A green build alone does not verify UI behavior. -- For user-visible features and UI fixes, use `.agents/skills/visual-acceptance/SKILL.md` before merging a PR or deploying to TestFlight. Attach inspected screenshots or a short feature demo to the PR description for the current head; choose the medium that proves the requested behavior. Deploy to TestFlight only after Danis explicitly approves deployment. +- Every screenshot in any PR description must follow the [compact media rules](docs/remote-ios-verification.md#compact-pr-media) for sizing, layout and rendered inspection. Publish reviewed CI screenshots through `scripts/pr-visual-evidence.py`. +- For user-visible features and UI fixes, use `.agents/skills/visual-acceptance/SKILL.md` before merging a PR or deploying to TestFlight. Attach inspected screenshots or a short feature demo to the PR description for the current head. Deploy to TestFlight only after Danis explicitly approves deployment. ## Code Style diff --git a/docs/remote-ios-verification.md b/docs/remote-ios-verification.md index 5a25e7bc..be96774c 100644 --- a/docs/remote-ios-verification.md +++ b/docs/remote-ios-verification.md @@ -106,12 +106,11 @@ on Linux (replace the example start and duration with the inspected segment): ```bash ffmpeg -ss 00:00:30 -i /tmp/EVIDENCE/ui-DEVICE-RUNTIME/journeys.mp4 -t 15 \ - -vf 'scale=720:-2' -c:v libx264 -crf 28 -preset fast -an -movflags +faststart \ + -c:v libx264 -crf 28 -preset fast -pix_fmt yuv420p -an -movflags +faststart \ /tmp/EVIDENCE/ui-DEVICE-RUNTIME/feature-demo.mp4 ``` -Inspect the resulting clip too. Keep each selected file below 10 MiB so it works -with GitHub's free-plan upload limit. Use screenshots from this run's exported +Inspect the resulting clip too. Use screenshots from this run's exported attachments, not checked-in comparison baselines. Write a temporary review summary that names the acceptance criteria demonstrated, the checkpoints/journey inspected, and any limits. Then publish, using real exported filenames: @@ -124,18 +123,18 @@ python3 scripts/pr-visual-evidence.py PR_NUMBER /tmp/EVIDENCE/ui-DEVICE-RUNTIME ``` Either `--image` or `--video` suffices; repeat the flags for additional relevant -media. The helper checks the PR head, latest successful CI run and attempt, and -artifact outcomes before uploading. It replaces only the marked visual acceptance -section, preserving the rest of the PR description, and records the full commit, -run link, simulator, and agent's review. Media is uploaded first through `gh api` -using the same native attachment endpoint as `gh --attach`; the helper then reads +media. Follow the [compact media rules](#compact-pr-media) for upload copies, +layout, and rendered inspection. The helper checks the PR head, latest successful +CI run and attempt, and artifact outcomes before uploading. It replaces only the +marked visual acceptance section, preserving the rest of the PR description, and +records the full commit, run link, simulator, and agent's review. Media is uploaded +first through `gh api` using the same native attachment endpoint as `gh --attach`; the helper then reads the latest description before replacing its evidence section. Upload failures leave the description untouched. GitHub does not support conditional PR writes, so an edit arriving between the final read and update can still race; coordinate -description edits during that final update. Read back the PR and check that -screenshots display and videos play. Inspect a failed PR-update request before -retrying. A concurrent push aborts publication or invalidates the helper's exact -section after publication, preserving fresh surrounding text. Retrieve and review +description edits during that final update. Inspect a failed PR-update request +before retrying. A concurrent push aborts publication or invalidates the helper's +exact section after publication, preserving fresh surrounding text. Retrieve and review evidence again before handoff. Upload errors are blockers, not a reason to substitute artifact download links for visible feature evidence. @@ -152,14 +151,42 @@ References: [GitHub CLI attachments](https://docs.github.com/en/github-cli/githu The upload request follows the [GitHub CLI implementation](https://github.com/cli/cli/blob/v2.102.0/internal/attachments/client.go); conditional-write limits are documented in [GitHub REST best practices](https://docs.github.com/en/rest/using-the-rest-api/best-practices-for-using-the-rest-api#use-conditional-requests). +## Compact PR media + +PR attachments fit within **320×640 pixels**, preserve aspect ratio, and never +upscale. The publisher requires local `ffmpeg` and its `ffprobe` tool for video timing; +install them with `brew install ffmpeg` +on macOS or `sudo apt install ffmpeg` on Ubuntu. It creates temporary upload copies +and leaves the full-resolution originals intact for visual comparison. Sources may +exceed 10 MiB; every copy must be nonempty, have a valid PNG or MP4 signature, and +fit within 10 MiB. All copies are converted and checked before the first upload. +Temporary copies are removed on success or failure. + +Videos retain the selected clip's complete duration and timing. Upload copies use +H.264/yuv420p MP4 with fast start and even dimensions within the same bounds. +Trimming selects the demonstrated interaction; preserve real timing for animation. + +Place two screenshot embeds in one Markdown paragraph, separated by a space, with +a blank line between pairs. Choose descriptive checkpoint filenames; the publisher +turns them into meaningful alt labels. Keep each native video URL on its own line +so GitHub renders its player. + +Inspect the original checkpoints and selected recording before publication. Then +inspect every rendered compact attachment in the PR at a **1280×800 viewport**: +confirm screenshot sharpness, compact bounds, and two-image grouping, and play +each native video to its end. Fix unreadable or poorly grouped evidence before +handoff; a successful upload alone does not complete visual acceptance. + ## Maintain the evidence tools +Install `ffmpeg` (including `ffprobe`) before running the helper tests. + ```bash bash -n scripts/run-ui-journeys.sh bash -n scripts/pr-evidence.sh PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s scripts/tests ``` -The focused tests exercise recorder finalization, failed-test status preservation, -build failures and portable report generation. Real simulator behavior is verified -by both macOS UI jobs. +The focused tests exercise real media conversion and publication, recorder +finalization, failed-test status preservation, build failures and portable report +generation. Real simulator behavior is verified by both macOS UI jobs. diff --git a/scripts/pr-visual-evidence.py b/scripts/pr-visual-evidence.py index 95f542c4..cbc9134f 100644 --- a/scripts/pr-visual-evidence.py +++ b/scripts/pr-visual-evidence.py @@ -3,8 +3,11 @@ import argparse import json +import re +import shutil import subprocess import tempfile +from fractions import Fraction from pathlib import Path from urllib.parse import urlencode @@ -12,6 +15,8 @@ START = "" END = "" MAX_BYTES = 10 * 1024 * 1024 +MAX_WIDTH = 320 +MAX_HEIGHT = 640 def gh(*arguments): @@ -34,11 +39,91 @@ def media_file(directory, relative, suffix): path = (directory / relative).resolve() if not path.is_relative_to(directory) or path.suffix.lower() != suffix: raise ValueError(f"Expected a {suffix} file inside the evidence directory: {relative}") - if not path.is_file() or not 0 < path.stat().st_size <= MAX_BYTES: - raise ValueError(f"Media must be nonempty and at most 10 MiB: {relative}") + if not path.is_file() or path.stat().st_size == 0: + raise ValueError(f"Media must be a nonempty file: {relative}") + with path.open("rb") as stream: + header = stream.read(12) + if suffix == ".png" and not header.startswith(b"\x89PNG\r\n\x1a\n"): + raise ValueError("Screenshot is not a PNG file") + if suffix == ".mp4" and header[4:8] != b"ftyp": + raise ValueError("Recording is not an MP4 file") return path +def compact_media(directory, images, videos, temporary): + sources = ([media_file(directory, name, ".png") for name in images] + + [media_file(directory, name, ".mp4") for name in videos]) + ffmpeg = shutil.which("ffmpeg") + if not ffmpeg: + raise ValueError("ffmpeg is required to create compact PR evidence; install it with `brew install ffmpeg` or `sudo apt install ffmpeg`.") + ffprobe = shutil.which("ffprobe") if videos else None + if videos and not ffprobe: + raise ValueError("ffprobe is required for video timing; install the ffmpeg package with `brew install ffmpeg` or `sudo apt install ffmpeg`.") + + compacted = [] + for index, source in enumerate(sources): + output_directory = temporary / f"{index:02d}" + output_directory.mkdir() + output = output_directory / source.name + if source.suffix.lower() == ".png": + command = [ffmpeg, "-nostdin", "-hide_banner", "-loglevel", "error", "-y", + "-i", str(source), "-frames:v", "1", "-vf", + f"scale=w='min({MAX_WIDTH},iw)':h='min({MAX_HEIGHT},ih)':force_original_aspect_ratio=decrease", + "-compression_level", "9", str(output)] + else: + # Passthrough preserves PTS; the final source packet fixes the encoder's trailing frame duration. + timing = json.loads(subprocess.check_output([ + ffprobe, "-v", "error", "-select_streams", "v:0", "-show_entries", + "stream=time_base:packet=pts,duration:format=start_time", + "-show_packets", "-show_streams", "-show_format", "-of", "json", str(source), + ], text=True)) + packets = timing.get("packets", []) + if not packets or not timing.get("streams"): + raise ValueError(f"Recording has no video timing: {source.name}") + # Packet order follows decoding, so select the last presented frame even with B-frames. + final_packet = max(packets, key=lambda packet: int(packet["pts"])) + time_base = Fraction(timing["streams"][0]["time_base"]) + duration = time_base * int(final_packet.get("duration", 0)) + if duration <= 0: + raise ValueError(f"Recording has no final frame duration: {source.name}") + final_pts = int(final_packet["pts"]) * time_base - Fraction(timing["format"]["start_time"]) + packet_timing = ( + "setts=pts=PTS:dts=DTS:duration=" + f"'if(eq(PTS,round(({final_pts})/TB)),({duration})/TB,DURATION)'" + ) + encoded_directory = output_directory / "encoded" + encoded_directory.mkdir() + encoded = encoded_directory / source.name + command = [ffmpeg, "-nostdin", "-hide_banner", "-loglevel", "error", "-y", + "-i", str(source), "-map", "0:v:0", "-map", "0:a?", "-vf", + f"scale=w='min({MAX_WIDTH},iw)':h='min({MAX_HEIGHT},ih)':force_original_aspect_ratio=decrease:force_divisible_by=2", + "-c:v", "libx264", "-preset", "veryfast", "-crf", "28", "-pix_fmt", "yuv420p", "-bf", "0", + "-c:a", "aac", "-b:a", "96k", + "-fps_mode", "passthrough", "-enc_time_base", str(time_base), "-bsf:v", packet_timing, + str(encoded)] + try: + subprocess.run(command, check=True, capture_output=True, text=True) + if source.suffix.lower() == ".mp4": + # Finalize movie metadata from encoded packet timing, preserving both streams without re-encoding. + subprocess.run([ + ffmpeg, "-nostdin", "-hide_banner", "-loglevel", "error", "-y", "-i", str(encoded), + "-map", "0", "-c", "copy", "-bsf:v", packet_timing, "-movflags", "+faststart", str(output), + ], check=True, capture_output=True, text=True) + except subprocess.CalledProcessError as error: + detail = error.stderr.strip() or "unknown conversion error" + raise ValueError(f"Could not create compact evidence for {source.name}: {detail}") from error + media_file(output_directory, output.name, source.suffix.lower()) + if output.stat().st_size > MAX_BYTES: + raise ValueError(f"Compact media must be at most 10 MiB: {source.name}") + compacted.append(output) + return compacted + + +def image_label(path): + label = " ".join(word.capitalize() for word in re.findall(r"[A-Za-z0-9]+", path.stem)) + return label or "Screenshot" + + def checked_run(repo, sha, metadata): workflow = metadata.get("workflow", "ci.yml") if workflow not in ("ci.yml", "app-store-screenshots.yml"): @@ -89,9 +174,9 @@ def publish(number, directory, summary, images, videos): raise ValueError("Describe the behavior inspected and any limits in the summary file.") if START in summary or END in summary: raise ValueError("Summary must not contain visual evidence markers.") - media = [media_file(directory, name, ".png") for name in images] - media += [media_file(directory, name, ".mp4") for name in videos] - if not 1 <= len(media) <= 50 or len(set(media)) != len(media): + sources = [media_file(directory, name, ".png") for name in images] + sources += [media_file(directory, name, ".mp4") for name in videos] + if not 1 <= len(sources) <= 50 or len(set(sources)) != len(sources): raise ValueError("Select 1–50 distinct screenshots or short videos.") repository = json.loads(gh("api", "repos/{owner}/{repo}")) repo = repository["full_name"] @@ -103,19 +188,23 @@ def publish(number, directory, summary, images, videos): checked_run(repo, sha, metadata) section(pr["body"] or "", "") # Reject malformed existing markers before uploading. - # Same native attachment endpoint used by gh --attach, without editing the PR during uploads. - attachments = [] - for path in media: - content_type = "video/mp4" if path.suffix.lower() == ".mp4" else "image/png" - query = urlencode({"repository_id": repository["id"], "name": path.name, "content_type": content_type}) - asset = json.loads(gh( - "api", f"https://uploads.github.com/user-attachments/assets?{query}", - "--method", "POST", "--header", "Content-Type: application/octet-stream", "--input", str(path), - )) - url = asset.get("url", "") - if not url.startswith("https://github.com/user-attachments/assets/"): - raise ValueError("GitHub did not return a native attachment URL.") - attachments.append((path, url)) + # Derive and validate every attachment before uploading; remove temporary copies on every exit. + with tempfile.TemporaryDirectory(prefix="growingup-compact-evidence-") as temporary_directory: + media = compact_media(directory, images, videos, Path(temporary_directory)) + + # Same native attachment endpoint used by gh --attach, without editing the PR during uploads. + attachments = [] + for path in media: + content_type = "video/mp4" if path.suffix.lower() == ".mp4" else "image/png" + query = urlencode({"repository_id": repository["id"], "name": path.name, "content_type": content_type}) + asset = json.loads(gh( + "api", f"https://uploads.github.com/user-attachments/assets?{query}", + "--method", "POST", "--header", "Content-Type: application/octet-stream", "--input", str(path), + )) + url = asset.get("url", "") + if not url.startswith("https://github.com/user-attachments/assets/"): + raise ValueError("GitHub did not return a native attachment URL.") + attachments.append((path, url)) run = checked_run(repo, sha, metadata) @@ -125,8 +214,13 @@ def publish(number, directory, summary, images, videos): f"{metadata['device']} / iOS {metadata['runtime']} / Xcode {metadata['xcode']}\n\n" f"{summary.strip()}\n\n" ) + image_attachments = [(path, url) for path, url in attachments if path.suffix.lower() == ".png"] + for index in range(0, len(image_attachments), 2): + row = image_attachments[index:index + 2] + evidence += " ".join(f"![{image_label(path)}]({url})" for path, url in row) + "\n\n" for path, url in attachments: - evidence += url + "\n\n" if path.suffix.lower() == ".mp4" else f"![Simulator checkpoint]({url})\n\n" + if path.suffix.lower() == ".mp4": + evidence += url + "\n\n" evidence += "TestFlight deployment awaits Danis's explicit approval.\n" + END # GitHub does not support conditional PR writes. Re-read after the slow uploads, immediately before PATCH. current = json.loads(gh("api", endpoint)) diff --git a/scripts/tests/test_pr_visual_evidence.py b/scripts/tests/test_pr_visual_evidence.py index a0a3bb60..9a8bdc6d 100644 --- a/scripts/tests/test_pr_visual_evidence.py +++ b/scripts/tests/test_pr_visual_evidence.py @@ -1,5 +1,7 @@ import importlib.util +import hashlib import json +import re import subprocess import tempfile import unittest @@ -26,8 +28,11 @@ def setUp(self): device="iPhone-17-Pro", runtime="26-5", xcode="26.6", ) self.image = self.directory / "checkpoint.png" - self.image.write_bytes(b"simulator screenshot") - (self.directory / "demo.mp4").write_bytes(b"trimmed simulator recording") + self.video = self.directory / "demo.mp4" + self.ffmpeg("-f", "lavfi", "-i", "testsrc=size=640x1280:rate=5", "-frames:v", "1", str(self.image)) + self.ffmpeg("-f", "lavfi", "-i", "testsrc=size=640x360:rate=5:duration=2", "-f", "lavfi", "-i", + "sine=frequency=1000:duration=2", "-map", "0:v:0", "-map", "1:a:0", "-c:v", "libx264", + "-pix_fmt", "yuv420p", "-c:a", "aac", "-shortest", "-movflags", "+faststart", str(self.video)) self.pr = dict( state="open", head={"sha": "current-head"}, body="Original description\n", html_url="https://github.com/owner/repo/pull/42", @@ -38,8 +43,18 @@ def setUp(self): ) self.edits = [] self.uploads = [] + self.upload_paths = [] self.queried_workflows = [] + def ffmpeg(self, *arguments): + subprocess.run(["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-y", *arguments], + check=True, capture_output=True, text=True) + + def probe(self, path): + result = subprocess.run(["ffprobe", "-v", "error", "-show_streams", "-show_format", "-show_frames", + "-of", "json", str(path)], check=True, capture_output=True, text=True) + return json.loads(result.stdout) + def gh(self, *args): if args == ("api", "repos/{owner}/{repo}"): return json.dumps({"full_name": "owner/repo", "id": 1234}) @@ -61,8 +76,11 @@ def upload(self, args): query = parse_qs(urlparse(args[1]).query) self.assertEqual(query["repository_id"], ["1234"]) self.assertIn(query["content_type"][0], ["image/png", "video/mp4"]) - self.assertTrue(Path(args[args.index("--input") + 1]).is_file()) - return json.dumps({"url": "https://github.com/user-attachments/assets/" + query["name"][0]}) + upload_path = Path(args[args.index("--input") + 1]) + self.assertTrue(upload_path.is_file()) + self.upload_paths.append(upload_path) + asset_name = re.sub(r"[^A-Za-z0-9-]", "-", query["name"][0]) + return json.dumps({"url": "https://github.com/user-attachments/assets/" + asset_name}) def edit(self, args): self.edits.append(args) @@ -85,8 +103,8 @@ def test_publication_embeds_media_and_preserves_other_sections_on_repeat(self): self.assertNotIn("Old evidence", body) self.assertIn("current-head", body) self.assertIn(self.run["html_url"], body) - self.assertIn("user-attachments/assets/checkpoint.png", body) - self.assertIn("user-attachments/assets/demo.mp4", body) + self.assertIn("user-attachments/assets/checkpoint-png", body) + self.assertIn("user-attachments/assets/demo-mp4", body) self.assertNotIn(self.temporary.name, body) def test_stale_head_or_attempt_never_uploads(self): @@ -194,6 +212,7 @@ def partial(args): self.assertEqual(len(self.uploads), 1) self.assertEqual(self.edits, []) self.assertEqual(self.pr["body"], "Original description\n") + self.assertTrue(all(not path.exists() for path in self.upload_paths)) def test_missing_published_section_is_not_reported_as_success(self): def missing(args): @@ -254,18 +273,170 @@ def test_summary_markers_are_rejected_before_upload(self): self.assertEqual(self.uploads, []) self.assertEqual(self.edits, []) - def test_media_rejects_external_files_and_oversize_uploads(self): + def test_media_rejects_external_files_and_invalid_signatures(self): with tempfile.TemporaryDirectory(prefix="growingup-external-") as other: external = Path(other) / "image.png" external.write_bytes(b"image") with self.assertRaises(ValueError): MODULE.media_file(self.directory, str(external), ".png") - with self.image.open("wb") as image: - image.truncate(MODULE.MAX_BYTES + 1) - with self.assertRaises(ValueError): + for path, invalid in ((self.image, b"not a PNG"), (self.video, b"not an MP4")): + with self.subTest(path=path): + original = path.read_bytes() + path.write_bytes(invalid) + with self.assertRaises(ValueError): + self.publish() + path.write_bytes(original) + self.assertEqual(self.edits, []) + self.assertEqual(self.uploads, []) + + def test_compact_derivatives_preserve_bounds_ratio_timing_and_originals(self): + originals = [self.image.read_bytes(), self.video.read_bytes()] + with tempfile.TemporaryDirectory(prefix="compact-output-") as temporary: + outputs = MODULE.compact_media(self.directory, ["checkpoint.png"], ["demo.mp4"], Path(temporary)) + self.assertEqual([self.image.read_bytes(), self.video.read_bytes()], originals) + for output, ratio in zip(outputs, (0.5, 640 / 360)): + info = self.probe(output) + video = next(stream for stream in info["streams"] if stream["codec_type"] == "video") + self.assertLessEqual(video["width"], 320) + self.assertLessEqual(video["height"], 640) + self.assertAlmostEqual(video["width"] / video["height"], ratio, places=2) + self.assertLessEqual(output.stat().st_size, MODULE.MAX_BYTES) + subprocess.run(["ffmpeg", "-v", "error", "-i", str(output), "-f", "null", "-"], + check=True, capture_output=True) + info = self.probe(outputs[1]) + video = next(stream for stream in info["streams"] if stream["codec_type"] == "video") + self.assertEqual(video["codec_name"], "h264") + self.assertEqual(video["pix_fmt"], "yuv420p") + self.assertEqual(video["width"] % 2, 0) + self.assertEqual(video["height"] % 2, 0) + self.assertAlmostEqual(float(info["format"]["duration"]), 2, delta=0.15) + self.assertIn("audio", [stream["codec_type"] for stream in info["streams"]]) + data = outputs[1].read_bytes() + self.assertLess(data.index(b"moov"), data.index(b"mdat")) + + def test_small_media_is_not_upscaled_and_variable_frame_timing_is_preserved(self): + image = self.directory / "small.png" + video = self.directory / "variable.mp4" + self.ffmpeg("-f", "lavfi", "-i", "testsrc=size=100x50:rate=1", "-frames:v", "1", str(image)) + cases = [("not(mod(n,3))", 0), ("not(mod(n,3))", 5), ("not(mod(n,4))+eq(n,18)", 0), + ("eq(n,0)+eq(n,2)+eq(n,5)+eq(n,9)+eq(n,13)+eq(n,17)+eq(n,19)", 0)] + for selection, offset in cases: + with self.subTest(selection=selection, offset=offset): + self.ffmpeg("-f", "lavfi", "-i", "testsrc=size=100x50:rate=10:duration=2", + "-vf", f"select='{selection}'", "-fps_mode", "vfr", "-c:v", "libx264", + "-pix_fmt", "yuv420p", "-output_ts_offset", str(offset), str(video)) + with tempfile.TemporaryDirectory(prefix="compact-output-") as temporary: + outputs = MODULE.compact_media(self.directory, [image.name], [video.name], Path(temporary)) + for output in outputs: + stream = self.probe(output)["streams"][0] + self.assertEqual((stream["width"], stream["height"]), (100, 50)) + original = self.probe(video) + compact = self.probe(outputs[1]) + self.assertGreater(original["streams"][0]["has_b_frames"], 0) + start = float(original["format"]["start_time"]) + packets = json.loads(subprocess.check_output([ + "ffprobe", "-v", "error", "-select_streams", "v:0", "-show_packets", "-show_entries", + "packet=pts,pts_time,duration_time", "-of", "json", str(video), + ], text=True))["packets"] + final_packet = max(packets, key=lambda packet: int(packet["pts"])) + expected_duration = float(final_packet["pts_time"]) + float(final_packet["duration_time"]) - start + self.assertAlmostEqual(float(compact["format"]["duration"]), expected_duration, delta=0.001) + self.assertEqual([f"{float(frame['best_effort_timestamp_time']) - start:.6f}" for frame in original["frames"]], + [frame["best_effort_timestamp_time"] for frame in compact["frames"]]) + + def test_oversized_png_source_is_compacted_and_preserved(self): + source = self.directory / "large.png" + self.ffmpeg("-f", "lavfi", "-i", "nullsrc=s=4800x2400:d=1,noise=alls=100:allf=t+u", + "-frames:v", "1", str(source)) + self.assertGreater(source.stat().st_size, MODULE.MAX_BYTES) + digest = hashlib.sha256(source.read_bytes()).digest() + with tempfile.TemporaryDirectory(prefix="compact-output-") as temporary: + output = MODULE.compact_media(self.directory, [source.name], [], Path(temporary))[0] + self.assertLessEqual(output.stat().st_size, MODULE.MAX_BYTES) + self.assertEqual(hashlib.sha256(source.read_bytes()).digest(), digest) + + def test_rational_frame_timing_preserves_video_duration_and_audio_offset(self): + source = self.directory / "audio-offset.mp4" + self.ffmpeg("-f", "lavfi", "-i", "testsrc=size=160x320:rate=30000/1001:duration=0.8", + "-f", "lavfi", "-i", "sine=sample_rate=44100:duration=0.8", "-map", "0:v:0", "-map", "1:a:0", + "-vf", "select='not(mod(n,4))+eq(n,22)'", "-fps_mode", "vfr", "-c:v", "libx264", + "-pix_fmt", "yuv420p", "-c:a", "aac", "-output_ts_offset", "3", str(source)) + original = self.probe(source) + original_video = next(stream for stream in original["streams"] if stream["codec_type"] == "video") + original_audio = next(stream for stream in original["streams"] if stream["codec_type"] == "audio") + self.assertGreater(original_video["has_b_frames"], 0) + tolerance = 1 / 30000 + with tempfile.TemporaryDirectory(prefix="compact-output-") as temporary: + output = MODULE.compact_media(self.directory, [], [source.name], Path(temporary))[0] + compact = self.probe(output) + video = next(stream for stream in compact["streams"] if stream["codec_type"] == "video") + audio = next(stream for stream in compact["streams"] if stream["codec_type"] == "audio") + # The stream interval reflects the MP4 sample timeline; demuxed final-packet duration can be wrong. + self.assertAlmostEqual(float(video["duration"]), float(original_video["duration"]), delta=tolerance) + self.assertAlmostEqual(float(video["start_time"]) - float(audio["start_time"]), + float(original_video["start_time"]) - float(original_audio["start_time"]), + delta=tolerance) + original_pts = [float(frame["best_effort_timestamp_time"]) - float(original["format"]["start_time"]) + for frame in original["frames"] if frame["media_type"] == "video"] + compact_pts = [float(frame["best_effort_timestamp_time"]) + for frame in compact["frames"] if frame["media_type"] == "video"] + self.assertEqual(len(compact_pts), len(original_pts)) + for expected, actual in zip(original_pts, compact_pts): + self.assertAlmostEqual(actual, expected, delta=tolerance) + + def test_derivative_names_cannot_collide_with_other_sources(self): + sources = [("a/02-frame.png", "red"), ("b/frame.png", "blue"), ("c/frame.png", "green")] + for relative, color in sources: + source = self.directory / relative + source.parent.mkdir(parents=True, exist_ok=True) + self.ffmpeg("-f", "lavfi", "-i", f"color=c={color}:s=320x640", "-frames:v", "1", str(source)) + with tempfile.TemporaryDirectory(prefix="compact-output-") as temporary: + outputs = MODULE.compact_media(self.directory, [relative for relative, _ in sources], [], Path(temporary)) + self.assertEqual(len({path.resolve() for path in outputs}), 3) + self.assertEqual([path.name for path in outputs], ["02-frame.png", "frame.png", "frame.png"]) + self.assertEqual(len({hashlib.sha256(path.read_bytes()).digest() for path in outputs}), 3) + + def test_failed_preflight_never_uploads_and_cleans_derivatives(self): + with patch.object(MODULE.shutil, "which", return_value=None), self.assertRaisesRegex(ValueError, "ffmpeg is required"): + self.publish() + ffmpeg = MODULE.shutil.which("ffmpeg") + with patch.object(MODULE.shutil, "which", side_effect=lambda tool: ffmpeg if tool == "ffmpeg" else None), \ + self.assertRaisesRegex(ValueError, "ffprobe is required"): + self.publish() + with patch.object(MODULE, "MAX_BYTES", 1), self.assertRaisesRegex(ValueError, "at most 10 MiB"): self.publish() + self.video.write_bytes(b"\x00\x00\x00\x20ftypisomnot really an mp4") + outputs = [] + original = MODULE.compact_media + def tracked(*args): + outputs.append(args[-1]) + return original(*args) + with patch.object(MODULE, "compact_media", side_effect=tracked), self.assertRaises(subprocess.CalledProcessError): + self.publish() + self.assertTrue(all(not path.exists() for path in outputs)) + self.assertEqual(self.uploads, []) self.assertEqual(self.edits, []) + def test_publisher_groups_pairs_with_checkpoint_labels_and_separate_video(self): + images = ["checkpoint.png", "people-list.png", "birthday-details.png", "person [edit]\\ draft.png", "saved.png"] + for name in images[1:]: + (self.directory / name).write_bytes(self.image.read_bytes()) + (self.directory / "metadata.json").write_text(json.dumps(self.metadata)) + with patch.object(MODULE, "gh", side_effect=self.gh): + MODULE.publish(42, self.directory, "Inspected evidence.", images, ["demo.mp4"]) + body = self.pr["body"] + self.assertRegex(body, r"!\[Checkpoint\]\([^\n]+\) !\[People List\]\([^\n]+\)\n\n" + r"!\[Birthday Details\]\([^\n]+\) !\[Person Edit Draft\]\([^\n]+\)\n\n" + r"!\[Saved\]\([^\n]+\)\n\nhttps://github.com/user-attachments/assets/demo-mp4\n\n") + self.assertTrue(all(not path.exists() for path in self.upload_paths)) + + def test_image_or_video_alone_still_suffices(self): + (self.directory / "metadata.json").write_text(json.dumps(self.metadata)) + with patch.object(MODULE, "gh", side_effect=self.gh): + MODULE.publish(42, self.directory, "Inspected checkpoint.", ["checkpoint.png"], []) + MODULE.publish(42, self.directory, "Inspected interaction.", [], ["demo.mp4"]) + self.assertEqual(len(self.uploads), 2) + def test_malformed_markers_do_not_destroy_body(self): for body in [MODULE.START, MODULE.END, MODULE.END + MODULE.START, MODULE.START * 2 + MODULE.END]: with self.subTest(body=body), self.assertRaises(ValueError):