dulwich/tests/test_partial_clone.py

# test_partial_clone.py -- Tests for partial clone filter specifications
# Copyright (C) 2024 Jelmer Vernooij <jelmer@jelmer.uk>
#
# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
# General Public License as published by the Free Software Foundation; version 2.0
# or (at your option) any later version. You can redistribute it and/or
# modify it under the terms of either of these two licenses.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# You should have received a copy of the licenses; if not, see
# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
# License, Version 2.0.
#

"""Tests for partial clone filter specifications."""

import os
import tempfile

from dulwich.object_store import MemoryObjectStore
from dulwich.objects import Blob, Tree
from dulwich.partial_clone import (
    BlobLimitFilter,
    BlobNoneFilter,
    CombineFilter,
    SparseOidFilter,
    TreeDepthFilter,
    filter_pack_objects,
    parse_filter_spec,
)
from dulwich.repo import Repo
from dulwich.tests.utils import make_commit

from . import TestCase


class ParseFilterSpecTests(TestCase):
    """Test parse_filter_spec function."""

    def test_parse_blob_none(self):
        """Test parsing 'blob:none' filter."""
        filter_spec = parse_filter_spec("blob:none")
        self.assertIsInstance(filter_spec, BlobNoneFilter)
        self.assertEqual("blob:none", filter_spec.to_spec_string())

    def test_parse_blob_none_bytes(self):
        """Test parsing 'blob:none' as bytes."""
        filter_spec = parse_filter_spec(b"blob:none")
        self.assertIsInstance(filter_spec, BlobNoneFilter)

    def test_parse_blob_limit_bytes(self):
        """Test parsing 'blob:limit=100' in bytes."""
        filter_spec = parse_filter_spec("blob:limit=100")
        self.assertIsInstance(filter_spec, BlobLimitFilter)
        self.assertEqual(100, filter_spec.limit)

    def test_parse_blob_limit_kb(self):
        """Test parsing 'blob:limit=10k'."""
        filter_spec = parse_filter_spec("blob:limit=10k")
        self.assertIsInstance(filter_spec, BlobLimitFilter)
        self.assertEqual(10 * 1024, filter_spec.limit)

    def test_parse_blob_limit_mb(self):
        """Test parsing 'blob:limit=5m'."""
        filter_spec = parse_filter_spec("blob:limit=5m")
        self.assertIsInstance(filter_spec, BlobLimitFilter)
        self.assertEqual(5 * 1024 * 1024, filter_spec.limit)

    def test_parse_blob_limit_gb(self):
        """Test parsing 'blob:limit=1g'."""
        filter_spec = parse_filter_spec("blob:limit=1g")
        self.assertIsInstance(filter_spec, BlobLimitFilter)
        self.assertEqual(1024 * 1024 * 1024, filter_spec.limit)

    def test_parse_tree_depth(self):
        """Test parsing 'tree:0' filter."""
        filter_spec = parse_filter_spec("tree:0")
        self.assertIsInstance(filter_spec, TreeDepthFilter)
        self.assertEqual(0, filter_spec.max_depth)

    def test_parse_tree_depth_nonzero(self):
        """Test parsing 'tree:3' filter."""
        filter_spec = parse_filter_spec("tree:3")
        self.assertIsInstance(filter_spec, TreeDepthFilter)
        self.assertEqual(3, filter_spec.max_depth)

    def test_parse_sparse_oid(self):
        """Test parsing 'sparse:oid=<oid>' filter."""
        oid = b"1234567890abcdef1234567890abcdef12345678"
        filter_spec = parse_filter_spec(f"sparse:oid={oid.decode('ascii')}")
        self.assertIsInstance(filter_spec, SparseOidFilter)
        self.assertEqual(oid, filter_spec.oid)

    def test_parse_combine(self):
        """Test parsing 'combine:blob:none+tree:0' filter."""
        filter_spec = parse_filter_spec("combine:blob:none+tree:0")
        self.assertIsInstance(filter_spec, CombineFilter)
        self.assertEqual(2, len(filter_spec.filters))
        self.assertIsInstance(filter_spec.filters[0], BlobNoneFilter)
        self.assertIsInstance(filter_spec.filters[1], TreeDepthFilter)

    def test_parse_combine_multiple(self):
        """Test parsing combine filter with 3+ filters."""
        filter_spec = parse_filter_spec("combine:blob:none+tree:0+blob:limit=1m")
        self.assertIsInstance(filter_spec, CombineFilter)
        self.assertEqual(3, len(filter_spec.filters))

    def test_parse_unknown_spec(self):
        """Test that unknown filter specs raise ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("unknown:spec")
        self.assertIn("Unknown filter specification", str(cm.exception))

    def test_parse_invalid_tree_depth(self):
        """Test that invalid tree depth raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("tree:invalid")
        self.assertIn("Invalid tree filter", str(cm.exception))

    def test_parse_invalid_blob_limit(self):
        """Test that invalid blob limit raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("blob:limit=invalid")
        self.assertIn("Invalid", str(cm.exception))

    def test_parse_empty_spec(self):
        """Test that empty filter spec raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("")
        self.assertIn("cannot be empty", str(cm.exception))

    def test_parse_blob_limit_no_value(self):
        """Test that blob:limit without value raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("blob:limit=")
        self.assertIn("requires a size value", str(cm.exception))

    def test_parse_tree_no_value(self):
        """Test that tree: without depth raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("tree:")
        self.assertIn("requires a depth value", str(cm.exception))

    def test_parse_tree_negative_depth(self):
        """Test that negative tree depth raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("tree:-1")
        self.assertIn("non-negative", str(cm.exception))

    def test_parse_sparse_oid_invalid_length(self):
        """Test that invalid OID length raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("sparse:oid=abc123")
        self.assertIn("40 or 64 hex chars", str(cm.exception))

    def test_parse_sparse_oid_invalid_hex(self):
        """Test that non-hex OID raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("sparse:oid=" + "x" * 40)
        self.assertIn("hexadecimal", str(cm.exception))

    def test_parse_combine_single_filter(self):
        """Test that combine with single filter raises ValueError."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("combine:blob:none")
        self.assertIn("at least two filters", str(cm.exception))

    def test_parse_unknown_with_helpful_message(self):
        """Test that unknown spec gives helpful error message."""
        with self.assertRaises(ValueError) as cm:
            parse_filter_spec("unknown:spec")
        error_msg = str(cm.exception)
        self.assertIn("Unknown filter specification", error_msg)
        self.assertIn("Supported formats", error_msg)
        self.assertIn("blob:none", error_msg)


class BlobNoneFilterTests(TestCase):
    """Test BlobNoneFilter class."""

    def test_should_include_blob(self):
        """Test that BlobNoneFilter excludes all blobs."""
        filter_spec = BlobNoneFilter()
        self.assertFalse(filter_spec.should_include_blob(0))
        self.assertFalse(filter_spec.should_include_blob(100))
        self.assertFalse(filter_spec.should_include_blob(1024 * 1024))

    def test_should_include_tree(self):
        """Test that BlobNoneFilter includes all trees."""
        filter_spec = BlobNoneFilter()
        self.assertTrue(filter_spec.should_include_tree(0))
        self.assertTrue(filter_spec.should_include_tree(1))
        self.assertTrue(filter_spec.should_include_tree(100))

    def test_to_spec_string(self):
        """Test conversion back to spec string."""
        filter_spec = BlobNoneFilter()
        self.assertEqual("blob:none", filter_spec.to_spec_string())

    def test_repr(self):
        """Test repr output."""
        filter_spec = BlobNoneFilter()
        self.assertEqual("BlobNoneFilter()", repr(filter_spec))


class BlobLimitFilterTests(TestCase):
    """Test BlobLimitFilter class."""

    def test_should_include_blob_under_limit(self):
        """Test that blobs under limit are included."""
        filter_spec = BlobLimitFilter(1024)
        self.assertTrue(filter_spec.should_include_blob(0))
        self.assertTrue(filter_spec.should_include_blob(512))
        self.assertTrue(filter_spec.should_include_blob(1024))

    def test_should_include_blob_over_limit(self):
        """Test that blobs over limit are excluded."""
        filter_spec = BlobLimitFilter(1024)
        self.assertFalse(filter_spec.should_include_blob(1025))
        self.assertFalse(filter_spec.should_include_blob(2048))

    def test_should_include_tree(self):
        """Test that BlobLimitFilter includes all trees."""
        filter_spec = BlobLimitFilter(1024)
        self.assertTrue(filter_spec.should_include_tree(0))
        self.assertTrue(filter_spec.should_include_tree(100))

    def test_to_spec_string_bytes(self):
        """Test conversion to spec string with bytes."""
        filter_spec = BlobLimitFilter(100)
        self.assertEqual("blob:limit=100", filter_spec.to_spec_string())

    def test_to_spec_string_kb(self):
        """Test conversion to spec string with KB."""
        filter_spec = BlobLimitFilter(10 * 1024)
        self.assertEqual("blob:limit=10k", filter_spec.to_spec_string())

    def test_to_spec_string_mb(self):
        """Test conversion to spec string with MB."""
        filter_spec = BlobLimitFilter(5 * 1024 * 1024)
        self.assertEqual("blob:limit=5m", filter_spec.to_spec_string())

    def test_to_spec_string_gb(self):
        """Test conversion to spec string with GB."""
        filter_spec = BlobLimitFilter(2 * 1024 * 1024 * 1024)
        self.assertEqual("blob:limit=2g", filter_spec.to_spec_string())

    def test_to_spec_string_not_round(self):
        """Test conversion to spec string with non-round size."""
        filter_spec = BlobLimitFilter(1500)
        self.assertEqual("blob:limit=1500", filter_spec.to_spec_string())

    def test_repr(self):
        """Test repr output."""
        filter_spec = BlobLimitFilter(1024)
        self.assertEqual("BlobLimitFilter(limit=1024)", repr(filter_spec))


class TreeDepthFilterTests(TestCase):
    """Test TreeDepthFilter class."""

    def test_should_include_blob(self):
        """Test that TreeDepthFilter includes all blobs."""
        filter_spec = TreeDepthFilter(0)
        self.assertTrue(filter_spec.should_include_blob(0))
        self.assertTrue(filter_spec.should_include_blob(1024))

    def test_should_include_tree_at_depth(self):
        """Test that trees at or below max_depth are included."""
        filter_spec = TreeDepthFilter(2)
        self.assertTrue(filter_spec.should_include_tree(0))
        self.assertTrue(filter_spec.should_include_tree(1))
        self.assertTrue(filter_spec.should_include_tree(2))

    def test_should_include_tree_beyond_depth(self):
        """Test that trees beyond max_depth are excluded."""
        filter_spec = TreeDepthFilter(2)
        self.assertFalse(filter_spec.should_include_tree(3))
        self.assertFalse(filter_spec.should_include_tree(10))

    def test_to_spec_string(self):
        """Test conversion back to spec string."""
        filter_spec = TreeDepthFilter(3)
        self.assertEqual("tree:3", filter_spec.to_spec_string())

    def test_repr(self):
        """Test repr output."""
        filter_spec = TreeDepthFilter(2)
        self.assertEqual("TreeDepthFilter(max_depth=2)", repr(filter_spec))


class SparseOidFilterTests(TestCase):
    """Test SparseOidFilter class."""

    def test_should_include_blob(self):
        """Test that SparseOidFilter includes all blobs."""
        oid = b"1234567890abcdef1234567890abcdef12345678"
        filter_spec = SparseOidFilter(oid)
        self.assertTrue(filter_spec.should_include_blob(0))
        self.assertTrue(filter_spec.should_include_blob(1024))

    def test_should_include_tree(self):
        """Test that SparseOidFilter includes all trees."""
        oid = b"1234567890abcdef1234567890abcdef12345678"
        filter_spec = SparseOidFilter(oid)
        self.assertTrue(filter_spec.should_include_tree(0))
        self.assertTrue(filter_spec.should_include_tree(10))

    def test_to_spec_string(self):
        """Test conversion back to spec string."""
        oid = b"1234567890abcdef1234567890abcdef12345678"
        filter_spec = SparseOidFilter(oid)
        expected = "sparse:oid=1234567890abcdef1234567890abcdef12345678"
        self.assertEqual(expected, filter_spec.to_spec_string())

    def test_repr(self):
        """Test repr output."""
        oid = b"1234567890abcdef1234567890abcdef12345678"
        filter_spec = SparseOidFilter(oid)
        self.assertIn("SparseOidFilter", repr(filter_spec))
        self.assertIn("1234567890abcdef1234567890abcdef12345678", repr(filter_spec))


class CombineFilterTests(TestCase):
    """Test CombineFilter class."""

    def test_should_include_blob_all_allow(self):
        """Test that blob is included when all filters allow it."""
        filters = [BlobLimitFilter(1024), BlobLimitFilter(2048)]
        filter_spec = CombineFilter(filters)
        self.assertTrue(filter_spec.should_include_blob(512))

    def test_should_include_blob_one_denies(self):
        """Test that blob is excluded when one filter denies it."""
        filters = [BlobLimitFilter(1024), BlobNoneFilter()]
        filter_spec = CombineFilter(filters)
        self.assertFalse(filter_spec.should_include_blob(512))

    def test_should_include_tree_all_allow(self):
        """Test that tree is included when all filters allow it."""
        filters = [TreeDepthFilter(2), TreeDepthFilter(3)]
        filter_spec = CombineFilter(filters)
        self.assertTrue(filter_spec.should_include_tree(1))

    def test_should_include_tree_one_denies(self):
        """Test that tree is excluded when one filter denies it."""
        filters = [TreeDepthFilter(2), TreeDepthFilter(1)]
        filter_spec = CombineFilter(filters)
        self.assertFalse(filter_spec.should_include_tree(2))

    def test_to_spec_string(self):
        """Test conversion back to spec string."""
        filters = [BlobNoneFilter(), TreeDepthFilter(0)]
        filter_spec = CombineFilter(filters)
        self.assertEqual("combine:blob:none+tree:0", filter_spec.to_spec_string())

    def test_repr(self):
        """Test repr output."""
        filters = [BlobNoneFilter()]
        filter_spec = CombineFilter(filters)
        self.assertIn("CombineFilter", repr(filter_spec))


class FilterPackObjectsTests(TestCase):
    """Test filter_pack_objects function."""

    def setUp(self):
        super().setUp()
        self.store = MemoryObjectStore()

        # Create test objects
        self.small_blob = Blob.from_string(b"small")
        self.large_blob = Blob.from_string(b"x" * 2000)
        self.tree = Tree()
        self.commit = make_commit(tree=self.tree.id)

        # Add objects to store
        self.store.add_object(self.small_blob)
        self.store.add_object(self.large_blob)
        self.store.add_object(self.tree)
        self.store.add_object(self.commit)

    def test_filter_blob_none(self):
        """Test that blob:none filter excludes all blobs."""
        object_ids = [
            self.small_blob.id,
            self.large_blob.id,
            self.tree.id,
            self.commit.id,
        ]

        filter_spec = BlobNoneFilter()
        filtered = filter_pack_objects(self.store, object_ids, filter_spec)

        # Should exclude both blobs but keep tree and commit
        self.assertNotIn(self.small_blob.id, filtered)
        self.assertNotIn(self.large_blob.id, filtered)
        self.assertIn(self.tree.id, filtered)
        self.assertIn(self.commit.id, filtered)

    def test_filter_blob_limit(self):
        """Test that blob:limit filter excludes blobs over size limit."""
        object_ids = [
            self.small_blob.id,
            self.large_blob.id,
            self.tree.id,
        ]

        # Set limit to 100 bytes
        filter_spec = BlobLimitFilter(100)
        filtered = filter_pack_objects(self.store, object_ids, filter_spec)

        # Should keep small blob but exclude large blob
        self.assertIn(self.small_blob.id, filtered)
        self.assertNotIn(self.large_blob.id, filtered)
        self.assertIn(self.tree.id, filtered)

    def test_filter_no_filter_keeps_all(self):
        """Test that without filtering all objects are kept."""
        # Create a filter that includes everything
        filter_spec = BlobLimitFilter(10000)  # Large limit

        object_ids = [
            self.small_blob.id,
            self.large_blob.id,
            self.tree.id,
            self.commit.id,
        ]

        filtered = filter_pack_objects(self.store, object_ids, filter_spec)

        # All objects should be included
        self.assertEqual(len(filtered), len(object_ids))
        for oid in object_ids:
            self.assertIn(oid, filtered)

    def test_filter_missing_object(self):
        """Test that missing objects are skipped without error."""
        from dulwich.objects import ObjectID

        fake_id = ObjectID(b"0" * 40)
        object_ids = [fake_id, self.small_blob.id]

        filter_spec = BlobNoneFilter()
        filtered = filter_pack_objects(self.store, object_ids, filter_spec)

        # Should skip the missing object
        self.assertNotIn(fake_id, filtered)

    def test_filter_combine(self):
        """Test combined filters."""
        object_ids = [
            self.small_blob.id,
            self.large_blob.id,
            self.tree.id,
        ]

        # Combine blob:limit with another filter
        filter_spec = CombineFilter([
            BlobLimitFilter(100),
            BlobNoneFilter(),  # This will exclude ALL blobs
        ])

        filtered = filter_pack_objects(self.store, object_ids, filter_spec)

        # Should exclude all blobs due to BlobNoneFilter
        self.assertNotIn(self.small_blob.id, filtered)
        self.assertNotIn(self.large_blob.id, filtered)
        self.assertIn(self.tree.id, filtered)


class PartialCloneIntegrationTests(TestCase):
    """Integration tests for partial clone with real repositories."""

    def setUp(self):
        super().setUp()
        self.repo_dir = tempfile.mkdtemp()
        self.addCleanup(self._cleanup)
        self.repo = Repo.init(self.repo_dir)

    def _cleanup(self):
        """Clean up test repository."""
        import shutil
        if os.path.exists(self.repo_dir):
            shutil.rmtree(self.repo_dir)

    def test_blob_none_filter_with_real_repo(self):
        """Test blob:none filter excludes blobs in real repository."""
        # Create a tree with files
        tree = Tree()

        # Add some blobs to the tree
        blob1 = Blob.from_string(b"file1 content")
        blob2 = Blob.from_string(b"file2 content")
        tree.add(b"file1.txt", 0o100644, blob1.id)
        tree.add(b"file2.txt", 0o100644, blob2.id)

        # Add objects to repo
        self.repo.object_store.add_object(blob1)
        self.repo.object_store.add_object(blob2)
        self.repo.object_store.add_object(tree)

        # Create commit
        commit = make_commit(tree=tree.id, message=b"Test commit")
        self.repo.object_store.add_object(commit)

        # Get all objects
        object_ids = [blob1.id, blob2.id, tree.id, commit.id]

        # Apply blob:none filter
        filter_spec = BlobNoneFilter()
        filtered = filter_pack_objects(
            self.repo.object_store, object_ids, filter_spec
        )

        # Verify blobs are excluded
        self.assertNotIn(blob1.id, filtered)
        self.assertNotIn(blob2.id, filtered)
        # But tree and commit are included
        self.assertIn(tree.id, filtered)
        self.assertIn(commit.id, filtered)

        # Verify we have only 2 objects (tree + commit)
        self.assertEqual(2, len(filtered))

    def test_blob_limit_filter_with_mixed_sizes(self):
        """Test blob:limit filter with mixed blob sizes."""
        tree = Tree()

        # Create blobs of different sizes
        small_blob = Blob.from_string(b"small")  # 5 bytes
        medium_blob = Blob.from_string(b"x" * 50)  # 50 bytes
        large_blob = Blob.from_string(b"y" * 500)  # 500 bytes

        tree.add(b"small.txt", 0o100644, small_blob.id)
        tree.add(b"medium.txt", 0o100644, medium_blob.id)
        tree.add(b"large.txt", 0o100644, large_blob.id)

        # Add to repo
        self.repo.object_store.add_object(small_blob)
        self.repo.object_store.add_object(medium_blob)
        self.repo.object_store.add_object(large_blob)
        self.repo.object_store.add_object(tree)

        commit = make_commit(tree=tree.id)
        self.repo.object_store.add_object(commit)

        # Test with 100 byte limit
        object_ids = [
            small_blob.id,
            medium_blob.id,
            large_blob.id,
            tree.id,
            commit.id,
        ]

        filter_spec = BlobLimitFilter(100)
        filtered = filter_pack_objects(
            self.repo.object_store, object_ids, filter_spec
        )

        # Small and medium should be included
        self.assertIn(small_blob.id, filtered)
        self.assertIn(medium_blob.id, filtered)
        # Large should be excluded
        self.assertNotIn(large_blob.id, filtered)
        # Tree and commit included
        self.assertIn(tree.id, filtered)
        self.assertIn(commit.id, filtered)

    def test_combined_filter_integration(self):
        """Test combined filters in real scenario."""
        tree = Tree()

        blob1 = Blob.from_string(b"content1")
        blob2 = Blob.from_string(b"x" * 1000)

        tree.add(b"file1.txt", 0o100644, blob1.id)
        tree.add(b"file2.txt", 0o100644, blob2.id)

        self.repo.object_store.add_object(blob1)
        self.repo.object_store.add_object(blob2)
        self.repo.object_store.add_object(tree)

        commit = make_commit(tree=tree.id)
        self.repo.object_store.add_object(commit)

        # Combine: limit to 500 bytes, but also apply blob:none
        # This should exclude ALL blobs (blob:none overrides limit)
        filter_spec = CombineFilter([
            BlobLimitFilter(500),
            BlobNoneFilter(),
        ])

        object_ids = [blob1.id, blob2.id, tree.id, commit.id]
        filtered = filter_pack_objects(
            self.repo.object_store, object_ids, filter_spec
        )

        # All blobs excluded
        self.assertNotIn(blob1.id, filtered)
        self.assertNotIn(blob2.id, filtered)
        # Only tree and commit
        self.assertEqual(2, len(filtered))