"""Steps for git_worktree_apply.feature.""" from __future__ import annotations import os import shutil import subprocess import tempfile from pathlib import Path from behave import given, then, when _PLAN_ID = "01TEST00000000000000000099" _BRANCH = f"cleveragents/plan-{_PLAN_ID}" def _git(args: list[str], cwd: str) -> subprocess.CompletedProcess[str]: return subprocess.run( ["git", *args], cwd=cwd, capture_output=True, text=True, check=True, timeout=10, ) # ── Given ────────────────────────────────────────────────── @given("a temp git project with 1 file for wt_apply") def step_git_project(context: object) -> None: d = tempfile.mkdtemp(prefix="wt-apply-") context.add_cleanup(shutil.rmtree, d, True) _git(["init", "-q"], d) _git(["config", "user.name", "T"], d) _git(["config", "user.email", "t@t"], d) _git(["config", "commit.gpgsign", "false"], d) Path(d, "main.py").write_text("print('hello')\n") _git(["add", "."], d) _git(["commit", "-q", "-m", "init"], d) context.wt_apply_project = d context.wt_apply_sandbox = None context.wt_apply_sandbox_obj = None @given("a temp non-git directory with 1 file for wt_apply") def step_non_git_dir(context: object) -> None: d = tempfile.mkdtemp(prefix="wt-apply-flat-") context.add_cleanup(shutil.rmtree, d, True) sandbox = os.path.join(d, "sandbox") os.makedirs(sandbox) Path(sandbox, "result.txt").write_text("output\n") context.wt_apply_project = d context.wt_apply_sandbox_flat = sandbox # ── When ─────────────────────────────────────────────────── @when("I create a sandbox for the project for wt_apply") @given("I create a sandbox for the project for wt_apply") def step_create_sandbox(context: object) -> None: from cleveragents.infrastructure.sandbox.git_worktree import ( GitWorktreeSandbox, ) sandbox = GitWorktreeSandbox( resource_id="res-wt-apply", original_path=context.wt_apply_project, ) ctx = sandbox.create(plan_id=_PLAN_ID) context.wt_apply_sandbox = ctx.sandbox_path context.wt_apply_sandbox_obj = sandbox context.add_cleanup(sandbox.cleanup) @when("I write {filename:S} to the sandbox for wt_apply") @given("I write {filename:S} to the sandbox for wt_apply") def step_write_file(context: object, filename: str) -> None: filename = filename.strip('"') Path(context.wt_apply_sandbox, filename).write_text("# fixed\n") @when("I commit the sandbox for wt_apply") @given("I commit the sandbox for wt_apply") def step_commit_sandbox(context: object) -> None: wt = context.wt_apply_sandbox _git(["add", "-A"], wt) _git(["commit", "-m", "test commit"], wt) @when("I merge the sandbox branch into main for wt_apply") def step_merge(context: object) -> None: repo = context.wt_apply_project _git(["merge", _BRANCH, "--no-edit", "-m", "merge test"], repo) @when("I cleanup the sandbox for wt_apply") def step_cleanup(context: object) -> None: context.wt_apply_sandbox_obj.cleanup() @when("I copy files from sandbox to project for wt_apply") def step_flat_copy(context: object) -> None: sandbox = context.wt_apply_sandbox_flat project = context.wt_apply_project for fname in os.listdir(sandbox): src = os.path.join(sandbox, fname) if os.path.isfile(src): shutil.copy2(src, os.path.join(project, fname)) # ── Then ─────────────────────────────────────────────────── @then("the sandbox path should be a directory for wt_apply") def step_sandbox_is_dir(context: object) -> None: assert os.path.isdir(context.wt_apply_sandbox) @then("the sandbox path should differ from the project path for wt_apply") def step_sandbox_differs(context: object) -> None: assert context.wt_apply_sandbox != context.wt_apply_project @then("the branch {branch_name:S} should exist for wt_apply") def step_branch_exists(context: object, branch_name: str) -> None: branch_name = branch_name.strip('"') result = subprocess.run( ["git", "rev-parse", "--verify", f"refs/heads/{branch_name}"], cwd=context.wt_apply_project, capture_output=True, check=False, timeout=10, ) assert result.returncode == 0, f"Branch {branch_name} does not exist" @then("the branch {branch_name:S} should not exist for wt_apply") def step_branch_not_exists(context: object, branch_name: str) -> None: branch_name = branch_name.strip('"') result = subprocess.run( ["git", "rev-parse", "--verify", f"refs/heads/{branch_name}"], cwd=context.wt_apply_project, capture_output=True, check=False, timeout=10, ) assert result.returncode != 0, f"Branch {branch_name} still exists" @then("{filename:S} should exist on branch {branch_name:S} for wt_apply") def step_file_on_branch(context: object, filename: str, branch_name: str) -> None: filename = filename.strip('"') branch_name = branch_name.strip('"') result = subprocess.run( ["git", "ls-tree", "--name-only", branch_name], cwd=context.wt_apply_project, capture_output=True, text=True, check=True, timeout=10, ) assert filename in result.stdout, ( f"{filename} not found on {branch_name}: {result.stdout}" ) @then("{filename:S} should exist in the project root for wt_apply") def step_file_in_root(context: object, filename: str) -> None: filename = filename.strip('"') assert os.path.isfile(os.path.join(context.wt_apply_project, filename)) @then("the sandbox path should not exist for wt_apply") def step_sandbox_gone(context: object) -> None: assert not os.path.exists(context.wt_apply_sandbox) @then("the project should have the copied file for wt_apply") def step_flat_copy_result(context: object) -> None: assert os.path.isfile(os.path.join(context.wt_apply_project, "result.txt"))