Jelajahi Sumber

Add prune cli command

Jelmer Vernooij 1 bulan lalu
induk
melakukan
a4ee7bcdbc
3 mengubah file dengan 81 tambahan dan 18 penghapusan
  1. 63 0
      dulwich/cli.py
  2. 2 2
      dulwich/porcelain.py
  3. 16 16
      tests/test_porcelain.py

+ 63 - 0
dulwich/cli.py

@@ -761,6 +761,68 @@ class cmd_unpack_objects(Command):
         print(f"Unpacked {count} objects")
 
 
+class cmd_prune(Command):
+    def run(self, args) -> Optional[int]:
+        import datetime
+        import time
+
+        from dulwich.object_store import DEFAULT_TEMPFILE_GRACE_PERIOD
+
+        parser = argparse.ArgumentParser(
+            description="Remove temporary pack files left behind by interrupted operations"
+        )
+        parser.add_argument(
+            "--expire",
+            nargs="?",
+            const="2.weeks.ago",
+            help="Only prune files older than the specified date (default: 2.weeks.ago)",
+        )
+        parser.add_argument(
+            "--dry-run",
+            "-n",
+            action="store_true",
+            help="Only report what would be removed",
+        )
+        parser.add_argument(
+            "--verbose",
+            "-v",
+            action="store_true",
+            help="Report all actions",
+        )
+        args = parser.parse_args(args)
+
+        # Parse expire grace period
+        grace_period = DEFAULT_TEMPFILE_GRACE_PERIOD
+        if args.expire:
+            try:
+                grace_period = parse_relative_time(args.expire)
+            except ValueError:
+                # Try to parse as absolute date
+                try:
+                    date = datetime.datetime.strptime(args.expire, "%Y-%m-%d")
+                    grace_period = int(time.time() - date.timestamp())
+                except ValueError:
+                    print(f"Error: Invalid expire date: {args.expire}", file=sys.stderr)
+                    return 1
+
+        # Progress callback
+        def progress(msg):
+            if args.verbose:
+                print(msg)
+
+        try:
+            porcelain.prune(
+                ".",
+                grace_period=grace_period,
+                dry_run=args.dry_run,
+                progress=progress if args.verbose else None,
+            )
+            return None
+        except porcelain.Error as e:
+            print(f"Error: {e}", file=sys.stderr)
+            return 1
+
+
 class cmd_pull(Command):
     def run(self, args) -> None:
         parser = argparse.ArgumentParser()
@@ -1491,6 +1553,7 @@ commands = {
     "notes": cmd_notes,
     "pack-objects": cmd_pack_objects,
     "pack-refs": cmd_pack_refs,
+    "prune": cmd_prune,
     "pull": cmd_pull,
     "push": cmd_push,
     "rebase": cmd_rebase,

+ 2 - 2
dulwich/porcelain.py

@@ -3444,10 +3444,10 @@ def prune(
     progress=None,
 ):
     """Prune/clean up a repository's object store.
-    
+
     This removes temporary files that were left behind by interrupted
     pack operations.
-    
+
     Args:
       repo: Path to the repository or a Repo object
       grace_period: Grace period in seconds for removing temporary files

+ 16 - 16
tests/test_porcelain.py

@@ -5513,17 +5513,17 @@ class PruneTests(PorcelainTestCase):
         tmp_pack_path = os.path.join(objects_dir, "tmp_pack_test")
         with open(tmp_pack_path, "wb") as f:
             f.write(b"old temporary data")
-        
+
         # Make it old
         old_time = time.time() - (DEFAULT_TEMPFILE_GRACE_PERIOD + 3600)
         os.utime(tmp_pack_path, (old_time, old_time))
-        
+
         # Run prune
         porcelain.prune(self.repo.path)
-        
+
         # Verify the file was removed
         self.assertFalse(os.path.exists(tmp_pack_path))
-    
+
     def test_prune_keeps_recent_tempfiles(self):
         """Test that prune keeps recent temporary files."""
         # Create a recent temporary file
@@ -5531,16 +5531,16 @@ class PruneTests(PorcelainTestCase):
         tmp_pack_path = os.path.join(objects_dir, "tmp_pack_recent")
         with open(tmp_pack_path, "wb") as f:
             f.write(b"recent temporary data")
-        
+
         # Run prune
         porcelain.prune(self.repo.path)
-        
+
         # Verify the file was NOT removed
         self.assertTrue(os.path.exists(tmp_pack_path))
-        
+
         # Clean up
         os.remove(tmp_pack_path)
-    
+
     def test_prune_with_custom_grace_period(self):
         """Test prune with custom grace period."""
         # Create a 1-hour-old temporary file
@@ -5548,17 +5548,17 @@ class PruneTests(PorcelainTestCase):
         tmp_pack_path = os.path.join(objects_dir, "tmp_pack_1hour")
         with open(tmp_pack_path, "wb") as f:
             f.write(b"1 hour old data")
-        
+
         # Make it 1 hour old
         old_time = time.time() - 3600
         os.utime(tmp_pack_path, (old_time, old_time))
-        
+
         # Prune with 30-minute grace period should remove it
         porcelain.prune(self.repo.path, grace_period=1800)
-        
+
         # Verify the file was removed
         self.assertFalse(os.path.exists(tmp_pack_path))
-    
+
     def test_prune_dry_run(self):
         """Test prune in dry-run mode."""
         # Create an old temporary file
@@ -5566,16 +5566,16 @@ class PruneTests(PorcelainTestCase):
         tmp_pack_path = os.path.join(objects_dir, "tmp_pack_dryrun")
         with open(tmp_pack_path, "wb") as f:
             f.write(b"old temporary data")
-        
+
         # Make it old
         old_time = time.time() - (DEFAULT_TEMPFILE_GRACE_PERIOD + 3600)
         os.utime(tmp_pack_path, (old_time, old_time))
-        
+
         # Run prune in dry-run mode
         porcelain.prune(self.repo.path, dry_run=True)
-        
+
         # Verify the file was NOT removed (dry run)
         self.assertTrue(os.path.exists(tmp_pack_path))
-        
+
         # Clean up
         os.remove(tmp_pack_path)