Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions src/specify_cli/commands/init.py
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ def init(
branch_numbering: str = typer.Option(None, "--branch-numbering", help="Branch numbering strategy: 'sequential' (001, 002, …, 1000, … — expands past 999 automatically) or 'timestamp' (YYYYMMDD-HHMMSS)"),
integration: str = typer.Option(None, "--integration", help="Use the new integration system (e.g. --integration copilot). Mutually exclusive with --ai."),
integration_options: str = typer.Option(None, "--integration-options", help='Options for the integration (e.g. --integration-options="--commands-dir .myagent/cmds")'),
workflow: str = typer.Option(None, "--workflow", help="Run a workflow YAML file after project initialization"),
):
"""
Initialize a new Specify project.
Expand Down Expand Up @@ -673,6 +674,45 @@ def init(
console.print(tracker.render())
console.print("\n[bold green]Project ready.[/bold green]")

if workflow:
workflow_path = Path(workflow)
if not workflow_path.is_absolute():
workflow_path = Path.cwd() / workflow_path
if not workflow_path.exists():
console.print(f"[red]Error:[/red] Workflow file not found: {workflow}")
raise typer.Exit(1)
console.print(f"\n[bold cyan]Running post-init workflow:[/bold cyan] {workflow}")
try:
from ..workflows.engine import WorkflowEngine
engine = WorkflowEngine(project_path)
engine.on_step_start = lambda sid, label: console.print(f" ▸ [{sid}] {label} …")
definition = engine.load_workflow(str(workflow_path))
errors = engine.validate(definition)
if errors:
console.print("[red]Workflow validation failed:[/red]")
for err in errors:
console.print(f" • {err}")
raise typer.Exit(1)
state = engine.execute(definition)
status_colors = {
"completed": "green",
"paused": "yellow",
"failed": "red",
"aborted": "red",
}
color = status_colors.get(state.status.value, "white")
console.print(f"\n[{color}]Workflow status: {state.status.value}[/{color}]")
if state.status.value == "paused":
console.print(f"\nResume with: [cyan]specify workflow resume {state.run_id}[/cyan]")
elif state.status.value in ("failed", "aborted"):
console.print("[red]Post-init workflow did not complete successfully.[/red]")
raise typer.Exit(1)
except (typer.Exit, SystemExit):
raise
except Exception as wf_exc:
console.print(f"[red]Post-init workflow failed:[/red] {wf_exc}")
raise typer.Exit(1)

agent_config = AGENT_CONFIG.get(selected_ai)
if agent_config:
agent_folder = ai_commands_dir if selected_ai == "generic" else agent_config["folder"]
Expand Down
133 changes: 133 additions & 0 deletions tests/test_init_workflow.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
"""Tests for --workflow flag on specify init."""

import os

import yaml


class TestInitWorkflowFlag:
"""Tests for the --workflow option on specify init."""

def test_workflow_flag_missing_file_errors(self, tmp_path):
from typer.testing import CliRunner
from specify_cli import app

runner = CliRunner()
project = tmp_path / "wf-test"
project.mkdir()
old_cwd = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, [
"init", "--here", "--integration", "copilot",
"--script", "sh", "--no-git",
"--workflow", str(tmp_path / "nonexistent.yml"),
], catch_exceptions=False)
finally:
os.chdir(old_cwd)
assert result.exit_code != 0
assert "Workflow file not found" in result.output

def test_workflow_flag_runs_workflow_after_init(self, tmp_path):
from typer.testing import CliRunner
from specify_cli import app

runner = CliRunner()
project = tmp_path / "wf-run-test"
project.mkdir()

# Create a minimal workflow YAML that runs a simple shell step
workflow_file = tmp_path / "post-init.yml"
workflow_content = {
"schema_version": "1.0",
"workflow": {
"id": "post-init-test",
"name": "Post Init Test",
"version": "1.0.0",
"description": "A test post-init workflow",
},
"steps": [
{
"id": "create-marker",
"type": "shell",
"run": "echo done > marker.txt",
},
],
}
workflow_file.write_text(yaml.dump(workflow_content), encoding="utf-8")

old_cwd = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, [
"init", "--here", "--integration", "copilot",
"--script", "sh", "--no-git",
"--workflow", str(workflow_file),
], catch_exceptions=False)
finally:
os.chdir(old_cwd)
assert result.exit_code == 0, f"init failed: {result.output}"
assert "Running post-init workflow" in result.output
assert "Workflow status: completed" in result.output
# The shell step should have created the marker file
assert (project / "marker.txt").exists()

def test_workflow_flag_failing_workflow_exits_nonzero(self, tmp_path):
from typer.testing import CliRunner
from specify_cli import app

runner = CliRunner()
project = tmp_path / "wf-fail-test"
project.mkdir()

# Create a workflow that fails
workflow_file = tmp_path / "bad-workflow.yml"
workflow_content = {
"schema_version": "1.0",
"workflow": {
"id": "bad-wf",
"name": "Bad Workflow",
"version": "1.0.0",
"description": "A failing workflow",
},
"steps": [
{
"id": "fail-step",
"type": "shell",
"run": "exit 1",
},
],
}
workflow_file.write_text(yaml.dump(workflow_content), encoding="utf-8")

old_cwd = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, [
"init", "--here", "--integration", "copilot",
"--script", "sh", "--no-git",
"--workflow", str(workflow_file),
], catch_exceptions=False)
finally:
os.chdir(old_cwd)
assert result.exit_code != 0

def test_init_without_workflow_flag_works_normally(self, tmp_path):
"""Ensure that omitting --workflow does not change existing behavior."""
from typer.testing import CliRunner
from specify_cli import app

runner = CliRunner()
project = tmp_path / "no-wf-test"
project.mkdir()
old_cwd = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, [
"init", "--here", "--integration", "copilot",
"--script", "sh", "--no-git",
], catch_exceptions=False)
finally:
os.chdir(old_cwd)
assert result.exit_code == 0, f"init failed: {result.output}"
assert "Running post-init workflow" not in result.output