diff --git a/docs/reference/workflows.md b/docs/reference/workflows.md index 1dfc5e6904..0097f8f62e 100644 --- a/docs/reference/workflows.md +++ b/docs/reference/workflows.md @@ -91,8 +91,192 @@ specify workflow add | `--dev` | Install from a local workflow YAML file or directory | | `--from ` | Install from a custom URL (`` names the expected workflow ID) | -Installs a workflow from the catalog, a URL (HTTPS required), or a local file path. +Installs a workflow from the catalog, a URL (HTTPS required), a local YAML file, or a local directory containing `workflow.yml`. +## Workflow Overlays + +Workflow overlays let a project extend or override an installed workflow without editing the installed `workflow.yml`. This keeps local customizations safe across `specify bundle update` or `specify workflow add` upgrades. + +When `specify workflow run ` loads a workflow, the engine composes the base workflow with all enabled overlays for that workflow id. The result is validated like any other workflow definition. + +### How Overlays Work + +An overlay is a YAML file that declares a set of edit operations against the step list of a base workflow. Overlays are applied in priority order (lowest first, highest last); equal-priority overlays are applied in source order (last applied wins). + +Project overlay files live at: + +| Location | Purpose | +| --- | --- | +| `.specify/workflows/overlays//*.yml` | Project-local customizations | + +### Overlay File Format + +The recommended edit format uses the operation name as the key and the anchor step id as the value: + +```yaml +id: "my-overlay" +extends: "speckit" +priority: 10 +enabled: true +edits: + - insert_after: implement + step: + id: run-lint + type: shell + run: "ruff check src/" + + - replace: review-spec + step: + id: review-spec + type: gate + message: "Review the generated spec (overlay override)." + options: [approve, reject] + on_reject: abort +``` + +The explicit form is also supported: + +```yaml +edits: + - operation: insert_after + anchor: implement + step: + id: run-lint + type: shell + run: "ruff check src/" +``` + +#### Fields + +| Field | Required | Description | +| --- | --- | --- | +| `id` | yes | Identifier for this overlay. Used in `specify workflow overlay *` commands. Must be lowercase letters, digits, and hyphens only; no dots, underscores, path separators, or `overlays`. | +| `extends` | yes | The workflow id this overlay applies to. Uses the same safe-id format as `id`; `overlays` is reserved. | +| `priority` | yes | Integer `>= 1`. Higher priority overlays are applied later and win conflicts. | +| `enabled` | no | Boolean. Defaults to `true`. Disabled overlays are ignored. | +| `edits` | yes | Non-empty list of edit operations. | + +#### Edit Operations + +| Operation | `step` required | Effect | +| --- | --- | --- | +| `insert_after` | yes | Insert `step` immediately after the anchor step. | +| `insert_before` | yes | Insert `step` immediately before the anchor step. | +| `replace` | yes | Replace the anchor step with `step`. | +| `remove` | no | Remove the anchor step from the list. | + +The `anchor` is the `id` of a step in the base workflow. Anchors are resolved recursively inside `then`, `else`, `steps`, `cases.*`, and `default` blocks, so nested base steps can also be targeted. Fan-out templates (`step` inside a `fan-out` step) are **not** valid anchors. + +Step ids must not contain `:` — that character is reserved for engine-generated nested ids. + +### Overlay CLI Commands + +#### Add a Project Overlay + +```bash +specify workflow overlay add --priority +``` + +Validates the overlay file and copies it to `.specify/workflows/overlays//.yml`. `--priority` overrides the `priority` field in the file. + +#### List Overlays + +```bash +specify workflow overlay list +``` + +Shows enabled overlays for the workflow, ordered by resolver precedence. Disabled overlays are ignored by the resolver and are not listed. + +#### Change Priority + +```bash +specify workflow overlay set-priority +``` + +#### Enable or Disable + +```bash +specify workflow overlay disable +specify workflow overlay enable +``` + +#### Remove + +```bash +specify workflow overlay remove +``` + +Removes the project overlay file. + +#### Inspect the Composed Workflow + +```bash +specify workflow resolve +``` + +Prints the layer stack (base + overlays) and the source attribution for each step after composition. Useful for debugging which overlay contributed or overrode a step. + +### Example: Adding Automated Linting after Implementation + +Given the built-in `speckit` workflow, create `project-overlay.yml`: + +```yaml +id: "add-lint" +extends: "speckit" +priority: 10 +edits: + - insert_after: implement + step: + id: run-lint + type: shell + run: "ruff check src/" +``` + +Install it: + +```bash +specify workflow overlay add project-overlay.yml --priority 10 +``` + +Run the workflow: + +```bash +specify workflow run speckit -i spec="Build a kanban board" +``` + +The composed workflow will now run the full SDD cycle and execute `ruff check src/` automatically after the `implement` step. + +### Example: Replacing a Gate + +```yaml +id: "skip-plan-review" +extends: "speckit" +priority: 20 +edits: + - replace: review-plan + step: + id: review-plan + type: command + command: speckit.plan + input: + args: "{{ inputs.spec }}" +``` + +Higher priority (`20`) means this overlay is applied after the `add-lint` overlay above. It replaces the `review-plan` gate with a non-interactive command. + +### Interaction with Bundles and Updates + +`specify workflow add ` installs `workflow.yml` from the local directory into `.specify/workflows//`. + +When an installed workflow is refreshed or reinstalled, project overlays in `.specify/workflows/overlays//` are preserved because they live outside the installed workflow directory. + +### Limitations + +- Overlays operate on the step list only. They cannot change workflow metadata (name, description, inputs, `requires`) or expression logic. +- Fan-out templates cannot be used as anchors. +- An overlay that targets a step id that does not exist in the base workflow will raise a validation error when the workflow is resolved. +- Overlays cannot target steps added by other overlays. +- Overlays cannot add new inputs or change the input schema of the base workflow. ## Update Workflows ```bash diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index 9ab199c023..09eabf9ad2 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -49,6 +49,13 @@ ) workflow_step_app.add_typer(workflow_step_catalog_app, name="catalog") +workflow_overlay_app = typer.Typer( + name="overlay", + help="Manage workflow overlays", + add_completion=False, +) +workflow_app.add_typer(workflow_overlay_app, name="overlay") + def _error_console(json_output: bool): """Console for error text: stderr under ``--json`` so the JSON stdout @@ -192,6 +199,10 @@ def _reject_unsafe_workflow_storage(project_root: Path) -> None: project_root / ".specify" / "workflows" / "runs", ".specify/workflows/runs", ) + _reject_unsafe_dir( + project_root / ".specify" / "workflows" / "overlays", + ".specify/workflows/overlays", + ) def _scan_for_workflow_owner(parts: tuple[str, ...]) -> int | None: @@ -366,7 +377,7 @@ def ownership_for(candidate: Path) -> tuple[Path, str] | None: _WORKFLOW_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$") -_RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"runs", "steps"}) +_RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"overlays", "runs", "steps"}) def _reject_insecure_download_redirect(old_url: str, new_url: str) -> None: @@ -3148,6 +3159,98 @@ def workflow_step_catalog_remove( console.print(f"[green]✓[/green] Step catalog source '{removed_name}' removed") +@workflow_overlay_app.command("add") +def workflow_overlay_add_cmd( + source: Path = typer.Argument(..., help="Path to overlay YAML file"), + priority: int | None = typer.Option( + None, "--priority", help="Overlay priority (higher wins, >= 1)" + ), +): + """Add a project-local overlay for a workflow.""" + from .overlays._commands import workflow_overlay_add + + project_root = _require_specify_project() + if workflow_overlay_add(project_root, source, priority) is None: + raise typer.Exit(1) + + +@workflow_overlay_app.command("set-priority") +def workflow_overlay_set_priority_cmd( + workflow_id: str = typer.Argument(..., help="Workflow ID the overlay extends"), + overlay_id: str = typer.Argument(..., help="Overlay ID"), + priority: int = typer.Argument(..., help="New priority (>= 1)"), +): + """Set the priority of a project-local overlay.""" + from .overlays._commands import workflow_overlay_set_priority + + project_root = _require_specify_project() + if not workflow_overlay_set_priority(project_root, workflow_id, overlay_id, priority): + raise typer.Exit(1) + + +@workflow_overlay_app.command("enable") +def workflow_overlay_enable_cmd( + workflow_id: str = typer.Argument(..., help="Workflow ID the overlay extends"), + overlay_id: str = typer.Argument(..., help="Overlay ID"), +): + """Enable a project-local overlay.""" + from .overlays._commands import workflow_overlay_enable + + project_root = _require_specify_project() + if not workflow_overlay_enable(project_root, workflow_id, overlay_id): + raise typer.Exit(1) + + +@workflow_overlay_app.command("disable") +def workflow_overlay_disable_cmd( + workflow_id: str = typer.Argument(..., help="Workflow ID the overlay extends"), + overlay_id: str = typer.Argument(..., help="Overlay ID"), +): + """Disable a project-local overlay.""" + from .overlays._commands import workflow_overlay_disable + + project_root = _require_specify_project() + if not workflow_overlay_disable(project_root, workflow_id, overlay_id): + raise typer.Exit(1) + + +@workflow_overlay_app.command("remove") +def workflow_overlay_remove_cmd( + workflow_id: str = typer.Argument(..., help="Workflow ID the overlay extends"), + overlay_id: str = typer.Argument(..., help="Overlay ID"), +): + """Remove a project-local overlay.""" + from .overlays._commands import workflow_overlay_remove + + project_root = _require_specify_project() + if not workflow_overlay_remove(project_root, workflow_id, overlay_id): + raise typer.Exit(1) + + +@workflow_overlay_app.command("list") +def workflow_overlay_list_cmd( + workflow_id: str = typer.Argument(..., help="Workflow ID"), +): + """List overlays for a workflow.""" + from .overlays._commands import workflow_overlay_list + + project_root = _require_specify_project() + if workflow_overlay_list(project_root, workflow_id) is None: + raise typer.Exit(1) + + +@workflow_app.command("resolve") +def workflow_resolve_cmd( + workflow_id: str = typer.Argument(..., help="Workflow ID to resolve"), +): + """Show layer attribution for a resolved workflow.""" + from .overlays._commands import workflow_resolve + + project_root = _require_specify_project() + if workflow_resolve(project_root, workflow_id) is None: + raise typer.Exit(1) + + def register(app: typer.Typer) -> None: """Attach the workflow command group to the root Typer app.""" app.add_typer(workflow_app, name="workflow") diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py index 91a9ac2ad5..c3987a1b04 100644 --- a/src/specify_cli/workflows/engine.py +++ b/src/specify_cli/workflows/engine.py @@ -727,13 +727,24 @@ def load_workflow(self, source: str | Path) -> WorkflowDefinition: ValueError: If the workflow YAML is invalid. """ + from .overlays import WorkflowResolver + path = Path(source).expanduser() # Try as a direct file path first if path.suffix.lower() in (".yml", ".yaml") and path.is_file(): return WorkflowDefinition.from_yaml(path) - # Try as an installed workflow ID + # Try as an installed workflow ID, resolving any overlays. + resolver = WorkflowResolver(self.project_root) + try: + return resolver.resolve(str(source)) + except FileNotFoundError: + # Fall back to the direct workflow.yml path so callers still get + # the original error when the workflow id is not installed. + pass + + # Legacy direct path check for workflows installed without registry entries. installed_path = ( self.project_root / ".specify" diff --git a/src/specify_cli/workflows/overlays/__init__.py b/src/specify_cli/workflows/overlays/__init__.py new file mode 100644 index 0000000000..40521a8feb --- /dev/null +++ b/src/specify_cli/workflows/overlays/__init__.py @@ -0,0 +1,75 @@ +"""Workflow overlay resolver — composes installed workflows from layers.""" + +from __future__ import annotations + +from pathlib import Path + +from ..engine import WorkflowDefinition +from .composer import StepListComposer +from .layer_sources import ( + BaseWorkflowSource, + Layer, + ProjectOverlaySource, +) +from .merge import ComposedStep + + +class WorkflowResolver: + """Resolves a workflow ID to its composed ``WorkflowDefinition``. + + Collects layers from two tiers: + - project-local overlays (``.specify/workflows/overlays//*.yml``) + - the base workflow itself (``.specify/workflows//workflow.yml``) + + Resolution is higher-wins: overlays with higher priority are applied + later and override earlier edits on the same anchors. + """ + + def __init__(self, project_root: Path) -> None: + self.project_root = project_root + self._sources = [ + ProjectOverlaySource(project_root), + BaseWorkflowSource(project_root), + ] + self._composer = StepListComposer() + + def collect_all_layers(self, workflow_id: str) -> list[Layer]: + """Collect all layers sorted by resolver precedence. + + Higher priority wins. Ties are broken so the actual winner appears first: + the composer applies equal-priority sources ascending (last wins), so we + reverse source ordering here to show the winner at the top of the list. + Two stable passes: source descending, then priority descending. + """ + all_layers: list[Layer] = [] + for source in self._sources: + all_layers.extend(source.collect(workflow_id)) + + # Pass 1: source descending — within each priority group the winning + # source (alphabetically last = applied last by the composer) rises first. + by_source = sorted(all_layers, key=lambda layer: layer.source, reverse=True) + # Pass 2: priority descending — overall ordering by precedence. + return sorted(by_source, key=lambda layer: layer.priority, reverse=True) + + def resolve(self, workflow_id: str) -> WorkflowDefinition: + """Resolve a workflow ID to its composed definition. + + Raises: + FileNotFoundError: if the workflow cannot be found. + ValueError: if the composed workflow is invalid. + """ + layers = self.collect_all_layers(workflow_id) + definition, _ = self._composer.compose(layers) + if definition is None: + raise FileNotFoundError(f"Workflow not found: {workflow_id}") + return definition + + def resolve_with_layers( + self, workflow_id: str + ) -> tuple[WorkflowDefinition, list[Layer], list[ComposedStep]]: + """Resolve a workflow and return its definition plus layer attribution.""" + layers = self.collect_all_layers(workflow_id) + definition, attribution = self._composer.compose(layers) + if definition is None: + raise FileNotFoundError(f"Workflow not found: {workflow_id}") + return definition, layers, attribution diff --git a/src/specify_cli/workflows/overlays/_commands.py b/src/specify_cli/workflows/overlays/_commands.py new file mode 100644 index 0000000000..dcba31160e --- /dev/null +++ b/src/specify_cli/workflows/overlays/_commands.py @@ -0,0 +1,389 @@ +"""CLI handlers for ``specify workflow overlay *`` and ``specify workflow resolve``.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import typer +import yaml + +from ..._console import console, err_console +from .._commands import _reject_unsafe_dir, _reject_unsafe_workflow_storage +from . import WorkflowResolver +from .schema import _RESERVED_OVERLAY_WORKFLOW_IDS, _SAFE_ID_PATTERN, validate_overlay_yaml + + +def _validate_overlay_id_or_exit(id_value: str, label: str) -> None: + """Validate a single-segment overlay/workflow id from CLI arguments.""" + if not isinstance(id_value, str) or not id_value: + err_console.print(f"[red]Error:[/red] {label} is required and must be a non-empty string.") + raise typer.Exit(1) + if not _SAFE_ID_PATTERN.match(id_value): + err_console.print( + f"[red]Error:[/red] Invalid {label} {id_value!r}: " + "only lowercase letters, digits, and hyphens are allowed." + ) + raise typer.Exit(1) + + +def _validate_workflow_id_or_exit(workflow_id: str) -> None: + """Validate a workflow id, treating the overlay root as reserved.""" + _validate_overlay_id_or_exit(workflow_id, "workflow ID") + if workflow_id in _RESERVED_OVERLAY_WORKFLOW_IDS: + err_console.print( + f"[red]Error:[/red] Invalid workflow ID {workflow_id!r}: " + "reserved name." + ) + raise typer.Exit(1) + + +def _overlay_root(project_root: Path) -> Path: + """Return the project-local overlay root after rejecting unsafe ancestors.""" + _reject_unsafe_workflow_storage(project_root) + root = project_root / ".specify" / "workflows" / "overlays" + _reject_unsafe_dir(root, ".specify/workflows/overlays") + return root + + +def _project_overlay_dir(project_root: Path, workflow_id: str) -> Path: + """Return the project-local overlay directory for a workflow id. + + Raises typer.Exit if the resolved path escapes the overlay root. + """ + _validate_workflow_id_or_exit(workflow_id) + root = _overlay_root(project_root) + target = root / workflow_id + return _ensure_contained_dir(target, root) + + +def _ensure_contained_dir(path: Path, root: Path) -> Path: + """Ensure *path* resolves inside *root* and is not a symlink. + + Returns *path* if safe. Raises typer.Exit on traversal or symlink. + """ + _reject_unsafe_dir(root, ".specify/workflows/overlays") + if path.is_symlink(): + err_console.print( + f"[red]Error:[/red] Refusing to use symlinked path {path}." + ) + raise typer.Exit(1) + if path.exists() and not path.is_dir(): + err_console.print( + f"[red]Error:[/red] Overlay directory path is not a directory: {path}." + ) + raise typer.Exit(1) + try: + resolved = path.resolve() + root_resolved = root.resolve() + resolved.relative_to(root_resolved) + except ValueError: + err_console.print( + f"[red]Error:[/red] Path traversal detected: {path} is outside the allowed directory." + ) + raise typer.Exit(1) + return path + + +def _find_overlay_file(project_root: Path, workflow_id: str, overlay_id: str) -> Path | None: + """Locate a project-local overlay file by workflow id and overlay id.""" + _validate_workflow_id_or_exit(workflow_id) + _validate_overlay_id_or_exit(overlay_id, "overlay ID") + overlay_dir = _project_overlay_dir(project_root, workflow_id) + for suffix in (".yml", ".yaml"): + candidate = overlay_dir / f"{overlay_id}{suffix}" + candidate = _ensure_contained_path(candidate, _overlay_root(project_root)) + if candidate.is_file(): + if candidate.is_symlink(): + err_console.print( + f"[red]Error:[/red] Refusing to read symlinked overlay file {candidate}." + ) + raise typer.Exit(1) + return candidate + return None + + +def _ensure_contained_path(path: Path, root: Path) -> Path: + """Return *path* only if it resolves inside *root*; otherwise raise typer.Exit.""" + _reject_unsafe_dir(root, ".specify/workflows/overlays") + if path.is_symlink(): + err_console.print( + f"[red]Error:[/red] Refusing to use symlinked path {path}." + ) + raise typer.Exit(1) + try: + resolved = path.resolve() + root_resolved = root.resolve() + resolved.relative_to(root_resolved) + except ValueError: + err_console.print( + f"[red]Error:[/red] Path traversal detected: {path} is outside the allowed directory." + ) + raise typer.Exit(1) + return path + + +def _validate_priority(priority: int) -> None: + """Validate a user-supplied priority value.""" + if isinstance(priority, bool) or not isinstance(priority, int): + err_console.print("[red]Error:[/red] Priority must be an integer.") + raise typer.Exit(1) + if priority < 1: + err_console.print("[red]Error:[/red] Priority must be >= 1.") + raise typer.Exit(1) + + +def _read_overlay(path: Path) -> tuple[dict[str, Any] | None, list[str]]: + """Read and parse an overlay YAML file, returning (data, errors).""" + try: + content = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + return None, [f"Failed to read {path}: {exc}"] + try: + data = yaml.safe_load(content) + except yaml.YAMLError as exc: + return None, [f"Invalid YAML in {path}: {exc}"] + if not isinstance(data, dict): + return None, [f"Overlay {path} must be a YAML mapping."] + return data, [] + + +def workflow_overlay_add( + project_root: Path, + source: Path, + priority: int | None = None, +) -> Path | None: + """Add a project-local overlay from a YAML file. + + Returns the path of the installed overlay file, or None on failure. + """ + _reject_unsafe_workflow_storage(project_root) + data, errors = _read_overlay(source) + if data is None: + for err in errors: + err_console.print(f"[red]Error:[/red] {err}") + return None + + # Apply --priority override before validation so a valid CLI priority + # can fix a missing or invalid priority in the file. + if priority is not None: + _validate_priority(priority) + data["priority"] = priority + + overlay, validation_errors = validate_overlay_yaml(data) + if overlay is None: + err_console.print("[red]Error:[/red] Overlay validation failed:") + for err in validation_errors: + err_console.print(f" \u2022 {err}") + return None + + target_dir = _project_overlay_dir(project_root, overlay.extends) + target_dir.mkdir(parents=True, exist_ok=True) + # Reuse an existing .yaml file so we don't create a duplicate .yml layer. + existing = _find_overlay_file(project_root, overlay.extends, overlay.id) + if existing is not None: + target_path = existing + else: + target_path = _ensure_contained_path( + target_dir / f"{overlay.id}.yml", _overlay_root(project_root) + ) + + try: + target_path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8") + except OSError as exc: + err_console.print(f"[red]Error:[/red] Failed to write overlay: {exc}") + return None + + console.print( + f"[green]\u2713[/green] Overlay '{overlay.id}' added for workflow '{overlay.extends}'" + ) + return target_path + + +def _update_overlay_field( + project_root: Path, + workflow_id: str, + overlay_id: str, + field: str, + value: Any, +) -> bool: + """Update a single field in a project-local overlay file.""" + _reject_unsafe_workflow_storage(project_root) + path = _find_overlay_file(project_root, workflow_id, overlay_id) + if path is None: + err_console.print( + f"[red]Error:[/red] Overlay '{overlay_id}' not found for workflow '{workflow_id}'" + ) + return False + + data, errors = _read_overlay(path) + if data is None: + for err in errors: + err_console.print(f"[red]Error:[/red] {err}") + return False + + data[field] = value + overlay, validation_errors = validate_overlay_yaml(data) + if overlay is None: + err_console.print("[red]Error:[/red] Overlay validation failed:") + for err in validation_errors: + err_console.print(f" \u2022 {err}") + return False + + try: + path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8") + except OSError as exc: + err_console.print(f"[red]Error:[/red] Failed to write overlay: {exc}") + return False + + return True + + +def workflow_overlay_set_priority( + project_root: Path, + workflow_id: str, + overlay_id: str, + priority: int, +) -> bool: + """Set the priority of a project-local overlay.""" + _validate_priority(priority) + if _update_overlay_field(project_root, workflow_id, overlay_id, "priority", priority): + console.print( + f"[green]\u2713[/green] Priority of overlay '{overlay_id}' set to {priority}" + ) + return True + return False + + +def workflow_overlay_enable( + project_root: Path, + workflow_id: str, + overlay_id: str, +) -> bool: + """Enable a project-local overlay.""" + if _update_overlay_field(project_root, workflow_id, overlay_id, "enabled", True): + console.print(f"[green]\u2713[/green] Overlay '{overlay_id}' enabled") + return True + return False + + +def workflow_overlay_disable( + project_root: Path, + workflow_id: str, + overlay_id: str, +) -> bool: + """Disable a project-local overlay.""" + if _update_overlay_field(project_root, workflow_id, overlay_id, "enabled", False): + console.print(f"[green]\u2713[/green] Overlay '{overlay_id}' disabled") + return True + return False + + +def workflow_overlay_remove( + project_root: Path, + workflow_id: str, + overlay_id: str, +) -> bool: + """Remove a project-local overlay file.""" + _reject_unsafe_workflow_storage(project_root) + path = _find_overlay_file(project_root, workflow_id, overlay_id) + if path is None: + err_console.print( + f"[red]Error:[/red] Overlay '{overlay_id}' not found for workflow '{workflow_id}'" + ) + return False + + try: + path.unlink() + except OSError as exc: + err_console.print(f"[red]Error:[/red] Failed to remove overlay: {exc}") + return False + + console.print(f"[green]\u2713[/green] Overlay '{overlay_id}' removed") + return True + + +def workflow_overlay_list(project_root: Path, workflow_id: str) -> list[dict[str, Any]] | None: + """List all overlays for a workflow and print a summary table. + + Returns the raw list data for machine-readable callers, or None on error. + """ + _reject_unsafe_workflow_storage(project_root) + _validate_workflow_id_or_exit(workflow_id) + resolver = WorkflowResolver(project_root) + try: + layers = resolver.collect_all_layers(workflow_id) + except ValueError as exc: + err_console.print(f"[red]Error:[/red] {exc}") + return None + overlays = [layer for layer in layers if layer.tier != "base"] + + if not overlays: + console.print(f"[yellow]No overlays found for workflow '{workflow_id}'.[/yellow]") + return [] + + console.print(f"Overlays for workflow '{workflow_id}':") + rows: list[dict[str, Any]] = [] + for layer in overlays: + overlay = layer.content + rows.append({ + "id": overlay.id, + "source": layer.source, + "tier": layer.tier, + "priority": overlay.priority, + "enabled": overlay.enabled, + "path": str(layer.path) if layer.path else None, + }) + enabled_marker = "enabled" if overlay.enabled else "disabled" + console.print( + f" \u2022 {overlay.id} (priority={overlay.priority}, " + f"source={layer.source}, {enabled_marker})" + ) + return rows + + +def workflow_resolve(project_root: Path, workflow_id: str) -> dict[str, Any] | None: + """Print layer attribution for a resolved workflow. + + Returns a serializable attribution payload. + """ + _reject_unsafe_workflow_storage(project_root) + _validate_workflow_id_or_exit(workflow_id) + resolver = WorkflowResolver(project_root) + try: + definition, layers, attribution = resolver.resolve_with_layers(workflow_id) + except FileNotFoundError: + err_console.print( + f"[red]Error:[/red] Workflow '{workflow_id}' not found" + ) + return None + except ValueError as exc: + err_console.print(f"[red]Error:[/red] {exc}") + return None + + console.print(f"Resolved workflow '{workflow_id}':") + console.print("Layers (highest precedence first):") + for layer in layers: + console.print( + f" \u2022 [{layer.tier}] {layer.source} (priority={layer.priority})" + ) + + console.print("Step attribution:") + for composed in attribution: + console.print(f" \u2022 {composed.step_id}: {composed.source}") + + return { + "workflow_id": workflow_id, + "layers": [ + { + "source": layer.source, + "tier": layer.tier, + "priority": layer.priority, + } + for layer in layers + ], + "attribution": [ + {"step_id": composed.step_id, "source": composed.source} + for composed in attribution + ], + } diff --git a/src/specify_cli/workflows/overlays/composer.py b/src/specify_cli/workflows/overlays/composer.py new file mode 100644 index 0000000000..8ce3d16440 --- /dev/null +++ b/src/specify_cli/workflows/overlays/composer.py @@ -0,0 +1,105 @@ +"""Workflow overlay composer — builds a WorkflowDefinition from layers.""" + +from __future__ import annotations + +from typing import Any + +from ..engine import WorkflowDefinition, validate_workflow +from .layer_sources import Layer +from .merge import OverlayLayer, merge_steps, validate_edits + + +class StepListComposer: + """Compose a workflow from a base layer and overlay layers. + + - The base layer (tier="base") provides the full step list. + - Overlay layers provide edit operations. + - Overlays are applied in merge order: lowest priority first, highest last; + ties are resolved so installed overlays are applied before project + overlays, letting project overlays win. + - ``validate_workflow`` runs on the composed result. + """ + + def compose( + self, layers: list[Layer] + ) -> tuple[WorkflowDefinition | None, list]: + """Compose a ``WorkflowDefinition`` from the given layers. + + Returns ``(None, [])`` when no base layer is present. + """ + base_layer: Layer | None = None + overlay_layers: list[Layer] = [] + for layer in layers: + if layer.tier == "base": + base_layer = layer + else: + overlay_layers.append(layer) + + if base_layer is None or base_layer.path is None: + return None, [] + + # Read the base workflow definition from disk. + base_definition = WorkflowDefinition.from_yaml(base_layer.path) + base_steps = base_definition.data.get("steps", []) + if not isinstance(base_steps, list): + base_steps = [] + + # Sort overlays into merge order. + merge_order = sorted( + overlay_layers, + key=lambda layer: ( + layer.priority, + 0 if layer.tier == "installed-overlay" else 1, + layer.source, + ), + ) + + # Validate edits against base anchors before mutation. + base_step_ids = self._collect_base_step_ids(base_steps) + for layer in merge_order: + edit_errors = validate_edits(layer.content.edits, base_step_ids) + if edit_errors: + raise ValueError( + f"Overlay '{layer.content.id}' has invalid edits:\n - " + + "\n - ".join(edit_errors) + ) + + composed_steps, attribution = merge_steps( + base_steps, + [OverlayLayer(layer.content, layer.source) for layer in merge_order], + ) + + # Build composed data while preserving all non-step fields from base. + composed_data: dict[str, Any] = dict(base_definition.data) + composed_data["steps"] = composed_steps + + composed_definition = WorkflowDefinition(composed_data, source_path=base_layer.path) + validation_errors = validate_workflow(composed_definition) + if validation_errors: + # Surface validation errors as a single exception with details. + raise ValueError( + "Composed workflow is invalid:\n - " + + "\n - ".join(validation_errors) + ) + + return composed_definition, attribution + + def _collect_base_step_ids(self, steps: list[dict[str, Any]]) -> set[str]: + """Collect all base step IDs reachable in the step tree.""" + ids: set[str] = set() + for step in steps: + if not isinstance(step, dict): + continue + step_id = step.get("id") + if isinstance(step_id, str): + ids.add(step_id) + for key in ("then", "else", "steps", "default"): + nested = step.get(key) + if isinstance(nested, list): + ids.update(self._collect_base_step_ids(nested)) + cases = step.get("cases") + if isinstance(cases, dict): + for case_steps in cases.values(): + if isinstance(case_steps, list): + ids.update(self._collect_base_step_ids(case_steps)) + return ids diff --git a/src/specify_cli/workflows/overlays/layer_sources.py b/src/specify_cli/workflows/overlays/layer_sources.py new file mode 100644 index 0000000000..0b631a6db9 --- /dev/null +++ b/src/specify_cli/workflows/overlays/layer_sources.py @@ -0,0 +1,153 @@ +"""Workflow overlay layer sources.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import yaml + +from .schema import Overlay, validate_overlay_yaml + + +@dataclass +class Layer: + """A single layer in the workflow overlay stack.""" + + content: Overlay + source: str + tier: str + priority: int + path: Path | None = None + + +class OverlayLoadError(ValueError): + """Raised when an overlay file cannot be loaded or validated.""" + + def __init__(self, path: Path, errors: list[str]) -> None: + self.path = path + self.errors = errors + super().__init__(f"Invalid overlay {path}:\n - " + "\n - ".join(errors)) + + +def _resolve_project_overlay_root(project_root: Path) -> Path: + """Return the unresolved overlay root after rejecting symlinked ancestors. + + This mirrors the older workflow step-directory hardening pattern locally. + The path-walk / containment logic should be centralized in a shared helper + in a follow-up DRY pass, but this stays inline here to keep the Cluster 1 + security fix small and low-risk. + """ + project_root_resolved = project_root.resolve() + overlays_root = project_root / ".specify" / "workflows" / "overlays" + + current = project_root + for part in (".specify", "workflows", "overlays"): + current = current / part + if current.is_symlink(): + raise OverlayLoadError( + current, + [f"Symlinked overlay directories are not allowed ({current})"], + ) + if current.exists() and not current.is_dir(): + raise OverlayLoadError( + current, + [f"Overlay directory path is not a directory ({current})"], + ) + + try: + overlays_root.resolve().relative_to(project_root_resolved) + except ValueError: + raise OverlayLoadError( + overlays_root, + ["Overlay directory escapes the project root"], + ) from None + return overlays_root + + +class ProjectOverlaySource: + """Project-local overlays: ``.specify/workflows/overlays//*.yml``.""" + + tier = "project-overlay" + + def __init__(self, project_root: Path) -> None: + self.project_root = project_root + self.overlays_dir = project_root / ".specify" / "workflows" / "overlays" + + def collect(self, workflow_id: str) -> list[Layer]: + """Collect all project-local overlays for the given workflow id.""" + self.overlays_dir = _resolve_project_overlay_root(self.project_root) + workflow_overlay_dir = self.overlays_dir / workflow_id + if workflow_overlay_dir.is_symlink(): + raise OverlayLoadError( + workflow_overlay_dir, + ["Symlinked overlay directories are not allowed"], + ) + if workflow_overlay_dir.exists() and not workflow_overlay_dir.is_dir(): + raise OverlayLoadError( + workflow_overlay_dir, + ["Overlay directory path is not a directory"], + ) + if not workflow_overlay_dir.is_dir(): + return [] + layers: list[Layer] = [] + for path in sorted(workflow_overlay_dir.iterdir()): + if not path.is_file() or path.suffix not in (".yml", ".yaml"): + continue + if path.is_symlink(): + raise OverlayLoadError(path, ["Symlinked overlay files are not allowed"]) + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + except yaml.YAMLError as exc: + raise OverlayLoadError(path, [f"Invalid YAML: {exc}"]) from exc + except (OSError, UnicodeDecodeError) as exc: + raise OverlayLoadError(path, [f"Cannot load overlay: {exc}"]) from exc + overlay, errors = validate_overlay_yaml(data) + if overlay is None or errors: + raise OverlayLoadError(path, errors) + if not overlay.enabled: + continue + if overlay.extends != workflow_id: + continue + layers.append( + Layer( + content=overlay, + source=f"project:{overlay.id}", + tier=self.tier, + priority=overlay.priority, + path=path, + ) + ) + return layers + + +class BaseWorkflowSource: + """Base workflow layer: ``.specify/workflows//workflow.yml``.""" + + tier = "base" + + def __init__(self, project_root: Path) -> None: + self.project_root = project_root + self.workflows_dir = project_root / ".specify" / "workflows" + + def collect(self, workflow_id: str) -> list[Layer]: + """Return the base workflow as a single layer if it exists.""" + path = self.workflows_dir / workflow_id / "workflow.yml" + if not path.is_file(): + return [] + # The base layer is represented by an Overlay with empty edits. + overlay = Overlay( + id=workflow_id, + extends=workflow_id, + priority=0, + edits=[], + ) + return [ + Layer( + content=overlay, + source="base", + tier=self.tier, + priority=0, + path=path, + ) + ] diff --git a/src/specify_cli/workflows/overlays/merge.py b/src/specify_cli/workflows/overlays/merge.py new file mode 100644 index 0000000000..a1a6fdd607 --- /dev/null +++ b/src/specify_cli/workflows/overlays/merge.py @@ -0,0 +1,391 @@ +"""Pure-function merge engine for workflow step lists.""" + +from __future__ import annotations + +import copy +from dataclasses import dataclass +from typing import Any + +from .schema import VALID_OPERATIONS, Overlay, OverlayEdit + + +@dataclass(frozen=True) +class ComposedStep: + """Attribution tracking for a single composed step.""" + + step_id: str + source: str + + +@dataclass(frozen=True) +class OverlayLayer: + """An overlay together with its layer source for attribution.""" + + overlay: Overlay + source: str + + +# Nested step keys that may contain a list of steps. +_NESTED_LIST_KEYS = ("then", "else", "steps", "default") + + +def find_step( + steps: list[dict[str, Any]], step_id: str +) -> tuple[list[dict[str, Any]], int] | None: + """Recursively locate a step by ID and return its (parent_list, index). + + Searches flat lists and nested lists inside ``then``, ``else``, ``steps``, + ``default``, and ``cases.*``. Does *not* descend into ``fan-out`` template + steps because those are runtime-multiplied stamps, not uniquely-addressable + nodes. + """ + for i, step in enumerate(steps): + if not isinstance(step, dict): + continue + if step.get("id") == step_id: + return (steps, i) + for key in _NESTED_LIST_KEYS: + nested = step.get(key) + if isinstance(nested, list): + result = find_step(nested, step_id) + if result is not None: + return result + cases = step.get("cases") + if isinstance(cases, dict): + for case_steps in cases.values(): + if isinstance(case_steps, list): + result = find_step(case_steps, step_id) + if result is not None: + return result + return None + + +def _all_base_step_ids(steps: list[dict[str, Any]]) -> set[str]: + """Collect all step IDs reachable in a step tree (excluding fan-out templates).""" + ids: set[str] = set() + for step in steps: + if not isinstance(step, dict): + continue + step_id = step.get("id") + if isinstance(step_id, str): + ids.add(step_id) + for key in _NESTED_LIST_KEYS: + nested = step.get(key) + if isinstance(nested, list): + ids.update(_all_base_step_ids(nested)) + cases = step.get("cases") + if isinstance(cases, dict): + for case_steps in cases.values(): + if isinstance(case_steps, list): + ids.update(_all_base_step_ids(case_steps)) + return ids + + +def _descendant_ids(step: dict[str, Any]) -> set[str]: + """Return all step IDs nested inside *step* (not including *step* itself).""" + ids: set[str] = set() + for key in _NESTED_LIST_KEYS: + nested = step.get(key) + if isinstance(nested, list): + ids.update(_all_base_step_ids(nested)) + cases = step.get("cases") + if isinstance(cases, dict): + for case_steps in cases.values(): + if isinstance(case_steps, list): + ids.update(_all_base_step_ids(case_steps)) + return ids + + +def _check_anchor_conflicts( + anchor_operations: dict[str, str], + base_steps: list[dict[str, Any]], +) -> list[str]: + """Return error messages for anchor pairs where one is an ancestor of the other. + + Only flags conflicts where the ancestor's winning edit is ``replace`` or + ``remove`` — operations that destroy the subtree and make any descendant + anchor unresolvable. Pure insert operations on an ancestor leave it intact, + so its descendants remain reachable regardless of processing order. + + Callers should raise on any returned errors before mutating the step tree. + """ + errors: list[str] = [] + for anchor, operation in sorted(anchor_operations.items()): + if operation in ("insert_after", "insert_before"): + # Inserts leave the ancestor step intact; descendants are unaffected. + continue + location = find_step(base_steps, anchor) + if location is None: + continue # missing anchors are reported by validate_edits + parent_list, idx = location + step = parent_list[idx] + conflicting = set(anchor_operations.keys()) & _descendant_ids(step) + for child_anchor in sorted(conflicting): + errors.append( + f"Anchor conflict: '{anchor}' is an ancestor of '{child_anchor}'. " + "Targeting both anchors in the same overlay set produces " + "order-dependent results; restructure edits to avoid nesting." + ) + return errors + + +def _init_sources_recursively( + steps: list[dict[str, Any]], sources: dict[str, str] +) -> None: + """Initialize attribution sources for all base steps, recursively.""" + for step in steps: + if not isinstance(step, dict): + continue + step_id = step.get("id") + if isinstance(step_id, str): + sources[step_id] = "base" + for key in _NESTED_LIST_KEYS: + nested = step.get(key) + if isinstance(nested, list): + _init_sources_recursively(nested, sources) + cases = step.get("cases") + if isinstance(cases, dict): + for case_steps in cases.values(): + if isinstance(case_steps, list): + _init_sources_recursively(case_steps, sources) + + +def _record_sources_recursively( + step: dict[str, Any], + source: str, + sources: dict[str, str], +) -> None: + """Record *source* for a step and all its nested child steps. + + Traverses ``then``, ``else``, ``steps``, ``default``, and ``cases.*`` + so that ``workflow resolve`` attributes every step inside a composite + insert or replacement to the correct overlay layer. + """ + step_id = step.get("id") + if isinstance(step_id, str): + sources[step_id] = source + for key in _NESTED_LIST_KEYS: + nested = step.get(key) + if isinstance(nested, list): + for child in nested: + if isinstance(child, dict): + _record_sources_recursively(child, source, sources) + cases = step.get("cases") + if isinstance(cases, dict): + for case_steps in cases.values(): + if isinstance(case_steps, list): + for child in case_steps: + if isinstance(child, dict): + _record_sources_recursively(child, source, sources) + + +def _remove_sources_recursively( + step: dict[str, Any], + sources: dict[str, str], +) -> None: + """Remove source entries for a step and all its nested child steps. + + Traverses the same nesting keys as ``_record_sources_recursively``. + """ + step_id = step.get("id") + if isinstance(step_id, str): + sources.pop(step_id, None) + for key in _NESTED_LIST_KEYS: + nested = step.get(key) + if isinstance(nested, list): + for child in nested: + if isinstance(child, dict): + _remove_sources_recursively(child, sources) + cases = step.get("cases") + if isinstance(cases, dict): + for case_steps in cases.values(): + if isinstance(case_steps, list): + for child in case_steps: + if isinstance(child, dict): + _remove_sources_recursively(child, sources) + + + +def _build_attribution( + steps: list[dict[str, Any]], + sources: dict[str, str], +) -> list[ComposedStep]: + """Build an ordered attribution list from the composed step tree.""" + result: list[ComposedStep] = [] + for step in steps: + if not isinstance(step, dict): + continue + step_id = step.get("id") + if isinstance(step_id, str): + result.append(ComposedStep(step_id, sources.get(step_id, "unknown"))) + for key in _NESTED_LIST_KEYS: + nested = step.get(key) + if isinstance(nested, list): + result.extend(_build_attribution(nested, sources)) + cases = step.get("cases") + if isinstance(cases, dict): + for case_steps in cases.values(): + if isinstance(case_steps, list): + result.extend(_build_attribution(case_steps, sources)) + return result + + +def _traverse_and_apply( + steps: list[dict[str, Any]], + edits_by_anchor: dict[str, list[tuple[OverlayLayer, OverlayEdit]]], + sources: dict[str, str], +) -> list[dict[str, Any]]: + """Walk the original step tree and apply overlay edits as each step is encountered. + + Edits are always resolved against the *original* structure — this function + traverses the unmodified list passed in, so a replacement step's new ID can + never be mistaken for a base anchor. Nested lists (``then``, ``else``, etc.) + are recursed into only for steps that survive the edit (not for replaced + steps). + + *edits* are expected to be in merge order (lowest priority first, highest + priority last); the winning edit for each anchor is ``edits[-1]``. + """ + result: list[dict[str, Any]] = [] + + for step in steps: + if not isinstance(step, dict): + result.append(step) + continue + + step_id = step.get("id") + edits = edits_by_anchor.get(step_id, []) if isinstance(step_id, str) else [] + winning_edit = edits[-1][1] if edits else None + + if winning_edit is not None and winning_edit.operation == "remove": + # Winning edit removes this step; ignore all other edits on this anchor. + _remove_sources_recursively(step, sources) + continue + + # Insert before (in merge order). + for layer, edit in edits: + if edit.operation == "insert_before": + new_step = copy.deepcopy(edit.step) + _record_sources_recursively(new_step, layer.source, sources) + result.append(new_step) + + if winning_edit is not None and winning_edit.operation == "replace": + winning_layer = edits[-1][0] + new_step = copy.deepcopy(winning_edit.step) + _remove_sources_recursively(step, sources) + _record_sources_recursively(new_step, winning_layer.source, sources) + result.append(new_step) + else: + # No replacement: keep this step and recurse into its nested lists. + for key in _NESTED_LIST_KEYS: + nested = step.get(key) + if isinstance(nested, list): + step[key] = _traverse_and_apply(nested, edits_by_anchor, sources) + cases = step.get("cases") + if isinstance(cases, dict): + for case_key, case_steps in cases.items(): + if isinstance(case_steps, list): + cases[case_key] = _traverse_and_apply(case_steps, edits_by_anchor, sources) + result.append(step) + + # Insert after (highest priority closest to anchor — reversed merge order). + for layer, edit in reversed(edits): + if edit.operation == "insert_after": + new_step = copy.deepcopy(edit.step) + _record_sources_recursively(new_step, layer.source, sources) + result.append(new_step) + + return result + + +def merge_steps( + base_steps: list[dict[str, Any]], + overlays: list[OverlayLayer], +) -> tuple[list[dict[str, Any]], list[ComposedStep]]: + """Apply overlays to base steps in merge order and return composed steps. + + *overlays* is expected to be sorted by merge order (lowest priority first, + highest priority last). The returned step list is a deep copy of the base; + base_steps is never mutated. + + Higher-wins semantics are enforced for edits that target the same base + anchor: the highest-priority edit (last in *overlays*) decides the fate of + the anchor. A lower-priority ``remove`` cannot prevent a higher-priority + ``replace`` or ``insert_*`` on the same anchor. + """ + steps = copy.deepcopy(base_steps) + sources: dict[str, str] = {} + _init_sources_recursively(steps, sources) + + # Group edits by anchor, preserving merge order. + edits_by_anchor: dict[str, list[tuple[OverlayLayer, OverlayEdit]]] = {} + for layer in overlays: + for edit in layer.overlay.edits: + edits_by_anchor.setdefault(edit.anchor, []).append((layer, edit)) + + # Raise early for non-remove edits that target anchors not present in the base. + # Overlays always apply to the original tree; they cannot target steps introduced + # by other overlays. + base_ids = _all_base_step_ids(base_steps) + for anchor, anchor_edits in edits_by_anchor.items(): + winning_op = anchor_edits[-1][1].operation + if winning_op != "remove" and anchor not in base_ids: + raise ValueError(f"Anchor '{anchor}' not found in workflow steps.") + + # Reject edits that target anchors with a parent/descendant relationship when + # the ancestor edit replaces or removes its subtree — those produce + # order-dependent results. Pure insert edits on an ancestor are safe because + # the ancestor step (and its descendants) remain intact. + anchor_winning_ops = { + anchor: anchor_edits[-1][1].operation + for anchor, anchor_edits in edits_by_anchor.items() + } + anchor_conflicts = _check_anchor_conflicts(anchor_winning_ops, base_steps) + if anchor_conflicts: + raise ValueError( + "Overlay anchor conflict(s) detected:\n - " + "\n - ".join(anchor_conflicts) + ) + + # Apply all overlay edits via a single-pass traversal of the original tree. + # Each edit is resolved against the original step structure, so a replacement + # step's new ID can never be mistaken for a base anchor in a later edit group. + result = _traverse_and_apply(steps, edits_by_anchor, sources) + + attribution = _build_attribution(result, sources) + return result, attribution + + +def validate_edits( + edits: list[OverlayEdit], + base_step_ids: set[str], +) -> list[str]: + """Validate overlay edits against a set of known base step IDs. + + Returns a list of human-readable error messages. Does not raise. + """ + errors: list[str] = [] + for idx, edit in enumerate(edits): + if edit.operation not in VALID_OPERATIONS: + errors.append(f"Edit {idx}: invalid operation {edit.operation!r}.") + continue + if edit.anchor not in base_step_ids: + errors.append( + f"Edit {idx}: anchor '{edit.anchor}' does not match any base step id." + ) + if edit.operation == "remove": + if edit.step is not None: + errors.append(f"Edit {idx}: 'remove' must not include a step.") + continue + if not isinstance(edit.step, dict): + errors.append(f"Edit {idx}: '{edit.operation}' requires a step mapping.") + continue + step_id = edit.step.get("id") + if not isinstance(step_id, str) or not step_id: + errors.append(f"Edit {idx}: step is missing required 'id'.") + continue + if ":" in step_id: + errors.append( + f"Edit {idx}: step id {step_id!r} contains ':' which is reserved " + "for engine-generated nested IDs." + ) + return errors diff --git a/src/specify_cli/workflows/overlays/schema.py b/src/specify_cli/workflows/overlays/schema.py new file mode 100644 index 0000000000..dc25c4a9ad --- /dev/null +++ b/src/specify_cli/workflows/overlays/schema.py @@ -0,0 +1,174 @@ +"""Workflow overlay schema — dataclasses and validation for overlay manifests.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any, Literal + + +# Safe single-segment identifiers: no path separators, no traversal, no dots. +_SAFE_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$") +_RESERVED_OVERLAY_WORKFLOW_IDS: frozenset[str] = frozenset({"overlays"}) + +VALID_OPERATIONS = frozenset({"insert_after", "insert_before", "replace", "remove"}) + +# Map shorthand keys to operation names. +_SHORTHAND_OPERATION_KEYS: frozenset[str] = VALID_OPERATIONS + + +@dataclass(frozen=True) +class OverlayEdit: + """A single edit operation on a workflow step list.""" + + operation: Literal["insert_after", "insert_before", "replace", "remove"] + anchor: str + step: dict[str, Any] | None = None + + +@dataclass +class Overlay: + """A declared overlay (one YAML file).""" + + id: str + extends: str + priority: int + edits: list[OverlayEdit] + enabled: bool = True + + +def _validate_safe_id(value: str, field_name: str, allow_reserved: bool = False) -> str | None: + """Return an error message if *value* is not a safe path segment ID.""" + if not isinstance(value, str) or not value: + return f"Overlay '{field_name}' is required and must be a non-empty string." + if not _SAFE_ID_PATTERN.fullmatch(value): + return ( + f"Overlay '{field_name}' {value!r} contains invalid characters; " + "only lowercase letters, digits, and hyphens are allowed." + ) + if not allow_reserved and value in _RESERVED_OVERLAY_WORKFLOW_IDS: + return f"Overlay '{field_name}' {value!r} is reserved." + return None + + +def _parse_edit(edit_raw: dict[str, Any], idx: int) -> tuple[OverlayEdit | None, str | None]: + """Parse a single edit dict into an OverlayEdit or an error string.""" + shorthand_keys = [key for key in _SHORTHAND_OPERATION_KEYS if key in edit_raw] + has_operation = "operation" in edit_raw + + operation: str | None = None + anchor: Any = None + + if shorthand_keys and has_operation: + return None, ( + f"Edit at index {idx} mixes shorthand operation key " + f"({shorthand_keys[0]!r}) with explicit 'operation' field." + ) + + if len(shorthand_keys) > 1: + return None, ( + f"Edit at index {idx} has multiple operation keys: " + f"{', '.join(repr(k) for k in shorthand_keys)}." + ) + + if shorthand_keys: + operation = shorthand_keys[0] + anchor = edit_raw[operation] + elif has_operation: + operation = edit_raw.get("operation") + anchor = edit_raw.get("anchor") + else: + return None, f"Edit at index {idx} has no operation; expected one of {sorted(VALID_OPERATIONS)}." + + if operation not in VALID_OPERATIONS: + return None, f"Edit at index {idx} has invalid operation {operation!r}." + + if not isinstance(anchor, str) or not anchor: + return None, f"Edit at index {idx} has invalid 'anchor'." + + step = edit_raw.get("step") + if operation == "remove": + if step is not None: + return None, f"Edit at index {idx} ('remove') must not include 'step'." + return OverlayEdit(operation=operation, anchor=anchor), None + + if not isinstance(step, dict): + return None, f"Edit at index {idx} ('{operation}') requires 'step' mapping." + step_id = step.get("id") + if not isinstance(step_id, str) or not step_id: + return None, f"Edit at index {idx} step is missing required 'id'." + if ":" in step_id: + return None, ( + f"Edit at index {idx} step id {step_id!r} contains ':' " + "which is reserved for engine-generated nested IDs." + ) + return OverlayEdit(operation=operation, anchor=anchor, step=step), None + + +def validate_overlay_yaml(data: dict[str, Any]) -> tuple[Overlay | None, list[str]]: + """Validate an overlay manifest dict and return (Overlay, errors). + + Errors are returned as a list of strings; validation never raises. + """ + errors: list[str] = [] + + if not isinstance(data, dict): + return None, ["Overlay manifest must be a mapping."] + + overlay_id = data.get("id") + if err := _validate_safe_id(overlay_id, "id"): + errors.append(err) + overlay_id = "" + + extends = data.get("extends") + if err := _validate_safe_id(extends, "extends"): + errors.append(err) + extends = "" + + priority = data.get("priority") + if isinstance(priority, bool) or not isinstance(priority, int): + errors.append( + f"Overlay 'priority' must be an integer >= 1, got {priority!r}." + ) + priority = 0 + elif priority < 1: + errors.append( + f"Overlay 'priority' must be an integer >= 1, got {priority!r}." + ) + + edits_raw = data.get("edits") + edits: list[OverlayEdit] = [] + if not isinstance(edits_raw, list): + errors.append("Overlay 'edits' is required and must be a list.") + elif not edits_raw: + errors.append("Overlay 'edits' must be a non-empty list.") + else: + for idx, edit_raw in enumerate(edits_raw): + if not isinstance(edit_raw, dict): + errors.append(f"Edit at index {idx} must be a mapping.") + continue + edit, err = _parse_edit(edit_raw, idx) + if err: + errors.append(err) + continue + if edit is not None: + edits.append(edit) + + enabled = data.get("enabled", True) + if not isinstance(enabled, bool): + errors.append("Overlay 'enabled' must be a boolean.") + enabled = bool(enabled) + + if errors: + return None, errors + + return ( + Overlay( + id=overlay_id, + extends=extends, + priority=priority, + edits=edits, + enabled=enabled, + ), + [], + ) diff --git a/tests/test_workflows.py b/tests/test_workflows.py index eab4bef540..1715845723 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -6916,7 +6916,7 @@ def test_remove_rejects_traversal_registry_key(self, project_dir, monkeypatch): assert "Invalid workflow ID" in result.output assert sentinel.read_text(encoding="utf-8") == "keep" - @pytest.mark.parametrize("workflow_id", ["runs", "steps"]) + @pytest.mark.parametrize("workflow_id", ["overlays", "runs", "steps"]) def test_remove_rejects_reserved_storage_ids( self, project_dir, monkeypatch, workflow_id ): @@ -7302,9 +7302,39 @@ def test_safe_workflow_id_dir_escapes_markup_in_invalid_id(self, temp_dir, capsy # Literal bracketed text survives; Rich did not consume it as a tag. assert "[red]evil[/red]" in out + def test_add_rejects_reserved_overlay_storage_id(self, temp_dir, monkeypatch): + """workflow add must not install into the overlay storage directory.""" + from typer.testing import CliRunner + from specify_cli import app + + (temp_dir / ".specify" / "workflows").mkdir(parents=True) + overlay_file = temp_dir / "incoming.yml" + overlay_file.write_text( + """ +schema_version: "1.0" +workflow: + id: "overlays" + name: "Bad Workflow" + version: "1.0.0" +steps: + - id: step-one + command: speckit.specify +""".strip() + + "\n", + encoding="utf-8", + ) + + monkeypatch.chdir(temp_dir) + result = CliRunner().invoke(app, ["workflow", "add", str(overlay_file)]) + + assert result.exit_code != 0 + assert "Invalid workflow ID" in result.output + assert not (temp_dir / ".specify" / "workflows" / "overlays" / "workflow.yml").exists() + @pytest.mark.parametrize( "workflow_id", [ + "overlays", "runs", "steps", "nested/workflow", diff --git a/tests/workflows/conftest.py b/tests/workflows/conftest.py new file mode 100644 index 0000000000..03d28890dd --- /dev/null +++ b/tests/workflows/conftest.py @@ -0,0 +1,26 @@ +"""Shared fixtures for workflow tests.""" + +from __future__ import annotations + +import shutil +import sys +import tempfile +from pathlib import Path + +import pytest + + +@pytest.fixture +def temp_dir(): + """Create a temporary directory for tests.""" + tmpdir = tempfile.mkdtemp() + yield Path(tmpdir) + shutil.rmtree(tmpdir, ignore_errors=(sys.platform == "win32")) + + +@pytest.fixture +def project_dir(temp_dir): + """Create a mock spec-kit project with ``.specify/workflows/`` directory.""" + workflows_dir = temp_dir / ".specify" / "workflows" + workflows_dir.mkdir(parents=True, exist_ok=True) + return temp_dir diff --git a/tests/workflows/test_overlay_commands.py b/tests/workflows/test_overlay_commands.py new file mode 100644 index 0000000000..f8dc0ba072 --- /dev/null +++ b/tests/workflows/test_overlay_commands.py @@ -0,0 +1,468 @@ +"""Tests for workflow overlay CLI commands.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml +from typer.testing import CliRunner + +from specify_cli import app + + +runner = CliRunner() + + +@pytest.fixture +def project_dir(tmp_path): + """Create a mock spec-kit project with ``.specify/workflows/`` directory.""" + workflows_dir = tmp_path / ".specify" / "workflows" + workflows_dir.mkdir(parents=True, exist_ok=True) + return tmp_path + + +def _write_workflow(project_root: Path, workflow_id: str, data: dict) -> Path: + wf_dir = project_root / ".specify" / "workflows" / workflow_id + wf_dir.mkdir(parents=True, exist_ok=True) + wf_path = wf_dir / "workflow.yml" + wf_path.write_text(yaml.safe_dump(data), encoding="utf-8") + return wf_path + + +def _write_overlay(project_root: Path, workflow_id: str, overlay_id: str, data: dict) -> Path: + ov_dir = project_root / ".specify" / "workflows" / "overlays" / workflow_id + ov_dir.mkdir(parents=True, exist_ok=True) + ov_path = ov_dir / f"{overlay_id}.yml" + ov_path.write_text(yaml.safe_dump(data), encoding="utf-8") + return ov_path + + +class TestOverlayCli: + """CLI-level tests for ``specify workflow overlay *``.""" + + def test_overlay_add(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + overlay_file = project_dir / "overlay.yml" + overlay_file.write_text( + yaml.safe_dump( + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + + result = runner.invoke( + app, ["workflow", "overlay", "add", str(overlay_file), "--priority", "5"] + ) + assert result.exit_code == 0, result.output + assert "Overlay 'ov1' added" in result.output + + installed = project_dir / ".specify" / "workflows" / "overlays" / "wf" / "ov1.yml" + assert installed.is_file() + data = yaml.safe_load(installed.read_text(encoding="utf-8")) + assert data["priority"] == 5 + + def test_overlay_add_reuses_yaml_extension(self, project_dir, monkeypatch): + """If .yaml already exists, overlay add must write to it instead of creating .yml.""" + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + # Pre-create the overlay using the .yaml extension. + existing_yaml = project_dir / ".specify" / "workflows" / "overlays" / "wf" / "ov1.yaml" + existing_yaml.parent.mkdir(parents=True, exist_ok=True) + existing_yaml.write_text( + yaml.safe_dump( + { + "id": "ov1", + "extends": "wf", + "priority": 1, + "edits": [{"remove": "a"}], + } + ), + encoding="utf-8", + ) + + overlay_file = project_dir / "overlay.yml" + overlay_file.write_text( + yaml.safe_dump( + { + "id": "ov1", + "extends": "wf", + "priority": 20, + "edits": [{"remove": "a"}], + } + ), + encoding="utf-8", + ) + + result = runner.invoke(app, ["workflow", "overlay", "add", str(overlay_file)]) + assert result.exit_code == 0, result.output + + # Should have written to the pre-existing .yaml file. + assert existing_yaml.is_file() + data = yaml.safe_load(existing_yaml.read_text(encoding="utf-8")) + assert data["priority"] == 20 + + # Must NOT have created a duplicate .yml alongside the .yaml. + duplicate_yml = existing_yaml.with_suffix(".yml") + assert not duplicate_yml.exists(), "duplicate .yml was created alongside existing .yaml" + + def test_overlay_add_with_priority_override_missing_in_file(self, project_dir, monkeypatch): + """--priority must fix a missing priority in the overlay file.""" + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + # Overlay file has NO priority field + overlay_file = project_dir / "overlay.yml" + overlay_file.write_text( + yaml.safe_dump( + { + "id": "ov1", + "extends": "wf", + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + + result = runner.invoke( + app, ["workflow", "overlay", "add", str(overlay_file), "--priority", "5"] + ) + assert result.exit_code == 0, result.output + assert "Overlay 'ov1' added" in result.output + + installed = project_dir / ".specify" / "workflows" / "overlays" / "wf" / "ov1.yml" + assert installed.is_file() + data = yaml.safe_load(installed.read_text(encoding="utf-8")) + assert data["priority"] == 5 + + def test_overlay_set_priority(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + result = runner.invoke( + app, ["workflow", "overlay", "set-priority", "wf", "ov1", "20"] + ) + assert result.exit_code == 0, result.output + data = yaml.safe_load( + ( + project_dir / ".specify" / "workflows" / "overlays" / "wf" / "ov1.yml" + ).read_text(encoding="utf-8") + ) + assert data["priority"] == 20 + + def test_overlay_disable_and_enable(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + result = runner.invoke(app, ["workflow", "overlay", "disable", "wf", "ov1"]) + assert result.exit_code == 0, result.output + data = yaml.safe_load( + ( + project_dir / ".specify" / "workflows" / "overlays" / "wf" / "ov1.yml" + ).read_text(encoding="utf-8") + ) + assert data["enabled"] is False + + result = runner.invoke(app, ["workflow", "overlay", "enable", "wf", "ov1"]) + assert result.exit_code == 0, result.output + data = yaml.safe_load( + ( + project_dir / ".specify" / "workflows" / "overlays" / "wf" / "ov1.yml" + ).read_text(encoding="utf-8") + ) + assert data["enabled"] is True + + def test_overlay_remove(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + result = runner.invoke(app, ["workflow", "overlay", "remove", "wf", "ov1"]) + assert result.exit_code == 0, result.output + assert not ( + project_dir / ".specify" / "workflows" / "overlays" / "wf" / "ov1.yml" + ).exists() + + def test_overlay_list(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + result = runner.invoke(app, ["workflow", "overlay", "list", "wf"]) + assert result.exit_code == 0, result.output + assert "ov1" in result.output + + def test_workflow_resolve(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + result = runner.invoke(app, ["workflow", "resolve", "wf"]) + assert result.exit_code == 0, result.output + assert "base" in result.output + assert "project:ov1" in result.output + assert "new" in result.output + + def test_workflow_resolve_equal_priority_winner_shown_first(self, project_dir, monkeypatch): + """Equal-priority overlays must be displayed with the actual winner (last applied) first.""" + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + # "zzz" sorts last alphabetically, so the composer applies it last → it wins. + # "aaa" sorts first, so it is applied first → it loses. + # The display must show the winner ("project:zzz") before the loser ("project:aaa"). + _write_overlay( + project_dir, + "wf", + "aaa", + { + "id": "aaa", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "aaa-step", "type": "command", "command": "echo"}, + } + ], + }, + ) + _write_overlay( + project_dir, + "wf", + "zzz", + { + "id": "zzz", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "zzz-step", "type": "command", "command": "echo"}, + } + ], + }, + ) + + result = runner.invoke(app, ["workflow", "resolve", "wf"]) + assert result.exit_code == 0, result.output + zzz_pos = result.output.index("project:zzz") + aaa_pos = result.output.index("project:aaa") + assert zzz_pos < aaa_pos, ( + "project:zzz (the winning source) should appear before project:aaa in the output" + ) + + def test_workflow_add_does_not_copy_overlays(self, project_dir, monkeypatch, tmp_path): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + source_dir = tmp_path / "source-wf" + source_dir.mkdir() + (source_dir / "workflow.yml").write_text( + yaml.safe_dump( + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + } + ), + encoding="utf-8", + ) + overlays_dir = source_dir / "overlays" + overlays_dir.mkdir() + (overlays_dir / "ov1.yml").write_text( + yaml.safe_dump( + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + + result = runner.invoke(app, ["workflow", "add", str(source_dir)]) + assert result.exit_code == 0, result.output + # Overlays in the source directory should NOT be copied — workflow add + # only installs the workflow.yml, not sibling overlays. + installed_overlay = ( + project_dir / ".specify" / "workflows" / "wf" / "overlays" / "ov1.yml" + ) + assert not installed_overlay.exists() diff --git a/tests/workflows/test_overlay_composer.py b/tests/workflows/test_overlay_composer.py new file mode 100644 index 0000000000..dc3b0e90d6 --- /dev/null +++ b/tests/workflows/test_overlay_composer.py @@ -0,0 +1,106 @@ +"""Tests for StepListComposer validation and error handling.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +from specify_cli.workflows.overlays import WorkflowResolver +from specify_cli.workflows.overlays.composer import StepListComposer +from specify_cli.workflows.overlays.layer_sources import BaseWorkflowSource, Layer +from specify_cli.workflows.overlays.schema import Overlay, OverlayEdit + + +def _write_workflow(project_root: Path, workflow_id: str, data: dict) -> Path: + wf_dir = project_root / ".specify" / "workflows" / workflow_id + wf_dir.mkdir(parents=True, exist_ok=True) + wf_path = wf_dir / "workflow.yml" + wf_path.write_text(yaml.safe_dump(data), encoding="utf-8") + return wf_path + + +class TestStepListComposerValidation: + """Composer validates edits before applying them.""" + + def test_composer_reports_invalid_anchors(self, project_dir): + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + base_layer = BaseWorkflowSource(project_dir).collect("wf")[0] + + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[OverlayEdit("insert_after", "missing", {"id": "new", "type": "command", "command": "echo"})], + ) + layer = Layer(content=overlay, source="project:ov", tier="project-overlay", priority=10) + composer = StepListComposer() + with pytest.raises(ValueError, match="does not match any base step id"): + composer.compose([base_layer, layer]) + + def test_composer_validates_edits_before_merge(self, project_dir): + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "a", {"id": "bad:id", "type": "command", "command": "echo"})], + ) + layer = Layer(content=overlay, source="project:ov", tier="project-overlay", priority=10) + composer = StepListComposer() + with pytest.raises(ValueError, match="bad:id"): + composer.compose([BaseWorkflowSource(project_dir).collect("wf")[0], layer]) + + def test_resolver_reports_invalid_anchor_as_validation_error(self, project_dir): + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + overlay_file = project_dir / "overlay.yml" + overlay_file.write_text( + yaml.safe_dump( + { + "id": "ov", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "missing", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + # Manually inject the overlay by writing it to disk in the correct location. + overlay_dir = project_dir / ".specify" / "workflows" / "overlays" / "wf" + overlay_dir.mkdir(parents=True, exist_ok=True) + (overlay_dir / "ov.yml").write_text(overlay_file.read_text(encoding="utf-8"), encoding="utf-8") + + resolver = WorkflowResolver(project_dir) + with pytest.raises(ValueError, match="missing"): + resolver.resolve("wf") diff --git a/tests/workflows/test_overlay_layer_sources.py b/tests/workflows/test_overlay_layer_sources.py new file mode 100644 index 0000000000..befebb4648 --- /dev/null +++ b/tests/workflows/test_overlay_layer_sources.py @@ -0,0 +1,60 @@ +"""Tests for ProjectOverlaySource file-read error handling.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + +import pytest +import yaml + +from specify_cli.workflows.overlays.layer_sources import ( + OverlayLoadError, + ProjectOverlaySource, +) + + +@pytest.fixture +def project_dir(tmp_path: Path) -> Path: + workflows_dir = tmp_path / ".specify" / "workflows" + workflows_dir.mkdir(parents=True, exist_ok=True) + return tmp_path + + +def _write_overlay_file(project_dir: Path, workflow_id: str, overlay_id: str, data: dict) -> Path: + ov_dir = project_dir / ".specify" / "workflows" / "overlays" / workflow_id + ov_dir.mkdir(parents=True, exist_ok=True) + path = ov_dir / f"{overlay_id}.yml" + path.write_text(yaml.safe_dump(data), encoding="utf-8") + return path + + +class TestProjectOverlaySourceFileReadErrors: + """File-read errors must be wrapped in OverlayLoadError, not leaked as raw tracebacks.""" + + def test_oserror_raises_overlay_load_error(self, project_dir: Path) -> None: + """An OSError from read_text (e.g. permission denied) is wrapped in OverlayLoadError.""" + _write_overlay_file( + project_dir, + "wf", + "ov1", + {"id": "ov1", "extends": "wf", "priority": 5, "edits": []}, + ) + source = ProjectOverlaySource(project_dir) + with patch.object(Path, "read_text", side_effect=OSError("Permission denied")): + with pytest.raises(OverlayLoadError) as exc_info: + source.collect("wf") + assert exc_info.value.errors, "OverlayLoadError must carry a non-empty errors list" + + def test_unicode_error_raises_overlay_load_error(self, project_dir: Path) -> None: + """A file containing non-UTF-8 bytes raises OverlayLoadError, not UnicodeDecodeError.""" + ov_dir = project_dir / ".specify" / "workflows" / "overlays" / "wf" + ov_dir.mkdir(parents=True, exist_ok=True) + # Write raw invalid UTF-8 bytes directly so read_text(encoding="utf-8") fails. + bad_file = ov_dir / "bad.yml" + bad_file.write_bytes(b"\xff\xfe invalid utf-8") + + source = ProjectOverlaySource(project_dir) + with pytest.raises(OverlayLoadError) as exc_info: + source.collect("wf") + assert exc_info.value.errors, "OverlayLoadError must carry a non-empty errors list" diff --git a/tests/workflows/test_overlay_merge.py b/tests/workflows/test_overlay_merge.py new file mode 100644 index 0000000000..b699b92c31 --- /dev/null +++ b/tests/workflows/test_overlay_merge.py @@ -0,0 +1,710 @@ +"""Tests for the workflow overlay merge engine.""" + +from __future__ import annotations + +import copy +from typing import Any + +import pytest + +from specify_cli.workflows.overlays.merge import ( + ComposedStep, + OverlayLayer, + find_step, + merge_steps, + validate_edits, +) +from specify_cli.workflows.overlays.schema import Overlay, OverlayEdit + + +def _step(id: str, **kwargs: Any) -> dict[str, Any]: # noqa: A002 + """Build a minimal step dict with the given id.""" + return {"id": id, "type": "command", "command": "speckit.specify", **kwargs} + + +def _layer(overlay: Overlay, source: str) -> OverlayLayer: + """Build an OverlayLayer for merge_steps.""" + return OverlayLayer(overlay, source) + + +class TestFindStep: + """Recursive anchor lookup across nested step lists.""" + + def test_find_step_flat(self): + steps = [_step("a"), _step("b"), _step("c")] + result = find_step(steps, "b") + assert result is not None + assert result[0] is steps + assert result[1] == 1 + + def test_find_step_missing(self): + steps = [_step("a"), _step("b")] + assert find_step(steps, "missing") is None + + def test_find_step_in_then(self): + steps = [ + { + "id": "if-1", + "type": "if", + "condition": "true", + "then": [_step("then-a")], + "else": [_step("else-b")], + }, + ] + result = find_step(steps, "then-a") + assert result is not None + assert result[0] is steps[0]["then"] + assert result[1] == 0 + + def test_find_step_in_else(self): + steps = [ + { + "id": "if-1", + "type": "if", + "condition": "true", + "then": [_step("then-a")], + "else": [_step("else-b")], + }, + ] + result = find_step(steps, "else-b") + assert result is not None + assert result[0] is steps[0]["else"] + assert result[1] == 0 + + def test_find_step_in_nested_steps(self): + steps = [ + { + "id": "while-1", + "type": "while", + "condition": "true", + "steps": [_step("inner-a"), _step("inner-b")], + }, + ] + result = find_step(steps, "inner-b") + assert result is not None + assert result[0] is steps[0]["steps"] + assert result[1] == 1 + + def test_find_step_in_switch_cases(self): + steps = [ + { + "id": "switch-1", + "type": "switch", + "expression": "{{ inputs.x }}", + "cases": { + "one": [_step("case-a")], + "two": [_step("case-b")], + }, + "default": [_step("default-c")], + }, + ] + assert find_step(steps, "case-b")[1] == 0 + assert find_step(steps, "default-c")[1] == 0 + + def test_find_step_not_in_fan_out_template(self): + steps = [ + { + "id": "fan-1", + "type": "fan-out", + "items": "{{ inputs.items }}", + "step": {"id": "template-x", "type": "command", "command": "echo"}, + }, + ] + assert find_step(steps, "template-x") is None + + +class TestMergeSteps: + """Composition of multiple overlays in merge order.""" + + def test_merge_steps_no_overlays(self): + base = [_step("a"), _step("b")] + steps, attribution = merge_steps(base, []) + assert [s["id"] for s in steps] == ["a", "b"] + assert attribution == [ComposedStep("a", "base"), ComposedStep("b", "base")] + + def test_merge_steps_single_overlay(self): + base = [_step("a"), _step("b")] + overlay = Overlay( + id="ov1", + extends="wf", + priority=10, + edits=[OverlayEdit("insert_after", "a", _step("new"))], + ) + steps, attribution = merge_steps(base, [_layer(overlay, "project:ov1")]) + assert [s["id"] for s in steps] == ["a", "new", "b"] + assert attribution == [ + ComposedStep("a", "base"), + ComposedStep("new", "project:ov1"), + ComposedStep("b", "base"), + ] + + def test_merge_steps_higher_priority_wins(self): + base = [_step("a")] + low = Overlay( + id="low", + extends="wf", + priority=5, + edits=[OverlayEdit("insert_after", "a", _step("low-step"))], + ) + high = Overlay( + id="high", + extends="wf", + priority=10, + edits=[OverlayEdit("insert_after", "a", _step("high-step"))], + ) + steps, attribution = merge_steps(base, [_layer(low, "project:low"), _layer(high, "project:high")]) + # low applied first, then high; both insert after 'a', so high-step ends + # closer to the anchor (higher priority wins the conflict). + assert [s["id"] for s in steps] == ["a", "high-step", "low-step"] + assert attribution == [ + ComposedStep("a", "base"), + ComposedStep("high-step", "project:high"), + ComposedStep("low-step", "project:low"), + ] + + def test_merge_steps_replace_wins_over_insert(self): + """Overlays apply to the original tree only; targeting an overlay-introduced step raises.""" + base = [_step("a")] + insert = Overlay( + id="insert", + extends="wf", + priority=5, + edits=[OverlayEdit("insert_after", "a", _step("inserted"))], + ) + replace = Overlay( + id="replace", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "inserted", _step("replaced"))], + ) + # "inserted" is not a base step — overlays cannot target each other's steps. + with pytest.raises(ValueError, match="Anchor 'inserted' not found"): + merge_steps(base, [_layer(insert, "project:insert"), _layer(replace, "project:replace")]) + + def test_merge_steps_does_not_mutate_base(self): + base = [_step("a")] + overlay = Overlay( + id="ov1", + extends="wf", + priority=10, + edits=[OverlayEdit("insert_after", "a", _step("new"))], + ) + original = copy.deepcopy(base) + merge_steps(base, [_layer(overlay, "project:ov1")]) + assert base == original + + def test_merge_steps_attribution_uses_source_not_overlay_id(self): + base = [_step("a")] + overlay = Overlay( + id="same-id", + extends="wf", + priority=10, + edits=[OverlayEdit("insert_after", "a", _step("new"))], + ) + steps, attribution = merge_steps(base, [_layer(overlay, "installed:same-id")]) + assert [s["id"] for s in steps] == ["a", "new"] + assert attribution == [ + ComposedStep("a", "base"), + ComposedStep("new", "installed:same-id"), + ] + + def test_merge_steps_nested_base_attribution(self): + base = [ + { + "id": "if-1", + "type": "if", + "condition": "true", + "then": [_step("then-a")], + "else": [_step("else-b")], + }, + ] + steps, attribution = merge_steps(base, []) + assert attribution == [ + ComposedStep("if-1", "base"), + ComposedStep("then-a", "base"), + ComposedStep("else-b", "base"), + ] + + def test_merge_steps_higher_replace_wins_lower_replace_same_anchor(self): + base = [_step("implement")] + low = Overlay( + id="low", + extends="wf", + priority=5, + edits=[OverlayEdit("replace", "implement", _step("low-implement"))], + ) + high = Overlay( + id="high", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "implement", _step("high-implement"))], + ) + steps, attribution = merge_steps(base, [_layer(low, "project:low"), _layer(high, "project:high")]) + assert [s["id"] for s in steps] == ["high-implement"] + assert any( + composed.step_id == "high-implement" and composed.source == "project:high" + for composed in attribution + ) + + def test_merge_steps_higher_replace_wins_after_lower_remove_same_anchor(self): + base = [_step("implement")] + low = Overlay( + id="low", + extends="wf", + priority=5, + edits=[OverlayEdit("remove", "implement")], + ) + high = Overlay( + id="high", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "implement", _step("high-implement"))], + ) + steps, attribution = merge_steps(base, [_layer(low, "project:low"), _layer(high, "project:high")]) + assert [s["id"] for s in steps] == ["high-implement"] + + def test_merge_steps_higher_insert_wins_after_lower_remove_same_anchor(self): + base = [_step("implement")] + low = Overlay( + id="low", + extends="wf", + priority=5, + edits=[OverlayEdit("remove", "implement")], + ) + high = Overlay( + id="high", + extends="wf", + priority=10, + edits=[OverlayEdit("insert_after", "implement", _step("high-after"))], + ) + steps, attribution = merge_steps(base, [_layer(low, "project:low"), _layer(high, "project:high")]) + assert [s["id"] for s in steps] == ["implement", "high-after"] + assert attribution == [ + ComposedStep("implement", "base"), + ComposedStep("high-after", "project:high"), + ] + + def test_merge_steps_later_overlay_wins_tie_same_anchor(self): + """When two overlays have the same priority, the one applied later wins.""" + base = [_step("a")] + first = Overlay( + id="first", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "a", _step("first-replace"))], + ) + second = Overlay( + id="second", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "a", _step("second-replace"))], + ) + # Merge order: first applied, then second wins tie. + steps, attribution = merge_steps( + base, + [ + _layer(first, "overlay:first"), + _layer(second, "overlay:second"), + ], + ) + assert [s["id"] for s in steps] == ["second-replace"] + assert any( + composed.step_id == "second-replace" and composed.source == "overlay:second" + for composed in attribution + ) + + def test_merge_steps_insert_after_then_replace_same_anchor_id_change(self): + """Inserts must be applied before the winning replace so the anchor still exists. + + Regression: when a replace changes the step ID, applying it before inserts + causes ``find_step`` to fail on the now-gone original anchor. + """ + base = [_step("build")] + low = Overlay( + id="low", + extends="wf", + priority=5, + edits=[OverlayEdit("insert_after", "build", _step("test"))], + ) + high = Overlay( + id="high", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "build", _step("compile"))], + ) + steps, attribution = merge_steps( + base, [_layer(low, "project:low"), _layer(high, "project:high")] + ) + # The insert should land after the original anchor position, then the + # anchor is replaced. Final order: ["compile", "test"]. + assert [s["id"] for s in steps] == ["compile", "test"] + assert attribution == [ + ComposedStep("compile", "project:high"), + ComposedStep("test", "project:low"), + ] + + def test_merge_steps_insert_before_then_replace_same_anchor_id_change(self): + """Same as above but with insert_before — anchor must still be findable.""" + base = [_step("build")] + low = Overlay( + id="low", + extends="wf", + priority=5, + edits=[OverlayEdit("insert_before", "build", _step("lint"))], + ) + high = Overlay( + id="high", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "build", _step("compile"))], + ) + steps, attribution = merge_steps( + base, [_layer(low, "project:low"), _layer(high, "project:high")] + ) + assert [s["id"] for s in steps] == ["lint", "compile"] + assert attribution == [ + ComposedStep("lint", "project:low"), + ComposedStep("compile", "project:high"), + ] + + def test_merge_steps_unknown_anchor_still_raises(self): + base = [_step("a")] + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "missing", _step("new"))], + ) + with pytest.raises(ValueError, match="Anchor 'missing' not found"): + merge_steps(base, [_layer(overlay, "project:ov")]) + + # ── composite step attribution ─────────────────────────────────────── + + def test_merge_insert_composite_if_attribution(self): + """Nested then/else children of an inserted 'if' step get the overlay source.""" + base = [_step("a")] + composite = { + "id": "if-1", + "type": "if", + "condition": "true", + "then": [_step("then-a")], + "else": [_step("else-b")], + } + overlay = Overlay( + id="ov", extends="wf", priority=10, + edits=[OverlayEdit("insert_after", "a", composite)], + ) + _steps, attribution = merge_steps( + base, [_layer(overlay, "project:ov")] + ) + assert attribution == [ + ComposedStep("a", "base"), + ComposedStep("if-1", "project:ov"), + ComposedStep("then-a", "project:ov"), + ComposedStep("else-b", "project:ov"), + ] + + def test_merge_insert_composite_switch_attribution(self): + """Nested cases/default children of an inserted 'switch' step get the overlay source.""" + base = [_step("a")] + composite = { + "id": "switch-1", + "type": "switch", + "expression": "{{inputs.x}}", + "cases": {"one": [_step("case-one")], "two": [_step("case-two")]}, + "default": [_step("default-z")], + } + overlay = Overlay( + id="ov", extends="wf", priority=10, + edits=[OverlayEdit("insert_before", "a", composite)], + ) + _steps, attribution = merge_steps( + base, [_layer(overlay, "project:ov")] + ) + assert attribution == [ + ComposedStep("switch-1", "project:ov"), + ComposedStep("default-z", "project:ov"), + ComposedStep("case-one", "project:ov"), + ComposedStep("case-two", "project:ov"), + ComposedStep("a", "base"), + ] + + def test_merge_replace_flat_with_composite_attribution(self): + """Replacing a flat step with a composite step attributes all nested children.""" + base = [_step("a")] + composite = { + "id": "if-1", + "type": "if", + "condition": "true", + "then": [_step("inner-x"), _step("inner-y")], + } + overlay = Overlay( + id="ov", extends="wf", priority=10, + edits=[OverlayEdit("replace", "a", composite)], + ) + _steps, attribution = merge_steps( + base, [_layer(overlay, "project:ov")] + ) + assert attribution == [ + ComposedStep("if-1", "project:ov"), + ComposedStep("inner-x", "project:ov"), + ComposedStep("inner-y", "project:ov"), + ] + + def test_merge_remove_composite_step_cleans_nested_sources(self): + """Removing a composite step also cleans its nested children from sources.""" + base = [ + { + "id": "if-1", + "type": "if", + "condition": "true", + "then": [_step("then-a")], + "else": [_step("else-b")], + }, + _step("a"), + ] + overlay = Overlay( + id="ov", extends="wf", priority=10, + edits=[OverlayEdit("remove", "if-1")], + ) + steps, attribution = merge_steps( + base, [_layer(overlay, "project:ov")] + ) + assert [s["id"] for s in steps] == ["a"] + assert attribution == [ComposedStep("a", "base")] + + def test_merge_insert_deeply_nested_composite_attribution(self): + """Deep nesting (if inside while) gets the overlay source at every level.""" + base = [_step("a")] + inner_if = { + "id": "inner-if", + "type": "if", + "condition": "true", + "then": [_step("deep-x")], + } + composite = { + "id": "while-1", + "type": "while", + "condition": "true", + "steps": [inner_if], + } + overlay = Overlay( + id="ov", extends="wf", priority=10, + edits=[OverlayEdit("insert_after", "a", composite)], + ) + _steps, attribution = merge_steps( + base, [_layer(overlay, "project:ov")] + ) + assert attribution == [ + ComposedStep("a", "base"), + ComposedStep("while-1", "project:ov"), + ComposedStep("inner-if", "project:ov"), + ComposedStep("deep-x", "project:ov"), + ] + + +class TestValidateEdits: + """Edit validation against known base step IDs.""" + + def test_valid_edits(self): + edits = [ + OverlayEdit("insert_after", "a", _step("new")), + OverlayEdit("remove", "b"), + ] + assert validate_edits(edits, {"a", "b"}) == [] + + def test_invalid_anchor(self): + edits = [OverlayEdit("insert_after", "missing", _step("new"))] + errors = validate_edits(edits, {"a"}) + assert any("missing" in e for e in errors) + + def test_step_id_contains_colon(self): + edits = [OverlayEdit("insert_after", "a", _step("bad:id"))] + errors = validate_edits(edits, {"a"}) + assert any("':'" in e for e in errors) + + def test_remove_requires_no_step(self): + edits = [OverlayEdit("remove", "a", _step("extra"))] + errors = validate_edits(edits, {"a"}) + assert len(errors) > 0 + + +class TestMergeStepsAncestorConflicts: + """merge_steps raises when two targeted anchors are in a parent/descendant relationship.""" + + def _if_step(self, parent_id: str, child_id: str) -> dict[str, Any]: + return { + "id": parent_id, + "type": "if", + "condition": "true", + "then": [_step(child_id)], + } + + def test_remove_parent_and_insert_after_child_raises(self): + """Removing a parent while inserting after its nested child is an anchor conflict.""" + parent_id = "if-step" + child_id = "then-child" + base = [self._if_step(parent_id, child_id)] + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[ + OverlayEdit("remove", parent_id), + OverlayEdit("insert_after", child_id, _step("new-step")), + ], + ) + with pytest.raises(ValueError, match="ancestor"): + merge_steps(base, [_layer(overlay, "project:ov")]) + + def test_replace_parent_and_remove_child_raises(self): + """Replacing a parent while also removing a nested child is an anchor conflict.""" + parent_id = "if-step" + child_id = "then-child" + base = [self._if_step(parent_id, child_id)] + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[ + OverlayEdit("replace", parent_id, _step("new-parent")), + OverlayEdit("remove", child_id), + ], + ) + with pytest.raises(ValueError, match="ancestor"): + merge_steps(base, [_layer(overlay, "project:ov")]) + + def test_conflict_across_multiple_overlays_raises(self): + """Conflict is detected even when conflicting anchors come from different overlays.""" + parent_id = "if-step" + child_id = "then-child" + base = [self._if_step(parent_id, child_id)] + overlay_a = Overlay( + id="ov-a", + extends="wf", + priority=5, + edits=[OverlayEdit("remove", parent_id)], + ) + overlay_b = Overlay( + id="ov-b", + extends="wf", + priority=10, + edits=[OverlayEdit("insert_after", child_id, _step("new-step"))], + ) + with pytest.raises(ValueError, match="ancestor"): + merge_steps( + base, + [_layer(overlay_a, "project:ov-a"), _layer(overlay_b, "project:ov-b")], + ) + + def test_sibling_anchors_not_conflicting(self): + """Anchors in sibling branches (not ancestor/descendant) are allowed.""" + base = [ + { + "id": "if-step", + "type": "if", + "condition": "true", + "then": [_step("then-child")], + "else": [_step("else-child")], + } + ] + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[ + OverlayEdit("insert_after", "then-child", _step("after-then")), + OverlayEdit("insert_after", "else-child", _step("after-else")), + ], + ) + # Should not raise — the two anchors are siblings, not ancestor/descendant. + steps, _ = merge_steps(base, [_layer(overlay, "project:ov")]) + step_ids = [s.get("id") for s in steps[0]["then"]] + [s.get("id") for s in steps[0]["else"]] + assert "after-then" in step_ids + assert "after-else" in step_ids + + def test_single_anchor_not_conflicting(self): + """A single anchor is never in conflict with itself.""" + parent_id = "if-step" + child_id = "then-child" + base = [self._if_step(parent_id, child_id)] + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[OverlayEdit("remove", parent_id)], + ) + steps, _ = merge_steps(base, [_layer(overlay, "project:ov")]) + assert steps == [] + + def test_child_not_targeted_no_conflict(self): + """Targeting a parent alone (child not in any edit) is allowed.""" + parent_id = "if-step" + child_id = "then-child" + base = [self._if_step(parent_id, child_id), _step("other")] + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[ + OverlayEdit("remove", parent_id), + OverlayEdit("insert_after", "other", _step("new-step")), + ], + ) + # "other" is not inside "if-step", so no ancestor conflict. + steps, _ = merge_steps(base, [_layer(overlay, "project:ov")]) + assert [s["id"] for s in steps] == ["other", "new-step"] + + def test_insert_only_on_ancestor_and_descendant_not_conflicting(self): + """insert_after on both a parent and its nested child is valid and order-independent.""" + parent_id = "if-step" + child_id = "then-child" + base = [self._if_step(parent_id, child_id)] + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[ + OverlayEdit("insert_after", parent_id, _step("after-parent")), + OverlayEdit("insert_after", child_id, _step("after-child")), + ], + ) + # Should not raise — inserts leave the ancestor intact. + steps, attribution = merge_steps(base, [_layer(overlay, "project:ov")]) + # "after-parent" is inserted at the top level after the if-step. + assert [s["id"] for s in steps] == [parent_id, "after-parent"] + # "after-child" is inserted inside the then list. + then_ids = [s["id"] for s in steps[0]["then"]] + assert then_ids == [child_id, "after-child"] + + +class TestMergeStepsIdCollision: + """merge_steps is deterministic when a replacement reuses a base step ID.""" + + def test_replace_with_reused_id_does_not_affect_original(self): + """Replacing A with new_step(id=B) must not interfere with editing original B. + + Before the fix, the remove-B anchor group would find the replacement step + (which now has id='b') instead of the original 'b' step, producing a + different result depending on dict iteration order. + """ + base = [_step("a"), _step("b"), _step("c")] + overlay = Overlay( + id="ov", + extends="wf", + priority=10, + edits=[ + # Replace "a" with a new step that reuses id "b". + OverlayEdit("replace", "a", {**_step("b"), "command": "speckit.replaced"}), + # Remove the original "b". + OverlayEdit("remove", "b"), + ], + ) + steps, _ = merge_steps(base, [_layer(overlay, "project:ov")]) + # The original "b" is removed; the replacement (also id="b") survives. + # "c" is untouched. + assert len(steps) == 2 + remaining_ids = [s["id"] for s in steps] + assert remaining_ids == ["b", "c"] + # The surviving "b" step is the replacement (has the custom command). + assert steps[0]["command"] == "speckit.replaced" diff --git a/tests/workflows/test_overlay_schema.py b/tests/workflows/test_overlay_schema.py new file mode 100644 index 0000000000..026797b0d8 --- /dev/null +++ b/tests/workflows/test_overlay_schema.py @@ -0,0 +1,244 @@ +"""Tests for overlay YAML schema normalization, especially shorthand edits.""" + +from __future__ import annotations + +import pytest + +from specify_cli.workflows.overlays.schema import ( + OverlayEdit, + validate_overlay_yaml, +) + + +class TestShorthandEdits: + """Requirements-compliant shorthand edit format.""" + + def test_shorthand_insert_after(self): + overlay, errors = validate_overlay_yaml( + { + "id": "lint", + "extends": "wf", + "priority": 10, + "edits": [ + { + "insert_after": "implement", + "step": {"id": "lint", "type": "shell", "command": "npm run lint"}, + } + ], + } + ) + assert not errors, errors + assert overlay is not None + assert overlay.edits == [ + OverlayEdit("insert_after", "implement", {"id": "lint", "type": "shell", "command": "npm run lint"}) + ] + + def test_shorthand_insert_before(self): + overlay, errors = validate_overlay_yaml( + { + "id": "ov", + "extends": "wf", + "priority": 10, + "edits": [ + { + "insert_before": "a", + "step": {"id": "b", "type": "command", "command": "echo"}, + } + ], + } + ) + assert not errors, errors + assert overlay is not None + assert overlay.edits == [ + OverlayEdit("insert_before", "a", {"id": "b", "type": "command", "command": "echo"}) + ] + + def test_shorthand_replace(self): + overlay, errors = validate_overlay_yaml( + { + "id": "ov", + "extends": "wf", + "priority": 10, + "edits": [ + { + "replace": "a", + "step": {"id": "a", "type": "command", "command": "echo"}, + } + ], + } + ) + assert not errors, errors + assert overlay is not None + assert overlay.edits == [ + OverlayEdit("replace", "a", {"id": "a", "type": "command", "command": "echo"}) + ] + + def test_shorthand_remove(self): + overlay, errors = validate_overlay_yaml( + { + "id": "ov", + "extends": "wf", + "priority": 10, + "edits": [{"remove": "a"}], + } + ) + assert not errors, errors + assert overlay is not None + assert overlay.edits == [OverlayEdit("remove", "a")] + + def test_explicit_operation_format_still_valid(self): + overlay, errors = validate_overlay_yaml( + { + "id": "ov", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "b", "type": "command", "command": "echo"}, + } + ], + } + ) + assert not errors, errors + assert overlay is not None + assert overlay.edits == [ + OverlayEdit("insert_after", "a", {"id": "b", "type": "command", "command": "echo"}) + ] + + def test_multiple_operation_fields_rejected(self): + overlay, errors = validate_overlay_yaml( + { + "id": "ov", + "extends": "wf", + "priority": 10, + "edits": [ + { + "insert_after": "a", + "remove": "a", + } + ], + } + ) + assert overlay is None + assert any("multiple" in e.lower() for e in errors), errors + + def test_invalid_operation_field_rejected(self): + overlay, errors = validate_overlay_yaml( + { + "id": "ov", + "extends": "wf", + "priority": 10, + "edits": [{"destroy": "a"}], + } + ) + assert overlay is None + assert any("operation" in e.lower() for e in errors), errors + + def test_shorthand_and_explicit_mixed_list(self): + overlay, errors = validate_overlay_yaml( + { + "id": "ov", + "extends": "wf", + "priority": 10, + "edits": [ + {"insert_after": "a", "step": {"id": "b", "type": "command", "command": "echo"}}, + { + "operation": "remove", + "anchor": "c", + }, + ], + } + ) + assert not errors, errors + assert overlay is not None + assert overlay.edits == [ + OverlayEdit("insert_after", "a", {"id": "b", "type": "command", "command": "echo"}), + OverlayEdit("remove", "c"), + ] + + def test_shorthand_remove_must_not_include_step(self): + overlay, errors = validate_overlay_yaml( + { + "id": "ov", + "extends": "wf", + "priority": 10, + "edits": [ + { + "remove": "a", + "step": {"id": "b", "type": "command", "command": "echo"}, + } + ], + } + ) + assert overlay is None + assert any("remove" in e.lower() and "step" in e.lower() for e in errors), errors + + +class TestOverlayIdValidation: + """Overlay and workflow IDs must be safe path segments.""" + + @pytest.mark.parametrize("overlay_id", ["../ov", "a/b", "a\\\\b", ".", "..", ""]) + def test_invalid_overlay_id_rejected(self, overlay_id): + overlay, errors = validate_overlay_yaml( + { + "id": overlay_id, + "extends": "wf", + "priority": 10, + "edits": [{"remove": "a"}], + } + ) + assert overlay is None + assert any("id" in e.lower() for e in errors), errors + + @pytest.mark.parametrize("extends", ["../wf", "a/b", "a\\\\b", ".", "..", ""]) + def test_invalid_extends_rejected(self, extends): + overlay, errors = validate_overlay_yaml( + { + "id": "ov", + "extends": extends, + "priority": 10, + "edits": [{"remove": "a"}], + } + ) + assert overlay is None + assert any("extends" in e.lower() for e in errors), errors + + def test_valid_dashed_id_accepted(self): + overlay, errors = validate_overlay_yaml( + { + "id": "my-overlay", + "extends": "my-workflow", + "priority": 10, + "edits": [{"remove": "a"}], + } + ) + assert not errors, errors + assert overlay is not None + + def test_validate_safe_id_rejects_trailing_newline(self): + """A trailing newline must not pass ID validation (fullmatch guard).""" + overlay, errors = validate_overlay_yaml( + { + "id": "overlay\n", + "extends": "wf", + "priority": 10, + "edits": [{"remove": "a"}], + } + ) + assert overlay is None + assert any("id" in e.lower() for e in errors), errors + + def test_validate_safe_id_rejects_embedded_newline(self): + """An embedded newline must not pass ID validation.""" + overlay, errors = validate_overlay_yaml( + { + "id": "ov\nerlay", + "extends": "wf", + "priority": 10, + "edits": [{"remove": "a"}], + } + ) + assert overlay is None + assert any("id" in e.lower() for e in errors), errors diff --git a/tests/workflows/test_overlay_security.py b/tests/workflows/test_overlay_security.py new file mode 100644 index 0000000000..505b438d2b --- /dev/null +++ b/tests/workflows/test_overlay_security.py @@ -0,0 +1,318 @@ +"""Security tests for workflow overlay path handling.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml +from typer.testing import CliRunner + +from specify_cli import app + + +runner = CliRunner() + + +@pytest.fixture +def project_dir(tmp_path): + """Create a mock spec-kit project with ``.specify/workflows/`` directory.""" + workflows_dir = tmp_path / ".specify" / "workflows" + workflows_dir.mkdir(parents=True, exist_ok=True) + return tmp_path + + +def _write_workflow(project_root: Path, workflow_id: str, data: dict) -> Path: + wf_dir = project_root / ".specify" / "workflows" / workflow_id + wf_dir.mkdir(parents=True, exist_ok=True) + wf_path = wf_dir / "workflow.yml" + wf_path.write_text(yaml.safe_dump(data), encoding="utf-8") + return wf_path + + +def _write_overlay(project_root: Path, workflow_id: str, overlay_id: str, data: dict) -> Path: + ov_dir = project_root / ".specify" / "workflows" / "overlays" / workflow_id + ov_dir.mkdir(parents=True, exist_ok=True) + ov_path = ov_dir / f"{overlay_id}.yml" + ov_path.write_text(yaml.safe_dump(data), encoding="utf-8") + return ov_path + + +class TestOverlayPathTraversal: + """Overlay CLI must stay inside the overlay directory.""" + + def test_overlay_add_rejects_traversal_in_workflow_id(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + overlay_file = project_dir / "overlay.yml" + overlay_file.write_text( + yaml.safe_dump( + { + "id": "ov1", + "extends": "../wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + + result = runner.invoke( + app, ["workflow", "overlay", "add", str(overlay_file), "--priority", "5"] + ) + assert result.exit_code != 0, result.output + assert "invalid" in result.output.lower() or "traversal" in result.output.lower() + + def test_overlay_add_rejects_traversal_in_overlay_id(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + overlay_file = project_dir / "overlay.yml" + overlay_file.write_text( + yaml.safe_dump( + { + "id": "../../ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + + result = runner.invoke( + app, ["workflow", "overlay", "add", str(overlay_file), "--priority", "5"] + ) + assert result.exit_code != 0, result.output + assert "invalid" in result.output.lower() or "traversal" in result.output.lower() + + def test_overlay_remove_cannot_escape_overlays_dir(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + # Create a base workflow file that would be the traversal target. + target = project_dir / ".specify" / "workflows" / "wf" / "workflow.yml" + assert target.is_file() + + result = runner.invoke( + app, ["workflow", "overlay", "remove", "wf", "../wf/workflow"] + ) + assert result.exit_code != 0, result.output + assert target.is_file() + assert "Invalid" in result.output or "traversal" in result.output.lower() + + def test_overlay_remove_rejects_symlink(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + overlay_dir = project_dir / ".specify" / "workflows" / "overlays" / "wf" + real_file = overlay_dir / "ov1.yml" + symlink_file = overlay_dir / "symlink.yml" + symlink_file.symlink_to(real_file) + + result = runner.invoke(app, ["workflow", "overlay", "remove", "wf", "symlink"]) + assert result.exit_code != 0, result.output + assert real_file.is_file() + assert "symlink" in result.output.lower() or "Invalid" in result.output + + def test_overlay_add_rejects_symlinked_target_file(self, project_dir, monkeypatch): + """overlay add must not overwrite through a symlinked overlay file target.""" + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + + overlay_dir = project_dir / ".specify" / "workflows" / "overlays" / "wf" + overlay_dir.mkdir(parents=True, exist_ok=True) + real_file = overlay_dir / "other.yml" + real_file.write_text("sentinel\n", encoding="utf-8") + (overlay_dir / "ov1.yml").symlink_to(real_file) + + overlay_file = project_dir / "overlay.yml" + overlay_file.write_text( + yaml.safe_dump( + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + + result = runner.invoke(app, ["workflow", "overlay", "add", str(overlay_file)]) + + assert result.exit_code != 0, result.output + assert "symlinked path" in result.output.lower() + assert real_file.read_text(encoding="utf-8") == "sentinel\n" + + def test_overlay_operations_reject_overlays_as_workflow_id(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + result = runner.invoke(app, ["workflow", "overlay", "list", "overlays"]) + assert result.exit_code != 0, result.output + assert "Invalid" in result.output or "reserved" in result.output.lower() + + def test_overlay_set_priority_rejects_traversal(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + result = runner.invoke( + app, ["workflow", "overlay", "set-priority", "wf", "../other", "10"] + ) + assert result.exit_code != 0, result.output + assert "invalid" in result.output.lower() or "traversal" in result.output.lower() + + def test_overlay_enable_rejects_traversal(self, project_dir, monkeypatch): + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + result = runner.invoke(app, ["workflow", "overlay", "enable", "wf", "../other"]) + assert result.exit_code != 0, result.output + assert "invalid" in result.output.lower() or "traversal" in result.output.lower() + + def test_overlay_rejects_symlinked_overlays_dir(self, project_dir, monkeypatch, tmp_path): + """Overlay commands must reject a symlinked .specify/workflows/overlays directory.""" + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + + # Create a symlinked overlays directory pointing outside the project + outside_dir = tmp_path / "outside" + outside_dir.mkdir() + overlays_dir = project_dir / ".specify" / "workflows" / "overlays" + overlays_dir.symlink_to(outside_dir) + + result = runner.invoke(app, ["workflow", "overlay", "list", "wf"]) + assert result.exit_code != 0, result.output + assert "symlink" in result.output.lower() + + def test_overlay_list_rejects_symlinked_per_workflow_dir(self, project_dir, monkeypatch, tmp_path): + """Overlay list must reject a symlinked per-workflow overlay directory.""" + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + + # Create a real overlay directory outside the project. + outside_dir = tmp_path / "outside_wf" + outside_dir.mkdir() + outside_dir.joinpath("evil.yml").write_text( + yaml.safe_dump( + { + "id": "evil", + "extends": "wf", + "priority": 100, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "evil-step", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + + # Symlink the per-workflow overlay directory to the outside location. + overlays_root = project_dir / ".specify" / "workflows" / "overlays" + overlays_root.mkdir(parents=True, exist_ok=True) + symlink_dir = overlays_root / "wf" + symlink_dir.symlink_to(outside_dir) + + result = runner.invoke(app, ["workflow", "overlay", "list", "wf"]) + assert result.exit_code != 0, result.output + assert "symlink" in result.output.lower() + + def test_overlay_list_reports_invalid_yaml_cleanly(self, project_dir, monkeypatch): + """Overlay list should surface malformed overlay YAML as a clean user error.""" + monkeypatch.setattr("specify_cli._require_specify_project", lambda: project_dir) + _write_workflow( + project_dir, + "wf", + { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "echo"}], + }, + ) + overlay_dir = project_dir / ".specify" / "workflows" / "overlays" / "wf" + overlay_dir.mkdir(parents=True, exist_ok=True) + (overlay_dir / "broken.yml").write_text("id: broken\nextends: wf\npriority: [\n", encoding="utf-8") + + result = runner.invoke(app, ["workflow", "overlay", "list", "wf"]) + + assert result.exit_code != 0, result.output + assert "Invalid YAML" in result.output diff --git a/tests/workflows/test_resolver_integration.py b/tests/workflows/test_resolver_integration.py new file mode 100644 index 0000000000..afd3461bd4 --- /dev/null +++ b/tests/workflows/test_resolver_integration.py @@ -0,0 +1,461 @@ +"""Integration tests for the WorkflowResolver.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + +from specify_cli.workflows.engine import WorkflowDefinition +from specify_cli.workflows.overlays import WorkflowResolver +from specify_cli.workflows.overlays.merge import ComposedStep + + +def _write_workflow(project_root: Path, workflow_id: str, data: dict) -> Path: + wf_dir = project_root / ".specify" / "workflows" / workflow_id + wf_dir.mkdir(parents=True, exist_ok=True) + wf_path = wf_dir / "workflow.yml" + wf_path.write_text(yaml.safe_dump(data), encoding="utf-8") + return wf_path + + +def _write_overlay(project_root: Path, workflow_id: str, overlay_id: str, data: dict) -> Path: + ov_dir = project_root / ".specify" / "workflows" / "overlays" / workflow_id + ov_dir.mkdir(parents=True, exist_ok=True) + ov_path = ov_dir / f"{overlay_id}.yml" + ov_path.write_text(yaml.safe_dump(data), encoding="utf-8") + return ov_path + + +class TestWorkflowResolver: + """End-to-end resolution of base workflows plus overlays.""" + + def test_resolve_without_overlays(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + + resolver = WorkflowResolver(project_dir) + definition = resolver.resolve("wf") + assert isinstance(definition, WorkflowDefinition) + assert definition.id == "wf" + assert [s["id"] for s in definition.steps] == ["a"] + + def test_resolve_with_project_overlay_insert(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [ + {"id": "a", "type": "command", "command": "speckit.specify"}, + {"id": "b", "type": "command", "command": "speckit.specify"}, + ], + } + _write_workflow(project_dir, "wf", data) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "speckit.plan"}, + } + ], + }, + ) + + resolver = WorkflowResolver(project_dir) + definition = resolver.resolve("wf") + assert [s["id"] for s in definition.steps] == ["a", "new", "b"] + + def test_resolve_higher_priority_wins(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + _write_overlay( + project_dir, + "wf", + "low", + { + "id": "low", + "extends": "wf", + "priority": 5, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "low-step", "type": "command", "command": "echo"}, + } + ], + }, + ) + _write_overlay( + project_dir, + "wf", + "high", + { + "id": "high", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "high-step", "type": "command", "command": "echo"}, + } + ], + }, + ) + + resolver = WorkflowResolver(project_dir) + definition = resolver.resolve("wf") + # Higher priority is applied later; both insert_after 'a', so high-step + # ends up closer to the anchor and wins the conflict. + assert [s["id"] for s in definition.steps] == ["a", "high-step", "low-step"] + + def test_resolve_with_layers_returns_attribution(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + resolver = WorkflowResolver(project_dir) + definition, layers, attribution = resolver.resolve_with_layers("wf") + assert [s["id"] for s in definition.steps] == ["a", "new"] + assert any(layer.tier == "base" for layer in layers) + assert attribution == [ComposedStep("a", "base"), ComposedStep("new", "project:ov1")] + + def test_resolve_attribution_for_nested_base_steps(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [ + { + "id": "if-1", + "type": "if", + "condition": "true", + "then": [{"id": "then-a", "type": "command", "command": "echo"}], + "else": [{"id": "else-b", "type": "command", "command": "echo"}], + } + ], + } + _write_workflow(project_dir, "wf", data) + + resolver = WorkflowResolver(project_dir) + definition, _layers, attribution = resolver.resolve_with_layers("wf") + assert [s["id"] for s in definition.steps] == ["if-1"] + sources = {c.step_id: c.source for c in attribution} + assert sources["if-1"] == "base" + assert sources["then-a"] == "base" + assert sources["else-b"] == "base" + + def test_resolve_invalid_project_overlay_fails(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + _write_overlay( + project_dir, + "wf", + "broken", + { + "id": "broken", + "extends": "wf", + "priority": 10, + "edits": "not-a-list", + }, + ) + + resolver = WorkflowResolver(project_dir) + with pytest.raises(ValueError): + resolver.resolve("wf") + + def test_resolve_disabled_overlay_is_skipped(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + _write_overlay( + project_dir, + "wf", + "disabled", + { + "id": "disabled", + "extends": "wf", + "priority": 10, + "enabled": False, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + resolver = WorkflowResolver(project_dir) + definition = resolver.resolve("wf") + assert [s["id"] for s in definition.steps] == ["a"] + + def test_resolve_invalid_anchor_raises(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "missing", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + resolver = WorkflowResolver(project_dir) + with pytest.raises(ValueError, match="anchor 'missing' does not match any base step id"): + resolver.resolve("wf") + + def test_resolve_missing_workflow(self, project_dir): + resolver = WorkflowResolver(project_dir) + with pytest.raises(FileNotFoundError, match="Workflow not found"): + resolver.resolve("missing") + + def test_resolve_validates_composed_result(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "replace", + "anchor": "a", + "step": {"id": "a", "type": "invalid-type", "command": "echo"}, + } + ], + }, + ) + + resolver = WorkflowResolver(project_dir) + with pytest.raises(ValueError, match="Composed workflow is invalid"): + resolver.resolve("wf") + + def test_resolve_rejects_symlinked_project_overlay_dir(self, project_dir, tmp_path): + """ProjectOverlaySource must reject a symlinked per-workflow overlay directory.""" + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + + # Create a real overlay directory outside the project with a malicious overlay. + outside_dir = tmp_path / "outside_overlays" / "wf" + outside_dir.mkdir(parents=True, exist_ok=True) + outside_dir.joinpath("evil.yml").write_text( + yaml.safe_dump( + { + "id": "evil", + "extends": "wf", + "priority": 100, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "evil-step", "type": "command", "command": "rm -rf /"}, + } + ], + } + ), + encoding="utf-8", + ) + + # Symlink the per-workflow overlay directory to the outside location. + overlays_root = project_dir / ".specify" / "workflows" / "overlays" + overlays_root.mkdir(parents=True, exist_ok=True) + symlink_dir = overlays_root / "wf" + symlink_dir.symlink_to(outside_dir) + + resolver = WorkflowResolver(project_dir) + with pytest.raises(ValueError, match="Symlinked overlay directories are not allowed"): + resolver.resolve("wf") + + def test_resolve_rejects_symlinked_overlay_root(self, project_dir, tmp_path): + """ProjectOverlaySource must reject a symlinked overlay root too.""" + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + + outside_root = tmp_path / "outside-overlays-root" + outside_root.mkdir(parents=True, exist_ok=True) + workflow_dir = outside_root / "wf" + workflow_dir.mkdir() + workflow_dir.joinpath("evil.yml").write_text( + yaml.safe_dump( + { + "id": "evil", + "extends": "wf", + "priority": 100, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "evil-step", "type": "command", "command": "rm -rf /"}, + } + ], + } + ), + encoding="utf-8", + ) + + overlays_root = project_dir / ".specify" / "workflows" / "overlays" + overlays_root.symlink_to(outside_root, target_is_directory=True) + + resolver = WorkflowResolver(project_dir) + with pytest.raises(ValueError, match="Symlinked overlay directories are not allowed"): + resolver.resolve("wf") + + def test_resolve_reports_invalid_overlay_yaml_cleanly(self, project_dir): + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + overlay_dir = project_dir / ".specify" / "workflows" / "overlays" / "wf" + overlay_dir.mkdir(parents=True, exist_ok=True) + (overlay_dir / "broken.yml").write_text("id: broken\nextends: wf\npriority: [\n", encoding="utf-8") + + resolver = WorkflowResolver(project_dir) + with pytest.raises(ValueError, match="Invalid YAML"): + resolver.resolve("wf") + + def test_resolve_attribution_for_inserted_composite_step(self, project_dir): + """Inserted composite steps must attribute nested children to the overlay source.""" + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": { + "id": "if-1", + "type": "if", + "condition": "true", + "then": [{"id": "then-x", "type": "command", "command": "echo"}], + "else": [{"id": "else-y", "type": "command", "command": "echo"}], + }, + } + ], + }, + ) + + resolver = WorkflowResolver(project_dir) + _definition, _layers, attribution = resolver.resolve_with_layers("wf") + sources = {c.step_id: c.source for c in attribution} + assert sources["a"] == "base" + assert sources["if-1"] == "project:ov1" + assert sources["then-x"] == "project:ov1" + assert sources["else-y"] == "project:ov1" + + def test_engine_load_workflow_uses_resolver(self, project_dir): + from specify_cli.workflows.engine import WorkflowEngine + + data = { + "schema_version": "1.0", + "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"}, + "steps": [{"id": "a", "type": "command", "command": "speckit.specify"}], + } + _write_workflow(project_dir, "wf", data) + _write_overlay( + project_dir, + "wf", + "ov1", + { + "id": "ov1", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "new", "type": "command", "command": "echo"}, + } + ], + }, + ) + + engine = WorkflowEngine(project_dir) + definition = engine.load_workflow("wf") + assert [s["id"] for s in definition.steps] == ["a", "new"]