From dbb6f818a38833081b4c9d22d349fc951449ab40 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 4 May 2026 17:09:45 +0000 Subject: [PATCH 1/3] fix(playwright_runner): transcode WebM bytes written into .mp4 paths (F1) Playwright's record_video_dir always writes WebM. Capture scripts that shutil.copy those bytes into a .mp4 path (to satisfy filename-based validators) end up with a WebM payload masquerading as MP4. The runner now sniffs the EBML header and transcodes in place via ffmpeg so downstream consumers always get a real ISO MP4. Real MP4 outputs pass through unchanged (back-compat preserved). Co-authored-by: John Menke --- src/docgen/playwright_runner.py | 56 ++++++++++++++++++++++++++ tests/test_playwright_runner.py | 70 +++++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+) diff --git a/src/docgen/playwright_runner.py b/src/docgen/playwright_runner.py index 7cebedf..c20f7f4 100644 --- a/src/docgen/playwright_runner.py +++ b/src/docgen/playwright_runner.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +import shutil import subprocess import sys from pathlib import Path @@ -16,6 +17,51 @@ class PlaywrightError(RuntimeError): """Raised when Playwright capture fails.""" +# EBML / Matroska / WebM files start with these four bytes. +_EBML_MAGIC = b"\x1a\x45\xdf\xa3" + + +def _looks_like_webm(path: Path) -> bool: + """Return True if the file's header is EBML (WebM/Matroska).""" + try: + with open(path, "rb") as f: + return f.read(4) == _EBML_MAGIC + except OSError: + return False + + +def _transcode_webm_to_mp4(src: Path, dst: Path) -> None: + """Transcode `src` (WebM) into `dst` (real MP4, libx264, +faststart). + + Used to fix F1: scripts that copy `.webm` bytes into the requested + `.mp4` path now get a real ISO MP4 emitted by docgen, so downstream + consumers don't have to know about the WebM-suffix mismatch. + """ + if shutil.which("ffmpeg") is None: + raise PlaywrightError( + "Playwright produced WebM but ffmpeg is not on PATH; " + "cannot transcode to MP4. Install ffmpeg or have the script " + "emit an ISO MP4 directly." + ) + tmp_dst = dst.with_suffix(dst.suffix + ".tmp.mp4") + cmd = [ + "ffmpeg", + "-y", + "-i", str(src), + "-c:v", "libx264", + "-pix_fmt", "yuv420p", + "-movflags", "+faststart", + "-c:a", "aac", + str(tmp_dst), + ] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode != 0: + raise PlaywrightError( + f"ffmpeg failed transcoding WebM → MP4: {proc.stderr[-400:]}" + ) + tmp_dst.replace(dst) + + class PlaywrightRunner: """Runs user-provided browser capture scripts for docgen segments.""" @@ -134,6 +180,16 @@ def capture( raise PlaywrightError( f"Playwright script finished but output is missing: {output_path}{hint}" ) + + # F1: Playwright's record_video_dir always writes WebM, but consumers + # historically `shutil.copy` those bytes into a .mp4 path to satisfy + # filename-based validators. Detect that by sniffing the file header + # (suffix is unreliable) and transcode in place when the path claims + # to be MP4. Files that are already real MP4 (or have a non-mp4 + # suffix) pass through untouched. + if output_path.suffix.lower() == ".mp4" and _looks_like_webm(output_path): + _transcode_webm_to_mp4(output_path, output_path) + return output_path def _resolve_path(self, value: Path | str) -> Path: diff --git a/tests/test_playwright_runner.py b/tests/test_playwright_runner.py index e1809ce..1effe48 100644 --- a/tests/test_playwright_runner.py +++ b/tests/test_playwright_runner.py @@ -51,6 +51,76 @@ def test_capture_runs_script_and_outputs_mp4(tmp_path: Path) -> None: assert output.read_bytes() == b"fake-mp4" +def test_capture_transcodes_webm_bytes_in_mp4_path( + tmp_path: Path, monkeypatch +) -> None: + """F1: scripts that copy WebM bytes into a .mp4 path get transcoded. + + Real Playwright `record_video_dir` always emits WebM. Many capture + scripts `shutil.copy` the .webm file into the .mp4 path the validator + expects; the runner now sniffs the EBML header and transcodes in + place so consumers always see a real ISO MP4 at the requested path. + """ + cfg = _write_cfg(tmp_path) + runner = PlaywrightRunner(cfg) + script = tmp_path / "capture.py" + script.write_text( + ( + "import os\n" + "from pathlib import Path\n" + "out = Path(os.environ['DOCGEN_PLAYWRIGHT_OUTPUT'])\n" + "out.parent.mkdir(parents=True, exist_ok=True)\n" + # EBML magic bytes — fake Playwright's WebM-into-mp4-path behavior. + "out.write_bytes(b'\\x1a\\x45\\xdf\\xa3' + b'rest of webm payload')\n" + ), + encoding="utf-8", + ) + + transcode_calls = [] + + def _fake_transcode(src: Path, dst: Path) -> None: + transcode_calls.append((src, dst)) + Path(dst).write_bytes(b"\x00\x00\x00\x18ftypisom....fake-mp4") + + monkeypatch.setattr( + "docgen.playwright_runner._transcode_webm_to_mp4", _fake_transcode + ) + + out = runner.capture(script=str(script), source="webm-in-mp4.mp4") + assert out.exists() + assert len(transcode_calls) == 1 + assert b"ftyp" in out.read_bytes() + + +def test_capture_passes_through_real_mp4_unchanged( + tmp_path: Path, monkeypatch +) -> None: + """A real ISO MP4 must NOT be transcoded (back-compat for F1).""" + cfg = _write_cfg(tmp_path) + runner = PlaywrightRunner(cfg) + script = tmp_path / "capture.py" + script.write_text( + ( + "import os\n" + "from pathlib import Path\n" + "out = Path(os.environ['DOCGEN_PLAYWRIGHT_OUTPUT'])\n" + "out.parent.mkdir(parents=True, exist_ok=True)\n" + "out.write_bytes(b'\\x00\\x00\\x00\\x18ftypisomalready-mp4')\n" + ), + encoding="utf-8", + ) + + def _fake_transcode(src: Path, dst: Path) -> None: # pragma: no cover + raise AssertionError("real MP4 should not be transcoded") + + monkeypatch.setattr( + "docgen.playwright_runner._transcode_webm_to_mp4", _fake_transcode + ) + + out = runner.capture(script=str(script), source="real.mp4") + assert b"ftyp" in out.read_bytes() + + def test_capture_builds_env_from_options(tmp_path: Path, monkeypatch) -> None: cfg = _write_cfg(tmp_path) runner = PlaywrightRunner(cfg) From 0b3b243e90d2864e4fa096266e1c50a24472b776 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 4 May 2026 17:10:05 +0000 Subject: [PATCH 2/3] feat(demo-function): add docgen demo-function subcommand MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds 'docgen demo-function' for rendering one short, single-purpose video per function (~the docs-site analogue of one Playwright test() per behavior). Inputs are declarative — either a *.docgen.yaml sidecar or a @pytest.mark.docgen(...) decorator on a Python test, read statically via ast (never imported / exec'd). Outputs are five files in --output-dir: rendered.mp4 (real ISO MP4), poster.png, fragment.txt (fn-), manifest.json (snapshot), cache-status.txt (hit/miss). Exit codes: 0 success, 1 invalid manifest / render failure, 2 missing ffmpeg / playwright (with install hint), 78 neutral skip (placeholder manifest with no demonstration.url) — mirrors Tekton's documented Skip exit code so CI does not flag placeholder-shaped manifests as failures. Action vocabulary (8 kinds): goto, click, fill, type, wait_for, wait_for_text, wait, screenshot. wait_for_text uses the native locator API instead of wait_for_function with embedded JS strings (F6). When OPENAI_API_KEY is set, the intent line is narrated via gpt-4o-mini-tts and muxed onto the video with -shortest. When unset, emits a documented warning to stderr and continues with a visual-only video (narration: null in manifest.json). Cache: when --cache-dir is given, content-addressed by sha256(identifier + intent + sorted(fixture_paths) + concat(fixture_contents))[:16]. Hits copy all four content files into output-dir verbatim and skip Playwright/ffmpeg/TTS entirely. Adds 31 unit tests covering manifest loading (both shapes), validation (resolution/duration/kind), action rendering, generated-script compile, fragment_id and cache_key stability, CLI exit codes, end-to-end render of kind=cli (ffmpeg-only, no Playwright), and the F7 docstring regression. Tests pass without Playwright or browsers installed. Co-authored-by: John Menke --- examples/lesson_compile.docgen.yaml | 21 + examples/sample_test.py | 37 + src/docgen/cli.py | 47 ++ src/docgen/demo_function.py | 1011 +++++++++++++++++++++++++++ tests/test_demo_function.py | 485 +++++++++++++ 5 files changed, 1601 insertions(+) create mode 100644 examples/lesson_compile.docgen.yaml create mode 100644 examples/sample_test.py create mode 100644 src/docgen/demo_function.py create mode 100644 tests/test_demo_function.py diff --git a/examples/lesson_compile.docgen.yaml b/examples/lesson_compile.docgen.yaml new file mode 100644 index 0000000..0f92f9c --- /dev/null +++ b/examples/lesson_compile.docgen.yaml @@ -0,0 +1,21 @@ +identifier: "course-builder/src/lessons/compileLesson.ts:compileLesson" +intent: | + Compiles a lesson markdown file into structured checkpoints and emits a status badge. +setup: + fixtures: + - tests/fixtures/lessons/intro.md +demonstration: + kind: playwright + url: "http://127.0.0.1:3000/lessons/new" + actions: + - {kind: type, selector: '[data-testid="title"]', value: "Intro to TS", delay_ms: 30} + - {kind: click, selector: '[data-testid="compile"]'} + - {kind: wait_for_text, selector: '[data-testid="status"]', text: "compiled", timeout_ms: 10000} + - {kind: wait, ms: 600} +assertions_to_surface: + - "lesson.status === 'compiled'" + - "checkpoint count = 3" +output_budget: + duration_seconds: 30 + segments: 1 + resolution: "1280x720" diff --git a/examples/sample_test.py b/examples/sample_test.py new file mode 100644 index 0000000..3940a7f --- /dev/null +++ b/examples/sample_test.py @@ -0,0 +1,37 @@ +"""Sample test demonstrating the @pytest.mark.docgen(...) marker shape. + +The marker is read statically by `docgen demo-function --manifest ::` +via `ast` (no import / no exec). The keyword args mirror the YAML keys. + +Note: this file does NOT need pytest to be installed at read time — the marker +is parsed from source. The triple-quoted text below intentionally talks about +``pytest.mark.docgen`` to verify the AST-based loader is not fooled by +docstring text (regression guard for F7). +""" + +import pytest + + +@pytest.mark.docgen( + identifier="course-builder/src/lessons/compileLesson.ts:compileLesson", + intent="Compiles a lesson markdown file into structured checkpoints and emits a status badge.", + setup={"fixtures": ["tests/fixtures/lessons/intro.md"]}, + demonstration={ + "kind": "playwright", + "url": "http://127.0.0.1:3000/lessons/new", + "actions": [ + {"kind": "type", "selector": '[data-testid="title"]', "value": "Intro to TS", "delay_ms": 30}, + {"kind": "click", "selector": '[data-testid="compile"]'}, + {"kind": "wait_for_text", "selector": '[data-testid="status"]', "text": "compiled", "timeout_ms": 10000}, + {"kind": "wait", "ms": 600}, + ], + }, + assertions_to_surface=[ + "lesson.status === 'compiled'", + "checkpoint count = 3", + ], + output_budget={"duration_seconds": 30, "segments": 1, "resolution": "1280x720"}, +) +def test_lesson_compile(): + """Render the compileLesson demo (sample for docgen demo-function).""" + pass diff --git a/src/docgen/cli.py b/src/docgen/cli.py index c7af63b..a8d3ef2 100644 --- a/src/docgen/cli.py +++ b/src/docgen/cli.py @@ -171,6 +171,53 @@ def playwright( click.echo(f"[playwright] captured: {video}") +@main.command("demo-function") +@click.option( + "--manifest", + "manifest_arg", + required=True, + help="Path to *.docgen.yaml sidecar OR .py:: for @pytest.mark.docgen.", +) +@click.option( + "--output-dir", + "output_dir_arg", + required=True, + type=click.Path(file_okay=False), + help="Directory to write rendered.mp4, poster.png, fragment.txt, manifest.json, cache-status.txt.", +) +@click.option( + "--cache-dir", + "cache_dir_arg", + default=None, + type=click.Path(file_okay=False), + help="Optional cache directory keyed by sha256(identifier+intent+fixtures).", +) +@click.option( + "--no-narration", + is_flag=True, + help="Skip TTS even if OPENAI_API_KEY is set.", +) +@click.pass_context +def demo_function( + ctx: click.Context, + manifest_arg: str, + output_dir_arg: str, + cache_dir_arg: str | None, + no_narration: bool, +) -> None: + """Render a single per-function demo video from a declarative manifest.""" + from docgen.demo_function import run_cli + + code = run_cli( + manifest_arg=manifest_arg, + output_dir_arg=output_dir_arg, + cache_dir_arg=cache_dir_arg, + no_narration=no_narration, + ) + if code != 0: + raise SystemExit(code) + + @main.command("tape-lint") @click.option("--tape", default=None, help="Lint a single tape name or pattern.") @click.pass_context diff --git a/src/docgen/demo_function.py b/src/docgen/demo_function.py new file mode 100644 index 0000000..c06cddd --- /dev/null +++ b/src/docgen/demo_function.py @@ -0,0 +1,1011 @@ +"""Per-function video docs subcommand: declarative Playwright + narration. + +This module implements `docgen demo-function`, which renders one short MP4 per +function from a declarative manifest (either a `*.docgen.yaml` sidecar or a +`@pytest.mark.docgen(...)` decorator on a Python test). The output is one +function → one ≤60s clip with a one-sentence narration, a poster frame, a +stable URL fragment, and a JSON manifest snapshot — designed for downstream +docs sites that want one video per function. + +Exit codes (used by `docgen.cli:demo_function`): + 0 render succeeded; all five artifacts written + 1 manifest invalid OR render failed + 2 required tooling missing (ffmpeg / playwright / browser) + 78 manifest is a placeholder (kind=playwright with no url) — neutral skip + +Supported action `kind`s: goto, click, fill, type, wait_for, wait_for_text, +wait, screenshot. Unknown kinds raise `ManifestError`. +""" + +from __future__ import annotations + +import ast +import contextlib +import hashlib +import json +import os +import re +import shutil +import subprocess +import sys +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Iterable + +import yaml + +# Exit code for "neutral skip" on placeholder manifests. Mirrors Tekton's +# documented Skip exit code so CI pipelines do not treat placeholder-shaped +# manifests as failures. +EXIT_NEUTRAL_SKIP = 78 +EXIT_TOOLING_MISSING = 2 +EXIT_INVALID = 1 +EXIT_OK = 0 + +HARD_CAP_SECONDS = 60 +DEFAULT_DURATION_SECONDS = 30 +DEFAULT_RESOLUTION = "1280x720" +RESOLUTION_RE = re.compile(r"^\d+x\d+$") +FRAGMENT_PREFIX_RE = re.compile(r"^fn-[a-z0-9-]+$") + +SUPPORTED_ACTION_KINDS = ( + "goto", + "click", + "fill", + "type", + "wait_for", + "wait_for_text", + "wait", + "screenshot", +) + +CACHED_ARTIFACTS = ("rendered.mp4", "poster.png", "fragment.txt", "manifest.json") + + +class ManifestError(ValueError): + """Raised when a manifest is malformed or violates the documented schema.""" + + +class PlaceholderManifest(Exception): + """Raised for placeholder manifests (kind=playwright with no url). + + The CLI translates this into exit code 78 (neutral skip). + """ + + +class ToolingMissingError(RuntimeError): + """Raised when a required external tool (ffmpeg, playwright) is missing. + + The CLI translates this into exit code 2 and prints the install hint that + accompanies the exception. + """ + + def __init__(self, message: str, install_hint: str) -> None: + super().__init__(message) + self.install_hint = install_hint + + +# --------------------------------------------------------------------------- +# Dataclasses +# --------------------------------------------------------------------------- + + +@dataclass +class Action: + """One declarative browser action. + + `kind` is one of `SUPPORTED_ACTION_KINDS`. `params` carries the rest of + the YAML mapping for the action (selector, value, ms, etc.). + """ + + kind: str + params: dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_mapping(cls, raw: Any) -> "Action": + if not isinstance(raw, dict): + raise ManifestError(f"action must be a mapping, got: {type(raw).__name__}") + kind = raw.get("kind") + if not isinstance(kind, str) or not kind: + raise ManifestError("action missing required field: 'kind'") + if kind not in SUPPORTED_ACTION_KINDS: + supported = ", ".join(SUPPORTED_ACTION_KINDS) + raise ManifestError( + f"unsupported action kind: '{kind}' (supported: {supported})" + ) + params = {k: v for k, v in raw.items() if k != "kind"} + return cls(kind=kind, params=params) + + +@dataclass +class Manifest: + """Normalised, validated manifest for one demo-function render.""" + + identifier: str + intent: str + kind: str + url: str | None = None + actions: list[Action] = field(default_factory=list) + fixtures: list[str] = field(default_factory=list) + assertions_to_surface: list[str] = field(default_factory=list) + duration_seconds: int = DEFAULT_DURATION_SECONDS + resolution: str = DEFAULT_RESOLUTION + source_path: Path | None = None + + @property + def viewport(self) -> tuple[int, int]: + w, h = self.resolution.split("x", 1) + return int(w), int(h) + + @property + def fragment_id(self) -> str: + slug = re.sub(r"[^a-z0-9]+", "-", self.identifier.lower()) + slug = re.sub(r"-+", "-", slug).strip("-") + return f"fn-{slug}" if slug else "fn-unknown" + + @property + def cache_key(self) -> str: + h = hashlib.sha256() + h.update(self.identifier.encode("utf-8")) + h.update(b"\x00") + h.update(self.intent.encode("utf-8")) + h.update(b"\x00") + for fixture in sorted(self.fixtures): + h.update(fixture.encode("utf-8")) + h.update(b"\x00") + # Concat fixture contents (relative to source_path's parent if known). + for fixture in sorted(self.fixtures): + content = self._read_fixture_bytes(fixture) + if content is not None: + h.update(content) + h.update(b"\x00") + return h.hexdigest()[:16] + + def _read_fixture_bytes(self, fixture: str) -> bytes | None: + candidates: list[Path] = [] + p = Path(fixture) + if p.is_absolute(): + candidates.append(p) + else: + if self.source_path is not None: + candidates.append((self.source_path.parent / p).resolve()) + candidates.append(Path.cwd() / p) + for c in candidates: + if c.exists() and c.is_file(): + try: + return c.read_bytes() + except OSError: + return None + return None + + +# --------------------------------------------------------------------------- +# Manifest loading +# --------------------------------------------------------------------------- + + +def load_manifest(spec: str | Path) -> Manifest: + """Load a manifest from either a YAML sidecar path or `path.py::test_name`. + + Raises `ManifestError` for invalid manifests, `FileNotFoundError` if the + path does not exist. + """ + if isinstance(spec, Path): + return _load_yaml_sidecar(spec) + text = str(spec) + if "::" in text: + path_part, _, test_name = text.partition("::") + py_path = Path(path_part) + if not py_path.exists(): + raise FileNotFoundError(f"manifest not found: {py_path}") + return _load_pytest_marker(py_path, test_name) + p = Path(text) + if not p.exists(): + raise FileNotFoundError(f"manifest not found: {p}") + if p.suffix == ".py": + raise ManifestError( + "Python manifest must use 'path.py::test_name' syntax to select a test" + ) + return _load_yaml_sidecar(p) + + +def _load_yaml_sidecar(path: Path) -> Manifest: + raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + if not isinstance(raw, dict): + raise ManifestError(f"manifest must be a mapping, got: {type(raw).__name__}") + return _coerce(raw, source_path=path) + + +def _load_pytest_marker(path: Path, test_name: str) -> Manifest: + """Read `@pytest.mark.docgen(...)` decorator on `test_name` via `ast`. + + Walking the AST avoids two common failure modes: + - `regex over source` matches markdown text inside module docstrings + that talk *about* the marker (F7). + - `import` / `exec` runs the test file's top-level code (and its + dependencies), which is unsafe and slow during static read. + """ + src = path.read_text(encoding="utf-8") + try: + tree = ast.parse(src, filename=str(path)) + except SyntaxError as exc: + raise ManifestError(f"could not parse {path}: {exc}") from exc + + target_fn: ast.FunctionDef | None = None + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef) and node.name == test_name: + target_fn = node + break + if target_fn is None: + raise ManifestError(f"function not found in {path}: {test_name}") + + marker_call: ast.Call | None = None + for dec in target_fn.decorator_list: + if not isinstance(dec, ast.Call): + continue + if _is_pytest_mark_docgen(dec.func): + marker_call = dec + break + if marker_call is None: + raise ManifestError( + f"{path}::{test_name} is missing @pytest.mark.docgen(...) decorator" + ) + + raw: dict[str, Any] = {} + for kw in marker_call.keywords: + if kw.arg is None: + continue + try: + raw[kw.arg] = ast.literal_eval(kw.value) + except (ValueError, SyntaxError) as exc: + raise ManifestError( + f"@pytest.mark.docgen({kw.arg}=...) must be a literal: {exc}" + ) from exc + return _coerce(raw, source_path=path) + + +def _is_pytest_mark_docgen(node: ast.AST) -> bool: + """Return True if `node` represents `pytest.mark.docgen` (or the bare + `docgen` mark imported from `pytest.mark`).""" + if isinstance(node, ast.Attribute) and node.attr == "docgen": + if isinstance(node.value, ast.Attribute) and node.value.attr == "mark": + inner = node.value.value + if isinstance(inner, ast.Name) and inner.id == "pytest": + return True + return False + + +def _coerce(raw: dict[str, Any], *, source_path: Path | None = None) -> Manifest: + """Validate `raw` and produce a normalised `Manifest`. + + Raises `ManifestError` for any schema violation. + """ + for required in ("identifier", "intent"): + if required not in raw or not str(raw.get(required, "")).strip(): + raise ManifestError(f"manifest missing required field: '{required}'") + + demonstration = raw.get("demonstration") + if not isinstance(demonstration, dict) or "kind" not in demonstration: + raise ManifestError("manifest missing required field: 'demonstration.kind'") + kind = str(demonstration.get("kind", "")).strip() + if kind not in {"playwright", "cli"}: + raise ManifestError( + f"demonstration.kind must be 'playwright' or 'cli', got: '{kind}'" + ) + + url = demonstration.get("url") + if url is not None: + url = str(url).strip() or None + + actions_raw = demonstration.get("actions") or [] + if not isinstance(actions_raw, list): + raise ManifestError("demonstration.actions must be a list") + actions = [Action.from_mapping(a) for a in actions_raw] + + setup = raw.get("setup") or {} + fixtures_raw = setup.get("fixtures", []) if isinstance(setup, dict) else [] + if not isinstance(fixtures_raw, list): + raise ManifestError("setup.fixtures must be a list of paths") + fixtures = [str(f) for f in fixtures_raw] + + assertions_raw = raw.get("assertions_to_surface") or [] + if not isinstance(assertions_raw, list): + raise ManifestError("assertions_to_surface must be a list of strings") + assertions: list[str] = [] + for a in assertions_raw: + s = str(a) + if len(s) > 60: + raise ManifestError( + f"assertions_to_surface entries must be ≤ 60 chars: '{s[:80]}'" + ) + assertions.append(s) + + output_budget = raw.get("output_budget") or {} + if not isinstance(output_budget, dict): + raise ManifestError("output_budget must be a mapping") + duration = int(output_budget.get("duration_seconds", DEFAULT_DURATION_SECONDS)) + if duration > HARD_CAP_SECONDS: + raise ManifestError( + f"output_budget.duration_seconds={duration} exceeds the 60s hard cap" + ) + if duration <= 0: + raise ManifestError("output_budget.duration_seconds must be positive") + resolution = str(output_budget.get("resolution", DEFAULT_RESOLUTION)) + if not RESOLUTION_RE.match(resolution): + raise ManifestError( + f"output_budget.resolution must match WxH (e.g. 1280x720), got: '{resolution}'" + ) + + return Manifest( + identifier=str(raw["identifier"]).strip(), + intent=str(raw["intent"]).strip(), + kind=kind, + url=url, + actions=actions, + fixtures=fixtures, + assertions_to_surface=assertions, + duration_seconds=duration, + resolution=resolution, + source_path=source_path, + ) + + +# --------------------------------------------------------------------------- +# Action -> Playwright source rendering +# --------------------------------------------------------------------------- + + +def _render_action(action: Action) -> str: + """Translate one action into a single line of Playwright sync_api code. + + Selectors and values are passed through `repr()` so embedded quotes are + escaped without any string-template gymnastics (F6). + """ + p = action.params + if action.kind == "goto": + url = p.get("url") or "" + return f"page.goto({url!r}, wait_until=\"networkidle\")" + if action.kind == "click": + return f"page.click({p['selector']!r})" + if action.kind == "fill": + return f"page.fill({p['selector']!r}, {p['value']!r})" + if action.kind == "type": + delay = int(p.get("delay_ms", 40)) + sel = p["selector"] + val = p["value"] + return ( + f"page.click({sel!r}); " + f"page.keyboard.type({val!r}, delay={delay})" + ) + if action.kind == "wait_for": + timeout = int(p.get("timeout_ms", 10000)) + return f"page.wait_for_selector({p['selector']!r}, timeout={timeout})" + if action.kind == "wait_for_text": + timeout = int(p.get("timeout_ms", 10000)) + sel = p["selector"] + text = p["text"] + return ( + f"page.locator({sel!r}).filter(has_text={text!r}).first." + f"wait_for(state=\"visible\", timeout={timeout})" + ) + if action.kind == "wait": + return f"page.wait_for_timeout({int(p['ms'])})" + if action.kind == "screenshot": + return f"page.screenshot(path={str(p['path'])!r})" + raise ManifestError(f"unsupported action kind: '{action.kind}'") + + +def generate_capture_script(manifest: Manifest, *, output_path: Path) -> str: + """Generate a standalone Playwright capture script for `manifest`. + + This is exposed (rather than being purely internal) so consumers can + inspect the generated script and so unit tests can assert that the + output compiles without launching Playwright. + """ + width, height = manifest.viewport + initial_url = manifest.url or "" + action_lines = [_render_action(a) for a in manifest.actions] + overlay_assertions = json.dumps(manifest.assertions_to_surface) + body = "\n ".join(action_lines) if action_lines else "pass" + return _SCRIPT_TEMPLATE.format( + output_path=str(output_path), + width=width, + height=height, + url=initial_url, + body=body, + overlay_assertions=overlay_assertions, + ) + + +_SCRIPT_TEMPLATE = '''"""Auto-generated Playwright capture script (docgen demo-function).""" + +from pathlib import Path + +from playwright.sync_api import sync_playwright + + +def main() -> None: + output_path = Path({output_path!r}) + output_path.parent.mkdir(parents=True, exist_ok=True) + with sync_playwright() as pw: + browser = pw.chromium.launch(headless=True) + context = browser.new_context( + viewport={{"width": {width}, "height": {height}}}, + record_video_dir=str(output_path.parent), + record_video_size={{"width": {width}, "height": {height}}}, + ) + page = context.new_page() + try: + initial_url = {url!r} + if initial_url: + page.goto(initial_url, wait_until="networkidle") + {body} + finally: + video_path = page.video.path() if page.video else None + context.close() + browser.close() + if video_path: + Path(video_path).rename(output_path) + + +if __name__ == "__main__": + main() +''' + + +# --------------------------------------------------------------------------- +# Cache +# --------------------------------------------------------------------------- + + +def _cache_lookup(cache_dir: Path, cache_key: str, output_dir: Path) -> bool: + """If a cache entry exists, copy artifacts into `output_dir`. Returns hit.""" + entry = cache_dir / cache_key + if not entry.exists() or not entry.is_dir(): + return False + for name in CACHED_ARTIFACTS: + f = entry / name + if not f.exists() or f.stat().st_size == 0: + return False + output_dir.mkdir(parents=True, exist_ok=True) + for name in CACHED_ARTIFACTS: + shutil.copy2(entry / name, output_dir / name) + return True + + +def _cache_store(cache_dir: Path, cache_key: str, output_dir: Path) -> None: + entry = cache_dir / cache_key + entry.mkdir(parents=True, exist_ok=True) + for name in CACHED_ARTIFACTS: + src = output_dir / name + if src.exists(): + shutil.copy2(src, entry / name) + + +# --------------------------------------------------------------------------- +# ffmpeg helpers +# --------------------------------------------------------------------------- + + +def _ensure_ffmpeg() -> None: + if shutil.which("ffmpeg") is None: + raise ToolingMissingError( + "ffmpeg not found on PATH", + install_hint="apt-get install -y ffmpeg # or: brew install ffmpeg", + ) + + +def _ensure_ffprobe() -> None: + if shutil.which("ffprobe") is None: + raise ToolingMissingError( + "ffprobe not found on PATH", + install_hint="apt-get install -y ffmpeg # or: brew install ffmpeg", + ) + + +def _transcode_to_mp4(src: Path, dst: Path, *, width: int, height: int) -> None: + """Transcode a video to MP4 (libx264, yuv420p, +faststart) at WxH.""" + _ensure_ffmpeg() + dst.parent.mkdir(parents=True, exist_ok=True) + cmd = [ + "ffmpeg", + "-y", + "-i", str(src), + "-vf", f"scale={width}:{height}:force_original_aspect_ratio=decrease," + f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2", + "-c:v", "libx264", + "-pix_fmt", "yuv420p", + "-movflags", "+faststart", + "-an", + str(dst), + ] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode != 0: + raise RuntimeError( + f"ffmpeg transcode failed: {proc.stderr[-400:]}" + ) + + +def _extract_poster(video: Path, poster: Path) -> None: + """Extract the last frame of `video` into `poster` (PNG).""" + _ensure_ffmpeg() + poster.parent.mkdir(parents=True, exist_ok=True) + cmd = [ + "ffmpeg", + "-y", + "-sseof", "-0.1", + "-i", str(video), + "-update", "1", + "-frames:v", "1", + str(poster), + ] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode != 0: + raise RuntimeError( + f"ffmpeg poster extraction failed: {proc.stderr[-400:]}" + ) + + +def _probe_audio_ms(audio_path: Path) -> int | None: + if shutil.which("ffprobe") is None: + return None + proc = subprocess.run( + [ + "ffprobe", "-v", "error", + "-show_entries", "format=duration", + "-of", "csv=p=0", + str(audio_path), + ], + capture_output=True, text=True, + ) + try: + return int(round(float(proc.stdout.strip()) * 1000)) + except ValueError: + return None + + +def _mux_audio(video: Path, audio: Path, dst: Path) -> None: + _ensure_ffmpeg() + cmd = [ + "ffmpeg", "-y", + "-i", str(video), + "-i", str(audio), + "-c:v", "copy", + "-c:a", "aac", + "-shortest", + "-movflags", "+faststart", + str(dst), + ] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode != 0: + raise RuntimeError(f"ffmpeg mux failed: {proc.stderr[-400:]}") + + +# --------------------------------------------------------------------------- +# Narration +# --------------------------------------------------------------------------- + + +@dataclass +class NarrationResult: + audio_path: Path + voice: str + model: str + ms: int + + +def _generate_narration( + intent: str, + work_dir: Path, + *, + voice: str = "coral", + model: str = "gpt-4o-mini-tts", +) -> NarrationResult: + """Generate narration MP3 from `intent` using the OpenAI TTS path. + + Mirrors `docgen.tts.TTSGenerator` but is self-contained — demo-function + runs are one-shot and don't need the segment-aware machinery. + """ + import openai + + out = work_dir / "narration.mp3" + client = openai.OpenAI() + response = client.audio.speech.create( + model=model, + voice=voice, + input=intent, + instructions=( + "You are narrating a per-function code demo. Speak in a calm, " + "professional tone. One sentence." + ), + ) + response.stream_to_file(str(out)) + ms = _probe_audio_ms(out) or 0 + return NarrationResult(audio_path=out, voice=voice, model=model, ms=ms) + + +# --------------------------------------------------------------------------- +# Render orchestration +# --------------------------------------------------------------------------- + + +@dataclass +class RenderResult: + output_dir: Path + cache_status: str # "hit" or "miss" + manifest: Manifest + narration: NarrationResult | None = None + + +def render( + manifest: Manifest, + output_dir: Path, + *, + cache_dir: Path | None = None, + no_narration: bool = False, + stderr=None, +) -> RenderResult: + """Render one demo-function manifest into `output_dir`. + + Raises: + PlaceholderManifest: if `kind=playwright` and no `url` was provided. + ToolingMissingError: if a required external tool (ffmpeg / playwright) + is missing. + ManifestError / RuntimeError: on render failures. + """ + if stderr is None: + stderr = sys.stderr + output_dir = Path(output_dir).resolve() + + if manifest.kind == "playwright" and not manifest.url: + raise PlaceholderManifest( + f"manifest is a placeholder (no demonstration.url): " + f"{manifest.identifier}" + ) + + output_dir.mkdir(parents=True, exist_ok=True) + + cache_key = manifest.cache_key + if cache_dir is not None: + cache_dir = Path(cache_dir).resolve() + cache_dir.mkdir(parents=True, exist_ok=True) + if _cache_lookup(cache_dir, cache_key, output_dir): + (output_dir / "cache-status.txt").write_text("hit\n", encoding="utf-8") + return RenderResult( + output_dir=output_dir, + cache_status="hit", + manifest=manifest, + ) + + rendered_mp4 = output_dir / "rendered.mp4" + poster_png = output_dir / "poster.png" + fragment_txt = output_dir / "fragment.txt" + manifest_json = output_dir / "manifest.json" + cache_status_txt = output_dir / "cache-status.txt" + + width, height = manifest.viewport + + with tempfile.TemporaryDirectory(prefix="docgen-demo-") as tmp: + tmp_path = Path(tmp) + _stage_fixtures(manifest, tmp_path, stderr=stderr) + + if manifest.kind == "playwright": + visual_mp4 = tmp_path / "visual.mp4" + _drive_playwright( + manifest, + output_path=visual_mp4, + work_dir=tmp_path, + stderr=stderr, + ) + elif manifest.kind == "cli": + visual_mp4 = tmp_path / "visual.mp4" + _render_cli_placeholder(manifest, visual_mp4) + else: + raise ManifestError(f"unsupported demonstration.kind: '{manifest.kind}'") + + narration: NarrationResult | None = None + api_key = os.environ.get("OPENAI_API_KEY", "").strip() + if no_narration: + pass + elif not api_key: + print( + "[demo-function] OPENAI_API_KEY not set; emitting visual-only " + "video. Pass --no-narration to silence this warning.", + file=stderr, + ) + else: + try: + narration = _generate_narration(manifest.intent, tmp_path) + except Exception as exc: # pragma: no cover - network-dependent + print( + f"[demo-function] narration failed ({exc}); " + "emitting visual-only video.", + file=stderr, + ) + narration = None + + if narration is not None: + _mux_audio(visual_mp4, narration.audio_path, rendered_mp4) + else: + shutil.move(str(visual_mp4), str(rendered_mp4)) + + _extract_poster(rendered_mp4, poster_png) + + fragment_txt.write_text(manifest.fragment_id, encoding="utf-8") + + snapshot = _manifest_snapshot(manifest, narration=narration) + manifest_json.write_text( + json.dumps(snapshot, indent=2) + "\n", + encoding="utf-8", + ) + + cache_status_txt.write_text("miss\n", encoding="utf-8") + + if cache_dir is not None: + _cache_store(cache_dir, cache_key, output_dir) + + return RenderResult( + output_dir=output_dir, + cache_status="miss", + manifest=manifest, + narration=narration, + ) + + +def _manifest_snapshot( + manifest: Manifest, + *, + narration: NarrationResult | None, +) -> dict[str, Any]: + return { + "identifier": manifest.identifier, + "intent": manifest.intent, + "fragment_id": manifest.fragment_id, + "cache_key": manifest.cache_key, + "duration_seconds": manifest.duration_seconds, + "resolution": manifest.resolution, + "assertions_to_surface": list(manifest.assertions_to_surface), + "narration": ( + None + if narration is None + else { + "voice": narration.voice, + "model": narration.model, + "ms": narration.ms, + } + ), + } + + +def _stage_fixtures( + manifest: Manifest, + work_dir: Path, + *, + stderr, +) -> None: + if not manifest.fixtures: + return + for fixture in manifest.fixtures: + src_candidates: list[Path] = [] + p = Path(fixture) + if p.is_absolute(): + src_candidates.append(p) + else: + if manifest.source_path is not None: + src_candidates.append((manifest.source_path.parent / p).resolve()) + src_candidates.append(Path.cwd() / p) + src = next((c for c in src_candidates if c.exists()), None) + if src is None: + print( + f"[demo-function] fixture not found, skipping: {fixture}", + file=stderr, + ) + continue + dst = work_dir / "fixtures" / Path(fixture).name + dst.parent.mkdir(parents=True, exist_ok=True) + if src.is_dir(): + shutil.copytree(src, dst, dirs_exist_ok=True) + else: + shutil.copy2(src, dst) + + +def _drive_playwright( + manifest: Manifest, + *, + output_path: Path, + work_dir: Path, + stderr, +) -> None: + """Drive Playwright directly (no shelled-out user script).""" + try: + from playwright.sync_api import sync_playwright # type: ignore[import-not-found] + except ImportError as exc: + raise ToolingMissingError( + "playwright is not installed", + install_hint="pip install playwright && playwright install chromium", + ) from exc + + width, height = manifest.viewport + raw_video = work_dir / "video" + raw_video.mkdir(parents=True, exist_ok=True) + + try: + with sync_playwright() as pw: + try: + browser = pw.chromium.launch(headless=True) + except Exception as exc: + raise ToolingMissingError( + f"failed to launch Chromium: {exc}", + install_hint="playwright install chromium", + ) from exc + context = browser.new_context( + viewport={"width": width, "height": height}, + record_video_dir=str(raw_video), + record_video_size={"width": width, "height": height}, + ) + page = context.new_page() + captured_video: Path | None = None + try: + if manifest.url: + page.goto(manifest.url, wait_until="networkidle") + _execute_actions(page, manifest.actions) + finally: + if page.video is not None: + try: + captured_video = Path(page.video.path()) + except Exception: + captured_video = None + with contextlib.suppress(Exception): + context.close() + with contextlib.suppress(Exception): + browser.close() + except ToolingMissingError: + raise + + if captured_video is None or not captured_video.exists(): + # Fallback: pick whatever video file was written. + candidates = sorted(raw_video.glob("*")) + candidates = [c for c in candidates if c.is_file()] + if not candidates: + raise RuntimeError("Playwright produced no video file") + captured_video = candidates[0] + + _transcode_to_mp4(captured_video, output_path, width=width, height=height) + + +def _execute_actions(page: Any, actions: Iterable[Action]) -> None: + """Run `actions` against a live Playwright `page`. Mirrors `_render_action`.""" + for action in actions: + p = action.params + if action.kind == "goto": + page.goto(p.get("url", ""), wait_until="networkidle") + elif action.kind == "click": + page.click(p["selector"]) + elif action.kind == "fill": + page.fill(p["selector"], p["value"]) + elif action.kind == "type": + page.click(p["selector"]) + page.keyboard.type(p["value"], delay=int(p.get("delay_ms", 40))) + elif action.kind == "wait_for": + page.wait_for_selector(p["selector"], timeout=int(p.get("timeout_ms", 10000))) + elif action.kind == "wait_for_text": + page.locator(p["selector"]).filter(has_text=p["text"]).first.wait_for( + state="visible", + timeout=int(p.get("timeout_ms", 10000)), + ) + elif action.kind == "wait": + page.wait_for_timeout(int(p["ms"])) + elif action.kind == "screenshot": + page.screenshot(path=str(p["path"])) + else: + raise ManifestError(f"unsupported action kind: '{action.kind}'") + + +def _render_cli_placeholder(manifest: Manifest, output_path: Path) -> None: + """Synthesize a tiny visual MP4 for `kind: cli` manifests via ffmpeg. + + `cli` support is intentionally minimal in v1 — it produces a black + background at the requested resolution for `duration_seconds`. Downstream + consumers can extend this for terminal-style demos later. + """ + _ensure_ffmpeg() + width, height = manifest.viewport + duration = max(1, int(manifest.duration_seconds)) + cmd = [ + "ffmpeg", "-y", + "-f", "lavfi", + "-i", f"color=c=black:s={width}x{height}:d={duration}:r=30", + "-c:v", "libx264", + "-pix_fmt", "yuv420p", + "-movflags", "+faststart", + str(output_path), + ] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode != 0: + raise RuntimeError(f"ffmpeg cli render failed: {proc.stderr[-400:]}") + + +# --------------------------------------------------------------------------- +# CLI entry point (called from docgen.cli:demo_function) +# --------------------------------------------------------------------------- + + +def run_cli( + manifest_arg: str, + output_dir_arg: str, + *, + cache_dir_arg: str | None = None, + no_narration: bool = False, + stderr=None, + stdout=None, +) -> int: + """Execute the `docgen demo-function` flow and return an exit code. + + Lives here (not in cli.py) to keep the runner testable without touching + Click's machinery. + """ + if stderr is None: + stderr = sys.stderr + if stdout is None: + stdout = sys.stdout + + try: + manifest = load_manifest(manifest_arg) + except FileNotFoundError as exc: + print(f"[demo-function] {exc}", file=stderr) + return EXIT_INVALID + except ManifestError as exc: + print(f"[demo-function] {exc}", file=stderr) + return EXIT_INVALID + + output_dir = Path(output_dir_arg) + cache_dir = Path(cache_dir_arg) if cache_dir_arg else None + + try: + result = render( + manifest, + output_dir, + cache_dir=cache_dir, + no_narration=no_narration, + stderr=stderr, + ) + except PlaceholderManifest as exc: + print(f"[demo-function] neutral skip: {exc}", file=stderr) + return EXIT_NEUTRAL_SKIP + except ToolingMissingError as exc: + print(f"[demo-function] {exc}", file=stderr) + print(f" install: {exc.install_hint}", file=stderr) + return EXIT_TOOLING_MISSING + except (ManifestError, RuntimeError) as exc: + print(f"[demo-function] render failed: {exc}", file=stderr) + return EXIT_INVALID + + print( + f"[demo-function] {result.cache_status}: " + f"{manifest.fragment_id} -> {result.output_dir}", + file=stdout, + ) + return EXIT_OK + + +__all__ = [ + "Action", + "Manifest", + "ManifestError", + "PlaceholderManifest", + "ToolingMissingError", + "RenderResult", + "NarrationResult", + "load_manifest", + "render", + "run_cli", + "generate_capture_script", + "EXIT_OK", + "EXIT_INVALID", + "EXIT_TOOLING_MISSING", + "EXIT_NEUTRAL_SKIP", + "SUPPORTED_ACTION_KINDS", + "CACHED_ARTIFACTS", + "HARD_CAP_SECONDS", +] diff --git a/tests/test_demo_function.py b/tests/test_demo_function.py new file mode 100644 index 0000000..fbc290e --- /dev/null +++ b/tests/test_demo_function.py @@ -0,0 +1,485 @@ +"""Unit tests for `docgen demo-function`. + +Most tests here are pure-data: they exercise manifest loading, validation, +action rendering, fragment/cache-key stability, and CLI exit-code mapping +without launching Playwright or ffmpeg. The few tests that need actual +rendering are guarded with `pytest.importorskip` / `shutil.which` checks so +the suite passes on CI runners without Playwright or ffmpeg installed. +""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path + +import pytest + +from docgen import demo_function as df +from docgen.demo_function import ( + Action, + ManifestError, + PlaceholderManifest, + SUPPORTED_ACTION_KINDS, + _coerce, + _render_action, + generate_capture_script, + load_manifest, + run_cli, +) + + +# --------------------------------------------------------------------------- +# Manifest loading — YAML sidecar (shape #1) +# --------------------------------------------------------------------------- + + +def _yaml_manifest_text(**overrides: object) -> str: + base = { + "identifier": "repo/path.ts:fn", + "intent": "Does the thing.", + "demonstration": { + "kind": "playwright", + "url": "http://127.0.0.1:3000/x", + "actions": [ + {"kind": "click", "selector": "[data-testid=\"go\"]"}, + ], + }, + "output_budget": {"duration_seconds": 10, "resolution": "1280x720"}, + } + base.update(overrides) + import yaml as _y + return _y.safe_dump(base) + + +def test_load_yaml_sidecar_minimal(tmp_path: Path) -> None: + p = tmp_path / "m.docgen.yaml" + p.write_text(_yaml_manifest_text(), encoding="utf-8") + m = load_manifest(p) + assert m.identifier == "repo/path.ts:fn" + assert m.intent == "Does the thing." + assert m.kind == "playwright" + assert m.url == "http://127.0.0.1:3000/x" + assert [a.kind for a in m.actions] == ["click"] + assert m.resolution == "1280x720" + + +def test_load_yaml_string_path(tmp_path: Path) -> None: + p = tmp_path / "m.docgen.yaml" + p.write_text(_yaml_manifest_text(), encoding="utf-8") + m = load_manifest(str(p)) + assert m.identifier == "repo/path.ts:fn" + + +# --------------------------------------------------------------------------- +# Manifest loading — Python @pytest.mark.docgen marker (shape #2) +# --------------------------------------------------------------------------- + + +_SAMPLE_PY = ''' +"""Module docstring that mentions @pytest.mark.docgen — should NOT match. + +A naive regex over this file would pick up text inside this docstring +and confuse it for a real marker; the AST-based loader must ignore it. +""" +import pytest + + +@pytest.mark.docgen( + identifier="repo/path.ts:fn", + intent="Does the thing.", + demonstration={ + "kind": "playwright", + "url": "http://127.0.0.1:3000/x", + "actions": [ + {"kind": "type", "selector": "#title", "value": "hi", "delay_ms": 30}, + {"kind": "click", "selector": "#go"}, + ], + }, + output_budget={"duration_seconds": 10, "resolution": "1280x720"}, + assertions_to_surface=["x.y === 1"], +) +def test_thing(): + pass +''' + + +def test_python_marker_basic(tmp_path: Path) -> None: + p = tmp_path / "sample_test.py" + p.write_text(_SAMPLE_PY, encoding="utf-8") + m = load_manifest(f"{p}::test_thing") + assert m.identifier == "repo/path.ts:fn" + assert m.intent == "Does the thing." + assert [a.kind for a in m.actions] == ["type", "click"] + assert m.assertions_to_surface == ["x.y === 1"] + + +def test_python_marker_ignores_docstring_text(tmp_path: Path) -> None: + """F7 regression: regex-over-source readers match docstring text. + + `_load_pytest_marker` uses `ast.walk` over the parsed module — a + docstring that *talks about* the marker must not be parsed as one. + The marker is on `test_thing`; a function without one must error. + """ + src = ''' +"""This file mentions pytest.mark.docgen(identifier=\\"X\\", intent=\\"Y\\") in prose.""" + +def test_no_marker(): + pass +''' + p = tmp_path / "no_marker.py" + p.write_text(src, encoding="utf-8") + with pytest.raises(ManifestError, match="missing @pytest.mark.docgen"): + load_manifest(f"{p}::test_no_marker") + + +def test_python_marker_unknown_function(tmp_path: Path) -> None: + p = tmp_path / "sample_test.py" + p.write_text(_SAMPLE_PY, encoding="utf-8") + with pytest.raises(ManifestError, match="function not found"): + load_manifest(f"{p}::missing_fn") + + +# --------------------------------------------------------------------------- +# Validation +# --------------------------------------------------------------------------- + + +def test_missing_identifier(tmp_path: Path) -> None: + p = tmp_path / "m.yaml" + p.write_text(_yaml_manifest_text(identifier=""), encoding="utf-8") + with pytest.raises(ManifestError, match="missing required field: 'identifier'"): + load_manifest(p) + + +def test_invalid_kind() -> None: + raw = { + "identifier": "x:y", + "intent": "z", + "demonstration": {"kind": "selenium"}, + } + with pytest.raises(ManifestError, match="must be 'playwright' or 'cli'"): + _coerce(raw) + + +def test_duration_hard_cap() -> None: + raw = { + "identifier": "x:y", + "intent": "z", + "demonstration": {"kind": "playwright", "url": "http://x"}, + "output_budget": {"duration_seconds": 120, "resolution": "1280x720"}, + } + with pytest.raises(ManifestError, match="exceeds the 60s hard cap"): + _coerce(raw) + + +def test_invalid_resolution() -> None: + raw = { + "identifier": "x:y", + "intent": "z", + "demonstration": {"kind": "playwright", "url": "http://x"}, + "output_budget": {"duration_seconds": 10, "resolution": "1280-720"}, + } + with pytest.raises(ManifestError, match="output_budget.resolution must match"): + _coerce(raw) + + +def test_unknown_action_kind() -> None: + raw = { + "identifier": "x:y", + "intent": "z", + "demonstration": { + "kind": "playwright", + "url": "http://x", + "actions": [{"kind": "teleport"}], + }, + } + with pytest.raises(ManifestError, match="unsupported action kind: 'teleport'"): + _coerce(raw) + + +def test_assertions_length_cap() -> None: + raw = { + "identifier": "x:y", + "intent": "z", + "demonstration": {"kind": "playwright", "url": "http://x"}, + "assertions_to_surface": ["x" * 61], + } + with pytest.raises(ManifestError, match="≤ 60 chars"): + _coerce(raw) + + +# --------------------------------------------------------------------------- +# Action -> Playwright source line +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "action,expected_substring", + [ + (Action("goto", {"url": "http://x"}), 'page.goto(\'http://x\''), + (Action("click", {"selector": "[data-testid=\"x\"]"}), "page.click(\'[data-testid=\"x\"]\'"), + (Action("fill", {"selector": "#a", "value": "v"}), "page.fill('#a', 'v')"), + (Action("type", {"selector": "#a", "value": "v"}), "page.keyboard.type('v', delay=40)"), + (Action("wait_for", {"selector": "#a"}), "page.wait_for_selector('#a', timeout=10000)"), + ( + Action("wait_for_text", {"selector": "[data-testid=\"x\"]", "text": "ok"}), + ".filter(has_text=\'ok\').first.wait_for(state=\"visible\"", + ), + (Action("wait", {"ms": 250}), "page.wait_for_timeout(250)"), + (Action("screenshot", {"path": "/tmp/x.png"}), "page.screenshot(path='/tmp/x.png')"), + ], +) +def test_render_action(action: Action, expected_substring: str) -> None: + line = _render_action(action) + assert expected_substring in line + + +def test_supported_action_kinds_set() -> None: + """The spec lists exactly these eight kinds — guard against accidental drift.""" + assert set(SUPPORTED_ACTION_KINDS) == { + "goto", "click", "fill", "type", "wait_for", + "wait_for_text", "wait", "screenshot", + } + + +def test_generated_script_compiles(tmp_path: Path) -> None: + """The generated capture script must be valid Python (compileable).""" + raw = { + "identifier": "x:y", + "intent": "z", + "demonstration": { + "kind": "playwright", + "url": "http://127.0.0.1/x", + "actions": [ + {"kind": "type", "selector": "#a", "value": "v", "delay_ms": 30}, + {"kind": "click", "selector": "[data-testid=\"go\"]"}, + {"kind": "wait_for_text", "selector": "#status", "text": "ok"}, + {"kind": "wait", "ms": 100}, + ], + }, + "output_budget": {"duration_seconds": 10, "resolution": "640x480"}, + } + m = _coerce(raw) + script = generate_capture_script(m, output_path=tmp_path / "out.mp4") + compile(script, "", "exec") + assert "record_video_size" in script + assert "viewport=" in script + assert "640" in script and "480" in script + + +# --------------------------------------------------------------------------- +# fragment_id and cache_key stability +# --------------------------------------------------------------------------- + + +def test_fragment_id_format() -> None: + raw = { + "identifier": "courseforge/Course Builder/src/Foo.ts:compileLesson!", + "intent": "z", + "demonstration": {"kind": "playwright", "url": "http://x"}, + } + m = _coerce(raw) + assert m.fragment_id == "fn-courseforge-course-builder-src-foo-ts-compilelesson" + import re as _re + assert _re.match(r"^fn-[a-z0-9-]+$", m.fragment_id) + + +def test_cache_key_stable_and_changes_with_input(tmp_path: Path) -> None: + raw1 = { + "identifier": "x:y", + "intent": "first", + "demonstration": {"kind": "playwright", "url": "http://x"}, + } + raw2 = dict(raw1, intent="second") + k1 = _coerce(raw1).cache_key + k1_again = _coerce(raw1).cache_key + k2 = _coerce(raw2).cache_key + assert k1 == k1_again + assert k1 != k2 + assert len(k1) == 16 + + +def test_cache_key_includes_fixture_contents(tmp_path: Path) -> None: + fixture = tmp_path / "fix.txt" + fixture.write_text("alpha", encoding="utf-8") + + src = tmp_path / "m.yaml" + src.write_text( + _yaml_manifest_text(setup={"fixtures": ["fix.txt"]}), + encoding="utf-8", + ) + m1 = load_manifest(src) + k1 = m1.cache_key + + fixture.write_text("beta", encoding="utf-8") + m2 = load_manifest(src) + k2 = m2.cache_key + assert k1 != k2 + + +# --------------------------------------------------------------------------- +# Placeholder skip + tooling-missing exit code mapping +# --------------------------------------------------------------------------- + + +def test_render_raises_placeholder(tmp_path: Path) -> None: + raw = { + "identifier": "x:y", + "intent": "z", + "demonstration": {"kind": "playwright"}, + } + m = _coerce(raw) + with pytest.raises(PlaceholderManifest): + df.render(m, tmp_path / "out", no_narration=True) + + +def test_run_cli_neutral_skip(tmp_path: Path, capsys) -> None: + p = tmp_path / "m.yaml" + p.write_text( + _yaml_manifest_text( + demonstration={"kind": "playwright", "actions": []}, + ), + encoding="utf-8", + ) + out_dir = tmp_path / "out" + code = run_cli(str(p), str(out_dir), no_narration=True) + assert code == df.EXIT_NEUTRAL_SKIP + err = capsys.readouterr().err + assert "neutral skip" in err + # Acceptance #9: writes nothing to output-dir. + assert not list(out_dir.iterdir()) if out_dir.exists() else True + + +def test_run_cli_invalid_manifest_duration(tmp_path: Path, capsys) -> None: + p = tmp_path / "m.yaml" + p.write_text( + _yaml_manifest_text( + output_budget={"duration_seconds": 120, "resolution": "1280x720"}, + ), + encoding="utf-8", + ) + code = run_cli(str(p), str(tmp_path / "out"), no_narration=True) + assert code == df.EXIT_INVALID + assert "exceeds the 60s hard cap" in capsys.readouterr().err + + +def test_run_cli_missing_manifest_file(tmp_path: Path, capsys) -> None: + code = run_cli(str(tmp_path / "no.yaml"), str(tmp_path / "out"), no_narration=True) + assert code == df.EXIT_INVALID + + +# --------------------------------------------------------------------------- +# End-to-end render (cli kind, ffmpeg-only — no Playwright required) +# --------------------------------------------------------------------------- + + +def _ffmpeg_present() -> bool: + return shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None + + +@pytest.mark.skipif(not _ffmpeg_present(), reason="ffmpeg / ffprobe not installed") +def test_render_cli_kind_emits_artifacts(tmp_path: Path, monkeypatch) -> None: + """End-to-end render using kind=cli (synthesizes a black video). + + Covers acceptance criteria #1, #2, #3, #4, #5, #6, #7, #11 (no narration). + """ + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + raw = { + "identifier": "course-builder/src/lessons/compileLesson.ts:compileLesson", + "intent": "Compiles a lesson into checkpoints.", + "demonstration": {"kind": "cli"}, + "output_budget": {"duration_seconds": 1, "resolution": "320x240"}, + "assertions_to_surface": ["lesson.status === 'compiled'"], + } + m = _coerce(raw) + + out_dir = tmp_path / "out" + cache_dir = tmp_path / "cache" + result = df.render(m, out_dir, cache_dir=cache_dir, no_narration=True) + assert result.cache_status == "miss" + + for name in ("rendered.mp4", "poster.png", "fragment.txt", "manifest.json", "cache-status.txt"): + assert (out_dir / name).exists(), f"missing artifact: {name}" + + fragment_text = (out_dir / "fragment.txt").read_text(encoding="utf-8") + assert fragment_text == m.fragment_id + assert not fragment_text.endswith("\n") + + snapshot = json.loads((out_dir / "manifest.json").read_text(encoding="utf-8")) + expected_keys = { + "identifier", "intent", "fragment_id", "cache_key", + "duration_seconds", "resolution", "assertions_to_surface", "narration", + } + assert expected_keys.issubset(snapshot.keys()) + assert snapshot["narration"] is None + + assert (out_dir / "cache-status.txt").read_text() == "miss\n" + + head = (out_dir / "rendered.mp4").read_bytes()[:12] + assert b"ftyp" in head, "rendered.mp4 should be a real ISO MP4" + + out2 = tmp_path / "out2" + result2 = df.render(m, out2, cache_dir=cache_dir, no_narration=True) + assert result2.cache_status == "hit" + assert (out2 / "cache-status.txt").read_text() == "hit\n" + for name in ("rendered.mp4", "poster.png", "fragment.txt", "manifest.json"): + assert (out2 / name).exists() + + +@pytest.mark.skipif(not _ffmpeg_present(), reason="ffmpeg / ffprobe not installed") +def test_render_warns_when_openai_key_missing(tmp_path: Path, monkeypatch, capsys) -> None: + """Acceptance #11 (first half): no key → warning + visual-only video.""" + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + raw = { + "identifier": "x:y", + "intent": "test", + "demonstration": {"kind": "cli"}, + "output_budget": {"duration_seconds": 1, "resolution": "320x240"}, + } + m = _coerce(raw) + df.render(m, tmp_path / "out") + err = capsys.readouterr().err + assert "OPENAI_API_KEY not set" in err + + +# --------------------------------------------------------------------------- +# Equivalence between YAML and Python marker shapes (acceptance #8) +# --------------------------------------------------------------------------- + + +def test_yaml_and_python_marker_produce_equivalent_manifests(tmp_path: Path) -> None: + yaml_path = tmp_path / "m.docgen.yaml" + yaml_path.write_text(_yaml_manifest_text(), encoding="utf-8") + py_path = tmp_path / "sample_test.py" + py_path.write_text(_SAMPLE_PY, encoding="utf-8") + + a = load_manifest(yaml_path) + + b_src = ''' +import pytest + +@pytest.mark.docgen( + identifier="repo/path.ts:fn", + intent="Does the thing.", + demonstration={ + "kind": "playwright", + "url": "http://127.0.0.1:3000/x", + "actions": [{"kind": "click", "selector": '[data-testid="go"]'}], + }, + output_budget={"duration_seconds": 10, "resolution": "1280x720"}, +) +def test_thing(): + pass +''' + b_path = tmp_path / "equiv_test.py" + b_path.write_text(b_src, encoding="utf-8") + b = load_manifest(f"{b_path}::test_thing") + + assert a.identifier == b.identifier + assert a.intent == b.intent + assert a.kind == b.kind + assert a.url == b.url + assert [act.kind for act in a.actions] == [act.kind for act in b.actions] + assert a.resolution == b.resolution + assert a.duration_seconds == b.duration_seconds From 1b2620f860bfccd9fc8a918f1c0ea74ba6d8cc9e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 4 May 2026 17:10:11 +0000 Subject: [PATCH 3/3] docs/version: README section for demo-function; bump to 0.2.0 Adds a README subsection describing 'docgen demo-function' (one paragraph + example invocation) and links to examples/. Bumps the package version to 0.2.0 since this adds a new public CLI surface. Co-authored-by: John Menke --- README.md | 27 +++++++++++++++++++++++++++ pyproject.toml | 2 +- 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 13ff38f..a5d05bc 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,7 @@ docgen validate --pre-push # validate all outputs before committing | `docgen manim [--scene StackDAGScene]` | Render Manim animations | | `docgen vhs [--tape 02-quickstart.tape] [--strict] [--timeout 120]` | Render VHS terminal recordings | | `docgen playwright --script scripts/capture.py --url http://localhost:3000 --source demo.mp4` | Capture browser demo video with Playwright script | +| `docgen demo-function --manifest --output-dir [--cache-dir ] [--no-narration]` | Render one short, single-purpose video per function from a declarative manifest | | `docgen tape-lint [--tape 02-quickstart.tape]` | Lint tapes for commands likely to hang in VHS | | `docgen sync-vhs [--segment 01] [--dry-run]` | Rewrite VHS `Sleep` values from `animations/timing.json` | | `docgen compose [01 02 03] [--ffmpeg-timeout 900]` | Compose segments (audio + video) | @@ -132,6 +133,32 @@ Script contract: `DOCGEN_PLAYWRIGHT_WIDTH`, `DOCGEN_PLAYWRIGHT_HEIGHT`, and optional segment metadata - must write an MP4 to the requested output path - should use headless Playwright for CI compatibility +### Per-function video docs (`docgen demo-function`) + +`docgen demo-function` renders **one short, single-purpose MP4 per function** — +the docs-site analogue of a single Playwright `test('…')` describing one +behavior. Inputs are declarative: either a `*.docgen.yaml` sidecar or a +`@pytest.mark.docgen(...)` decorator on a Python test (read statically via +`ast` — never imported / `exec`'d). Outputs are five files in `--output-dir`: +`rendered.mp4` (real ISO MP4), `poster.png`, `fragment.txt` (`fn-`), +`manifest.json` (snapshot), and `cache-status.txt` (`hit` / `miss`). + +```bash +docgen demo-function \ + --manifest examples/lesson_compile.docgen.yaml \ + --output-dir /tmp/out \ + --cache-dir /tmp/docgen-cache +``` + +Exit codes: `0` success, `1` invalid manifest / render failure, `2` missing +`ffmpeg` / `playwright`, `78` neutral skip (placeholder manifest with no +`url` — useful in CI). When `OPENAI_API_KEY` is set, the intent line is +narrated via `gpt-4o-mini-tts` and muxed onto the video; pass +`--no-narration` to skip TTS even if the key is set. See +[`examples/lesson_compile.docgen.yaml`](examples/lesson_compile.docgen.yaml) +and [`examples/sample_test.py`](examples/sample_test.py) for both input +shapes. + ### VHS safety: avoid real long-running commands in tapes VHS executes commands in a real shell session. For demos, prefer simulated output with `echo` diff --git a/pyproject.toml b/pyproject.toml index 9f772ca..b44b3a7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "docgen" -version = "0.1.0" +version = "0.2.0" description = "Reusable demo generation pipeline: TTS, Manim, VHS, ffmpeg composition, validation, GitHub Pages publishing" readme = "README.md" requires-python = ">=3.10"