diff --git a/src/git_pulsar/cli.py b/src/git_pulsar/cli.py index 99551ab..fa1adbd 100644 --- a/src/git_pulsar/cli.py +++ b/src/git_pulsar/cli.py @@ -1,6 +1,5 @@ import argparse import datetime -import fcntl import logging import os import subprocess @@ -368,7 +367,7 @@ def show_status() -> None: # Display global repository count if not currently in a repository. elif REGISTRY_FILE.exists(): - count = len(system.get_registered_repos()) + count = len(system.get_registered_repos(REGISTRY_FILE)) console.print(f"[dim]Watching {count} repositories.[/dim]") @@ -399,13 +398,16 @@ def list_repos() -> None: console.print("[yellow]Registry is empty.[/yellow]") return + repos = system.get_registered_repos(REGISTRY_FILE) + if not repos: + console.print("[yellow]Registry is empty.[/yellow]") + return + table = Table(show_header=True, header_style="bold magenta") table.add_column("Repository", style="cyan") table.add_column("Status") table.add_column("Last Backup", justify="right", style="dim") - repos = system.get_registered_repos() - for path in repos: display_path = str(path).replace(str(Path.home()), "~") @@ -447,39 +449,18 @@ def list_repos() -> None: def unregister_repo() -> None: """Removes the current working directory from the Git Pulsar registry.""" - cwd = str(Path.cwd()) + cwd = Path.cwd() if not REGISTRY_FILE.exists(): console.print("Registry is empty.", style="yellow") return - tmp_file = REGISTRY_FILE.with_suffix(".tmp") - try: - with open(REGISTRY_FILE, "r+") as f: - fcntl.flock(f.fileno(), fcntl.LOCK_EX) - try: - current_paths = [line.strip() for line in f if line.strip()] - if cwd not in current_paths: - console.print( - f"Current path not registered: [cyan]{cwd}[/cyan]", - style="yellow", - ) - return - - with open(tmp_file, "w") as tf: - for path in current_paths: - if path != cwd: - tf.write(f"{path}\n") - tf.flush() - os.fsync(tf.fileno()) - - os.replace(tmp_file, REGISTRY_FILE) - finally: - fcntl.flock(f.fileno(), fcntl.LOCK_UN) + if system.remove_repo_from_registry(cwd, registry_path=REGISTRY_FILE): console.print(f"✔ Unregistered: [cyan]{cwd}[/cyan]", style="green") - except OSError as e: - logger.error(f"Failed to unregister repository: {e}") - if tmp_file.exists(): - tmp_file.unlink() + else: + console.print( + f"Current path not registered: [cyan]{cwd}[/cyan]", + style="yellow", + ) def _check_systemd_linger() -> str | None: @@ -575,13 +556,9 @@ def run_doctor() -> None: ) def clean_registry() -> bool: - try: - with open(REGISTRY_FILE, "w") as f: - f.write("\n".join(valid_lines) + "\n") - return True - except Exception as e: - logger.error(f"Registry cleanup failed: {e}") - return False + return system.write_registered_repos( + valid_lines, registry_path=REGISTRY_FILE + ) actions.append( DoctorAction( @@ -702,10 +679,8 @@ def sync_drift() -> bool: # 1. Check the health of registered repositories (State Check + Hook Interference). is_healthy = True with console.status("[bold blue]Checking Repository Health...", spinner="dots"): - if REGISTRY_FILE.exists(): - with open(REGISTRY_FILE) as f: - paths = [Path(line.strip()) for line in f if line.strip()] - + paths = system.get_registered_repos(REGISTRY_FILE) + if paths: issues = [] for p in paths: if p.exists(): @@ -952,24 +927,10 @@ def setup_repo(registry_path: Path = REGISTRY_FILE) -> None: # Register the repository path. console.print("Registering path...", style="dim") - if not registry_path.exists(): - registry_path.parent.mkdir(parents=True, exist_ok=True) - registry_path.touch() - - with open(registry_path, "r+") as f: - fcntl.flock(f.fileno(), fcntl.LOCK_EX) - try: - content = f.read() - if str(cwd) not in [line.strip() for line in content.splitlines()]: - f.seek(0, os.SEEK_END) - f.write(f"{cwd}\n") - f.flush() - os.fsync(f.fileno()) - console.print(f"Registered: [cyan]{cwd}[/cyan]", style="green") - else: - console.print("Already registered.", style="dim") - finally: - fcntl.flock(f.fileno(), fcntl.LOCK_UN) + if system.add_repo_to_registry(cwd, registry_path=registry_path): + console.print(f"Registered: [cyan]{cwd}[/cyan]", style="green") + else: + console.print("Already registered.", style="dim") console.print("\n[bold green]✔ Pulsar Active.[/bold green]") diff --git a/src/git_pulsar/daemon.py b/src/git_pulsar/daemon.py index 91b50b0..b92e82e 100644 --- a/src/git_pulsar/daemon.py +++ b/src/git_pulsar/daemon.py @@ -1,6 +1,5 @@ import atexit import datetime -import fcntl import logging import os import signal @@ -197,41 +196,11 @@ def prune_registry(original_path_str: str) -> None: Args: original_path_str (str): The path string to remove. """ - if not REGISTRY_FILE.exists(): - return - - target = original_path_str.strip() - tmp_file = REGISTRY_FILE.with_suffix(".tmp") - - try: - # 1. Read existing registry and rewrite atomically while holding exclusive lock. - with open(REGISTRY_FILE, "r+") as f: - fcntl.flock(f.fileno(), fcntl.LOCK_EX) - try: - lines = f.readlines() - # 2. Write valid lines to temp file. - with open(tmp_file, "w") as tf: - for line in lines: - clean_line = line.strip() - if clean_line and clean_line != target: - tf.write(clean_line + "\n") - tf.flush() - os.fsync(tf.fileno()) # Force write to disk. - - # 3. Atomic Swap. - os.replace(tmp_file, REGISTRY_FILE) - finally: - fcntl.flock(f.fileno(), fcntl.LOCK_UN) - + if system.remove_repo_from_registry(original_path_str, registry_path=REGISTRY_FILE): repo_name = Path(original_path_str).name logger.info(f"PRUNED: {original_path_str} removed from registry.") SYSTEM.notify("Backup Stopped", f"Removed missing repo: {repo_name}") - except OSError as e: - logger.error(f"ERROR: Could not prune registry. {e}") - if tmp_file.exists(): - tmp_file.unlink() - def _should_skip(repo_path: Path, config: Config, interactive: bool) -> str | None: """Determines if the backup for a given repository should be skipped. diff --git a/src/git_pulsar/system.py b/src/git_pulsar/system.py index b961807..53f0fdc 100644 --- a/src/git_pulsar/system.py +++ b/src/git_pulsar/system.py @@ -1,3 +1,4 @@ +import contextlib import fcntl import logging import os @@ -5,6 +6,7 @@ import socket import subprocess import sys +from collections.abc import Sequence from pathlib import Path from rich.console import Console @@ -22,12 +24,13 @@ logger = logging.getLogger(APP_NAME) -def get_registered_repos() -> list[Path]: +def get_registered_repos(registry_path: Path | None = None) -> list[Path]: """Reads the registry file and returns a list of registered repository paths.""" - if not REGISTRY_FILE.exists(): + target_path = registry_path if registry_path is not None else REGISTRY_FILE + if not target_path.exists(): return [] try: - with open(REGISTRY_FILE) as f: + with open(target_path) as f: fcntl.flock(f.fileno(), fcntl.LOCK_SH) try: return [Path(line.strip()) for line in f if line.strip()] @@ -38,6 +41,144 @@ def get_registered_repos() -> list[Path]: return [] +def add_repo_to_registry( + repo_path: Path | str, registry_path: Path | None = None +) -> bool: + """Atomically adds a repository path to the registry if not already present. + + Args: + repo_path (Path | str): Path of the repository to register. + registry_path (Path | None): Path to the registry file. Defaults to REGISTRY_FILE. + + Returns: + bool: True if newly added, False if already registered or on failure. + """ + target_path = registry_path if registry_path is not None else REGISTRY_FILE + raw_target = str(repo_path).strip() + try: + target = str(Path(repo_path).resolve()) + except Exception: + target = raw_target + + target_path.parent.mkdir(parents=True, exist_ok=True) + if not target_path.exists(): + target_path.touch() + + try: + with open(target_path, "r+") as f: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + try: + content = f.read() + current_paths = [ + line.strip() for line in content.splitlines() if line.strip() + ] + if target in current_paths or raw_target in current_paths: + return False + + f.seek(0, os.SEEK_END) + f.write(f"{target}\n") + f.flush() + os.fsync(f.fileno()) + return True + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + except OSError as e: + logger.error(f"Failed to add repository to registry: {e}") + return False + + +def remove_repo_from_registry( + repo_path: Path | str, registry_path: Path | None = None +) -> bool: + """Atomically removes a repository path from the registry. + + Args: + repo_path (Path | str): Path of the repository to unregister. + registry_path (Path | None): Path to the registry file. Defaults to REGISTRY_FILE. + + Returns: + bool: True if removed, False if path was not registered or on failure. + """ + target_path = registry_path if registry_path is not None else REGISTRY_FILE + if not target_path.exists(): + return False + + raw_target = str(repo_path).strip() + try: + target = str(Path(repo_path).resolve()) + except Exception: + target = raw_target + + tmp_file = target_path.with_suffix(".tmp") + + try: + with open(target_path, "r+") as f: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + try: + lines = [line.strip() for line in f if line.strip()] + remaining = [p for p in lines if p != target and p != raw_target] + removed = len(remaining) != len(lines) + + with open(tmp_file, "w") as tf: + for p in remaining: + tf.write(f"{p}\n") + tf.flush() + os.fsync(tf.fileno()) + + os.replace(tmp_file, target_path) + return removed + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + except OSError as e: + logger.error(f"Failed to remove repository from registry: {e}") + if tmp_file.exists(): + with contextlib.suppress(OSError): + tmp_file.unlink() + return False + + +def write_registered_repos( + repos: Sequence[Path | str], registry_path: Path | None = None +) -> bool: + """Atomically replaces the registry content with the provided repository paths. + + Args: + repos (Sequence[Path | str]): Repository paths to write. + registry_path (Path | None): Path to the registry file. Defaults to REGISTRY_FILE. + + Returns: + bool: True on success, False on error. + """ + target_path = registry_path if registry_path is not None else REGISTRY_FILE + target_path.parent.mkdir(parents=True, exist_ok=True) + if not target_path.exists(): + target_path.touch() + + tmp_file = target_path.with_suffix(".tmp") + try: + with open(target_path, "r+") as f: + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + try: + with open(tmp_file, "w") as tf: + for repo in repos: + clean = str(repo).strip() + if clean: + tf.write(f"{clean}\n") + tf.flush() + os.fsync(tf.fileno()) + + os.replace(tmp_file, target_path) + return True + finally: + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + except OSError as e: + logger.error(f"Failed to write registry: {e}") + if tmp_file.exists(): + with contextlib.suppress(OSError): + tmp_file.unlink() + return False + + class SystemStrategy: """Base class defining the interface for system-level interactions.""" diff --git a/tests/test_system.py b/tests/test_system.py index 87596fc..7617efb 100644 --- a/tests/test_system.py +++ b/tests/test_system.py @@ -326,3 +326,76 @@ def test_get_registered_repos_missing(tmp_path: Path, mocker: MagicMock) -> None """Verifies get_registered_repos returns empty list if registry file does not exist.""" mocker.patch("git_pulsar.system.REGISTRY_FILE", tmp_path / "nonexistent") assert system.get_registered_repos() == [] + + +def test_add_repo_to_registry(tmp_path: Path) -> None: + """Verifies add_repo_to_registry adds paths and enforces uniqueness.""" + reg_file = tmp_path / "custom_state" / "registry" + repo1 = tmp_path / "repo1" + repo1.mkdir() + repo2 = tmp_path / "repo2" + repo2.mkdir() + + # Add first repo -> returns True + assert system.add_repo_to_registry(repo1, registry_path=reg_file) is True + assert system.get_registered_repos(reg_file) == [repo1.resolve()] + + # Add duplicate repo -> returns False + assert system.add_repo_to_registry(repo1, registry_path=reg_file) is False + assert len(system.get_registered_repos(reg_file)) == 1 + + # Add second repo -> returns True + assert system.add_repo_to_registry(repo2, registry_path=reg_file) is True + assert system.get_registered_repos(reg_file) == [repo1.resolve(), repo2.resolve()] + + +def test_remove_repo_from_registry(tmp_path: Path) -> None: + """Verifies remove_repo_from_registry removes paths atomically and handles missing entries.""" + reg_file = tmp_path / "registry" + repo1 = tmp_path / "repo1" + repo2 = tmp_path / "repo2" + repo3 = tmp_path / "repo3" + for r in (repo1, repo2, repo3): + r.mkdir() + + # Registry does not exist -> returns False + assert system.remove_repo_from_registry(repo1, registry_path=reg_file) is False + + system.add_repo_to_registry(repo1, registry_path=reg_file) + system.add_repo_to_registry(repo2, registry_path=reg_file) + + # Remove non-existent entry -> returns False + assert system.remove_repo_from_registry(repo3, registry_path=reg_file) is False + assert len(system.get_registered_repos(reg_file)) == 2 + + # Remove existing entry -> returns True + assert system.remove_repo_from_registry(repo1, registry_path=reg_file) is True + assert system.get_registered_repos(reg_file) == [repo2.resolve()] + + +def test_write_registered_repos(tmp_path: Path) -> None: + """Verifies write_registered_repos writes list of repos atomically.""" + reg_file = tmp_path / "nested" / "registry" + repo1 = tmp_path / "repo1" + repo2 = tmp_path / "repo2" + + assert system.write_registered_repos([repo1, repo2], registry_path=reg_file) is True + assert system.get_registered_repos(reg_file) == [ + Path(str(repo1)), + Path(str(repo2)), + ] + + +def test_registry_oserror_handling(tmp_path: Path, mocker: MagicMock) -> None: + """Verifies OSError handling in add/remove/write registry functions.""" + reg_file = tmp_path / "registry" + reg_file.touch() + + mocker.patch("builtins.open", side_effect=OSError("Disk read-only")) + assert system.add_repo_to_registry("/some/path", registry_path=reg_file) is False + assert ( + system.remove_repo_from_registry("/some/path", registry_path=reg_file) is False + ) + assert ( + system.write_registered_repos(["/some/path"], registry_path=reg_file) is False + )