diff --git a/docs/reference/workflows.md b/docs/reference/workflows.md index 16bbe0893..37e3de791 100644 --- a/docs/reference/workflows.md +++ b/docs/reference/workflows.md @@ -86,7 +86,195 @@ Lists workflows installed in the current project. specify workflow add ``` -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`. Local workflow directories may include an optional `overlays/` subdirectory, which is installed alongside the workflow as shipped overlays. + +## 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); at the same priority, installed overlays are applied before project overlays, so project overlays win ties. + +Overlay files live in two locations: + +| Location | Purpose | Tier | +| --- | --- | --- | +| `.specify/workflows//overlays/*.yml` | Shipped with the installed workflow | `installed-overlay` | +| `.specify/workflows/overlays//*.yml` | Project-local customizations | `project-overlay` | + +Project overlays always take precedence over installed overlays at the same priority. + +### 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. Installed overlays shipped with a workflow are removed by `specify workflow remove `, which deletes the entire workflow directory. + +#### 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 ` copies `workflow.yml` and an optional `overlays/` subdirectory into `.specify/workflows//`. Overlays shipped this way are discovered automatically as `installed-overlay` layers. + +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. ## Remove a Workflow diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index e1d29b47d..384f08c5c 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -49,6 +49,13 @@ workflow_step_catalog_app = typer.Typer( ) 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 @@ -625,6 +632,17 @@ def workflow_add( dest_dir.mkdir(parents=True, exist_ok=True) import shutil shutil.copy2(yaml_path, dest_dir / "workflow.yml") + + # If the source is a directory that also contains an overlays/ subdirectory, + # copy it alongside the workflow.yml so installed overlays are preserved. + source_dir = yaml_path.parent + source_overlays = source_dir / "overlays" + if source_overlays.is_dir() and not source_overlays.is_symlink(): + dest_overlays = dest_dir / "overlays" + if dest_overlays.exists(): + shutil.rmtree(dest_overlays) + shutil.copytree(source_overlays, dest_overlays) + registry.add(definition.id, { "name": definition.name, "version": definition.version, @@ -1724,6 +1742,97 @@ 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() + workflow_overlay_list(project_root, workflow_id) + + +@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 6025c30c5..01ea73d2a 100644 --- a/src/specify_cli/workflows/engine.py +++ b/src/specify_cli/workflows/engine.py @@ -625,13 +625,24 @@ class WorkflowEngine: 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 000000000..67929b0a6 --- /dev/null +++ b/src/specify_cli/workflows/overlays/__init__.py @@ -0,0 +1,80 @@ +"""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, + InstalledOverlaySource, + Layer, + ProjectOverlaySource, +) +from .merge import ComposedStep + + +class WorkflowResolver: + """Resolves a workflow ID to its composed ``WorkflowDefinition``. + + Collects layers from three tiers: + - project-local overlays (``.specify/workflows/overlays//*.yml``) + - installed overlays shipped with a workflow (``.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), + InstalledOverlaySource(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 project overlays rank above + installed overlays (they are listed first in higher-wins order). + """ + all_layers: list[Layer] = [] + for source in self._sources: + all_layers.extend(source.collect(workflow_id)) + + return sorted( + all_layers, + key=lambda layer: ( + -layer.priority, + 0 if layer.tier == "project-overlay" else 1, + layer.source, + ), + ) + + 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 000000000..bccb7872b --- /dev/null +++ b/src/specify_cli/workflows/overlays/_commands.py @@ -0,0 +1,364 @@ +"""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 . 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 directory.""" + return project_root / ".specify" / "workflows" / "overlays" + + +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. + """ + 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 _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.""" + 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 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. + """ + data, errors = _read_overlay(source) + if data is None: + for err in errors: + err_console.print(f"[red]Error:[/red] {err}") + return None + + 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 + + if priority is not None: + _validate_priority(priority) + data["priority"] = priority + # Re-validate after mutation. + 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) + 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.""" + 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.""" + 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]]: + """List all overlays for a workflow and print a summary table. + + Returns the raw list data for machine-readable callers. + """ + _validate_workflow_id_or_exit(workflow_id) + resolver = WorkflowResolver(project_root) + layers = resolver.collect_all_layers(workflow_id) + 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. + """ + _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 000000000..8ce3d1644 --- /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 000000000..8f4082c61 --- /dev/null +++ b/src/specify_cli/workflows/overlays/layer_sources.py @@ -0,0 +1,142 @@ +"""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)) + + +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.""" + workflow_overlay_dir = self.overlays_dir / workflow_id + 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"]) + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + 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 InstalledOverlaySource: + """Installed overlays: ``.specify/workflows//overlays/*.yml``.""" + + tier = "installed-overlay" + + 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]: + """Collect all installed overlays shipped with the given workflow.""" + installed_overlay_dir = self.workflows_dir / workflow_id / "overlays" + if not installed_overlay_dir.is_dir(): + return [] + layers: list[Layer] = [] + for path in sorted(installed_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"]) + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + 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"installed:{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 000000000..73a9c9f3e --- /dev/null +++ b/src/specify_cli/workflows/overlays/merge.py @@ -0,0 +1,272 @@ +"""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 _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 apply_edit( + steps: list[dict[str, Any]], + edit: OverlayEdit, + source: str, +) -> tuple[list[dict[str, Any]], ComposedStep | None, str | None]: + """Apply a single edit to a step list and return the mutated list. + + The returned list is the same list object as *steps* but deep-copied first + so callers can avoid mutating the base. Raises ``ValueError`` if the anchor + is not found. + + Returns: + ``(steps, composed_step, replaced_or_removed_id)``. *composed_step* is + ``None`` for ``remove`` operations; *replaced_or_removed_id* is the + previous step id for ``replace``/``remove`` operations, otherwise + ``None``. + """ + location = find_step(steps, edit.anchor) + if location is None: + raise ValueError(f"Anchor '{edit.anchor}' not found in workflow steps.") + + parent_list, index = location + old_step = parent_list[index] + old_step_id = old_step.get("id") if isinstance(old_step, dict) else None + + if edit.operation == "insert_after": + new_step = copy.deepcopy(edit.step) + parent_list.insert(index + 1, new_step) + step_id = str(new_step.get("id")) if isinstance(new_step, dict) else "" + return steps, ComposedStep(step_id, source), None + if edit.operation == "insert_before": + new_step = copy.deepcopy(edit.step) + parent_list.insert(index, new_step) + step_id = str(new_step.get("id")) if isinstance(new_step, dict) else "" + return steps, ComposedStep(step_id, source), None + if edit.operation == "replace": + new_step = copy.deepcopy(edit.step) + parent_list[index] = new_step + step_id = str(new_step.get("id")) if isinstance(new_step, dict) else "" + return steps, ComposedStep(step_id, source), old_step_id + if edit.operation == "remove": + del parent_list[index] + return steps, None, old_step_id + raise ValueError(f"Unsupported edit operation: {edit.operation}") + + +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 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)) + + for anchor, edits in edits_by_anchor.items(): + # Highest-priority edit is last because *overlays* is already sorted. + _, winning_edit = edits[-1] + + if winning_edit.operation == "remove": + # Anchor is removed; ignore all other edits on this anchor. + location = find_step(steps, anchor) + if location is not None: + parent_list, index = location + removed_step = parent_list[index] + removed_id = removed_step.get("id") if isinstance(removed_step, dict) else None + del parent_list[index] + if isinstance(removed_id, str): + sources.pop(removed_id, None) + continue + + # For replace/insert_*, the anchor survives. Only the highest-priority + # replace is applied; lower-priority replaces on the same anchor are + # skipped. Inserts are applied in merge order. + if winning_edit.operation == "replace": + winning_layer, _ = edits[-1] + steps, composed, replaced_id = apply_edit(steps, winning_edit, winning_layer.source) + if isinstance(replaced_id, str): + sources.pop(replaced_id, None) + if composed is not None: + sources[composed.step_id] = composed.source + + for layer, edit in edits: + if edit.operation in ("insert_after", "insert_before"): + steps, composed, replaced_id = apply_edit(steps, edit, layer.source) + if replaced_id is not None: + sources.pop(replaced_id, None) + if composed is not None: + sources[composed.step_id] = composed.source + + attribution = _build_attribution(steps, sources) + return steps, 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 000000000..5652d3f71 --- /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.match(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/workflows/conftest.py b/tests/workflows/conftest.py new file mode 100644 index 000000000..03d28890d --- /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 000000000..3e4aad913 --- /dev/null +++ b/tests/workflows/test_overlay_commands.py @@ -0,0 +1,314 @@ +"""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_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_add_copies_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 + installed_overlay = ( + project_dir / ".specify" / "workflows" / "wf" / "overlays" / "ov1.yml" + ) + assert installed_overlay.is_file() diff --git a/tests/workflows/test_overlay_composer.py b/tests/workflows/test_overlay_composer.py new file mode 100644 index 000000000..dc3b0e90d --- /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_merge.py b/tests/workflows/test_overlay_merge.py new file mode 100644 index 000000000..01be81bf4 --- /dev/null +++ b/tests/workflows/test_overlay_merge.py @@ -0,0 +1,402 @@ +"""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, + apply_edit, + 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 TestApplyEdit: + """Single edit operations against a step list.""" + + def test_insert_after(self): + steps = [_step("a"), _step("b")] + new_step = _step("new") + result, composed, replaced = apply_edit(steps, OverlayEdit("insert_after", "a", new_step), "project") + assert [s["id"] for s in result] == ["a", "new", "b"] + assert composed == ComposedStep("new", "project") + assert replaced is None + + def test_insert_before(self): + steps = [_step("a"), _step("b")] + new_step = _step("new") + result, composed, replaced = apply_edit(steps, OverlayEdit("insert_before", "b", new_step), "project") + assert [s["id"] for s in result] == ["a", "new", "b"] + assert replaced is None + + def test_replace(self): + steps = [_step("a"), _step("b")] + new_step = _step("replaced") + result, composed, replaced = apply_edit(steps, OverlayEdit("replace", "b", new_step), "installed:x") + assert [s["id"] for s in result] == ["a", "replaced"] + assert composed == ComposedStep("replaced", "installed:x") + assert replaced == "b" + + def test_remove(self): + steps = [_step("a"), _step("b"), _step("c")] + result, composed, replaced = apply_edit(steps, OverlayEdit("remove", "b"), "project") + assert [s["id"] for s in result] == ["a", "c"] + assert composed is None + assert replaced == "b" + + def test_apply_edit_anchors_nested(self): + steps = [ + { + "id": "if-1", + "type": "if", + "condition": "true", + "then": [_step("then-a")], + }, + ] + new_step = _step("new") + result, _, _ = apply_edit(steps, OverlayEdit("insert_after", "then-a", new_step), "project") + assert [s["id"] for s in result[0]["then"]] == ["then-a", "new"] + + def test_apply_edit_missing_anchor(self): + steps = [_step("a")] + new_step = _step("new") + with pytest.raises(ValueError, match="Anchor 'missing' not found"): + apply_edit(steps, OverlayEdit("insert_after", "missing", new_step), "project") + + +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): + 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"))], + ) + steps, _ = merge_steps(base, [_layer(insert, "project:insert"), _layer(replace, "project:replace")]) + assert [s["id"] for s in steps] == ["a", "replaced"] + + 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_project_wins_tie_over_installed_same_anchor(self): + base = [_step("a")] + installed = Overlay( + id="same", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "a", _step("installed-replace"))], + ) + project = Overlay( + id="same", + extends="wf", + priority=10, + edits=[OverlayEdit("replace", "a", _step("project-replace"))], + ) + # Merge order: installed first, then project (project wins tie). + steps, attribution = merge_steps( + base, + [ + _layer(installed, "installed:same"), + _layer(project, "project:same"), + ], + ) + assert [s["id"] for s in steps] == ["project-replace"] + assert any( + composed.step_id == "project-replace" and composed.source == "project:same" + for composed in attribution + ) + + 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")]) + + +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 diff --git a/tests/workflows/test_overlay_schema.py b/tests/workflows/test_overlay_schema.py new file mode 100644 index 000000000..72595b6d2 --- /dev/null +++ b/tests/workflows/test_overlay_schema.py @@ -0,0 +1,218 @@ +"""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 diff --git a/tests/workflows/test_overlay_security.py b/tests/workflows/test_overlay_security.py new file mode 100644 index 000000000..3871d8921 --- /dev/null +++ b/tests/workflows/test_overlay_security.py @@ -0,0 +1,204 @@ +"""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_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() diff --git a/tests/workflows/test_resolver_integration.py b/tests/workflows/test_resolver_integration.py new file mode 100644 index 000000000..cb14ec340 --- /dev/null +++ b/tests/workflows/test_resolver_integration.py @@ -0,0 +1,485 @@ +"""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_with_installed_overlay(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) + installed_dir = project_dir / ".specify" / "workflows" / "wf" / "overlays" + installed_dir.mkdir(parents=True, exist_ok=True) + installed_dir.joinpath("shipped.yml").write_text( + yaml.safe_dump( + { + "id": "shipped", + "extends": "wf", + "priority": 5, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "shipped-new", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + + resolver = WorkflowResolver(project_dir) + definition = resolver.resolve("wf") + assert [s["id"] for s in definition.steps] == ["a", "shipped-new"] + + 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_project_wins_tie_with_installed(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) + installed_dir = project_dir / ".specify" / "workflows" / "wf" / "overlays" + installed_dir.mkdir(parents=True, exist_ok=True) + installed_dir.joinpath("shipped.yml").write_text( + yaml.safe_dump( + { + "id": "shipped", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "installed-step", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + _write_overlay( + project_dir, + "wf", + "project", + { + "id": "project", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "project-step", "type": "command", "command": "echo"}, + } + ], + }, + ) + + resolver = WorkflowResolver(project_dir) + definition = resolver.resolve("wf") + # Same priority: installed applied first, then project wins tie by + # ending up closer to the anchor. + assert [s["id"] for s in definition.steps] == ["a", "project-step", "installed-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_distinguishes_project_and_installed_same_id(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) + installed_dir = project_dir / ".specify" / "workflows" / "wf" / "overlays" + installed_dir.mkdir(parents=True, exist_ok=True) + installed_dir.joinpath("same.yml").write_text( + yaml.safe_dump( + { + "id": "same", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "installed-new", "type": "command", "command": "echo"}, + } + ], + } + ), + encoding="utf-8", + ) + _write_overlay( + project_dir, + "wf", + "same", + { + "id": "same", + "extends": "wf", + "priority": 10, + "edits": [ + { + "operation": "insert_after", + "anchor": "a", + "step": {"id": "project-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", "project-new", "installed-new"] + sources = {c.step_id: c.source for c in attribution} + assert sources["project-new"] == "project:same" + assert sources["installed-new"] == "installed:same" + + 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_invalid_installed_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) + installed_dir = project_dir / ".specify" / "workflows" / "wf" / "overlays" + installed_dir.mkdir(parents=True, exist_ok=True) + installed_dir.joinpath("broken.yml").write_text( + yaml.safe_dump( + { + "id": "broken", + "extends": "wf", + "priority": 10, + "edits": "not-a-list", + } + ), + encoding="utf-8", + ) + + 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_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"]