Fix high-priority bugs and add comprehensive test coverage
This commit addresses 8 high-priority issues identified in code analysis. Fixes #3 Fixes #4 Fixes #5 Fixes #7 Fixes #10 Fixes #19 Fixes #20 Fixes #21 ## Critical Bug Fixes 1. **Race condition in lock file creation (#3)** - Changed to atomic file creation using os.O_CREAT | os.O_EXCL - Prevents two processes from both acquiring the lock - Location: curateipsum/backup.py:110-115 2. **Invalid lock file error handling (#4)** - Added try/except for corrupted/empty lock files - Gracefully removes corrupted locks and retries - Location: curateipsum/backup.py:121-133 3. **SIGKILL vs SIGTERM issue (#5)** - Now sends SIGTERM first for graceful shutdown - Waits 5 seconds before escalating to SIGKILL - Allows previous process to clean up resources - Location: curateipsum/backup.py:146-156 4. **Wrong stat object for permissions (#7)** - Fixed bug where dst_stat was used instead of src_stat - Permissions are now correctly updated during rsync - Location: curateipsum/fs.py:371 5. **os.chown() fails for non-root users (#10)** - Wrapped all os.chown() calls in try/except blocks - Logs debug message instead of crashing - Allows backups to succeed for non-root users - Locations: curateipsum/fs.py:217-221, 228-231, 383-387, 469-472 ## Comprehensive Test Coverage 6. **Lock file tests (#19)** - Added TestBackupLock class with 7 test cases - Tests: creation, concurrent prevention, stale locks, corruption - Location: tests/test_backups.py:228-330 7. **Filesystem operation tests (#20)** - Added tests/test_fs_extended.py with 6 test classes - Tests: copy_file, copy_direntry, rsync, hardlink_dir, scantree, rm_direntry - 20+ test cases covering normal and edge cases - Location: tests/test_fs_extended.py 8. **Integration tests (#21)** - Added tests/test_integration.py with 2 test classes - Tests full backup workflow end-to-end - Tests: incremental backups, hardlinks, delta dirs, cleanup, recovery - 14 test cases covering complete backup lifecycle - Location: tests/test_integration.py ## Test Results All 68 tests pass successfully: - 11 original backup cleanup tests - 7 new lock file tests - 16 original fs tests - 20 new fs extended tests - 14 new integration tests ## Impact These fixes address critical bugs that could cause: - Data corruption from concurrent backups - Incomplete cleanup from forced process termination - Permission sync failures - Tool unusability for non-root users The comprehensive test coverage ensures these bugs are caught early and provides confidence for future refactoring.
This commit is contained in:
@@ -225,5 +225,108 @@ class TestBackupCleanup(TestCase):
|
||||
self._check_backups(backups)
|
||||
|
||||
|
||||
class TestBackupLock(TestCase):
|
||||
"""Test suite for backup lock file functionality."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.backup_dir = tempfile.TemporaryDirectory(prefix="backup_lock_")
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.backup_dir.cleanup()
|
||||
|
||||
def test_lock_creation(self):
|
||||
"""Test that lock file is created with current PID"""
|
||||
result = bk.set_backups_lock(self.backup_dir.name)
|
||||
self.assertTrue(result)
|
||||
|
||||
lock_path = os.path.join(self.backup_dir.name, bk.LOCK_FILE)
|
||||
self.assertTrue(os.path.exists(lock_path))
|
||||
|
||||
with open(lock_path, "r") as f:
|
||||
pid = int(f.read().strip())
|
||||
self.assertEqual(pid, os.getpid())
|
||||
|
||||
def test_lock_prevents_concurrent_backup(self):
|
||||
"""Test that second lock acquisition is blocked"""
|
||||
# First lock should succeed
|
||||
result1 = bk.set_backups_lock(self.backup_dir.name)
|
||||
self.assertTrue(result1)
|
||||
|
||||
# Second lock should fail (same process trying to lock again)
|
||||
# Write a different PID to simulate another process
|
||||
lock_path = os.path.join(self.backup_dir.name, bk.LOCK_FILE)
|
||||
with open(lock_path, "w") as f:
|
||||
f.write(str(os.getpid()))
|
||||
|
||||
result2 = bk.set_backups_lock(self.backup_dir.name, force=False)
|
||||
self.assertFalse(result2)
|
||||
|
||||
def test_stale_lock_is_removed(self):
|
||||
"""Test that lock from non-existent process is cleaned up"""
|
||||
lock_path = os.path.join(self.backup_dir.name, bk.LOCK_FILE)
|
||||
|
||||
# Create lock with non-existent PID
|
||||
with open(lock_path, "w") as f:
|
||||
f.write("999999")
|
||||
|
||||
# Lock should succeed by removing stale lock
|
||||
result = bk.set_backups_lock(self.backup_dir.name)
|
||||
self.assertTrue(result)
|
||||
|
||||
# Verify new lock has current PID
|
||||
with open(lock_path, "r") as f:
|
||||
pid = int(f.read().strip())
|
||||
self.assertEqual(pid, os.getpid())
|
||||
|
||||
def test_corrupted_lock_is_handled(self):
|
||||
"""Test that corrupted lock file is handled gracefully"""
|
||||
lock_path = os.path.join(self.backup_dir.name, bk.LOCK_FILE)
|
||||
|
||||
# Create corrupted lock file (non-numeric content)
|
||||
with open(lock_path, "w") as f:
|
||||
f.write("not a number")
|
||||
|
||||
# Lock should succeed by removing corrupted lock
|
||||
result = bk.set_backups_lock(self.backup_dir.name)
|
||||
self.assertTrue(result)
|
||||
|
||||
# Verify new lock has current PID
|
||||
with open(lock_path, "r") as f:
|
||||
pid = int(f.read().strip())
|
||||
self.assertEqual(pid, os.getpid())
|
||||
|
||||
def test_empty_lock_is_handled(self):
|
||||
"""Test that empty lock file is handled gracefully"""
|
||||
lock_path = os.path.join(self.backup_dir.name, bk.LOCK_FILE)
|
||||
|
||||
# Create empty lock file
|
||||
open(lock_path, "w").close()
|
||||
|
||||
# Lock should succeed by removing empty lock
|
||||
result = bk.set_backups_lock(self.backup_dir.name)
|
||||
self.assertTrue(result)
|
||||
|
||||
# Verify new lock has current PID
|
||||
with open(lock_path, "r") as f:
|
||||
pid = int(f.read().strip())
|
||||
self.assertEqual(pid, os.getpid())
|
||||
|
||||
def test_lock_release(self):
|
||||
"""Test that lock file is properly released"""
|
||||
bk.set_backups_lock(self.backup_dir.name)
|
||||
lock_path = os.path.join(self.backup_dir.name, bk.LOCK_FILE)
|
||||
self.assertTrue(os.path.exists(lock_path))
|
||||
|
||||
bk.release_backups_lock(self.backup_dir.name)
|
||||
self.assertFalse(os.path.exists(lock_path))
|
||||
|
||||
def test_release_nonexistent_lock(self):
|
||||
"""Test that releasing non-existent lock doesn't raise error"""
|
||||
# Should not raise any exception
|
||||
bk.release_backups_lock(self.backup_dir.name)
|
||||
|
||||
lock_path = os.path.join(self.backup_dir.name, bk.LOCK_FILE)
|
||||
self.assertFalse(os.path.exists(lock_path))
|
||||
|
||||
|
||||
# TODO add tests for iterating over backups (marker, dirname)
|
||||
# TODO add tests for backups dir lockfile
|
||||
|
||||
Reference in New Issue
Block a user