""" Basic tests for dfdr functionality. """ import pytest from pathlib import Path import tempfile import shutil import os from dfdr.config import Config from dfdr.checksum import calculate_md5, save_checksum, load_checksum, load_checksum_info from dfdr.exceptions import ConfigError, ChecksumError from dfdr.fetcher import Fetcher from dfdr.storage import Storage class TestConfig: """Test configuration management.""" def test_config_initialization(self): """Test that config initializes correctly.""" with tempfile.TemporaryDirectory() as tmpdir: config = Config(Path(tmpdir)) assert config.project_root == Path(tmpdir) assert config.dfdr_dir.exists() assert config.storage_dir.exists() def test_add_remote(self): """Test adding a local registry.""" with tempfile.TemporaryDirectory() as tmpdir: config = Config(Path(tmpdir)) registry_path = Path(tmpdir) / "test_registry" registry_path.mkdir() config.add_remote("test", str(registry_path)) remotes = config.list_remotes() assert "test" in remotes assert remotes["test"] == registry_path def test_duplicate_remote(self): """Test that adding duplicate local registry raises error.""" with tempfile.TemporaryDirectory() as tmpdir: config = Config(Path(tmpdir)) registry_path = Path(tmpdir) / "test_registry" registry_path.mkdir() config.add_remote("test", str(registry_path)) with pytest.raises(ConfigError): config.add_remote("test", str(Path(tmpdir) / "other_registry")) def test_remove_remote(self): """Test removing a local registry.""" with tempfile.TemporaryDirectory() as tmpdir: config = Config(Path(tmpdir)) registry_path = Path(tmpdir) / "test_registry" registry_path.mkdir() config.add_remote("test", str(registry_path)) config.remove_remote("test") remotes = config.list_remotes() assert "test" not in remotes def test_remove_nonexistent_remote(self): """Test that removing nonexistent remote raises error.""" with tempfile.TemporaryDirectory() as tmpdir: config = Config(Path(tmpdir)) with pytest.raises(ConfigError): config.remove_remote("nonexistent") class TestChecksum: """Test checksum functionality.""" def test_calculate_md5(self): """Test MD5 calculation.""" with tempfile.NamedTemporaryFile(mode='w', delete=False) as f: f.write("test content") f.flush() checksum = calculate_md5(Path(f.name)) # MD5 of "test content" expected = "9473fdd0d880a43c21b7778d34872157" assert checksum == expected assert len(checksum) == 32 # MD5 is 32 hex chars assert isinstance(checksum, str) Path(f.name).unlink() def test_save_and_load_checksum(self): """Test saving and loading checksums.""" with tempfile.TemporaryDirectory() as tmpdir: test_file = Path(tmpdir) / "test.txt" test_file.write_text("test content") # Calculate and save checksum checksum = calculate_md5(test_file) save_checksum(test_file, checksum, "test_registry", "test.txt") # Load and verify loaded_info = load_checksum_info(test_file) assert loaded_info is not None assert loaded_info["checksum"] == checksum assert loaded_info["registry_name"] == "test_registry" assert loaded_info["original_path"] == "test.txt" # Test the simple load_checksum function loaded_checksum = load_checksum(test_file) assert loaded_checksum == checksum # Check that .dfdr file exists dfdr_file = Path(f"{test_file}.dfdr") assert dfdr_file.exists() def test_load_nonexistent_checksum(self): """Test loading checksum for nonexistent file.""" with tempfile.TemporaryDirectory() as tmpdir: test_file = Path(tmpdir) / "nonexistent.txt" checksum = load_checksum(test_file) assert checksum is None class TestFetcher: """Test Fetcher functionality.""" def test_fetch_index(self): """Test fetching index from local registry.""" with tempfile.TemporaryDirectory() as tmpdir: registry_path = Path(tmpdir) / "test_registry" registry_path.mkdir() (registry_path / "file1.txt").touch() (registry_path / "file2.txt").touch() fetcher = Fetcher() files = fetcher.fetch_index(registry_path) assert set(files) == {"file1.txt", "file2.txt"} def test_fetch_file(self): """Test fetching a file from local registry.""" with tempfile.TemporaryDirectory() as tmpdir: registry_path = Path(tmpdir) / "test_registry" registry_path.mkdir() source_file = registry_path / "test_file.txt" source_file.write_text("test content") output_path = Path(tmpdir) / "output.txt" fetcher = Fetcher() fetcher.fetch_file(registry_path, "test_file.txt", output_path) assert output_path.exists() assert output_path.read_text() == "test content" class TestStorage: """Test Storage functionality.""" def test_add_file(self): """Test adding a file from local registry to working copy.""" with tempfile.TemporaryDirectory() as tmpdir: project_root = Path(tmpdir) / "project" project_root.mkdir() config = Config(project_root) registry_path = Path(tmpdir) / "test_registry" registry_path.mkdir() source_file = registry_path / "test_file.txt" source_file.write_text("test content") config.add_remote("test", str(registry_path)) storage = Storage(config) storage.add_file("test", "test_file.txt") working_copy_file = project_root / "test_file.txt" assert working_copy_file.exists() assert working_copy_file.read_text() == "test content" def test_fetch_all(self): """Test fetching all files from local registries.""" with tempfile.TemporaryDirectory() as tmpdir: project_root = Path(tmpdir) / "project" project_root.mkdir() config = Config(project_root) registry1_path = Path(tmpdir) / "registry1" registry1_path.mkdir() (registry1_path / "file1.txt").write_text("content1") registry2_path = Path(tmpdir) / "registry2" registry2_path.mkdir() (registry2_path / "file2.txt").write_text("content2") config.add_remote("reg1", str(registry1_path)) config.add_remote("reg2", str(registry2_path)) storage = Storage(config) fetched_files = storage.fetch_all() assert set(fetched_files.keys()) == {"reg1", "reg2"} assert set(fetched_files["reg1"]) == {"file1.txt"} assert set(fetched_files["reg2"]) == {"file2.txt"}