268 lines
8.9 KiB
Python
268 lines
8.9 KiB
Python
"""Tests for Git bundle generation and application."""
|
|
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from radicle_reticulum.git_bundle import (
|
|
BundleMetadata,
|
|
BundleType,
|
|
GitBundle,
|
|
GitBundleGenerator,
|
|
GitBundleApplicator,
|
|
RADICLE_REF_PATTERNS,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_git_repo():
|
|
"""Create a temporary Git repository for testing."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir) / "test_repo"
|
|
repo_path.mkdir()
|
|
|
|
# Initialize repo
|
|
subprocess.run(["git", "init"], cwd=repo_path, check=True, capture_output=True)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "test@test.com"],
|
|
cwd=repo_path, check=True, capture_output=True
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "Test User"],
|
|
cwd=repo_path, check=True, capture_output=True
|
|
)
|
|
|
|
# Create initial commit
|
|
(repo_path / "README.md").write_text("# Test Repo\n")
|
|
subprocess.run(["git", "add", "README.md"], cwd=repo_path, check=True, capture_output=True)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "Initial commit"],
|
|
cwd=repo_path, check=True, capture_output=True
|
|
)
|
|
|
|
yield repo_path
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_bare_repo():
|
|
"""Create a temporary bare Git repository for testing."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir) / "bare_repo.git"
|
|
subprocess.run(["git", "init", "--bare", str(repo_path)], check=True, capture_output=True)
|
|
yield repo_path
|
|
|
|
|
|
class TestBundleMetadata:
|
|
"""Test BundleMetadata encoding/decoding."""
|
|
|
|
def test_encode_decode_full_bundle(self):
|
|
"""Test encode/decode of full bundle metadata."""
|
|
metadata = BundleMetadata(
|
|
bundle_type=BundleType.FULL,
|
|
repository_id="rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5",
|
|
source_node="did:key:z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK",
|
|
timestamp=1234567890123,
|
|
refs_included=["refs/heads/main", "refs/rad/id"],
|
|
prerequisites=[],
|
|
size_bytes=1024,
|
|
checksum=b"\x00" * 32,
|
|
)
|
|
|
|
encoded = metadata.encode()
|
|
decoded, consumed = BundleMetadata.decode(encoded)
|
|
|
|
assert decoded.bundle_type == metadata.bundle_type
|
|
assert decoded.repository_id == metadata.repository_id
|
|
assert decoded.source_node == metadata.source_node
|
|
assert decoded.timestamp == metadata.timestamp
|
|
assert decoded.refs_included == metadata.refs_included
|
|
assert decoded.prerequisites == metadata.prerequisites
|
|
assert decoded.size_bytes == metadata.size_bytes
|
|
assert decoded.checksum == metadata.checksum
|
|
|
|
def test_encode_decode_incremental_bundle(self):
|
|
"""Test encode/decode of incremental bundle metadata."""
|
|
metadata = BundleMetadata(
|
|
bundle_type=BundleType.INCREMENTAL,
|
|
repository_id="rad:test",
|
|
source_node="did:key:test",
|
|
timestamp=1000,
|
|
refs_included=["refs/heads/feature"],
|
|
prerequisites=["abc123", "def456"],
|
|
size_bytes=512,
|
|
checksum=b"\xff" * 32,
|
|
)
|
|
|
|
encoded = metadata.encode()
|
|
decoded, _ = BundleMetadata.decode(encoded)
|
|
|
|
assert decoded.bundle_type == BundleType.INCREMENTAL
|
|
assert decoded.prerequisites == ["abc123", "def456"]
|
|
|
|
|
|
class TestGitBundle:
|
|
"""Test GitBundle encoding/decoding."""
|
|
|
|
def test_encode_decode_roundtrip(self):
|
|
"""Test bundle encode/decode roundtrip."""
|
|
import hashlib
|
|
|
|
bundle_data = b"fake git bundle data"
|
|
metadata = BundleMetadata(
|
|
bundle_type=BundleType.FULL,
|
|
repository_id="rad:test",
|
|
source_node="did:key:test",
|
|
timestamp=1000,
|
|
refs_included=["refs/heads/main"],
|
|
prerequisites=[],
|
|
size_bytes=len(bundle_data),
|
|
checksum=hashlib.sha256(bundle_data).digest(),
|
|
)
|
|
|
|
bundle = GitBundle(metadata=metadata, data=bundle_data)
|
|
encoded = bundle.encode()
|
|
decoded = GitBundle.decode(encoded)
|
|
|
|
assert decoded.data == bundle_data
|
|
assert decoded.metadata.repository_id == metadata.repository_id
|
|
|
|
def test_checksum_verification_fails_on_corruption(self):
|
|
"""Test that checksum verification catches corruption."""
|
|
import hashlib
|
|
|
|
bundle_data = b"original data"
|
|
metadata = BundleMetadata(
|
|
bundle_type=BundleType.FULL,
|
|
repository_id="rad:test",
|
|
source_node="did:key:test",
|
|
timestamp=1000,
|
|
refs_included=[],
|
|
prerequisites=[],
|
|
size_bytes=len(bundle_data),
|
|
checksum=hashlib.sha256(bundle_data).digest(),
|
|
)
|
|
|
|
bundle = GitBundle(metadata=metadata, data=bundle_data)
|
|
encoded = bytearray(bundle.encode())
|
|
|
|
# Corrupt the data portion
|
|
encoded[-1] ^= 0xFF
|
|
|
|
with pytest.raises(ValueError, match="checksum mismatch"):
|
|
GitBundle.decode(bytes(encoded))
|
|
|
|
|
|
class TestGitBundleGenerator:
|
|
"""Test GitBundleGenerator."""
|
|
|
|
def test_get_refs(self, temp_git_repo):
|
|
"""Test getting refs from repository."""
|
|
generator = GitBundleGenerator(temp_git_repo)
|
|
refs = generator.get_refs(["refs/heads/*"])
|
|
|
|
assert "refs/heads/main" in refs or "refs/heads/master" in refs
|
|
|
|
def test_create_full_bundle(self, temp_git_repo):
|
|
"""Test creating a full bundle."""
|
|
generator = GitBundleGenerator(temp_git_repo)
|
|
|
|
bundle = generator.create_full_bundle(
|
|
repository_id="rad:test",
|
|
source_node="did:key:test",
|
|
)
|
|
|
|
assert bundle is not None
|
|
assert bundle.metadata.bundle_type == BundleType.FULL
|
|
assert bundle.metadata.size_bytes > 0
|
|
assert len(bundle.data) > 0
|
|
|
|
def test_create_incremental_bundle_no_changes(self, temp_git_repo):
|
|
"""Test that incremental bundle returns None when no changes."""
|
|
generator = GitBundleGenerator(temp_git_repo)
|
|
|
|
# Get current refs as basis
|
|
current_refs = generator.get_refs()
|
|
|
|
# Create incremental with same refs - should be None
|
|
bundle = generator.create_incremental_bundle(
|
|
repository_id="rad:test",
|
|
source_node="did:key:test",
|
|
basis_refs=current_refs,
|
|
)
|
|
|
|
assert bundle is None
|
|
|
|
def test_create_incremental_bundle_with_changes(self, temp_git_repo):
|
|
"""Test creating incremental bundle with new commits."""
|
|
generator = GitBundleGenerator(temp_git_repo)
|
|
|
|
# Get current refs as basis
|
|
basis_refs = generator.get_refs()
|
|
|
|
# Add new commit
|
|
(temp_git_repo / "new_file.txt").write_text("new content")
|
|
subprocess.run(["git", "add", "new_file.txt"], cwd=temp_git_repo, check=True, capture_output=True)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "Add new file"],
|
|
cwd=temp_git_repo, check=True, capture_output=True
|
|
)
|
|
|
|
# Create incremental
|
|
bundle = generator.create_incremental_bundle(
|
|
repository_id="rad:test",
|
|
source_node="did:key:test",
|
|
basis_refs=basis_refs,
|
|
)
|
|
|
|
assert bundle is not None
|
|
assert bundle.metadata.bundle_type == BundleType.INCREMENTAL
|
|
|
|
def test_invalid_repo_path(self):
|
|
"""Test that invalid repo path raises error."""
|
|
with pytest.raises(ValueError, match="Not a Git repository"):
|
|
GitBundleGenerator(Path("/nonexistent/path"))
|
|
|
|
|
|
class TestGitBundleApplicator:
|
|
"""Test GitBundleApplicator."""
|
|
|
|
def test_apply_bundle(self, temp_git_repo, temp_bare_repo):
|
|
"""Test applying a bundle to a repository."""
|
|
# Create bundle from source repo
|
|
generator = GitBundleGenerator(temp_git_repo)
|
|
bundle = generator.create_full_bundle(
|
|
repository_id="rad:test",
|
|
source_node="did:key:test",
|
|
)
|
|
|
|
# Apply to bare repo
|
|
applicator = GitBundleApplicator(temp_bare_repo)
|
|
applied_refs = applicator.apply_bundle(bundle)
|
|
|
|
assert len(applied_refs) > 0
|
|
assert any("main" in ref or "master" in ref for ref in applied_refs)
|
|
|
|
def test_verify_bundle(self, temp_git_repo, temp_bare_repo):
|
|
"""Test bundle verification."""
|
|
generator = GitBundleGenerator(temp_git_repo)
|
|
bundle = generator.create_full_bundle(
|
|
repository_id="rad:test",
|
|
source_node="did:key:test",
|
|
)
|
|
|
|
applicator = GitBundleApplicator(temp_bare_repo)
|
|
ok, msg = applicator.verify_bundle(bundle)
|
|
|
|
assert ok
|
|
assert "verified" in msg.lower() or msg == ""
|
|
|
|
def test_get_current_refs(self, temp_git_repo):
|
|
"""Test getting current refs."""
|
|
applicator = GitBundleApplicator(temp_git_repo)
|
|
refs = applicator.get_current_refs(["refs/heads/*"])
|
|
|
|
assert len(refs) > 0
|