From 698f42641d5582881a23312bd47908cbc6e77e86 Mon Sep 17 00:00:00 2001 From: wangchenguang Date: Tue, 25 Aug 2026 20:08:57 +0800 Subject: [PATCH] Preview init changes before a forced merge can alter a project Dry-run stages the target and invokes the public initializer in an isolated child process, then reports create, overwrite, and preserve actions without writing to the requested project. This keeps previews aligned with integration-specific installation behavior. --- src/specify_cli/commands/bundle/__init__.py | 2 + src/specify_cli/commands/init.py | 268 ++++++++++++++++++-- tests/test_init_dry_run.py | 214 ++++++++++++++++ 3 files changed, 464 insertions(+), 20 deletions(-) create mode 100644 tests/test_init_dry_run.py diff --git a/src/specify_cli/commands/bundle/__init__.py b/src/specify_cli/commands/bundle/__init__.py index 165f674a36..65271145af 100644 --- a/src/specify_cli/commands/bundle/__init__.py +++ b/src/specify_cli/commands/bundle/__init__.py @@ -126,6 +126,8 @@ def _run_init(integration: str, *, script_type: str, offline: bool = False) -> N integration_options=None, extensions=None, trust_extension_urls=False, + dry_run=False, + json_output=False, ) except typer.Exit as exc: if exc.exit_code: diff --git a/src/specify_cli/commands/init.py b/src/specify_cli/commands/init.py index 4af9427bfa..4b826b7b30 100644 --- a/src/specify_cli/commands/init.py +++ b/src/specify_cli/commands/init.py @@ -2,11 +2,14 @@ from __future__ import annotations +import hashlib +import json import os import shlex import shutil import subprocess import sys +import tempfile from pathlib import Path from typing import Any @@ -53,6 +56,191 @@ def _ext_spec_is_url(ext_spec: str) -> bool: return False +def _snapshot_files(root: Path) -> dict[str, str]: + """Return SHA-256 digests for regular files below *root*.""" + if not root.exists(): + return {} + + files: dict[str, str] = {} + for path in root.rglob("*"): + if not path.is_file() or path.is_symlink(): + continue + digest = hashlib.sha256(path.read_bytes()).hexdigest() + files[path.relative_to(root).as_posix()] = digest + return files + + +def _preview_manifest_provenance(staged_root: Path) -> dict[str, str]: + """Map manifest-tracked staged paths to their installation source.""" + provenance: dict[str, str] = {} + manifests = staged_root / ".specify" / "integrations" + if not manifests.is_dir(): + return provenance + + for manifest_path in manifests.glob("*.manifest.json"): + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + key = str(manifest.get("key", manifest_path.stem.removesuffix(".manifest"))) + source = "core" if key == "speckit" else f"integration:{key}" + for relative_path in manifest.get("files", {}): + provenance[str(relative_path)] = source + except (OSError, TypeError, ValueError): + continue + return provenance + + +def _preview_default_provenance(relative_path: str) -> str: + if relative_path.startswith(".specify/workflows/"): + return "workflow" + if relative_path.startswith(".specify/extensions/"): + return "extension" + if relative_path.startswith(".specify/presets/"): + return "preset" + if relative_path.startswith(".specify/"): + return "core" + return "integration" + + +def _build_preview_actions( + initial_files: dict[str, str], staged_root: Path +) -> list[dict[str, str]]: + """Classify files produced by a staged initialization.""" + staged_files = _snapshot_files(staged_root) + provenance = _preview_manifest_provenance(staged_root) + candidates = { + path + for path, digest in staged_files.items() + if initial_files.get(path) != digest + } + candidates.update(path for path in provenance if path in staged_files) + + actions: list[dict[str, str]] = [] + for path in sorted(candidates): + staged_digest = staged_files[path] + initial_digest = initial_files.get(path) + action = ( + "create" + if initial_digest is None + else "overwrite" + if initial_digest != staged_digest + else "preserve" + ) + actions.append( + { + "action": action, + "path": path, + "provenance": provenance.get(path, _preview_default_provenance(path)), + } + ) + return actions + + +def _emit_dry_run_preview(payload: dict[str, Any], *, json_output: bool) -> None: + """Render a stable human or machine-readable initialization preview.""" + if json_output: + typer.echo(json.dumps(payload, sort_keys=True)) + return + + console.print("\n[bold cyan]Initialization preview[/bold cyan]") + if payload["conflict"]: + console.print( + "[yellow]conflict[/yellow] target directory is non-empty; rerun with --force " + "to preview a forced merge" + ) + return + + for record in payload["actions"]: + console.print( + f"{record['action']:<10} {record['path']} " + f"[dim]({record['provenance']})[/dim]" + ) + + +def _preview_init( + *, + project_path: Path, + directory_conflict: bool, + script_type: str, + selected_integration: str, + ignore_agent_tools: bool, + preset: str | None, + integration_options: str | None, + extensions: list[str] | None, + trust_extension_urls: bool, + json_output: bool, +) -> None: + """Run the canonical initializer in staging and report its file plan.""" + payload: dict[str, Any] = { + "dry_run": True, + "target": str(project_path), + "conflict": directory_conflict, + "actions": [], + } + if directory_conflict: + _emit_dry_run_preview(payload, json_output=json_output) + return + + initial_files = _snapshot_files(project_path) + url_extensions = [spec for spec in extensions or [] if _ext_spec_is_url(spec)] + staged_extensions = [spec for spec in extensions or [] if not _ext_spec_is_url(spec)] + + with tempfile.TemporaryDirectory(prefix="specify-init-preview-") as tmp_dir: + staged_root = Path(tmp_dir) / "project" + if project_path.exists(): + shutil.copytree(project_path, staged_root, symlinks=True) + + # Run the same public CLI path in a child process. Besides preventing + # mutations of the target root, this isolates Rich's Live output from + # the preview's human/JSON output contract. + command = [ + sys.executable, + "-c", + "from specify_cli import main; main()", + "init", + str(staged_root), + "--force", + "--non-interactive", + "--integration", + selected_integration, + "--script", + script_type, + ] + if ignore_agent_tools: + command.append("--ignore-agent-tools") + if integration_options: + command.extend(["--integration-options", integration_options]) + if preset: + command.extend(["--preset", preset]) + for extension in staged_extensions: + command.extend(["--extension", extension]) + if trust_extension_urls: + command.append("--trust-extension-urls") + + result = subprocess.run( + command, + cwd=Path.cwd(), + capture_output=True, + text=True, + check=False, + ) + if result.returncode: + details = (result.stderr or result.stdout).strip().replace("\n", " ") + raise RuntimeError(f"staged initialization failed: {details[:240]}") + + payload["actions"] = _build_preview_actions(initial_files, staged_root) + + for spec in url_extensions: + payload["actions"].append( + { + "action": "unresolved", + "path": spec, + "provenance": "extension:url", + } + ) + payload["actions"].sort(key=lambda action: action["path"]) + _emit_dry_run_preview(payload, json_output=json_output) + + def _confirm_extension_url_trust( url_specs: list[str], *, @@ -336,6 +524,16 @@ def init( "--trust-extension-urls", help="Pre-authorize installing extensions from external URLs without the interactive trust prompt (required for non-interactive URL installs).", ), + dry_run: bool = typer.Option( + False, + "--dry-run", + help="Preview initialization changes without writing to the target project.", + ), + json_output: bool = typer.Option( + False, + "--json", + help="Emit the dry-run preview as a single JSON document.", + ), ): """ Initialize a new Specify project. @@ -391,7 +589,12 @@ def init( _write_integration_json, ) - show_banner() + if not (dry_run and json_output): + show_banner() + + if json_output and not dry_run: + console.print("[red]Error:[/red] --json requires --dry-run") + raise typer.Exit(1) from ..integrations import INTEGRATION_REGISTRY, get_integration @@ -423,6 +626,7 @@ def init( raise typer.Exit(1) dir_existed_before = False + directory_conflict = False if here: project_name = Path.cwd().name project_path = Path.cwd() @@ -430,17 +634,21 @@ def init( existing_items = list(project_path.iterdir()) if existing_items: - console.print( - f"[yellow]Warning:[/yellow] Current directory is not empty ({len(existing_items)} items)" - ) - if force: - # Proceeding: the merge/overwrite warning is accurate here. + if not (dry_run and json_output): console.print( - "[yellow]Template files will be merged with existing content and may overwrite existing files[/yellow]" - ) - console.print( - "[cyan]--force supplied: skipping confirmation and proceeding with merge[/cyan]" + f"[yellow]Warning:[/yellow] Current directory is not empty ({len(existing_items)} items)" ) + if dry_run and not force: + directory_conflict = True + elif force: + # Proceeding: the merge/overwrite warning is accurate here. + if not (dry_run and json_output): + console.print( + "[yellow]Template files will be merged with existing content and may overwrite existing files[/yellow]" + ) + console.print( + "[cyan]--force supplied: skipping confirmation and proceeding with merge[/cyan]" + ) elif non_interactive: console.print( "[red]Error:[/red] Current directory is not empty and " @@ -492,17 +700,20 @@ def init( ) raise typer.Exit(1) existing_items = list(project_path.iterdir()) - if force: - if existing_items: + if dry_run and not force: + directory_conflict = True + elif force: + if existing_items and not (dry_run and json_output): console.print( f"[yellow]Warning:[/yellow] Directory '{safe_name}' is not empty ({len(existing_items)} items)" ) console.print( "[yellow]Template files will be merged with existing content and may overwrite existing files[/yellow]" ) - console.print( - f"[cyan]--force supplied: merging into existing directory '[cyan]{safe_name}[/cyan]'[/cyan]" - ) + if not (dry_run and json_output): + console.print( + f"[cyan]--force supplied: merging into existing directory '[cyan]{safe_name}[/cyan]'[/cyan]" + ) else: error_panel = Panel( f"Directory already exists: '[cyan]{safe_name}[/cyan]'\n" @@ -568,9 +779,10 @@ def init( f"{'Target Path':<15} [dim]{_escape_markup(str(project_path))}[/dim]" ) - console.print( - Panel("\n".join(setup_lines), border_style="cyan", padding=(1, 2)) - ) + if not (dry_run and json_output): + console.print( + Panel("\n".join(setup_lines), border_style="cyan", padding=(1, 2)) + ) if not ignore_agent_tools: agent_config = AGENT_CONFIG.get(selected_ai) @@ -610,8 +822,24 @@ def init( else: selected_script = default_script - console.print(f"[cyan]Selected coding agent integration:[/cyan] {selected_ai}") - console.print(f"[cyan]Selected script type:[/cyan] {selected_script}") + if not (dry_run and json_output): + console.print(f"[cyan]Selected coding agent integration:[/cyan] {selected_ai}") + console.print(f"[cyan]Selected script type:[/cyan] {selected_script}") + + if dry_run: + _preview_init( + project_path=project_path, + directory_conflict=directory_conflict, + script_type=selected_script, + selected_integration=selected_ai, + ignore_agent_tools=ignore_agent_tools, + preset=preset, + integration_options=integration_options, + extensions=extensions, + trust_extension_urls=trust_extension_urls, + json_output=json_output, + ) + return tracker = StepTracker("Initialize Specify Project") diff --git a/tests/test_init_dry_run.py b/tests/test_init_dry_run.py new file mode 100644 index 0000000000..835b719e23 --- /dev/null +++ b/tests/test_init_dry_run.py @@ -0,0 +1,214 @@ +"""CLI contract tests for ``specify init --dry-run``.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from typer.testing import CliRunner + +from specify_cli import app +from specify_cli.commands.init import _snapshot_files + + +def test_dry_run_reports_new_project_files_without_creating_target(tmp_path: Path) -> None: + target = tmp_path / "preview-project" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "Initialization preview" in result.output + assert ".github/skills/speckit-plan/SKILL.md" in result.output + assert not target.exists() + + +def test_dry_run_json_is_machine_readable_and_has_no_target_writes(tmp_path: Path) -> None: + target = tmp_path / "json-preview-project" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["dry_run"] is True + assert {action["path"] for action in payload["actions"]} >= { + ".github/skills/speckit-plan/SKILL.md" + } + assert not target.exists() + + +def test_forced_dry_run_reports_overwrite_without_changing_existing_file( + tmp_path: Path, +) -> None: + target = tmp_path / "existing-project" + command = target / ".github" / "skills" / "speckit-plan" / "SKILL.md" + command.parent.mkdir(parents=True) + command.write_text("user-owned content\n", encoding="utf-8") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--force", + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert { + (action["action"], action["path"]) + for action in payload["actions"] + } >= {("overwrite", ".github/skills/speckit-plan/SKILL.md")} + assert command.read_text(encoding="utf-8") == "user-owned content\n" + + +def test_non_forced_dry_run_reports_existing_target_conflict(tmp_path: Path) -> None: + target = tmp_path / "nonempty-project" + target.mkdir() + existing = target / "keep.txt" + existing.write_text("keep\n", encoding="utf-8") + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["conflict"] is True + assert payload["actions"] == [] + assert existing.read_text(encoding="utf-8") == "keep\n" + + +def test_dry_run_leaves_url_extension_unresolved_without_creating_target( + tmp_path: Path, +) -> None: + target = tmp_path / "url-extension-preview" + extension_url = "https://example.com/spec-kit-extension.zip" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--extension", + extension_url, + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert { + (action["action"], action["path"]) + for action in payload["actions"] + } >= {("unresolved", extension_url)} + assert not target.exists() + + +def test_dry_run_changed_paths_match_a_forced_real_initialization(tmp_path: Path) -> None: + target = tmp_path / "parity-project" + command = target / ".github" / "skills" / "speckit-plan" / "SKILL.md" + command.parent.mkdir(parents=True) + command.write_text("user-owned content\n", encoding="utf-8") + before = _snapshot_files(target) + arguments = [ + "init", + str(target), + "--force", + "--integration", + "copilot", + "--script", + "sh", + ] + + preview = CliRunner().invoke( + app, [*arguments, "--dry-run", "--json"], catch_exceptions=False + ) + assert preview.exit_code == 0, preview.output + predicted = { + action["path"] + for action in json.loads(preview.output)["actions"] + if action["action"] in {"create", "overwrite"} + } + + actual = CliRunner().invoke(app, arguments, catch_exceptions=False) + assert actual.exit_code == 0, actual.output + after = _snapshot_files(target) + changed = {path for path, digest in after.items() if before.get(path) != digest} + + assert predicted == changed + + +def test_dry_run_includes_bundled_extension_artifacts(tmp_path: Path) -> None: + target = tmp_path / "extension-preview-project" + + result = CliRunner().invoke( + app, + [ + "init", + str(target), + "--dry-run", + "--json", + "--integration", + "copilot", + "--script", + "sh", + "--extension", + "git", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert any(action["provenance"] == "extension" for action in payload["actions"]) + assert not target.exists()