"""Step definitions for sandbox _fs_utils BDD coverage tests.""" from __future__ import annotations import os import shutil import stat import tempfile from typing import Any from behave import given, then, when from cleveragents.infrastructure.sandbox._fs_utils import ( backup_directory, safe_restore, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _create_file(path: str, content: str, mode: int = 0o644) -> None: """Create a file with given content and permissions.""" with open(path, "w") as f: f.write(content) os.chmod(path, mode) # Set a known mtime so we can verify preservation os.utime(path, (1000000.0, 1000000.0)) def _register_tmpdir_cleanup(context: Any) -> None: """Register a cleanup handler to remove _fs_tmpdir after the scenario.""" tmpdir = context._fs_tmpdir context.add_cleanup(lambda: shutil.rmtree(tmpdir, ignore_errors=True)) # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given("a source directory with files having varied permissions") def step_given_source_dir_with_varied_perms(context: Any) -> None: context._fs_tmpdir = tempfile.mkdtemp(prefix="ca-fs-test-") _register_tmpdir_cleanup(context) context._fs_src = os.path.join(context._fs_tmpdir, "src") os.makedirs(context._fs_src) _create_file(os.path.join(context._fs_src, "readable.txt"), "hello", 0o644) _create_file(os.path.join(context._fs_src, "executable.sh"), "#!/bin/sh", 0o755) sub = os.path.join(context._fs_src, "sub") os.makedirs(sub) _create_file(os.path.join(sub, "nested.txt"), "nested", 0o600) # Set distinctive directory timestamps so that the "directory # timestamps should match" assertion actually verifies preservation # rather than passing by coincidence when both dirs are "now". os.utime(sub, (2000000.0, 2000000.0)) os.utime(context._fs_src, (2000000.0, 2000000.0)) @given("an empty destination directory") def step_given_empty_dest_dir(context: Any) -> None: context._fs_dst = tempfile.mkdtemp(prefix="ca-fs-dst-", dir=context._fs_tmpdir) @given("a source directory with a symlink") def step_given_source_dir_with_symlink(context: Any) -> None: context._fs_tmpdir = tempfile.mkdtemp(prefix="ca-fs-test-") _register_tmpdir_cleanup(context) context._fs_src = os.path.join(context._fs_tmpdir, "src") os.makedirs(context._fs_src) _create_file(os.path.join(context._fs_src, "real.txt"), "real") os.symlink("real.txt", os.path.join(context._fs_src, "link.txt")) @given("a source directory with restricted permissions") def step_given_source_dir_restricted_perms(context: Any) -> None: context._fs_tmpdir = tempfile.mkdtemp(prefix="ca-fs-test-") _register_tmpdir_cleanup(context) context._fs_src = os.path.join(context._fs_tmpdir, "src") os.makedirs(context._fs_src, mode=0o750) _create_file(os.path.join(context._fs_src, "data.txt"), "data") @given("an original directory with known content") def step_given_original_dir_known_content(context: Any) -> None: context._fs_tmpdir = tempfile.mkdtemp(prefix="ca-fs-test-") _register_tmpdir_cleanup(context) context._fs_original = os.path.join(context._fs_tmpdir, "original") os.makedirs(context._fs_original) _create_file(os.path.join(context._fs_original, "file.txt"), "original-content") context._fs_original_content = "original-content" @given("a backup directory with different content") def step_given_backup_dir_different_content(context: Any) -> None: context._fs_backup = os.path.join(context._fs_tmpdir, "backup") os.makedirs(context._fs_backup) _create_file(os.path.join(context._fs_backup, "file.txt"), "backup-content") @given("a backup path that will cause rename to fail") def step_given_backup_path_that_fails(context: Any) -> None: # Use a non-existent path as backup — os.rename will fail context._fs_backup = os.path.join(context._fs_tmpdir, "nonexistent-backup") @given("a stale atomic-rollback-old directory exists") def step_given_stale_rollback_dir(context: Any) -> None: # Legacy step preserved for backward compatibility. stale = context._fs_original + ".atomic-rollback-old" os.makedirs(stale, exist_ok=True) _create_file(os.path.join(stale, "stale.txt"), "stale") context._fs_stale_path = stale # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when("I call backup_directory from source to destination") def step_when_backup_directory(context: Any) -> None: context._fs_error = None try: backup_directory(context._fs_src, context._fs_dst) except Exception as exc: context._fs_error = exc @when("I call safe_restore from backup to original") def step_when_safe_restore(context: Any) -> None: context._fs_error = None try: safe_restore(context._fs_backup, context._fs_original) except Exception as exc: context._fs_error = exc @when("I call safe_restore expecting an error") def step_when_safe_restore_expecting_error(context: Any) -> None: context._fs_error = None try: safe_restore(context._fs_backup, context._fs_original) except Exception as exc: context._fs_error = exc # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then("the destination should contain all source files") def step_then_dest_contains_all_files(context: Any) -> None: src_files: set[str] = set() for dirpath, _dirnames, filenames in os.walk(context._fs_src): for f in filenames: rel = os.path.relpath(os.path.join(dirpath, f), context._fs_src) src_files.add(rel) dst_files: set[str] = set() for dirpath, _dirnames, filenames in os.walk(context._fs_dst): for f in filenames: rel = os.path.relpath(os.path.join(dirpath, f), context._fs_dst) dst_files.add(rel) assert src_files == dst_files, f"File mismatch: src={src_files}, dst={dst_files}" @then("file permissions should match between source and destination") def step_then_permissions_match(context: Any) -> None: for dirpath, _dirnames, filenames in os.walk(context._fs_src): for f in filenames: src_path = os.path.join(dirpath, f) rel = os.path.relpath(src_path, context._fs_src) dst_path = os.path.join(context._fs_dst, rel) if os.path.islink(src_path): continue src_mode = stat.S_IMODE(os.stat(src_path).st_mode) dst_mode = stat.S_IMODE(os.stat(dst_path).st_mode) assert src_mode == dst_mode, ( f"Permission mismatch for {rel}: " f"src={oct(src_mode)}, dst={oct(dst_mode)}" ) @then("file timestamps should match between source and destination") def step_then_timestamps_match(context: Any) -> None: for dirpath, _dirnames, filenames in os.walk(context._fs_src): for f in filenames: src_path = os.path.join(dirpath, f) rel = os.path.relpath(src_path, context._fs_src) dst_path = os.path.join(context._fs_dst, rel) if os.path.islink(src_path): continue src_mtime = os.stat(src_path).st_mtime dst_mtime = os.stat(dst_path).st_mtime assert abs(src_mtime - dst_mtime) < 1.0, ( f"Timestamp mismatch for {rel}: src={src_mtime}, dst={dst_mtime}" ) @then("directory timestamps should match between source and destination") def step_then_dir_timestamps_match(context: Any) -> None: for dirpath, dirnames, _filenames in os.walk(context._fs_src): for dname in dirnames: src_dir = os.path.join(dirpath, dname) rel = os.path.relpath(src_dir, context._fs_src) dst_dir = os.path.join(context._fs_dst, rel) if os.path.islink(src_dir): continue src_mtime = os.stat(src_dir).st_mtime dst_mtime = os.stat(dst_dir).st_mtime assert abs(src_mtime - dst_mtime) < 1.0, ( f"Directory timestamp mismatch for {rel}: " f"src={src_mtime}, dst={dst_mtime}" ) # Also check root directory src_mtime = os.stat(context._fs_src).st_mtime dst_mtime = os.stat(context._fs_dst).st_mtime assert abs(src_mtime - dst_mtime) < 1.0, ( f"Root directory timestamp mismatch: src={src_mtime}, dst={dst_mtime}" ) @then("the destination should contain a symlink with the same target") def step_then_dest_has_symlink(context: Any) -> None: link_path = os.path.join(context._fs_dst, "link.txt") assert os.path.islink(link_path), f"Expected symlink at {link_path}" target = os.readlink(link_path) assert target == "real.txt", f"Expected symlink target 'real.txt', got '{target}'" @then("the destination root permissions should match the source root permissions") def step_then_root_permissions_match(context: Any) -> None: src_mode = stat.S_IMODE(os.stat(context._fs_src).st_mode) dst_mode = stat.S_IMODE(os.stat(context._fs_dst).st_mode) assert src_mode == dst_mode, ( f"Root permission mismatch: src={oct(src_mode)}, dst={oct(dst_mode)}" ) @then("the original should contain the backup content") def step_then_original_has_backup_content(context: Any) -> None: assert context._fs_error is None, f"Unexpected error: {context._fs_error}" file_path = os.path.join(context._fs_original, "file.txt") assert os.path.isfile(file_path), f"Expected file at {file_path}" with open(file_path) as f: content = f.read() assert content == "backup-content", f"Expected 'backup-content', got '{content}'" @then("the backup directory should have been removed") def step_then_backup_removed(context: Any) -> None: assert not os.path.exists(context._fs_backup), ( f"Backup directory should have been removed: {context._fs_backup}" ) @then("the original directory should still contain its original content") def step_then_original_still_has_original_content(context: Any) -> None: file_path = os.path.join(context._fs_original, "file.txt") assert os.path.isfile(file_path), f"Expected file at {file_path}" with open(file_path) as f: content = f.read() assert content == context._fs_original_content, ( f"Expected '{context._fs_original_content}', got '{content}'" ) @then("no stale rollback directory should remain") def step_then_no_stale_rollback_dir(context: Any) -> None: stale = context._fs_original + ".atomic-rollback-old" assert not os.path.exists(stale), ( f"Stale rollback directory should have been removed: {stale}" ) @then("no atomic-rollback-old directories should remain in the parent") def step_then_no_rollback_dirs_in_parent(context: Any) -> None: """Verify that no .atomic-rollback-old-* temp directories remain.""" parent_dir = os.path.dirname(context._fs_original) leftovers = [ entry for entry in os.listdir(parent_dir) if entry.startswith(".atomic-rollback-old-") ] assert len(leftovers) == 0, ( f"Found leftover rollback directories in {parent_dir}: {leftovers}" )