diff --git a/src/vcspull/cli/add.py b/src/vcspull/cli/add.py index f79b5397..95f7ff0d 100644 --- a/src/vcspull/cli/add.py +++ b/src/vcspull/cli/add.py @@ -17,7 +17,6 @@ expand_dir, find_home_config_files, merge_duplicate_workspace_roots, - normalize_workspace_roots, save_config_yaml, save_config_yaml_with_items, workspace_root_label, @@ -120,7 +119,8 @@ def _resolve_workspace_path( if workspace_root: return canonicalize_workspace_path(workspace_root, cwd=cwd) if repo_path_str: - return expand_dir(pathlib.Path(repo_path_str), cwd) + repo_path = expand_dir(pathlib.Path(repo_path_str), cwd) + return repo_path.parent return cwd @@ -165,6 +165,50 @@ def _normalize_detected_url(remote: str | None) -> tuple[str, str]: return display_url, config_url +def _build_ordered_items( + top_level_items: list[tuple[str, t.Any]] | None, + raw_config: dict[str, t.Any], +) -> list[dict[str, t.Any]]: + """Return deep-copied top-level items preserving original ordering.""" + source: list[tuple[str, t.Any]] = top_level_items or list(raw_config.items()) + + ordered: list[dict[str, t.Any]] = [] + for label, section in source: + ordered.append({"label": label, "section": copy.deepcopy(section)}) + return ordered + + +def _aggregate_from_ordered_items( + items: list[dict[str, t.Any]], +) -> dict[str, t.Any]: + """Collapse ordered top-level items into a mapping grouped by label.""" + aggregated: dict[str, t.Any] = {} + for entry in items: + label = entry["label"] + section = entry["section"] + if isinstance(section, dict): + workspace_section = aggregated.setdefault(label, {}) + for repo_name, repo_config in section.items(): + workspace_section[repo_name] = copy.deepcopy(repo_config) + else: + aggregated[label] = copy.deepcopy(section) + return aggregated + + +def _collect_duplicate_sections( + items: list[dict[str, t.Any]], +) -> dict[str, list[t.Any]]: + """Return mapping of labels to their repeated sections (>= 2 occurrences).""" + occurrences: dict[str, list[t.Any]] = {} + for entry in items: + label = entry["label"] + occurrences.setdefault(label, []).append(copy.deepcopy(entry["section"])) + + return { + label: sections for label, sections in occurrences.items() if len(sections) > 1 + } + + def handle_add_command(args: argparse.Namespace) -> None: """Entry point for the ``vcspull add`` CLI command.""" repo_input = getattr(args, "repo_path", None) @@ -369,26 +413,102 @@ def add_repo( display_config_path, ) - config_items: list[tuple[str, t.Any]] = ( - [(label, copy.deepcopy(section)) for label, section in top_level_items] - if top_level_items - else [(label, copy.deepcopy(section)) for label, section in raw_config.items()] + cwd = pathlib.Path.cwd() + home = pathlib.Path.home() + + workspace_path = _resolve_workspace_path( + workspace_root_path, + path, + cwd=cwd, ) - def _aggregate_items(items: list[tuple[str, t.Any]]) -> dict[str, t.Any]: - aggregated: dict[str, t.Any] = {} - for label, section in items: - if isinstance(section, dict): - workspace_section = aggregated.setdefault(label, {}) - for repo_name, repo_config in section.items(): - workspace_section[repo_name] = copy.deepcopy(repo_config) - else: - aggregated[label] = copy.deepcopy(section) - return aggregated + explicit_dot = workspace_root_path in {".", "./"} + + preferred_label = workspace_root_label( + workspace_path, + cwd=cwd, + home=home, + preserve_cwd_label=explicit_dot, + ) - if not merge_duplicates: - raw_config = _aggregate_items(config_items) + new_repo_entry = {"repo": url} + def _ensure_workspace_label_for_merge( + config_data: dict[str, t.Any], + ) -> tuple[str, bool]: + workspace_map: dict[pathlib.Path, str] = {} + for label, section in config_data.items(): + if not isinstance(section, dict): + continue + try: + path_key = canonicalize_workspace_path(label, cwd=cwd) + except Exception: + continue + workspace_map[path_key] = label + + existing_label = workspace_map.get(workspace_path) + relabelled = False + + if explicit_dot: + workspace_label = "./" + if existing_label and existing_label != "./": + config_data["./"] = config_data.pop(existing_label) + relabelled = True + else: + config_data.setdefault("./", {}) + else: + if existing_label is None: + workspace_label = preferred_label + config_data.setdefault(workspace_label, {}) + else: + workspace_label = existing_label + + if workspace_label not in config_data: + config_data[workspace_label] = {} + + return workspace_label, relabelled + + def _prepare_no_merge_items( + items: list[dict[str, t.Any]], + ) -> tuple[str, int, bool]: + matching_indexes: list[int] = [] + for idx, entry in enumerate(items): + section = entry["section"] + if not isinstance(section, dict): + continue + try: + path_key = canonicalize_workspace_path(entry["label"], cwd=cwd) + except Exception: + continue + if path_key == workspace_path: + matching_indexes.append(idx) + + relabelled = False + + if explicit_dot: + if matching_indexes: + for idx in matching_indexes: + if items[idx]["label"] != "./": + items[idx]["label"] = "./" + relabelled = True + target_index = matching_indexes[-1] + else: + items.append({"label": "./", "section": {}}) + target_index = len(items) - 1 + workspace_label = items[target_index]["label"] + return workspace_label, target_index, relabelled + + if not matching_indexes: + workspace_label = preferred_label + items.append({"label": workspace_label, "section": {}}) + target_index = len(items) - 1 + return workspace_label, target_index, relabelled + + target_index = matching_indexes[-1] + workspace_label = items[target_index]["label"] + return workspace_label, target_index, relabelled + + config_was_relabelled = False duplicate_merge_conflicts: list[str] = [] duplicate_merge_changes = 0 duplicate_merge_details: list[tuple[str, int]] = [] @@ -417,109 +537,146 @@ def _aggregate_items(items: list[tuple[str, t.Any]]) -> dict[str, t.Any]: label, Style.RESET_ALL, ) - else: - if duplicate_root_occurrences: - duplicate_merge_details = [ - (label, len(values)) - for label, values in duplicate_root_occurrences.items() - ] - for label, occurrence_count in duplicate_merge_details: - log.warning( - "%s•%s Duplicate workspace root %s%s%s appears %s%d%s time%s; " - "skipping merge because --no-merge was provided.", - Fore.BLUE, - Style.RESET_ALL, - Fore.MAGENTA, - label, - Style.RESET_ALL, + + workspace_label, relabelled = _ensure_workspace_label_for_merge(raw_config) + config_was_relabelled = relabelled + workspace_section = raw_config.get(workspace_label) + if not isinstance(workspace_section, dict): + log.error( + "Workspace root '%s' in configuration is not a dictionary. Aborting.", + workspace_label, + ) + return + + existing_config = workspace_section.get(name) + if existing_config is not None: + if isinstance(existing_config, str): + current_url = existing_config + elif isinstance(existing_config, dict): + repo_value = existing_config.get("repo") + url_value = existing_config.get("url") + current_url = repo_value or url_value or "unknown" + else: + current_url = str(existing_config) + + log.warning( + "Repository '%s' already exists under '%s'. Current URL: %s. " + "To update, remove and re-add, or edit the YAML file manually.", + name, + workspace_label, + current_url, + ) + + if (duplicate_merge_changes > 0 or config_was_relabelled) and not dry_run: + try: + save_config_yaml(config_file_path, raw_config) + log.info( + "%s✓%s Workspace label adjustments saved to %s%s%s.", + Fore.GREEN, + Style.RESET_ALL, + Fore.BLUE, + display_config_path, + Style.RESET_ALL, + ) + except Exception: + log.exception("Error saving config to %s", config_file_path) + if log.isEnabledFor(logging.DEBUG): + traceback.print_exc() + elif (duplicate_merge_changes > 0 or config_was_relabelled) and dry_run: + log.info( + "%s→%s Would save workspace label adjustments to %s%s%s.", Fore.YELLOW, - occurrence_count, Style.RESET_ALL, - "" if occurrence_count == 1 else "s", + Fore.BLUE, + display_config_path, + Style.RESET_ALL, ) + return - duplicate_merge_conflicts = [] - - cwd = pathlib.Path.cwd() - home = pathlib.Path.home() + workspace_section[name] = copy.deepcopy(new_repo_entry) - aggregated_config = ( - raw_config if merge_duplicates else _aggregate_items(config_items) - ) + if dry_run: + log.info( + "%s→%s Would add %s'%s'%s (%s%s%s) to %s%s%s under '%s%s%s'.", + Fore.YELLOW, + Style.RESET_ALL, + Fore.CYAN, + name, + Style.RESET_ALL, + Fore.YELLOW, + url, + Style.RESET_ALL, + Fore.BLUE, + display_config_path, + Style.RESET_ALL, + Fore.MAGENTA, + workspace_label, + Style.RESET_ALL, + ) + return - if merge_duplicates: - ( - raw_config, - workspace_map, - merge_conflicts, - merge_changes, - ) = normalize_workspace_roots( - aggregated_config, - cwd=cwd, - home=home, - ) - config_was_normalized = (merge_changes + duplicate_merge_changes) > 0 - else: - ( - _normalized_preview, - workspace_map, - merge_conflicts, - _merge_changes, - ) = normalize_workspace_roots( - aggregated_config, - cwd=cwd, - home=home, - ) - config_was_normalized = False + try: + save_config_yaml(config_file_path, raw_config) + log.info( + "%s✓%s Successfully added %s'%s'%s (%s%s%s) to %s%s%s under '%s%s%s'.", + Fore.GREEN, + Style.RESET_ALL, + Fore.CYAN, + name, + Style.RESET_ALL, + Fore.YELLOW, + url, + Style.RESET_ALL, + Fore.BLUE, + display_config_path, + Style.RESET_ALL, + Fore.MAGENTA, + workspace_label, + Style.RESET_ALL, + ) + except Exception: + log.exception("Error saving config to %s", config_file_path) + if log.isEnabledFor(logging.DEBUG): + traceback.print_exc() + return - for message in merge_conflicts: - log.warning(message) + ordered_items = _build_ordered_items(top_level_items, raw_config) - workspace_path = _resolve_workspace_path( - workspace_root_path, - path, - cwd=cwd, - ) - workspace_label = workspace_map.get(workspace_path) + workspace_label, target_index, relabelled = _prepare_no_merge_items(ordered_items) + config_was_relabelled = relabelled - if workspace_root_path is None: - preserve_workspace_label = path is None - else: - preserve_workspace_label = workspace_root_path in {".", "./"} - - if workspace_label is None: - workspace_label = workspace_root_label( - workspace_path, - cwd=cwd, - home=home, - preserve_cwd_label=preserve_workspace_label, + duplicate_sections = _collect_duplicate_sections(ordered_items) + for label, sections in duplicate_sections.items(): + occurrence_count = len(sections) + log.warning( + "%s•%s Duplicate workspace root %s%s%s appears %s%d%s time%s; " + "skipping merge because --no-merge was provided.", + Fore.BLUE, + Style.RESET_ALL, + Fore.MAGENTA, + label, + Style.RESET_ALL, + Fore.YELLOW, + occurrence_count, + Style.RESET_ALL, + "" if occurrence_count == 1 else "s", ) - workspace_map[workspace_path] = workspace_label - raw_config.setdefault(workspace_label, {}) - if not merge_duplicates: - config_items.append((workspace_label, {})) - - if workspace_label not in raw_config: - raw_config[workspace_label] = {} - if not merge_duplicates: - config_items.append((workspace_label, {})) - elif not isinstance(raw_config[workspace_label], dict): + + raw_config_view = _aggregate_from_ordered_items(ordered_items) + workspace_section_view = raw_config_view.get(workspace_label) + if workspace_section_view is None: + workspace_section_view = {} + raw_config_view[workspace_label] = workspace_section_view + + if not isinstance(workspace_section_view, dict): log.error( "Workspace root '%s' in configuration is not a dictionary. Aborting.", workspace_label, ) return - workspace_sections: list[tuple[int, dict[str, t.Any]]] = [ - (idx, section) - for idx, (label, section) in enumerate(config_items) - if label == workspace_label and isinstance(section, dict) - ] - - # Check if repo already exists - if name in raw_config[workspace_label]: - existing_config = raw_config[workspace_label][name] - # Handle both string and dict formats - current_url: str + + existing_config = workspace_section_view.get(name) + if existing_config is not None: if isinstance(existing_config, str): current_url = existing_config elif isinstance(existing_config, dict): @@ -536,10 +693,11 @@ def _aggregate_items(items: list[tuple[str, t.Any]]) -> dict[str, t.Any]: workspace_label, current_url, ) - if config_was_normalized: + + if config_was_relabelled: if dry_run: log.info( - "%s→%s Would save normalized workspace roots to %s%s%s.", + "%s→%s Would save workspace label adjustments to %s%s%s.", Fore.YELLOW, Style.RESET_ALL, Fore.BLUE, @@ -548,9 +706,12 @@ def _aggregate_items(items: list[tuple[str, t.Any]]) -> dict[str, t.Any]: ) else: try: - save_config_yaml(config_file_path, raw_config) + save_config_yaml_with_items( + config_file_path, + [(entry["label"], entry["section"]) for entry in ordered_items], + ) log.info( - "%s✓%s Normalized workspace roots saved to %s%s%s.", + "%s✓%s Workspace label adjustments saved to %s%s%s.", Fore.GREEN, Style.RESET_ALL, Fore.BLUE, @@ -563,21 +724,17 @@ def _aggregate_items(items: list[tuple[str, t.Any]]) -> dict[str, t.Any]: traceback.print_exc() return - # Add the repository in verbose format - new_repo_entry = {"repo": url} - if merge_duplicates: - raw_config[workspace_label][name] = new_repo_entry - else: - target_section: dict[str, t.Any] - if workspace_sections: - _, target_section = workspace_sections[-1] - else: - target_section = {} - config_items.append((workspace_label, target_section)) - target_section[name] = copy.deepcopy(new_repo_entry) - raw_config[workspace_label][name] = copy.deepcopy(new_repo_entry) + target_section = ordered_items[target_index]["section"] + if not isinstance(target_section, dict): + log.error( + "Workspace root '%s' in configuration is not a dictionary. Aborting.", + ordered_items[target_index]["label"], + ) + return + + target_section[name] = copy.deepcopy(new_repo_entry) + workspace_section_view[name] = copy.deepcopy(new_repo_entry) - # Save or preview config if dry_run: log.info( "%s→%s Would add %s'%s'%s (%s%s%s) to %s%s%s under '%s%s%s'.", @@ -596,31 +753,31 @@ def _aggregate_items(items: list[tuple[str, t.Any]]) -> dict[str, t.Any]: workspace_label, Style.RESET_ALL, ) - else: - try: - if merge_duplicates: - save_config_yaml(config_file_path, raw_config) - else: - save_config_yaml_with_items(config_file_path, config_items) - log.info( - "%s✓%s Successfully added %s'%s'%s (%s%s%s) to %s%s%s under '%s%s%s'.", - Fore.GREEN, - Style.RESET_ALL, - Fore.CYAN, - name, - Style.RESET_ALL, - Fore.YELLOW, - url, - Style.RESET_ALL, - Fore.BLUE, - display_config_path, - Style.RESET_ALL, - Fore.MAGENTA, - workspace_label, - Style.RESET_ALL, - ) - except Exception: - log.exception("Error saving config to %s", config_file_path) - if log.isEnabledFor(logging.DEBUG): - traceback.print_exc() - return + return + + try: + save_config_yaml_with_items( + config_file_path, + [(entry["label"], entry["section"]) for entry in ordered_items], + ) + log.info( + "%s✓%s Successfully added %s'%s'%s (%s%s%s) to %s%s%s under '%s%s%s'.", + Fore.GREEN, + Style.RESET_ALL, + Fore.CYAN, + name, + Style.RESET_ALL, + Fore.YELLOW, + url, + Style.RESET_ALL, + Fore.BLUE, + display_config_path, + Style.RESET_ALL, + Fore.MAGENTA, + workspace_label, + Style.RESET_ALL, + ) + except Exception: + log.exception("Error saving config to %s", config_file_path) + if log.isEnabledFor(logging.DEBUG): + traceback.print_exc() diff --git a/src/vcspull/config.py b/src/vcspull/config.py index 1c06e079..25ab97f6 100644 --- a/src/vcspull/config.py +++ b/src/vcspull/config.py @@ -597,6 +597,9 @@ def workspace_root_label( if preserve_cwd_label and workspace_path == cwd: return "./" + if workspace_path == home: + return "~/" + try: relative_to_home = workspace_path.relative_to(home) label = f"~/{relative_to_home.as_posix()}" diff --git a/tests/cli/__snapshots__/test_add.ambr b/tests/cli/__snapshots__/test_add.ambr index e642f1f6..04525835 100644 --- a/tests/cli/__snapshots__/test_add.ambr +++ b/tests/cli/__snapshots__/test_add.ambr @@ -49,6 +49,36 @@ 'test_id': 'path-dry-run-shows-tilde-config', }) # --- +# name: test_handle_add_command_path_mode[path-explicit-dot-workspace-no-merge] + dict({ + 'log': ''' + INFO vcspull.cli.add:add.py: Found new repository to import: + INFO vcspull.cli.add:add.py: + pytest-docker (https://github.com/example/dot-nomerge) + INFO vcspull.cli.add:add.py: • workspace: ./ + INFO vcspull.cli.add:add.py: ↳ path: ~/study/python/pytest-docker + INFO vcspull.cli.add:add.py: ? Import this repository? [y/N]: y (auto-confirm) + INFO vcspull.cli.add:add.py: Config file ~/.vcspull.yaml not found. A new one will be created. + INFO vcspull.cli.add:add.py: ✓ Successfully added 'pytest-docker' (git+https://github.com/example/dot-nomerge) to ~/.vcspull.yaml under './'. + + ''', + 'test_id': 'path-explicit-dot-workspace-no-merge', + }) +# --- +# name: test_handle_add_command_path_mode[path-explicit-dot-workspace] + dict({ + 'log': ''' + INFO vcspull.cli.add:add.py: Found new repository to import: + INFO vcspull.cli.add:add.py: + pytest-docker (https://github.com/example/dot) + INFO vcspull.cli.add:add.py: • workspace: ./ + INFO vcspull.cli.add:add.py: ↳ path: ~/study/python/pytest-docker + INFO vcspull.cli.add:add.py: ? Import this repository? [y/N]: y (auto-confirm) + INFO vcspull.cli.add:add.py: Config file ~/.vcspull.yaml not found. A new one will be created. + INFO vcspull.cli.add:add.py: ✓ Successfully added 'pytest-docker' (git+https://github.com/example/dot) to ~/.vcspull.yaml under './'. + + ''', + 'test_id': 'path-explicit-dot-workspace', + }) +# --- # name: test_handle_add_command_path_mode[path-explicit-url] dict({ 'log': ''' diff --git a/tests/cli/test_add.py b/tests/cli/test_add.py index eddafd0f..2fa5d386 100644 --- a/tests/cli/test_add.py +++ b/tests/cli/test_add.py @@ -55,12 +55,12 @@ def init_git_repo(repo_path: pathlib.Path, remote_url: str | None) -> None: name="myproject", url="git+https://github.com/user/myproject.git", workspace_root=None, - path_relative=None, + path_relative="myproject", dry_run=False, use_default_config=False, preexisting_config=None, expected_in_config={ - "./": { + "~/": { "myproject": {"repo": "git+https://github.com/user/myproject.git"}, }, }, @@ -99,12 +99,12 @@ def init_git_repo(repo_path: pathlib.Path, remote_url: str | None) -> None: name="autoproject", url="git+https://github.com/user/autoproject.git", workspace_root=None, - path_relative=None, + path_relative="autoproject", dry_run=False, use_default_config=True, preexisting_config=None, expected_in_config={ - "./": { + "~/": { "autoproject": { "repo": "git+https://github.com/user/autoproject.git", }, @@ -125,7 +125,7 @@ def init_git_repo(repo_path: pathlib.Path, remote_url: str | None) -> None: use_default_config=False, preexisting_config=None, expected_in_config={ - "~/code/lib/": { + "~/code/": { "lib": {"repo": "git+https://github.com/user/lib.git"}, }, }, @@ -136,7 +136,7 @@ def init_git_repo(repo_path: pathlib.Path, remote_url: str | None) -> None: name="extra", url="git+https://github.com/user/extra.git", workspace_root=None, - path_relative=None, + path_relative="extra", dry_run=False, use_default_config=False, preexisting_config={ @@ -145,10 +145,10 @@ def init_git_repo(repo_path: pathlib.Path, remote_url: str | None) -> None: }, }, expected_in_config={ - "~/code/": { + "~/code": { "existing": {"repo": "git+https://github.com/user/existing.git"}, }, - "./": { + "~/": { "extra": {"repo": "git+https://github.com/user/extra.git"}, }, }, @@ -260,13 +260,15 @@ def test_add_repo_duplicate_warning( monkeypatch.chdir(tmp_path) config_file = tmp_path / ".vcspull.yaml" + repo_path = tmp_path / "myproject" + repo_path.mkdir() # Add repo first time add_repo( name="myproject", url="git+https://github.com/user/myproject.git", config_file_path_str=str(config_file), - path=None, + path=str(repo_path), workspace_root_path=None, dry_run=False, ) @@ -279,7 +281,7 @@ def test_add_repo_duplicate_warning( name="myproject", url="git+https://github.com/user/myproject-v2.git", config_file_path_str=str(config_file), - path=None, + path=str(repo_path), workspace_root_path=None, dry_run=False, ) @@ -299,11 +301,14 @@ def test_add_repo_creates_new_file( config_file = tmp_path / ".vcspull.yaml" assert not config_file.exists() + repo_path = tmp_path / "newrepo" + repo_path.mkdir() + add_repo( name="newrepo", url="git+https://github.com/user/newrepo.git", config_file_path_str=str(config_file), - path=None, + path=str(repo_path), workspace_root_path=None, dry_run=False, ) @@ -315,8 +320,8 @@ def test_add_repo_creates_new_file( with config_file.open() as f: config = yaml.safe_load(f) - assert "./" in config - assert "newrepo" in config["./"] + assert "~/" in config + assert "newrepo" in config["~/"] class AddDuplicateMergeFixture(t.NamedTuple): @@ -598,6 +603,42 @@ class PathAddFixture(t.NamedTuple): expected_workspace_label="~/study/python/", preserve_config_path_in_log=True, ), + PathAddFixture( + test_id="path-explicit-dot-workspace", + remote_url="https://github.com/example/dot", + explicit_url=None, + assume_yes=True, + prompt_response=None, + dry_run=False, + expected_written=True, + expected_url_kind="remote", + override_name=None, + expected_warning=None, + merge_duplicates=True, + preexisting_yaml=None, + use_relative_repo_path=False, + workspace_override="./", + expected_workspace_label="./", + preserve_config_path_in_log=False, + ), + PathAddFixture( + test_id="path-explicit-dot-workspace-no-merge", + remote_url="https://github.com/example/dot-nomerge", + explicit_url=None, + assume_yes=True, + prompt_response=None, + dry_run=False, + expected_written=True, + expected_url_kind="remote", + override_name=None, + expected_warning=None, + merge_duplicates=False, + preexisting_yaml=None, + use_relative_repo_path=False, + workspace_override="./", + expected_workspace_label="./", + preserve_config_path_in_log=False, + ), ] @@ -932,3 +973,102 @@ def test_handle_add_command_workspace_label_from_workspace_root( config_data = yaml.safe_load(fh) assert expected_label in config_data + + +@pytest.mark.parametrize("merge_duplicates", [True, False]) +def test_handle_add_command_workspace_label_variants( + merge_duplicates: bool, + tmp_path: pathlib.Path, + monkeypatch: MonkeyPatch, + caplog: t.Any, +) -> None: + """Path-first adds should keep tilde workspaces regardless of merge flag.""" + caplog.set_level(logging.INFO) + + monkeypatch.setenv("HOME", str(tmp_path)) + + workspace_root = tmp_path / "study/python" + repo_path = workspace_root / "pytest-docker" + init_git_repo(repo_path, remote_url="https://github.com/avast/pytest-docker") + + monkeypatch.chdir(workspace_root) + + config_file = tmp_path / ".vcspull.yaml" + + args = argparse.Namespace( + repo_path=str(repo_path), + url=None, + override_name=None, + config=str(config_file), + workspace_root_path=None, + dry_run=False, + assume_yes=True, + merge_duplicates=merge_duplicates, + ) + + handle_add_command(args) + + expected_label = "~/study/python/" + + assert expected_label in caplog.text + + import yaml + + with config_file.open(encoding="utf-8") as fh: + config_data = yaml.safe_load(fh) or {} + + assert expected_label in config_data + assert "./" not in config_data + + +def test_handle_add_command_preserves_existing_dot_workspace_section( + tmp_path: pathlib.Path, + monkeypatch: MonkeyPatch, + caplog: t.Any, +) -> None: + """Existing './' sections should be preserved when they match the workspace.""" + caplog.set_level(logging.INFO) + + monkeypatch.setenv("HOME", str(tmp_path)) + + workspace_root = tmp_path / "study/python" + repo_path = workspace_root / "pytest-docker" + init_git_repo(repo_path, remote_url="https://github.com/avast/pytest-docker") + + monkeypatch.chdir(workspace_root) + + config_file = tmp_path / ".vcspull.yaml" + import yaml + + config_file.write_text( + yaml.dump( + { + "./": { + "existing": { + "repo": "git+https://github.com/example/existing.git", + }, + }, + }, + ), + encoding="utf-8", + ) + + args = argparse.Namespace( + repo_path=str(repo_path), + url=None, + override_name=None, + config=str(config_file), + workspace_root_path=None, + dry_run=False, + assume_yes=True, + merge_duplicates=True, + ) + + handle_add_command(args) + + with config_file.open(encoding="utf-8") as fh: + config_data = yaml.safe_load(fh) or {} + + assert "./" in config_data + assert "existing" in config_data["./"] + assert "pytest-docker" in config_data["./"]