refactor: Address CodeRabbit review feedback

hyaku0121 · hyaku0121 · commit da14bd9bdba8 · 2025-12-14T19:47:25.000+09:00
- manager.py: Harden directory permissions, atomic writes, robust metadata
- cleaner.py: Add sudo -n flag, fix freed bytes calculation, use safe param
- scheduler.py: Pass safe_mode to systemd/cron setup
- cli.py: Add error message when no subcommand specified
- test_cleanup_scanner.py: Reduce test file size from 150MB to 2MB
- test_cleanup_cleaner.py: Fix test for safe mode behavior
- CLEANUP_GUIDE.md: Fix markdownlint issues
diff --git a/cortex/cleanup/cleaner.py b/cortex/cleanup/cleaner.py
@@ -39,8 +39,8 @@ def clean_package_cache(self) -> int:
         if self.dry_run:
             return size_freed
             
-        # Run apt-get clean
-        cmd = "sudo apt-get clean"
+        # Run apt-get clean (use -n for non-interactive mode)
+        cmd = "sudo -n apt-get clean"
         result = run_command(cmd, validate=True)
         
         if result.success:
@@ -65,7 +65,8 @@ def remove_orphaned_packages(self, packages: List[str]) -> int:
         if self.dry_run:
             return 0 # Size is estimated in scanner
             
-        cmd = "sudo apt-get autoremove -y"
+        # Use -n for non-interactive mode
+        cmd = "sudo -n apt-get autoremove -y"
         result = run_command(cmd, validate=True)
         
         freed_bytes = 0
@@ -122,33 +123,25 @@ def clean_temp_files(self, files: List[str]) -> int:
             filepath = Path(filepath_str)
             if not filepath.exists():
                 continue
+            
+            # Get size before any operation
+            try:
+                size = filepath.stat().st_size
+            except OSError:
+                size = 0
                 
             if self.dry_run:
-                try:
-                    freed_bytes += filepath.stat().st_size
-                except OSError:
-                    pass
+                freed_bytes += size
                 continue
                 
             # Move to quarantine
             item_id = self.manager.quarantine_file(str(filepath))
             if item_id:
-                # We assume success means we freed the file size.
-                # Ideally we should get the size from the manager or check before.
-                # But for now we don't have the size unless we check again.
-                # Let's assume the scanner's size is accurate enough for reporting,
-                # or check size before quarantine if we really need to return exact freed bytes here.
-                # But manager.quarantine_file moves it, so it's gone from original location.
-                pass
+                freed_bytes += size
             else:
-                # Failed to quarantine
-                pass
+                logger.warning(f"Failed to quarantine temp file: {filepath}")
                 
-        # Returning 0 here because calculating exact freed bytes per file during deletion 
-        # without re-statting every file (which might fail if moved) is complex.
-        # The CLI can use the scan result for total potential, or we can improve this.
-        # For now, let's return 0 and rely on the scan result or improve manager to return size.
-        return freed_bytes 
+        return freed_bytes
 
     def compress_logs(self, files: List[str]) -> int:
         """
@@ -216,7 +209,9 @@ def run_cleanup(self, scan_results: List[ScanResult], safe: bool = True) -> Dict
                 summary["Package Cache"] = self.clean_package_cache()
                 
             elif result.category == "Orphaned Packages":
-                summary["Orphaned Packages"] = self.remove_orphaned_packages(result.items)
+                # Only remove orphaned packages in non-safe mode
+                if not safe:
+                    summary["Orphaned Packages"] = self.remove_orphaned_packages(result.items)
                 
             elif result.category == "Temporary Files":
                 summary["Temporary Files"] = self.clean_temp_files(result.items)
diff --git a/cortex/cleanup/manager.py b/cortex/cleanup/manager.py
@@ -29,30 +29,42 @@ class CleanupManager:
     """
     Manages the quarantine (undo) system for cleaned files.
     """
-    def __init__(self):
+    def __init__(self) -> None:
+        """Initialize quarantine storage and metadata paths."""
         self.quarantine_dir = Path.home() / ".cortex" / "trash"
         self.metadata_file = self.quarantine_dir / "metadata.json"
         self._ensure_dir()
 
-    def _ensure_dir(self):
+    def _ensure_dir(self) -> None:
         """Ensure quarantine directory exists with secure permissions."""
-        if not self.quarantine_dir.exists():
-            self.quarantine_dir.mkdir(parents=True, mode=0o700)
+        self.quarantine_dir.mkdir(parents=True, exist_ok=True)
+        try:
+            # Ensure privacy even if pre-existing
+            self.quarantine_dir.chmod(0o700)
+        except OSError:
+            # Best-effort; callers still handle failures later
+            pass
 
     def _load_metadata(self) -> Dict[str, dict]:
         """Load metadata from JSON file."""
         if not self.metadata_file.exists():
             return {}
         try:
-            with open(self.metadata_file, 'r') as f:
+            with self.metadata_file.open("r", encoding="utf-8") as f:
                 return json.load(f)
-        except json.JSONDecodeError:
+        except (json.JSONDecodeError, OSError, ValueError):
             return {}
 
-    def _save_metadata(self, metadata: Dict[str, dict]):
-        """Save metadata to JSON file."""
-        with open(self.metadata_file, 'w') as f:
+    def _save_metadata(self, metadata: Dict[str, dict]) -> None:
+        """Save metadata to JSON file atomically."""
+        tmp = self.metadata_file.with_suffix(".json.tmp")
+        with tmp.open("w", encoding="utf-8") as f:
             json.dump(metadata, f, indent=2)
+        os.replace(tmp, self.metadata_file)
+        try:
+            self.metadata_file.chmod(0o600)
+        except OSError:
+            pass
 
     def quarantine_file(self, filepath_str: str) -> Optional[str]:
         """
@@ -147,20 +159,27 @@ def list_items(self) -> List[QuarantineItem]:
             items.append(QuarantineItem(**v))
         return sorted(items, key=lambda x: x.timestamp, reverse=True)
 
-    def cleanup_old_items(self, days: int = 30):
+    def cleanup_old_items(self, days: int = 30) -> None:
         """
         Remove quarantine items older than X days.
         
         Args:
             days (int): Age in days to expire items.
+            
+        Raises:
+            ValueError: If days is negative.
         """
+        if days < 0:
+            raise ValueError("days must be >= 0")
+            
         metadata = self._load_metadata()
         now = time.time()
         cutoff = now - (days * 86400)
         
         to_remove = []
         for item_id, data in metadata.items():
-            if data['timestamp'] < cutoff:
+            ts = data.get("timestamp")
+            if isinstance(ts, (int, float)) and ts < cutoff:
                 to_remove.append(item_id)
                 
         for item_id in to_remove:
@@ -174,3 +193,4 @@ def cleanup_old_items(self, days: int = 30):
             
         if to_remove:
             self._save_metadata(metadata)
+
diff --git a/cortex/cleanup/scheduler.py b/cortex/cleanup/scheduler.py
@@ -137,7 +137,7 @@ def enable_schedule(
         )
         
         # Try to set up systemd timer first
-        systemd_result = self._setup_systemd_timer(interval)
+        systemd_result = self._setup_systemd_timer(interval, safe_mode)
         if systemd_result["success"]:
             self.save_config(config)
             return {
@@ -147,7 +147,7 @@ def enable_schedule(
             }
         
         # Fall back to cron
-        cron_result = self._setup_cron(interval)
+        cron_result = self._setup_cron(interval, safe_mode)
         if cron_result["success"]:
             self.save_config(config)
             return {
@@ -237,12 +237,13 @@ def _get_cron_schedule(self, interval: ScheduleInterval) -> str:
         else:  # monthly
             return "0 3 1 * *"  # 3 AM 1st of month
     
-    def _setup_systemd_timer(self, interval: ScheduleInterval) -> Dict[str, Any]:
+    def _setup_systemd_timer(self, interval: ScheduleInterval, safe_mode: bool = True) -> Dict[str, Any]:
         """
         Set up systemd timer for scheduling.
         
         Args:
             interval: Scheduling interval.
+            safe_mode: If True, run with --safe flag; otherwise --force.
             
         Returns:
             dict: Result with success status.
@@ -258,14 +259,17 @@ def _setup_systemd_timer(self, interval: ScheduleInterval) -> Dict[str, Any]:
             if result.returncode not in (0, 1):  # 1 is "degraded" which is OK
                 return {"success": False, "error": "systemd not available"}
             
+            # Determine cleanup mode flag
+            mode_flag = "--safe" if safe_mode else "--force"
+            
             # Create service file
             service_content = f"""[Unit]
 Description=Cortex Disk Cleanup Service
 After=network.target
 
 [Service]
 Type=oneshot
-ExecStart=/usr/bin/env cortex cleanup run --safe --yes
+ExecStart=/usr/bin/env cortex cleanup run {mode_flag} --yes
 """
             
             # Create timer file
@@ -350,19 +354,21 @@ def _check_systemd_timer(self) -> bool:
         except (subprocess.TimeoutExpired, OSError):
             return False
     
-    def _setup_cron(self, interval: ScheduleInterval) -> Dict[str, Any]:
+    def _setup_cron(self, interval: ScheduleInterval, safe_mode: bool = True) -> Dict[str, Any]:
         """
         Set up cron job for scheduling.
         
         Args:
             interval: Scheduling interval.
+            safe_mode: If True, run with --safe flag; otherwise --force.
             
         Returns:
             dict: Result with success status.
         """
         try:
             cron_schedule = self._get_cron_schedule(interval)
-            cron_command = f"{cron_schedule} /usr/bin/env cortex cleanup run --safe --yes # cortex-cleanup"
+            mode_flag = "--safe" if safe_mode else "--force"
+            cron_command = f"{cron_schedule} /usr/bin/env cortex cleanup run {mode_flag} --yes # cortex-cleanup"
             
             # Get current crontab
             result = subprocess.run(
diff --git a/cortex/cli.py b/cortex/cli.py
@@ -530,6 +530,10 @@ def cleanup(self, args):
         from rich.prompt import Confirm
         from datetime import datetime
 
+        if not hasattr(args, 'cleanup_action') or args.cleanup_action is None:
+            self._print_error("Please specify a subcommand (scan/run/undo/schedule)")
+            return 1
+
         if args.cleanup_action == 'scan':
             scanner = CleanupScanner()
             self._print_status("🔍", "Scanning for cleanup opportunities...")
diff --git a/docs/CLEANUP_GUIDE.md b/docs/CLEANUP_GUIDE.md
@@ -15,13 +15,13 @@ Cortex provides intelligent disk cleanup capabilities that can:
 
 ```bash
 # Scan for cleanup opportunities
-$ cortex cleanup scan
+cortex cleanup scan
 
 # Run cleanup (with confirmation)
-$ cortex cleanup run
+cortex cleanup run
 
 # Run cleanup without confirmation (safe mode)
-$ cortex cleanup run --safe --yes
+cortex cleanup run --safe --yes
 ```
 
 ## Commands
@@ -31,11 +31,12 @@ $ cortex cleanup run --safe --yes
 Identify cleanup opportunities without making any changes:
 
 ```bash
-$ cortex cleanup scan
+cortex cleanup scan
 ```
 
 **Output example:**
-```
+
+```text
 💾 Cleanup Opportunities:
 
 Category           Items    Size
@@ -53,13 +54,13 @@ Execute cleanup operations:
 
 ```bash
 # Safe mode (default) - with confirmation
-$ cortex cleanup run
+cortex cleanup run
 
 # Safe mode - skip confirmation
-$ cortex cleanup run --safe --yes
+cortex cleanup run --safe --yes
 
 # Force mode - clean all items (use with caution)
-$ cortex cleanup run --force --yes
+cortex cleanup run --force --yes
 ```
 
 **Options:**
@@ -76,14 +77,15 @@ Restore files that were cleaned:
 
 ```bash
 # List restorable items
-$ cortex cleanup undo
+cortex cleanup undo
 
 # Restore a specific item
-$ cortex cleanup undo <item-id>
+cortex cleanup undo <item-id>
 ```
 
 **Example:**
-```bash
+
+```text
 $ cortex cleanup undo
 ID       File              Size       Date
 abc123   temp_file.txt     1.2 MB     2024-01-15 10:30
@@ -101,19 +103,19 @@ Configure automatic cleanup:
 
 ```bash
 # Show current schedule status
-$ cortex cleanup schedule --show
+cortex cleanup schedule --show
 
 # Enable weekly cleanup (default)
-$ cortex cleanup schedule --enable
+cortex cleanup schedule --enable
 
 # Enable daily cleanup
-$ cortex cleanup schedule --enable --interval daily
+cortex cleanup schedule --enable --interval daily
 
 # Enable monthly cleanup
-$ cortex cleanup schedule --enable --interval monthly
+cortex cleanup schedule --enable --interval monthly
 
 # Disable scheduled cleanup
-$ cortex cleanup schedule --disable
+cortex cleanup schedule --disable
 ```
 
 **Supported intervals:**
@@ -186,7 +188,7 @@ Some cleanup operations require root privileges:
 
 ```bash
 # Clean system package cache
-$ sudo cortex cleanup run
+sudo cortex cleanup run
 ```
 
 ### No Space Reclaimed
@@ -218,4 +220,4 @@ Default settings for cleanup:
 
 ---
 
-For more information, visit: https://cortexlinux.com/docs/cleanup
+For more information, visit: [Cortex Cleanup Documentation](https://cortexlinux.com/docs/cleanup)
diff --git a/tests/test_cleanup_cleaner.py b/tests/test_cleanup_cleaner.py
@@ -188,7 +188,7 @@ def test_compress_logs_dry_run(self, dry_run_cleaner, tmp_path):
         assert log_file.exists()
     
     def test_run_cleanup_all_categories(self, cleaner):
-        """Test run_cleanup with all categories."""
+        """Test run_cleanup with all categories (non-safe mode)."""
         scan_results = [
             ScanResult("Package Cache", 1000, 5, []),
             ScanResult("Orphaned Packages", 2000, 3, ["pkg1"]),
@@ -201,7 +201,8 @@ def test_run_cleanup_all_categories(self, cleaner):
              patch.object(cleaner, 'clean_temp_files', return_value=500), \
              patch.object(cleaner, 'compress_logs', return_value=800):
             
-            summary = cleaner.run_cleanup(scan_results)
+            # Use safe=False to include orphaned packages
+            summary = cleaner.run_cleanup(scan_results, safe=False)
             
             assert summary["Package Cache"] == 1000
             assert summary["Orphaned Packages"] == 2000
diff --git a/tests/test_cleanup_scanner.py b/tests/test_cleanup_scanner.py
