204 lines
7.5 KiB
Python
204 lines
7.5 KiB
Python
"""
|
|
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"}
|