mirror of
https://github.com/github/spec-kit.git
synced 2026-08-15 14:02:04 +02:00
feat(workflows): add standalone WorkflowResolver and overlay subsystem
Implement PR 1 of the workflow-overlays plan: a concrete, standalone WorkflowResolver for downstream workflow extensibility without touching the Preset subsystem. - Add overlay manifest schema (Overlay, OverlayEdit, validate_overlay_yaml) - Add pure-function merge engine (find_step, apply_edit, merge_steps, validate_edits) with recursive anchor search and higher-wins semantics - Add StepListComposer and tiered layer sources (project, installed, base) - Add WorkflowResolver facade with inline HIGHER_WINS priority sorting - Add CLI verbs: workflow overlay add/set-priority/enable/disable/remove/list and workflow resolve <id> - Wire WorkflowEngine.load_workflow through WorkflowResolver - Extend workflow add to copy optional overlays/ subdirectory from local workflow directories - Add comprehensive unit, integration, and security tests Refs: discussion #3473 (https://github.com/github/spec-kit/discussions/3473) Assisted-by: Kimi (model: opencode-go/kimi-k2.7-code, autonomous)
This commit is contained in:
Markus
+189
-1
@@ -86,7 +86,195 @@ Lists workflows installed in the current project.
|
||||
specify workflow add <source>
|
||||
```
|
||||
|
||||
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 <workflow-id>` 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/<id>/overlays/*.yml` | Shipped with the installed workflow | `installed-overlay` |
|
||||
| `.specify/workflows/overlays/<id>/*.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 <path-to-overlay.yml> --priority <n>
|
||||
```
|
||||
|
||||
Validates the overlay file and copies it to `.specify/workflows/overlays/<extends>/<id>.yml`. `--priority` overrides the `priority` field in the file.
|
||||
|
||||
#### List Overlays
|
||||
|
||||
```bash
|
||||
specify workflow overlay list <workflow-id>
|
||||
```
|
||||
|
||||
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 <workflow-id> <overlay-id> <n>
|
||||
```
|
||||
|
||||
#### Enable or Disable
|
||||
|
||||
```bash
|
||||
specify workflow overlay disable <workflow-id> <overlay-id>
|
||||
specify workflow overlay enable <workflow-id> <overlay-id>
|
||||
```
|
||||
|
||||
#### Remove
|
||||
|
||||
```bash
|
||||
specify workflow overlay remove <workflow-id> <overlay-id>
|
||||
```
|
||||
|
||||
Removes the project overlay file. Installed overlays shipped with a workflow are removed by `specify workflow remove <workflow-id>`, which deletes the entire workflow directory.
|
||||
|
||||
#### Inspect the Composed Workflow
|
||||
|
||||
```bash
|
||||
specify workflow resolve <workflow-id>
|
||||
```
|
||||
|
||||
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 <local-directory>` copies `workflow.yml` and an optional `overlays/` subdirectory into `.specify/workflows/<id>/`. 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/<id>/` 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
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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/<id>/*.yml``)
|
||||
- installed overlays shipped with a workflow (``.specify/workflows/<id>/overlays/*.yml``)
|
||||
- the base workflow itself (``.specify/workflows/<id>/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
|
||||
|
||||
@@ -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
|
||||
],
|
||||
}
|
||||
@@ -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
|
||||
@@ -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/<id>/*.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/<id>/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/<id>/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,
|
||||
)
|
||||
]
|
||||
@@ -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
|
||||
@@ -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,
|
||||
),
|
||||
[],
|
||||
)
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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")
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user