Implement git maintenance register and unregister subcommands

The register subcommand adds a repository to the global maintenance.repo
configuration and sets up the incremental maintenance strategy.

dulwich/cli.py

@@ -4363,13 +4363,18 @@ class cmd_maintenance(Command):
         # maintenance stop subcommand (placeholder)
         subparsers.add_parser("stop", help="Stop background maintenance")
 
-        # maintenance register subcommand (placeholder)
+        # maintenance register subcommand
         subparsers.add_parser("register", help="Register repository for maintenance")
 
-        # maintenance unregister subcommand (placeholder)
-        subparsers.add_parser(
+        # maintenance unregister subcommand
+        unregister_parser = subparsers.add_parser(
             "unregister", help="Unregister repository from maintenance"
         )
+        unregister_parser.add_argument(
+            "--force",
+            action="store_true",
+            help="Don't error if repository is not registered",
+        )
 
         parsed_args = parser.parse_args(args)
 
@@ -4408,7 +4413,18 @@ class cmd_maintenance(Command):
             except porcelain.Error as e:
                 logger.error("%s", e)
                 return 1
-        elif parsed_args.subcommand in ("start", "stop", "register", "unregister"):
+        elif parsed_args.subcommand == "register":
+            porcelain.maintenance_register(".")
+            logger.info("Repository registered for background maintenance")
+        elif parsed_args.subcommand == "unregister":
+            try:
+                force = getattr(parsed_args, "force", False)
+                porcelain.maintenance_unregister(".", force=force)
+            except ValueError as e:
+                logger.error(str(e))
+                return 1
+            logger.info("Repository unregistered from background maintenance")
+        elif parsed_args.subcommand in ("start", "stop"):
             # TODO: Implement background maintenance scheduling
             logger.error(
                 f"The '{parsed_args.subcommand}' subcommand is not yet implemented"

dulwich/maintenance.py

@@ -5,13 +5,14 @@ and maintaining Git repositories.
 """
 
 import logging
+import os
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import TYPE_CHECKING, Callable, Optional
 
 if TYPE_CHECKING:
-    from .repo import BaseRepo
+    from .repo import BaseRepo, Repo
 
 logger = logging.getLogger(__name__)
 
@@ -382,3 +383,129 @@ def run_maintenance(
             logger.error(f"Task {task_name} failed: {e}")
 
     return result
+
+
+def register_repository(repo: "Repo") -> None:
+    """Register a repository for background maintenance.
+
+    This adds the repository to the global maintenance.repo config and sets
+    up recommended configuration for scheduled maintenance.
+
+    Args:
+        repo: Repository to register
+    """
+    from .config import ConfigFile
+
+    repo_path = os.path.abspath(repo.path)
+
+    # Get global config path
+    global_config_path = os.path.expanduser("~/.gitconfig")
+    try:
+        global_config = ConfigFile.from_path(global_config_path)
+    except FileNotFoundError:
+        # Create new config file if it doesn't exist
+        global_config = ConfigFile()
+        global_config.path = global_config_path
+
+    # Add repository to maintenance.repo list
+    # Check if already registered
+    repo_path_bytes = repo_path.encode()
+    try:
+        existing_repos = list(global_config.get_multivar((b"maintenance",), b"repo"))
+    except KeyError:
+        existing_repos = []
+
+    if repo_path_bytes in existing_repos:
+        # Already registered
+        return
+
+    # Add to global config
+    global_config.set((b"maintenance",), b"repo", repo_path_bytes)
+
+    # Set up incremental strategy in global config if not already set
+    try:
+        global_config.get((b"maintenance",), b"strategy")
+    except KeyError:
+        global_config.set((b"maintenance",), b"strategy", b"incremental")
+
+    # Configure task schedules for incremental strategy
+    schedule_config = {
+        b"commit-graph": b"hourly",
+        b"prefetch": b"hourly",
+        b"loose-objects": b"daily",
+        b"incremental-repack": b"daily",
+    }
+
+    for task, schedule in schedule_config.items():
+        try:
+            global_config.get((b"maintenance", task), b"schedule")
+        except KeyError:
+            global_config.set((b"maintenance", task), b"schedule", schedule)
+
+    global_config.write_to_path()
+
+    # Disable foreground auto maintenance in the repository
+    repo_config = repo.get_config()
+    repo_config.set((b"maintenance",), b"auto", False)
+    repo_config.write_to_path()
+
+
+def unregister_repository(repo: "Repo", force: bool = False) -> None:
+    """Unregister a repository from background maintenance.
+
+    This removes the repository from the global maintenance.repo config.
+
+    Args:
+        repo: Repository to unregister
+        force: If True, don't error if repository is not registered
+
+    Raises:
+        ValueError: If repository is not registered and force is False
+    """
+    from .config import ConfigFile
+
+    repo_path = os.path.abspath(repo.path)
+
+    # Get global config
+    global_config_path = os.path.expanduser("~/.gitconfig")
+    try:
+        global_config = ConfigFile.from_path(global_config_path)
+    except FileNotFoundError:
+        if not force:
+            raise ValueError(
+                f"Repository {repo_path} is not registered for maintenance"
+            )
+        return
+
+    # Check if repository is registered
+    repo_path_bytes = repo_path.encode()
+    try:
+        existing_repos = list(global_config.get_multivar((b"maintenance",), b"repo"))
+    except KeyError:
+        if not force:
+            raise ValueError(
+                f"Repository {repo_path} is not registered for maintenance"
+            )
+        return
+
+    if repo_path_bytes not in existing_repos:
+        if not force:
+            raise ValueError(
+                f"Repository {repo_path} is not registered for maintenance"
+            )
+        return
+
+    # Remove from list
+    existing_repos.remove(repo_path_bytes)
+
+    # Delete the maintenance section and recreate it with remaining repos
+    try:
+        del global_config[(b"maintenance",)]
+    except KeyError:
+        pass
+
+    # Re-add remaining repos
+    for remaining_repo in existing_repos:
+        global_config.set((b"maintenance",), b"repo", remaining_repo)
+
+    global_config.write_to_path()

dulwich/porcelain.py

@@ -6411,6 +6411,36 @@ def maintenance_run(
         return run_maintenance(r, tasks=tasks, auto=auto, progress=progress)
 
 
+def maintenance_register(repo: RepoPath) -> None:
+    """Register a repository for background maintenance.
+
+    This adds the repository to the global maintenance.repo config and sets
+    up recommended configuration for scheduled maintenance.
+
+    Args:
+      repo: Path to the repository or repository object
+    """
+    from .maintenance import register_repository
+
+    with open_repo_closing(repo) as r:
+        register_repository(r)
+
+
+def maintenance_unregister(repo: RepoPath, force: bool = False) -> None:
+    """Unregister a repository from background maintenance.
+
+    This removes the repository from the global maintenance.repo config.
+
+    Args:
+      repo: Path to the repository or repository object
+      force: If True, don't error if repository is not registered
+    """
+    from .maintenance import unregister_repository
+
+    with open_repo_closing(repo) as r:
+        unregister_repository(r, force=force)
+
+
 def count_objects(repo: RepoPath = ".", verbose: bool = False) -> CountObjectsResult:
     """Count unpacked objects and their disk usage.
 

tests/test_cli.py

@@ -34,14 +34,14 @@ from unittest.mock import MagicMock, patch
 
 from dulwich import cli
 from dulwich.cli import (
+    AutoFlushBinaryIOWrapper,
+    AutoFlushTextIOWrapper,
+    _should_auto_flush,
     detect_terminal_width,
     format_bytes,
     launch_editor,
     parse_relative_time,
     write_columns,
-    _should_auto_flush,
-    AutoFlushTextIOWrapper,
-    AutoFlushBinaryIOWrapper,
 )
 from dulwich.repo import Repo
 from dulwich.tests.utils import (
@@ -3778,6 +3778,13 @@ class GitFlushTest(TestCase):
 class MaintenanceCommandTest(DulwichCliTestCase):
     """Tests for maintenance command."""
 
+    def setUp(self):
+        super().setUp()
+        # Set up a temporary HOME for testing global config
+        self.temp_home = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.temp_home)
+        self.overrideEnv("HOME", self.temp_home)
+
     def test_maintenance_run_default(self):
         """Test maintenance run with default tasks."""
         result, _stdout, _stderr = self._run_cli("maintenance", "run")
@@ -3812,9 +3819,33 @@ class MaintenanceCommandTest(DulwichCliTestCase):
         result, _stdout, _stderr = self._run_cli("maintenance")
         self.assertEqual(result, 1)
 
+    def test_maintenance_register(self):
+        """Test maintenance register subcommand."""
+        result, _stdout, _stderr = self._run_cli("maintenance", "register")
+        self.assertIsNone(result)
+
+    def test_maintenance_unregister(self):
+        """Test maintenance unregister subcommand."""
+        # First register
+        _result, _stdout, _stderr = self._run_cli("maintenance", "register")
+
+        # Then unregister
+        result, _stdout, _stderr = self._run_cli("maintenance", "unregister")
+        self.assertIsNone(result)
+
+    def test_maintenance_unregister_not_registered(self):
+        """Test unregistering a repository that is not registered."""
+        result, _stdout, _stderr = self._run_cli("maintenance", "unregister")
+        self.assertEqual(result, 1)
+
+    def test_maintenance_unregister_force(self):
+        """Test unregistering with --force flag."""
+        result, _stdout, _stderr = self._run_cli("maintenance", "unregister", "--force")
+        self.assertIsNone(result)
+
     def test_maintenance_unimplemented_subcommand(self):
         """Test unimplemented maintenance subcommands."""
-        for subcommand in ["start", "stop", "register", "unregister"]:
+        for subcommand in ["start", "stop"]:
             result, _stdout, _stderr = self._run_cli("maintenance", subcommand)
             self.assertEqual(result, 1)
 

tests/test_maintenance.py

@@ -233,3 +233,82 @@ class PorcelainMaintenanceTest(MaintenanceTaskTestCase):
         result = porcelain.maintenance_run(self.test_dir, tasks=["pack-refs"])
         self.assertEqual(result.tasks_run, ["pack-refs"])
         self.assertEqual(result.tasks_succeeded, ["pack-refs"])
+
+
+class MaintenanceRegisterTest(MaintenanceTaskTestCase):
+    """Tests for maintenance register/unregister."""
+
+    def setUp(self):
+        super().setUp()
+        # Set up a temporary HOME for testing global config
+        self.temp_home = tempfile.mkdtemp()
+        self.addCleanup(self._cleanup_temp_home)
+        self.overrideEnv("HOME", self.temp_home)
+
+    def _cleanup_temp_home(self):
+        import shutil
+
+        shutil.rmtree(self.temp_home)
+
+    def test_register_repository(self):
+        """Test registering a repository for maintenance."""
+        porcelain.maintenance_register(self.test_dir)
+
+        # Verify repository was added to global config
+        import os
+
+        from dulwich.config import ConfigFile
+
+        global_config_path = os.path.expanduser("~/.gitconfig")
+        global_config = ConfigFile.from_path(global_config_path)
+
+        repos = list(global_config.get_multivar((b"maintenance",), b"repo"))
+        self.assertIn(self.test_dir.encode(), repos)
+
+        # Verify strategy was set
+        strategy = global_config.get((b"maintenance",), b"strategy")
+        self.assertEqual(strategy, b"incremental")
+
+        # Verify auto maintenance was disabled in repo
+        repo_config = self.repo.get_config()
+        auto = repo_config.get_boolean((b"maintenance",), b"auto")
+        self.assertFalse(auto)
+
+    def test_register_already_registered(self):
+        """Test registering an already registered repository."""
+        porcelain.maintenance_register(self.test_dir)
+        # Should not error when registering again
+        porcelain.maintenance_register(self.test_dir)
+
+    def test_unregister_repository(self):
+        """Test unregistering a repository."""
+        # First register
+        porcelain.maintenance_register(self.test_dir)
+
+        # Then unregister
+        porcelain.maintenance_unregister(self.test_dir)
+
+        # Verify repository was removed from global config
+        import os
+
+        from dulwich.config import ConfigFile
+
+        global_config_path = os.path.expanduser("~/.gitconfig")
+        global_config = ConfigFile.from_path(global_config_path)
+
+        try:
+            repos = list(global_config.get_multivar((b"maintenance",), b"repo"))
+            self.assertNotIn(self.test_dir.encode(), repos)
+        except KeyError:
+            # No repos registered, which is fine
+            pass
+
+    def test_unregister_not_registered(self):
+        """Test unregistering a repository that is not registered."""
+        with self.assertRaises(ValueError):
+            porcelain.maintenance_unregister(self.test_dir)
+
+    def test_unregister_not_registered_force(self):
+        """Test unregistering with force flag."""
+        # Should not error with force=True
+        porcelain.maintenance_unregister(self.test_dir, force=True)