import fcntl import json import os import subprocess import sys import threading import time import pytest from unittest import mock from datetime import datetime from curateipsum import backup as bk class TestBackupCleanup: """Tests for backup cleanup and retention policies.""" def test_no_backups(self, backup_dir, run_cleanup): """Test behaviour with no available backups""" backup_dir.mkdir() bk.cleanup_old_backups(str(backup_dir)) assert not os.listdir(str(backup_dir)) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_only_one_backup(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test the only backup will not be removed in any case""" mock_datetime.now.return_value = datetime(2021, 10, 20) # very old backup only_backup = add_backup("20010101_0000") run_cleanup(keep_all=1) check_backups([only_backup]) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_at_least_one_should_be_left(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test at least one backup should be left""" mock_datetime.now.return_value = datetime(2021, 10, 20) backups = [ add_backup("20211103_0300"), # latest, should be kept add_backup("20201216_0100"), # rest should be removed add_backup("20200716_0100"), add_backup("20181116_0100"), ] expected_backups = [backups[0]] run_cleanup() check_backups(expected_backups) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_keep_all_threshold_only(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test threshold for keeping all backups""" mock_datetime.now.return_value = datetime(2021, 10, 20) backups = [ add_backup("20211019_0300"), # keep add_backup("20211017_0100"), # keep add_backup("20211016_2300"), # remove, older than 3 days ] expected_backups = backups[:2] run_cleanup(keep_all=3) check_backups(expected_backups) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_keep_daily_threshold_only(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test threshold for keeping daily backups""" mock_datetime.now.return_value = datetime(2021, 10, 20) backups = [ add_backup("20211019_0300"), # keep, first daily at 2021-10-19 add_backup("20211017_2100"), # remove, not first daily add_backup("20211017_0100"), # remove, not first daily add_backup("20211017_0030"), # keep, first daily at 2021-10-17 add_backup("20211016_2300"), # remove, older than 3 days add_backup("20211016_0100"), # remove, older than 3 days ] expected_backups = [backups[0], backups[3]] run_cleanup(keep_daily=3) check_backups(expected_backups) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_keep_all_and_daily_thresholds(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test threshold for keeping all and daily backups""" mock_datetime.now.return_value = datetime(2021, 10, 20) backups = [ add_backup("20211019_0300"), # keep, newer than 3 days add_backup("20211017_0200"), # keep, newer than 3 days add_backup("20211017_0100"), # keep, newer than 3 days add_backup("20211016_2300"), # remove, not first daily add_backup("20211016_2200"), # keep, first daily at 2021-10-16 add_backup("20211015_2200"), # remove, not first daily add_backup("20211015_1500"), # remove, not first daily add_backup("20211015_0200"), # keep, first daily at 2021-10-15 add_backup("20211014_2200"), # remove, older than 5 days add_backup("20211014_2000"), # remove, older than 5 days add_backup("20211014_1232"), # remove, older than 5 days ] expected_backups = backups[0:3] + [backups[4]] + [backups[7]] run_cleanup(keep_all=3, keep_daily=5) check_backups(expected_backups) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_keep_weekly_threshold_only(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test threshold for keeping weekly backups""" mock_datetime.now.return_value = datetime(2021, 11, 11) backups = [ add_backup("20211111_0300"), # remove, not first weekly (Thu) add_backup("20211110_0300"), # remove, not first weekly (Wed) add_backup("20211108_0100"), # keep, first weekly 2021-11-08 (Mon) add_backup("20211107_2300"), # remove, not first weekly (Sun) add_backup("20211107_0100"), # keep, first weekly 2021-11-07 (Sun) add_backup("20211031_0100"), # remove, not first weekly (Sun) add_backup("20211025_0100"), # keep, first weekly 2021-10-25 (Mon) add_backup("20211024_0100"), # remove, not first weekly (Sun) add_backup("20211023_0100"), # remove, not first weekly (Sat) add_backup("20211022_0100"), # keep, first weekly 2021-10-22 (Fri) add_backup("20211008_0100"), # remove, not first weekly (Fri) add_backup("20211007_0100"), # remove, not first weekly (Thu) add_backup("20211004_0100"), # keep, first weekly 2021-10-04 (Mon) add_backup("20211003_0100"), # remove, older than 5 weeks add_backup("20211002_0100"), # remove, older than 5 weeks ] expected_backups = [backups[2], backups[4], backups[6], backups[9], backups[12]] run_cleanup(keep_weekly=5) check_backups(expected_backups) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_keep_weekly_threshold_inclusive(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test threshold for keeping weekly backups""" mock_datetime.now.return_value = datetime(2021, 11, 11) backups = [ add_backup("20211111_0300"), # remove, not first weekly (Thu) add_backup("20211110_0300"), # keep, first weekly (Wed) add_backup("20211107_0100"), # remove, not first weekly (Sun) add_backup("20211102_0100"), # keep, first weekly (Tue) ] expected_backups = [backups[1], backups[3]] run_cleanup(keep_weekly=5) check_backups(expected_backups) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_keep_monthly_threshold_only(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test threshold for keeping monthly backups""" mock_datetime.now.return_value = datetime(2021, 11, 11) backups = [ add_backup("20211103_0300"), # keep, first monthly at 2021-11 add_backup("20211019_0300"), # remove, not first monthly add_backup("20211017_2100"), # remove, not first monthly add_backup("20211017_0100"), # keep, first monthly at 2021-10 add_backup("20210916_2300"), # remove, not first monthly add_backup("20210916_0100"), # keep, first monthly at 2021-09 add_backup("20210816_0100"), # remove, not first monthly add_backup("20210810_0000"), # keep, first monthly at 2021-08 add_backup("20210716_0100"), # remove, older than 3 months add_backup("20210715_0100"), # remove, older than 3 months ] expected_backups = [backups[0], backups[3], backups[5], backups[7]] run_cleanup(keep_monthly=3) check_backups(expected_backups) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_keep_yearly_threshold_only(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test threshold for keeping yearly backups""" mock_datetime.now.return_value = datetime(2021, 11, 11) backups = [ add_backup("20211103_0300"), # remove, not first yearly in 2021 add_backup("20210810_0000"), # remove, not first yearly in 2021 add_backup("20210716_0100"), # keep, first yearly in 2021 add_backup("20201216_0100"), # remove, not first yearly in 2020 add_backup("20200716_0100"), # keep, first yearly in 2020 add_backup("20191216_0100"), # remove, not first yearly in 2019 add_backup("20190316_0100"), # keep, first yearly in 2019 add_backup("20181216_0100"), # remove, not first yearly in 2018 add_backup("20181116_0100"), # keep, first yearly in 2018 add_backup("20171116_0100"), # remove, older than 3 years add_backup("20171115_0100"), # remove, older than 3 years ] expected_backups = [backups[2], backups[4], backups[6], backups[8]] run_cleanup(keep_yearly=3) check_backups(expected_backups) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_dry_run(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test dry run does not remove anything""" mock_datetime.now.return_value = datetime(2021, 11, 11) backups = [ add_backup("20211103_0300"), add_backup("20210810_0000"), add_backup("20210716_0100"), add_backup("20200716_0100"), add_backup("20181116_0100"), ] run_cleanup(keep_all=2, dry_run=True) check_backups(backups) @pytest.mark.xfail(reason="Bug #32: coarser retention tiers override finer " "tiers, violating retention guarantees", strict=True) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_combined_retention_policies(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test all retention policies working together Tests EXPECTED behavior where retention policies honor their documented guarantees. Currently fails due to bug where coarser retention tiers (weekly/monthly/yearly) can override finer tiers (all/daily/weekly/monthly), violating retention guarantees. Bug affects multiple tier interactions (see issue #32): - Weekly overrides daily: Nov 4 removed by weekly - Monthly overrides weekly: Oct 11 removed by monthly - Yearly overrides monthly: May 15 removed by yearly Once fixed, this test will pass and validate that each tier protects its threshold range from coarser tiers. """ # Current: 2021-11-11 (Thursday) # Policies: keep_all=2, keep_daily=7, keep_weekly=4, # keep_monthly=6, keep_yearly=3 # Thresholds: all=Nov 9, daily=Nov 4, weekly=Oct 11, # monthly=May 1, yearly=Jan 1 2018 mock_datetime.now.return_value = datetime(2021, 11, 11) # Use dict for readability b = {} # Keep-all range (all backups after Nov 9) b["nov_11"] = add_backup("20211111_0300") b["nov_10_23h"] = add_backup("20211110_2300") b["nov_10_12h"] = add_backup("20211110_1200") b["nov_10_01h"] = add_backup("20211110_0100") b["nov_09"] = add_backup("20211109_0300") # Daily range (one per day: Nov 4-8) b["nov_08_dupe"] = add_backup("20211108_2200") # removed: not oldest b["nov_08"] = add_backup("20211108_0100") b["nov_07"] = add_backup("20211107_0100") b["nov_06_dupe"] = add_backup("20211106_1800") # removed: not oldest b["nov_06"] = add_backup("20211106_0100") b["nov_05"] = add_backup("20211105_0100") b["nov_04"] = add_backup("20211104_0100") # BUG: removed by weekly! # Weekly range (one per week: Oct 11-Nov 7) b["nov_03"] = add_backup("20211103_0300") # removed: same week b["nov_01_dupe"] = add_backup("20211101_1500") # removed: not oldest b["nov_01"] = add_backup("20211101_0100") b["oct_31"] = add_backup("20211031_0100") # removed: same week b["oct_25_dupe"] = add_backup("20211025_2000") # removed: not oldest b["oct_25"] = add_backup("20211025_0100") b["oct_18"] = add_backup("20211018_0100") b["oct_11"] = add_backup("20211011_0100") # BUG: removed by monthly! # Monthly range (one per month: May-Oct) b["oct_04"] = add_backup("20211004_0100") # removed: same month b["oct_01"] = add_backup("20211001_0100") b["sep_15_dupe"] = add_backup("20210915_2000") # removed: not oldest b["sep_15"] = add_backup("20210915_0100") b["aug_15"] = add_backup("20210815_0100") b["jul_15_dupe"] = add_backup("20210715_1200") # removed: not oldest b["jul_15"] = add_backup("20210715_0100") b["jun_15"] = add_backup("20210615_0100") b["may_15"] = add_backup("20210515_0100") # BUG: removed by yearly! # Yearly range (one per year: 2018-2021) b["apr_15"] = add_backup("20210415_0100") # removed: same year b["jan_15_2021"] = add_backup("20210115_0100") b["dec_15_2020"] = add_backup("20201215_0100") # removed: same year b["jul_15_2020"] = add_backup("20200715_0100") # removed: same year b["jan_15_2020"] = add_backup("20200115_0100") b["dec_15_2019_dupe"] = add_backup("20191215_2300") # removed: not oldest b["dec_15_2019"] = add_backup("20191215_0100") # removed: same year b["jul_15_2019"] = add_backup("20190715_0100") # removed: same year b["jan_15_2019"] = add_backup("20190115_0100") b["nov_15_2018"] = add_backup("20181115_0100") # removed: same year b["jan_15_2018"] = add_backup("20180115_0100") # Beyond all thresholds b["jan_15_2017"] = add_backup("20170115_0100") # removed: too old # Expected: each tier protects its range expected = [ # keep_all: all 5 backups b["nov_11"], b["nov_10_23h"], b["nov_10_12h"], b["nov_10_01h"], b["nov_09"], # daily: oldest per day (Nov 4-8) = 5 backups # BUG: nov_04 missing - removed by weekly (same week as nov_03) b["nov_08"], b["nov_07"], b["nov_06"], b["nov_05"], b["nov_04"], # weekly: oldest per week (Oct 11 - Nov 7) = 4 backups # BUG: oct_11 missing - removed by monthly (same month as oct_01) b["nov_01"], b["oct_25"], b["oct_18"], b["oct_11"], # monthly: oldest per month (May-Oct) = 6 backups # BUG: may_15 missing - removed by yearly (same year as jan_15_2021) b["oct_01"], b["sep_15"], b["aug_15"], b["jul_15"], b["jun_15"], b["may_15"], # yearly: oldest per year (2018-2021) = 4 backups b["jan_15_2021"], b["jan_15_2020"], b["jan_15_2019"], b["jan_15_2018"], ] run_cleanup(keep_all=2, keep_daily=7, keep_weekly=4, keep_monthly=6, keep_yearly=3) check_backups(expected) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_iso_week_53_year_boundary(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test ISO week 53 spanning year boundary (2020-2021) 2020 had 53 ISO weeks. Both Dec 31, 2020 and Jan 1-3, 2021 belong to ISO week 53 of 2020. These should be treated as the same week. """ # Current: 2021-01-10 (Sunday, ISO week 1 of 2021) mock_datetime.now.return_value = datetime(2021, 1, 10) backups = [ add_backup("20210103_1200"), # Sun, ISO 2020-W53 add_backup("20210102_1200"), # Sat, ISO 2020-W53 add_backup("20210101_1200"), # Fri, ISO 2020-W53, same week add_backup("20201231_1200"), # Thu, ISO 2020-W53, first in week add_backup("20201230_1200"), # Wed, ISO 2020-W53 add_backup("20201228_1200"), # Mon, ISO 2020-W53 ] # Keep one week: should keep oldest backup from ISO week 53 expected = [backups[5]] # 20201228 run_cleanup(keep_weekly=1) check_backups(expected) @pytest.mark.xfail(reason="Bug #33: Weekly retention bug: same ISO week " "number across years treated as duplicates", strict=True) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_iso_week_number_across_years(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test same ISO week number in different years Tests that week 1 in 2022 and week 1 in 2023 are treated as different weeks. Both backups are within retention window to isolate the ISO year comparison bug. """ # Current: 2023-01-15 (Sunday, ISO week 2 of 2023) mock_datetime.now.return_value = datetime(2023, 1, 15) backups = [ add_backup("20230103_1200"), # Tue, ISO week 1 of 2023 add_backup("20220104_1200"), # Tue, ISO week 1 of 2022 ] # Both are within 60-week window, but different ISO years expected = backups run_cleanup(keep_weekly=60) check_backups(expected) @mock.patch(f"{bk.__name__}.datetime", wraps=datetime) def test_late_december_in_week_1_next_year(self, mock_datetime, add_backup, run_cleanup, check_backups): """Test late December dates in week 1 of next year In some years, Dec 29-31 belong to ISO week 1 of the next year (e.g., 2024-12-30 is in ISO 2025-W01). These should be grouped with early January dates of the same ISO week. """ # Current: 2025-01-15 (Wednesday, ISO week 3) mock_datetime.now.return_value = datetime(2025, 1, 15) backups = [ add_backup("20241230_1200"), # Mon, ISO 2025-W01 (late Dec!) add_backup("20241231_1200"), # Tue, ISO 2025-W01 add_backup("20250102_1200"), # Thu, ISO 2025-W01 add_backup("20250104_1200"), # Sat, ISO 2025-W01 add_backup("20250108_1200"), # Wed, ISO 2025-W02 add_backup("20250112_1200"), # Sun, ISO 2025-W02 ] # Keep weekly: oldest from week 1 (Dec 30) and week 2 (Jan 8) expected = [backups[0], backups[4]] run_cleanup(keep_weekly=10) check_backups(expected) class TestBackupLock: """Test suite for the fcntl.flock()-based backup lock.""" @staticmethod def _read_lock_pid(lock_path: str) -> int: with open(lock_path, "r") as f: return int(f.readline().strip()) def test_lock_creation(self, backup_dir): """Test that lock file is created with current PID""" backup_dir.mkdir() result = bk.set_backups_lock(str(backup_dir)) assert result lock_path = os.path.join(str(backup_dir), bk.LOCK_FILE) assert os.path.exists(lock_path) assert self._read_lock_pid(lock_path) == os.getpid() bk.release_backups_lock(str(backup_dir)) def test_lock_metadata_includes_start_time(self, backup_dir): """Test that lock file records a parseable start timestamp""" backup_dir.mkdir() bk.set_backups_lock(str(backup_dir)) lock_path = os.path.join(str(backup_dir), bk.LOCK_FILE) with open(lock_path, "r") as f: lines = f.read().splitlines() assert int(lines[0]) == os.getpid() datetime.fromisoformat(lines[1]) # must not raise bk.release_backups_lock(str(backup_dir)) def test_lock_prevents_concurrent_backup(self, backup_dir): """ Test that a second holder of the kernel lock blocks a non-forced acquisition attempt, using a real flock() held on a separate file descriptor to stand in for another process. """ backup_dir.mkdir() lock_path = os.path.join(str(backup_dir), bk.LOCK_FILE) other_fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o644) fcntl.flock(other_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) os.write(other_fd, b"12345\n2024-01-01T00:00:00\n") try: result = bk.set_backups_lock(str(backup_dir), force=False) assert not result finally: fcntl.flock(other_fd, fcntl.LOCK_UN) os.close(other_fd) @pytest.mark.parametrize("garbage", ["999999", "not a number", ""]) def test_lock_acquired_despite_garbage_content(self, backup_dir, garbage): """ A leftover lock file with stale/corrupted/empty content must not block acquisition: the kernel lock, not the file content, is what decides ownership. The old process's death already released it. """ backup_dir.mkdir() lock_path = os.path.join(str(backup_dir), bk.LOCK_FILE) with open(lock_path, "w") as f: f.write(garbage) result = bk.set_backups_lock(str(backup_dir)) assert result assert self._read_lock_pid(lock_path) == os.getpid() bk.release_backups_lock(str(backup_dir)) def test_lock_release(self, backup_dir): """Test that a held lock can be released and re-acquired""" backup_dir.mkdir() assert bk.set_backups_lock(str(backup_dir)) lock_path = os.path.join(str(backup_dir), bk.LOCK_FILE) assert os.path.exists(lock_path) bk.release_backups_lock(str(backup_dir)) # lock file itself is intentionally left in place (see comment in # release_backups_lock); the kernel lock must be free though assert bk.set_backups_lock(str(backup_dir)) bk.release_backups_lock(str(backup_dir)) def test_release_nonexistent_lock(self, backup_dir): """Releasing a lock never acquired by this process is a no-op""" backup_dir.mkdir() # Should not raise any exception bk.release_backups_lock(str(backup_dir)) def test_reacquiring_own_lock_does_not_leak_fd(self, backup_dir): """ A second set_backups_lock() call for a directory this process already holds must fail cleanly instead of silently overwriting the tracked descriptor (which would leak the original fd and let release_backups_lock() close only the newest one, unlocking nothing). """ backup_dir.mkdir() assert bk.set_backups_lock(str(backup_dir)) try: assert not bk.set_backups_lock(str(backup_dir)) finally: bk.release_backups_lock(str(backup_dir)) # The lock must be fully free after a single release. assert bk.set_backups_lock(str(backup_dir)) bk.release_backups_lock(str(backup_dir)) def test_release_only_releases_own_lock(self, backup_dir): """ release_backups_lock() must only release a lock this process acquired via set_backups_lock(), never one it merely knows about through the lock file's on-disk content. """ backup_dir.mkdir() lock_path = os.path.join(str(backup_dir), bk.LOCK_FILE) other_fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o644) fcntl.flock(other_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) try: # We never successfully acquired this lock ourselves. bk.release_backups_lock(str(backup_dir)) # The other holder's lock must still be intact: a fresh # non-blocking acquisition attempt must still fail. assert not bk.set_backups_lock(str(backup_dir), force=False) finally: fcntl.flock(other_fd, fcntl.LOCK_UN) os.close(other_fd) def test_force_waits_for_lock_release_without_killing(self, backup_dir): """ force=True must block until the current holder releases the lock on its own, and must never send a signal to do it. """ backup_dir.mkdir() lock_path = os.path.join(str(backup_dir), bk.LOCK_FILE) other_fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o644) fcntl.flock(other_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) os.write(other_fd, b"12345\n2024-01-01T00:00:00\n") def _release_other_after_delay(): time.sleep(0.5) fcntl.flock(other_fd, fcntl.LOCK_UN) os.close(other_fd) releaser = threading.Thread(target=_release_other_after_delay) with mock.patch("os.kill") as mock_kill: releaser.start() start = time.monotonic() result = bk.set_backups_lock(str(backup_dir), force=True) elapsed = time.monotonic() - start releaser.join() assert result # Must have actually waited for the release, not raced past it. assert elapsed >= 0.4 mock_kill.assert_not_called() assert self._read_lock_pid(lock_path) == os.getpid() bk.release_backups_lock(str(backup_dir)) def test_multiprocess_only_one_holder_and_crash_releases_lock( self, backup_dir ): """ End-to-end proof with a real second process (not a mock): - only one process can hold the lock at a time. - killing the holder with SIGKILL (simulating a crash) releases the kernel lock automatically, with no PID/metadata inspection needed on our side, and no other process is ever signaled. """ backup_dir.mkdir() lock_path = os.path.join(str(backup_dir), bk.LOCK_FILE) holder_script = ( "import fcntl, os, sys, time;" "fd = os.open(sys.argv[1], os.O_CREAT | os.O_RDWR, 0o644);" "fcntl.flock(fd, fcntl.LOCK_EX);" "print('locked', flush=True);" "time.sleep(30)" ) holder = subprocess.Popen( [sys.executable, "-c", holder_script, lock_path], stdout=subprocess.PIPE, text=True, ) try: # Wait for the holder to actually acquire the lock. line = holder.stdout.readline() assert line.strip() == "locked" # A second, non-forced acquisition attempt must fail while the # holder is alive. assert not bk.set_backups_lock(str(backup_dir), force=False) # Simulate the holder crashing. Do this outside the os.kill # patch below: Popen.kill() itself calls os.kill() and we only # want to observe *our* code's behavior, not intercept the # test's own process teardown. holder.kill() holder.wait(timeout=5) with mock.patch("os.kill") as mock_kill: # The kernel released the lock on process exit; we should # be able to acquire it now without having sent any signal # to the (now-dead) holder PID ourselves. assert bk.set_backups_lock(str(backup_dir), force=False) mock_kill.assert_not_called() finally: bk.release_backups_lock(str(backup_dir)) if holder.poll() is None: holder.kill() holder.wait(timeout=5) class TestBackupIteration: """Tests for internal backup iteration and validation functions.""" def test_is_backup_valid_backup(self, add_backup): """Test _is_backup recognizes valid backup directory""" backup = add_backup("20210101_120000") assert bk._is_backup(backup) def test_is_backup_missing_marker(self, backup_dir, tmp_path): """Test _is_backup rejects directory without marker""" backup_dir.mkdir() backup_path = backup_dir / "20210101_120000" backup_path.mkdir() # Create content but no marker (backup_path / "file.txt").write_text("content") entry = os.scandir(str(backup_dir)) backup = next(entry) entry.close() assert not bk._is_backup(backup) def test_is_backup_only_marker_no_content(self, backup_dir): """Test _is_backup rejects directory with only marker file""" backup_dir.mkdir() backup_path = backup_dir / "20210101_120000" backup_path.mkdir() # Create only marker, no content marker_name = f"{bk.BACKUP_MARKER}_20210101_120000" (backup_path / marker_name).touch() entry = os.scandir(str(backup_dir)) backup = next(entry) entry.close() assert not bk._is_backup(backup) def test_is_backup_invalid_name_format(self, backup_dir): """Test _is_backup rejects invalid directory name""" backup_dir.mkdir() backup_path = backup_dir / "not-a-backup" backup_path.mkdir() # Create marker and content marker_name = f"{bk.BACKUP_MARKER}_not-a-backup" (backup_path / marker_name).touch() (backup_path / "file.txt").write_text("content") entry = os.scandir(str(backup_dir)) backup = next(entry) entry.close() assert not bk._is_backup(backup) def test_iterate_backups_empty_directory(self, backup_dir): """Test _iterate_backups on empty directory""" backup_dir.mkdir() backups = list(bk._iterate_backups(str(backup_dir))) assert backups == [] def test_iterate_backups_mixed_contents(self, backup_dir, add_backup): """Test _iterate_backups filters non-backup entries""" # Create valid backups backup1 = add_backup("20210101_120000") backup2 = add_backup("20210102_120000") # Create invalid entries (backup_dir / "random_file.txt").write_text("not a backup") (backup_dir / "invalid_dir").mkdir() (backup_dir / bk.LOCK_FILE).touch() backups = sorted(bk._iterate_backups(str(backup_dir)), key=lambda e: e.name) assert len(backups) == 2 assert backups[0].name == backup1.name assert backups[1].name == backup2.name def test_iterate_backups_incomplete_backup(self, backup_dir): """Test _iterate_backups skips backup without marker""" backup_dir.mkdir() # Create complete backup complete = backup_dir / "20210101_120000" complete.mkdir() (complete / "file.txt").write_text("content") marker_name = f"{bk.BACKUP_MARKER}_20210101_120000" (complete / marker_name).touch() # Create incomplete backup (no marker) incomplete = backup_dir / "20210102_120000" incomplete.mkdir() (incomplete / "file.txt").write_text("content") backups = list(bk._iterate_backups(str(backup_dir))) assert len(backups) == 1 assert backups[0].name == "20210101_120000" def test_get_latest_backup_returns_most_recent(self, backup_dir, add_backup): """Test _get_latest_backup returns most recent backup""" add_backup("20210101_120000") add_backup("20210102_120000") latest = add_backup("20210103_120000") result = bk._get_latest_backup(str(backup_dir)) assert result is not None assert result.name == latest.name def test_get_latest_backup_empty_directory(self, backup_dir): """Test _get_latest_backup returns None for empty directory""" backup_dir.mkdir() result = bk._get_latest_backup(str(backup_dir)) assert result is None def test_get_latest_backup_no_valid_backups(self, backup_dir): """Test _get_latest_backup returns None with no valid backups""" backup_dir.mkdir() # Create incomplete backup incomplete = backup_dir / "20210101_120000" incomplete.mkdir() (incomplete / "file.txt").write_text("content") # no marker result = bk._get_latest_backup(str(backup_dir)) assert result is None def test_set_backup_marker_creates_marker(self, backup_dir): """Test set_backup_marker creates marker file""" backup_dir.mkdir() backup_path = backup_dir / "20210101_120000" backup_path.mkdir() backup_entry = bk.fs.PseudoDirEntry(str(backup_path)) bk.set_backup_marker(backup_entry) marker_name = f"{bk.BACKUP_MARKER}_20210101_120000" marker_path = backup_path / marker_name assert marker_path.exists() def test_set_backup_marker_records_repo_id(self, backup_dir): """Test set_backup_marker records the repository identity.""" backup_dir.mkdir() backup_path = backup_dir / "20210101_120000" backup_path.mkdir() backup_entry = bk.fs.PseudoDirEntry(str(backup_path)) bk.set_backup_marker(backup_entry) marker_path = backup_path / f"{bk.BACKUP_MARKER}_20210101_120000" manifest = bk._read_manifest(str(marker_path)) assert manifest["repo_id"] == bk._ensure_repo_id(str(backup_dir)) assert manifest["snapshot"] == "20210101_120000" def test_set_backup_marker_idempotent(self, backup_dir): """Test set_backup_marker is idempotent""" backup_dir.mkdir() backup_path = backup_dir / "20210101_120000" backup_path.mkdir() backup_entry = bk.fs.PseudoDirEntry(str(backup_path)) bk.set_backup_marker(backup_entry) # Call again - should not fail bk.set_backup_marker(backup_entry) marker_name = f"{bk.BACKUP_MARKER}_20210101_120000" marker_path = backup_path / marker_name assert marker_path.exists() class TestValidateTopology: """Tests for validate_topology source/backups_dir boundary checks.""" def test_disjoint_sources_ok(self, tmp_path): src1 = tmp_path / "src1" src2 = tmp_path / "src2" backups = tmp_path / "backups" src1.mkdir() src2.mkdir() backups.mkdir() bk.validate_topology([str(src1), str(src2)], str(backups)) def test_source_inside_backups_dir_rejected(self, tmp_path): backups = tmp_path / "backups" src = backups / "src" backups.mkdir() src.mkdir() with pytest.raises(bk.BackupFailedError): bk.validate_topology([str(src)], str(backups)) def test_backups_dir_inside_source_rejected(self, tmp_path): src = tmp_path / "src" backups = src / "backups" src.mkdir() backups.mkdir() with pytest.raises(bk.BackupFailedError): bk.validate_topology([str(src)], str(backups)) def test_source_equals_backups_dir_rejected(self, tmp_path): shared = tmp_path / "shared" shared.mkdir() with pytest.raises(bk.BackupFailedError): bk.validate_topology([str(shared)], str(shared)) def test_duplicate_source_via_symlink_rejected(self, tmp_path): src = tmp_path / "src" backups = tmp_path / "backups" alias = tmp_path / "alias" src.mkdir() backups.mkdir() alias.symlink_to(src) with pytest.raises(bk.BackupFailedError): bk.validate_topology([str(src), str(alias)], str(backups)) def test_duplicate_basename_rejected(self, tmp_path): parent_a = tmp_path / "a" parent_b = tmp_path / "b" backups = tmp_path / "backups" (parent_a / "data").mkdir(parents=True) (parent_b / "data").mkdir(parents=True) backups.mkdir() with pytest.raises(bk.BackupFailedError): bk.validate_topology( [str(parent_a / "data"), str(parent_b / "data")], str(backups), ) def test_parent_traversal_rejected(self, tmp_path): backups = tmp_path / "backups" outside = tmp_path / "outside" backups.mkdir() outside.mkdir() traversal_src = str(backups / ".." / "outside") bk.validate_topology([traversal_src], str(backups)) # sanity: same directory reached directly is rejected as an alias with pytest.raises(bk.BackupFailedError): bk.validate_topology([traversal_src, str(outside)], str(backups)) class TestDeletionBoundarySafety: """Tests that recursive deletion refuses unsafe targets.""" def test_refuses_to_delete_symlinked_snapshot_dir( self, backup_dir, add_backup, run_cleanup, tmp_path ): """A symlink standing in for a snapshot dir must never be recursively deleted, since shutil.rmtree would follow it.""" real_target = tmp_path / "outside_target" real_target.mkdir() (real_target / "canary").write_text("do-not-delete") add_backup("20220101_0000") # kept, forces the old one to be # evaluated for removal old_name = "20200101_0000" old_marker = f"{bk.BACKUP_MARKER}_{old_name}" os.symlink(str(real_target), os.path.join(str(backup_dir), old_name)) # marker must exist for _is_backup(), even though this is a symlink os.makedirs( os.path.join(str(real_target)), exist_ok=True ) (real_target / old_marker).write_text("") run_cleanup(keep_all=0) assert os.path.lexists(os.path.join(str(backup_dir), old_name)) assert (real_target / "canary").exists() def test_refuses_to_delete_snapshot_with_wrong_repo_id( self, backup_dir, add_backup, run_cleanup ): add_backup("20220101_0000") old = add_backup("20200101_0000") marker_path = bk._get_backup_marker(old).path with open(marker_path, "w") as f: f.write(json.dumps({"repo_id": "forged", "snapshot": old.name})) run_cleanup(keep_all=0) assert os.path.isdir(old.path) def test_deletes_legitimate_old_backup( self, backup_dir, add_backup, run_cleanup ): add_backup("20220101_0000") old = add_backup("20200101_0000") run_cleanup(keep_all=0) assert not os.path.exists(old.path) def test_cleanup_stale_staging_ignores_symlinked_staging_dir( self, backup_dir, tmp_path ): backup_dir.mkdir() real_target = tmp_path / "outside_target" real_target.mkdir() (real_target / "canary").write_text("do-not-delete") staging_name = f"{bk.STAGING_PREFIX}20220101_0000" os.symlink(str(real_target), os.path.join(str(backup_dir), staging_name)) bk.cleanup_stale_staging_dirs(str(backup_dir)) assert os.path.lexists(os.path.join(str(backup_dir), staging_name)) assert (real_target / "canary").exists() def test_deletes_legacy_backup_with_empty_marker( self, backup_dir, add_backup, run_cleanup ): """A pre-migration empty marker (no manifest) is adopted, not permanently protected, since it can't be a forged one.""" add_backup("20220101_0000") old = add_backup("20200101_0000") marker_path = bk._get_backup_marker(old).path open(marker_path, "w").close() # simulate pre-migration marker run_cleanup(keep_all=0) assert not os.path.exists(old.path)