orchard/backend/tests/test_ref_count.py

"""
Unit and integration tests for reference counting behavior.

Tests cover:
- ref_count is set correctly for new artifacts
- ref_count increments on duplicate uploads
- ref_count query correctly identifies existing artifacts
- Artifact lookup by SHA256 hash works correctly
"""

import pytest
import io
from tests.conftest import (
    compute_sha256,
    upload_test_file,
    TEST_CONTENT_HELLO,
    TEST_HASH_HELLO,
)


class TestRefCountQuery:
    """Tests for ref_count querying and artifact lookup."""

    @pytest.mark.integration
    def test_artifact_lookup_by_sha256(self, integration_client, test_package):
        """Test artifact lookup by SHA256 hash (primary key) works correctly."""
        project, package = test_package
        content = b"unique content for lookup test"
        expected_hash = compute_sha256(content)

        # Upload a file
        upload_result = upload_test_file(
            integration_client, project, package, content, tag="v1"
        )
        assert upload_result["artifact_id"] == expected_hash

        # Look up artifact by ID (SHA256)
        response = integration_client.get(f"/api/v1/artifact/{expected_hash}")
        assert response.status_code == 200

        artifact = response.json()
        assert artifact["id"] == expected_hash
        assert artifact["sha256"] == expected_hash
        assert artifact["size"] == len(content)

    @pytest.mark.integration
    def test_ref_count_query_identifies_existing_artifact(
        self, integration_client, test_package
    ):
        """Test ref_count query correctly identifies existing artifacts by hash."""
        project, package = test_package
        content = b"content for ref count query test"
        expected_hash = compute_sha256(content)

        # Upload a file with a tag
        upload_result = upload_test_file(
            integration_client, project, package, content, tag="v1"
        )

        # Query artifact and check ref_count
        response = integration_client.get(f"/api/v1/artifact/{expected_hash}")
        assert response.status_code == 200

        artifact = response.json()
        assert artifact["ref_count"] >= 1  # At least 1 from the tag

    @pytest.mark.integration
    def test_ref_count_set_to_1_for_new_artifact_with_tag(
        self, integration_client, test_package, unique_test_id
    ):
        """Test ref_count is set to 1 for new artifacts when created with a tag."""
        project, package = test_package
        content = f"brand new content for ref count test {unique_test_id}".encode()
        expected_hash = compute_sha256(content)

        # Upload a new file with a tag
        upload_result = upload_test_file(
            integration_client, project, package, content, tag="initial"
        )

        assert upload_result["artifact_id"] == expected_hash
        assert upload_result["ref_count"] == 1
        assert upload_result["deduplicated"] is False

    @pytest.mark.integration
    def test_ref_count_increments_on_duplicate_upload_with_tag(
        self, integration_client, test_package, unique_test_id
    ):
        """Test ref_count is incremented when duplicate content is uploaded with a new tag."""
        project, package = test_package
        content = f"content that will be uploaded twice {unique_test_id}".encode()
        expected_hash = compute_sha256(content)

        # First upload with tag
        result1 = upload_test_file(
            integration_client, project, package, content, tag="v1"
        )
        assert result1["ref_count"] == 1
        assert result1["deduplicated"] is False

        # Second upload with different tag (same content)
        result2 = upload_test_file(
            integration_client, project, package, content, tag="v2"
        )
        assert result2["artifact_id"] == expected_hash
        assert result2["ref_count"] == 2
        assert result2["deduplicated"] is True

    @pytest.mark.integration
    def test_ref_count_after_multiple_tags(self, integration_client, test_package):
        """Test ref_count correctly reflects number of tags pointing to artifact."""
        project, package = test_package
        content = b"content for multiple tag test"
        expected_hash = compute_sha256(content)

        # Upload with multiple tags
        tags = ["v1", "v2", "v3", "latest"]
        for i, tag in enumerate(tags):
            result = upload_test_file(
                integration_client, project, package, content, tag=tag
            )
            assert result["artifact_id"] == expected_hash
            assert result["ref_count"] == i + 1

        # Verify final ref_count via artifact endpoint
        response = integration_client.get(f"/api/v1/artifact/{expected_hash}")
        assert response.status_code == 200
        assert response.json()["ref_count"] == len(tags)


class TestRefCountWithDeletion:
    """Tests for ref_count behavior when tags are deleted."""

    @pytest.mark.integration
    def test_ref_count_decrements_on_tag_delete(self, integration_client, test_package):
        """Test ref_count decrements when a tag is deleted."""
        project, package = test_package
        content = b"content for delete test"
        expected_hash = compute_sha256(content)

        # Upload with two tags
        upload_test_file(integration_client, project, package, content, tag="v1")
        upload_test_file(integration_client, project, package, content, tag="v2")

        # Verify ref_count is 2
        response = integration_client.get(f"/api/v1/artifact/{expected_hash}")
        assert response.json()["ref_count"] == 2

        # Delete one tag
        delete_response = integration_client.delete(
            f"/api/v1/project/{project}/{package}/tags/v1"
        )
        assert delete_response.status_code == 204

        # Verify ref_count is now 1
        response = integration_client.get(f"/api/v1/artifact/{expected_hash}")
        assert response.json()["ref_count"] == 1

    @pytest.mark.integration
    def test_ref_count_zero_after_all_tags_deleted(
        self, integration_client, test_package
    ):
        """Test ref_count goes to 0 when all tags are deleted."""
        project, package = test_package
        content = b"content that will be orphaned"
        expected_hash = compute_sha256(content)

        # Upload with one tag
        upload_test_file(integration_client, project, package, content, tag="only-tag")

        # Delete the tag
        integration_client.delete(f"/api/v1/project/{project}/{package}/tags/only-tag")

        # Verify ref_count is 0
        response = integration_client.get(f"/api/v1/artifact/{expected_hash}")
        assert response.json()["ref_count"] == 0