feat(copy): add --work-dir option and extract shared repo_utils

.changelog/022.yaml

@@ -0,0 +1,18 @@
+name: resolve_repo_path
+ts: 2026-02-17 17:11:39.210022+00:00
+type: keep_private
+full_path: _internal.repo_utils.resolve_repo_path
+---
+name: ensure_repo
+ts: 2026-02-17 17:11:39.210111+00:00
+type: keep_private
+full_path: _internal.repo_utils.ensure_repo
+---
+name: CopyOptions
+ts: 2026-02-17 17:11:39.671624+00:00
+type: additional_change
+auto_generated: true
+change_kind: optional_field_added
+details: 'added optional field ''work_dir'' (default: '''')'
+field_name: work_dir
+group: copy

docs/copy/copyoptions.md

@@ -2,7 +2,7 @@
 
 <!-- === DO_NOT_EDIT: pkg-ext copyoptions_def === -->
 ## class: CopyOptions
-- [source](../../path_sync/_internal/cmd_copy.py#L48)
+- [source](../../path_sync/_internal/cmd_copy.py#L49)
 > **Since:** 0.3.0
 
 ```python
@@ -18,6 +18,7 @@ class CopyOptions(BaseModel):
     skip_verify: bool = False
     no_wait: bool = False
     no_auto_merge: bool = False
+    work_dir: str = ''
     pr_title: str = ''
     labels: list[str] | None = None
     reviewers: list[str] | None = None
@@ -47,6 +48,7 @@ class CopyOptions(BaseModel):
 
 | Version | Change |
 |---------|--------|
+| unreleased | added optional field 'work_dir' (default: '') |
 | 0.7.0 | added optional field 'no_auto_merge' (default: False) |
 | 0.7.0 | added optional field 'no_wait' (default: False) |
 | 0.6.0 | added optional field 'skip_commit' (default: False) |

docs/copy/index.md

@@ -13,11 +13,11 @@
 <a id="copy_def"></a>
 
 ### cli_command: `copy`
-- [source](../../path_sync/_internal/cmd_copy.py#L66)
+- [source](../../path_sync/_internal/cmd_copy.py#L68)
 > **Since:** 0.4.1
 
 ```python
-def copy(*, name: str = '', config_path_opt: str = '', src_root_opt: str = '', dest_filter: str = '', dry_run: bool = False, force_overwrite: bool = False, detailed_exit_code: bool = False, no_checkout: bool = False, checkout_from_default: bool = False, skip_commit: bool = False, no_prompt: bool = False, no_pr: bool = False, pr_title: str = '', pr_labels: str = '', pr_reviewers: str = '', pr_assignees: str = '', skip_orphan_cleanup: bool = False, skip_verify: bool = False, no_wait: bool = False, no_auto_merge: bool = False) -> None:
+def copy(*, name: str = '', config_path_opt: str = '', src_root_opt: str = '', dest_filter: str = '', work_dir: str = '', dry_run: bool = False, force_overwrite: bool = False, detailed_exit_code: bool = False, no_checkout: bool = False, checkout_from_default: bool = False, skip_commit: bool = False, no_prompt: bool = False, no_pr: bool = False, pr_title: str = '', pr_labels: str = '', pr_reviewers: str = '', pr_assignees: str = '', skip_orphan_cleanup: bool = False, skip_verify: bool = False, no_wait: bool = False, no_auto_merge: bool = False) -> None:
     ...
 ```
 
@@ -31,6 +31,7 @@ Copy files from SRC to DEST repositories.
 | `-c`, `--config-path` | `str` | `''` | Full path to config file (alternative to --name) |
 | `--src-root` | `str` | `''` | Source repo root (default: find git root from cwd) |
 | `-d`, `--dest` | `str` | `''` | Filter destinations (comma-separated) |
+| `--work-dir` | `str` | `''` | Clone repos here (overrides dest_path_relative) |
 | `--dry-run` | `bool` | `False` | Preview without writing |
 | `--force-overwrite` | `bool` | `False` | Overwrite files even if header removed (opted out) |
 | `--detailed-exit-code` | `bool` | `False` | Exit 0=no changes, 1=changes, 2=error |

path_sync/_internal/cmd_copy.py

@@ -23,6 +23,7 @@
     pr_already_synced,
     resolve_config_path,
 )
+from path_sync._internal.repo_utils import ensure_repo, resolve_repo_path
 from path_sync._internal.typer_app import app
 from path_sync._internal.verify import StepFailure, VerifyResult, VerifyStatus
 from path_sync._internal.yaml_utils import load_yaml_model
@@ -57,6 +58,7 @@ class CopyOptions(BaseModel):
     skip_verify: bool = False
     no_wait: bool = False
     no_auto_merge: bool = False
+    work_dir: str = ""
     pr_title: str = ""
     labels: list[str] | None = None
     reviewers: list[str] | None = None
@@ -78,6 +80,7 @@ def copy(
         help="Source repo root (default: find git root from cwd)",
     ),
     dest_filter: str = typer.Option("", "-d", "--dest", help="Filter destinations (comma-separated)"),
+    work_dir: str = typer.Option("", "--work-dir", help="Clone repos here (overrides dest_path_relative)"),
     dry_run: bool = typer.Option(False, "--dry-run", help="Preview without writing"),
     force_overwrite: bool = typer.Option(
         False,
@@ -174,6 +177,7 @@ def copy(
         skip_verify=skip_verify,
         no_wait=no_wait,
         no_auto_merge=no_auto_merge,
+        work_dir=work_dir,
         pr_title=pr_title or config.pr_defaults.title,
         labels=cmd_options.split_csv(pr_labels) or config.pr_defaults.labels,
         reviewers=cmd_options.split_csv(pr_reviewers) or config.pr_defaults.reviewers,
@@ -251,12 +255,8 @@ def _sync_destination(
     opts: CopyOptions,
     read_log: Callable[[], str],
 ) -> tuple[int, PRRef | None]:
-    dest_root = (src_root / dest.dest_path_relative).resolve()
-
-    if opts.dry_run and not dest_root.exists():
-        raise ValueError(f"Destination repo not found: {dest_root}. Clone it first or run without --dry-run.")
-
-    dest_repo = _ensure_dest_repo(dest, dest_root, opts.dry_run)
+    dest_root = resolve_repo_path(dest, src_root, opts.work_dir)
+    dest_repo = ensure_repo(dest, dest_root, dry_run=opts.dry_run)
     copy_branch = dest.resolved_copy_branch(config.name)
 
     if _skip_already_synced(dest.name, dest_root, copy_branch, commit_ts, opts, config):
@@ -314,16 +314,6 @@ def _print_sync_summary(dest: Destination, result: SyncResult) -> None:
         typer.echo(f"\n{result.total} changes ready.", err=True)
 
 
-def _ensure_dest_repo(dest: Destination, dest_root: Path, dry_run: bool):
-    if not dest_root.exists():
-        if dry_run:
-            raise ValueError(f"Destination repo not found: {dest_root}. Clone it first or run without --dry-run.")
-        if not dest.repo_url:
-            raise ValueError(f"Dest {dest.name} not found and no repo_url configured")
-        git_ops.clone_repo(dest.repo_url, dest_root)
-    return git_ops.get_repo(dest_root)
-
-
 def _sync_paths(
     config: SrcConfig,
     dest: Destination,

path_sync/_internal/cmd_dep_update.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-import shutil
 import subprocess
 from dataclasses import dataclass, field
 from enum import StrEnum
@@ -10,7 +9,7 @@
 import typer
 from git import Repo
 
-from path_sync._internal import cmd_options, git_ops, prompt_utils, verify
+from path_sync._internal import cmd_options, git_ops, verify
 from path_sync._internal.auto_merge import PRRef, handle_auto_merge
 from path_sync._internal.log_capture import capture_log
 from path_sync._internal.models import Destination, OnFailStrategy, find_repo_root
@@ -19,6 +18,7 @@
     UpdateEntry,
     resolve_dep_config_path,
 )
+from path_sync._internal.repo_utils import ensure_repo, resolve_repo_path
 from path_sync._internal.typer_app import app
 from path_sync._internal.verify import StepFailure, VerifyStatus
 from path_sync._internal.yaml_utils import load_yaml_model
@@ -154,8 +154,8 @@ def _process_single_repo_inner(
     opts: DepUpdateOptions,
 ) -> RepoResult:
     logger.info(f"Processing {dest.name}...")
-    repo_path = _resolve_repo_path(dest, src_root, work_dir)
-    repo = _ensure_repo(dest, repo_path, dest.default_branch)
+    repo_path = resolve_repo_path(dest, src_root, work_dir)
+    repo = ensure_repo(dest, repo_path)
     git_ops.prepare_copy_branch(repo, dest.default_branch, config.pr.branch, from_default=True)
 
     if failure := _run_updates(config.updates, repo_path):
@@ -219,29 +219,6 @@ def _create_prs(config: DepConfig, results: list[RepoResult], opts: DepUpdateOpt
     return pr_refs
 
 
-def _resolve_repo_path(dest: Destination, src_root: Path, work_dir: str) -> Path:
-    if work_dir:
-        return Path(work_dir) / dest.name
-    if dest.dest_path_relative:
-        return (src_root / dest.dest_path_relative).resolve()
-    raise typer.BadParameter(f"No dest_path_relative for {dest.name}, --work-dir required")
-
-
-def _ensure_repo(dest: Destination, repo_path: Path, default_branch: str) -> Repo:
-    if repo_path.exists():
-        if git_ops.is_git_repo(repo_path):
-            repo = git_ops.get_repo(repo_path)
-            git_ops.fetch_and_reset_to_default(repo, default_branch)
-            return repo
-        logger.warning(f"Invalid git repo at {repo_path}")
-        if not prompt_utils.prompt_confirm(f"Remove {repo_path} and re-clone?"):
-            raise typer.Abort()
-        shutil.rmtree(repo_path)
-    if not dest.repo_url:
-        raise ValueError(f"Dest {dest.name} not found at {repo_path} and no repo_url configured")
-    return git_ops.clone_repo(dest.repo_url, repo_path)
-
-
 def _build_pr_body(log_content: str, failures: list[StepFailure]) -> str:
     body = "Automated dependency update."
 

path_sync/_internal/cmd_dep_update_test.py

@@ -24,6 +24,7 @@
     PRConfig,
     UpdateEntry,
 )
+from path_sync._internal.repo_utils import ensure_repo
 from path_sync._internal.verify import StepFailure
 
 MODULE = _process_single_repo.__module__
@@ -66,10 +67,9 @@ def test_process_single_repo_no_changes_skips(dest: Destination, config: DepConf
 
     with (
         patch(f"{MODULE}.git_ops") as git_ops,
+        patch(f"{MODULE}.{ensure_repo.__name__}", return_value=mock_repo),
         patch(f"{VERIFY_MODULE}.{verify_module.run_command.__name__}"),
     ):
-        git_ops.get_repo.return_value = mock_repo
-        git_ops.is_git_repo.return_value = True
         git_ops.has_changes.return_value = False
 
         result = _process_single_repo(config, dest, tmp_path, "", opts)
@@ -85,11 +85,10 @@ def test_process_single_repo_update_fails_returns_skipped(
     opts = DepUpdateOptions()
 
     with (
-        patch(f"{MODULE}.git_ops") as git_ops,
+        patch(f"{MODULE}.git_ops"),
+        patch(f"{MODULE}.{ensure_repo.__name__}", return_value=mock_repo),
         patch(f"{VERIFY_MODULE}.{verify_module.run_command.__name__}") as run_cmd,
     ):
-        git_ops.get_repo.return_value = mock_repo
-        git_ops.is_git_repo.return_value = True
         run_cmd.side_effect = subprocess.CalledProcessError(1, "uv lock")
 
         result = _process_single_repo(config, dest, tmp_path, "", opts)
@@ -108,10 +107,9 @@ def test_process_single_repo_changes_with_skip_verify_passes(
 
     with (
         patch(f"{MODULE}.git_ops") as git_ops,
+        patch(f"{MODULE}.{ensure_repo.__name__}", return_value=mock_repo),
         patch(f"{VERIFY_MODULE}.{verify_module.run_command.__name__}"),
     ):
-        git_ops.get_repo.return_value = mock_repo
-        git_ops.is_git_repo.return_value = True
         git_ops.has_changes.return_value = True
 
         result = _process_single_repo(config, dest, tmp_path, "", opts)
@@ -133,10 +131,9 @@ def test_process_single_repo_verify_runs_when_changes_present(dest: Destination,
 
     with (
         patch(f"{MODULE}.git_ops") as git_ops,
+        patch(f"{MODULE}.{ensure_repo.__name__}", return_value=mock_repo),
         patch(f"{VERIFY_MODULE}.{verify_module.run_command.__name__}") as run_cmd,
     ):
-        git_ops.get_repo.return_value = mock_repo
-        git_ops.is_git_repo.return_value = True
         git_ops.has_changes.return_value = True
 
         result = _process_single_repo(config, dest, tmp_path, "", opts)
@@ -190,10 +187,9 @@ def test_update_and_validate_keeps_pr_when_config_flag_set(
 
     with (
         patch(f"{MODULE}.git_ops") as git_ops,
+        patch(f"{MODULE}.{ensure_repo.__name__}", return_value=mock_repo),
         patch(f"{VERIFY_MODULE}.{verify_module.run_command.__name__}"),
     ):
-        git_ops.get_repo.return_value = mock_repo
-        git_ops.is_git_repo.return_value = True
         git_ops.has_changes.return_value = False
 
         results = _update_and_validate(config, [dest], tmp_path, "", opts)

path_sync/_internal/repo_utils.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import logging
+import shutil
+from pathlib import Path
+
+from git import Repo
+
+from path_sync._internal import git_ops, prompt_utils
+from path_sync._internal.models import Destination
+
+logger = logging.getLogger(__name__)
+
+
+def resolve_repo_path(dest: Destination, src_root: Path, work_dir: str) -> Path:
+    if work_dir:
+        return Path(work_dir) / dest.name
+    if dest.dest_path_relative:
+        return (src_root / dest.dest_path_relative).resolve()
+    raise ValueError(f"No dest_path_relative for {dest.name}, use --work-dir")
+
+
+def ensure_repo(dest: Destination, repo_path: Path, dry_run: bool = False) -> Repo:
+    if repo_path.exists():
+        if git_ops.is_git_repo(repo_path):
+            return git_ops.get_repo(repo_path)
+        logger.warning(f"Invalid git repo at {repo_path}")
+        if dry_run:
+            raise ValueError(f"Invalid git repo at {repo_path}, cannot re-clone in dry-run mode")
+        if not prompt_utils.prompt_confirm(f"Remove {repo_path} and re-clone?"):
+            raise ValueError(f"Aborted: user declined to remove invalid repo at {repo_path}")
+        shutil.rmtree(repo_path)
+    if dry_run:
+        raise ValueError(f"Destination repo not found: {repo_path}. Clone it first or run without --dry-run.")
+    if not dest.repo_url:
+        raise ValueError(f"Dest {dest.name} not found at {repo_path} and no repo_url configured")
+    return git_ops.clone_repo(dest.repo_url, repo_path)

path_sync/_internal/repo_utils_test.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from path_sync._internal.models import Destination
+from path_sync._internal.repo_utils import ensure_repo, resolve_repo_path
+
+MODULE = ensure_repo.__module__
+
+
+@pytest.fixture
+def dest() -> Destination:
+    return Destination(name="my-repo", dest_path_relative="code/my-repo", repo_url="https://github.com/org/my-repo")
+
+
+def test_resolve_repo_path_work_dir(dest: Destination):
+    result = resolve_repo_path(dest, Path("/src"), "/tmp/work")
+    assert result == Path("/tmp/work/my-repo")
+
+
+def test_resolve_repo_path_relative(dest: Destination):
+    result = resolve_repo_path(dest, Path("/src"), "")
+    assert result == Path("/src/code/my-repo").resolve()
+
+
+def test_resolve_repo_path_no_relative_no_workdir():
+    dest = Destination(name="x", dest_path_relative="")
+    with pytest.raises(ValueError, match="--work-dir"):
+        resolve_repo_path(dest, Path("/src"), "")
+
+
+def test_ensure_repo_clones_when_missing(dest: Destination, tmp_path: Path):
+    repo_path = tmp_path / "missing"
+    with patch(f"{MODULE}.git_ops") as git_ops:
+        mock_repo = MagicMock()
+        git_ops.clone_repo.return_value = mock_repo
+        result = ensure_repo(dest, repo_path)
+        assert result is mock_repo
+        git_ops.clone_repo.assert_called_once_with(dest.repo_url, repo_path)
+
+
+def test_ensure_repo_dry_run_raises_when_missing(dest: Destination, tmp_path: Path):
+    repo_path = tmp_path / "missing"
+    with pytest.raises(ValueError, match="dry-run"):
+        ensure_repo(dest, repo_path, dry_run=True)
+
+
+def test_ensure_repo_returns_existing(dest: Destination, tmp_path: Path):
+    repo_path = tmp_path / "existing"
+    repo_path.mkdir()
+    mock_repo = MagicMock()
+
+    with patch(f"{MODULE}.git_ops") as git_ops:
+        git_ops.is_git_repo.return_value = True
+        git_ops.get_repo.return_value = mock_repo
+        result = ensure_repo(dest, repo_path)
+        assert result is mock_repo
+        git_ops.get_repo.assert_called_once_with(repo_path)

path_sync/cmd_copy_test.py

@@ -9,7 +9,6 @@
     CopyOptions,
     _cleanup_orphans,
     _close_stale_pr,
-    _ensure_dest_repo,
     _skip_already_synced,
     _sync_path,
 )
@@ -24,6 +23,7 @@
     VerifyConfig,
     VerifyStep,
 )
+from path_sync._internal.repo_utils import ensure_repo
 from path_sync._internal.verify import VerifyStatus, run_verify_steps
 
 CONFIG_NAME = "test-config"
@@ -172,11 +172,11 @@ def test_sync_with_sections_skip(tmp_path):
     assert "keep this" in (dest_root / "file.sh").read_text()
 
 
-def test_ensure_dest_repo_dry_run_errors_if_missing(tmp_path):
+def test_ensure_repo_dry_run_errors_if_missing(tmp_path):
     dest = _make_dest()
     dest_root = tmp_path / "missing_repo"
-    with pytest.raises(ValueError, match="Destination repo not found"):
-        _ensure_dest_repo(dest, dest_root, dry_run=True)
+    with pytest.raises(ValueError, match="dry-run"):
+        ensure_repo(dest, dest_root, dry_run=True)
 
 
 def test_copy_options_defaults():