test_crud_content_unit.py

import pytest

from urllib.parse import urljoin
from pypi_simple import PyPISimple

from pulpcore.tests.functional.utils import PulpTaskError
from pulp_python.tests.functional.constants import (
    PYTHON_FIXTURES_URL,
    PYTHON_PACKAGE_DATA,
    PYTHON_EGG_FILENAME,
    PYTHON_EGG_URL,
    PYTHON_SM_FIXTURE_CHECKSUMS,
    PYTHON_WHEEL_FILENAME,
    PYTHON_WHEEL_URL,
)
from pulp_python.tests.functional.utils import ensure_metadata


def test_content_crud(
    python_bindings, pulpcore_bindings, python_repo_factory, download_python_file, monitor_task
):
    """Test CRUD python content unit."""
    monitor_task(pulpcore_bindings.OrphansCleanupApi.cleanup({"orphan_protection_time": 0}).task)
    python_file = download_python_file(PYTHON_EGG_FILENAME, PYTHON_EGG_URL)

    artifact = pulpcore_bindings.ArtifactsApi.create(python_file)

    # Test create
    content_body = {"relative_path": PYTHON_EGG_FILENAME, "artifact": artifact.pulp_href}
    response = python_bindings.ContentPackagesApi.create(**content_body)
    task = monitor_task(response.task)
    content = python_bindings.ContentPackagesApi.read(task.created_resources[0])
    for k, v in PYTHON_PACKAGE_DATA.items():
        assert getattr(content, k) == v

    # Test read
    result = python_bindings.ContentPackagesApi.list(filename=content.filename)
    assert result.count == 1
    assert result.results[0] == content

    # Test partial update
    with pytest.raises(AttributeError) as e:
        python_bindings.ContentPackagesApi.partial_update(content.pulp_href, {"filename": "te"})
    assert "object has no attribute 'partial_update'" in e.value.args[0]

    # Test delete
    with pytest.raises(AttributeError) as e:
        python_bindings.ContentPackagesApi.delete(content.pulp_href)
    assert "object has no attribute 'delete'" in e.value.args[0]

    monitor_task(pulpcore_bindings.OrphansCleanupApi.cleanup({"orphan_protection_time": 0}).task)

    # Test create w/ file
    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file": python_file}
    response = python_bindings.ContentPackagesApi.create(**content_body)
    task = monitor_task(response.task)
    content = python_bindings.ContentPackagesApi.read(task.created_resources[0])
    for k, v in PYTHON_PACKAGE_DATA.items():
        assert getattr(content, k) == v

    monitor_task(pulpcore_bindings.OrphansCleanupApi.cleanup({"orphan_protection_time": 0}).task)

    # Test create w/ file & repository
    repo = python_repo_factory()
    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
    task = monitor_task(response.task)
    assert len(task.created_resources) == 2
    content_search = python_bindings.ContentPackagesApi.list(
        repository_version_added=task.created_resources[0]
    )
    content = python_bindings.ContentPackagesApi.read(content_search.results[0].pulp_href)
    for k, v in PYTHON_PACKAGE_DATA.items():
        assert getattr(content, k) == v

    # Test duplicate upload
    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file": python_file}
    response = python_bindings.ContentPackagesApi.create(**content_body)
    task = monitor_task(response.task)
    assert task.created_resources[0] == content.pulp_href

    # Test upload same filename w/ different artifact
    second_python_url = urljoin(urljoin(PYTHON_FIXTURES_URL, "packages/"), "aiohttp-3.3.0.tar.gz")
    second_python_file = download_python_file("aiohttp-3.3.0.tar.gz", second_python_url)
    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file": second_python_file}
    response = python_bindings.ContentPackagesApi.create(**content_body)
    task = monitor_task(response.task)
    content2 = python_bindings.ContentPackagesApi.read(task.created_resources[0])
    assert content2.pulp_href != content.pulp_href

    # Test upload same filename w/ different artifacts in same repo
    # repo already has EGG_FILENAME w/ EGG_ARTIFACT, not upload EGG_FILENAME w/ AIO_ARTIFACT
    # and see that repo will only have the second content unit inside after upload
    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
    task = monitor_task(response.task)
    assert len(task.created_resources) == 2
    assert content2.pulp_href in task.created_resources
    repo_ver2 = task.created_resources[0]
    content_list = python_bindings.ContentPackagesApi.list(repository_version=repo_ver2)
    assert content_list.count == 1
    assert content_list.results[0].pulp_href == content2.pulp_href

    # Test upload w/ mismatched sha256
    # If we don't perform orphan cleanup here, the upload will fail with a different error :hmm:
    monitor_task(python_bindings.RepositoriesPythonApi.delete(repo.pulp_href).task)
    monitor_task(pulpcore_bindings.OrphansCleanupApi.cleanup({"orphan_protection_time": 0}).task)
    mismatch_sha256 = PYTHON_SM_FIXTURE_CHECKSUMS["aiohttp-3.3.0.tar.gz"]
    content_body = {
        "relative_path": PYTHON_EGG_FILENAME,
        "file": python_file,
        "sha256": mismatch_sha256,
    }
    with pytest.raises(PulpTaskError) as e:
        response = python_bindings.ContentPackagesApi.create(**content_body)
        monitor_task(response.task)
    msg = "[PLP0003]"
    assert msg in e.value.task.error["description"]


def test_content_create_new_metadata(
    delete_orphans_pre, download_python_file, monitor_task, python_bindings
):
    """
    Test the creation of python content unit with newly added core metadata (provides_extras,
    dynamic, license_expression, license_file).
    """
    python_egg_filename = "setuptools-80.9.0.tar.gz"
    python_egg_url = urljoin(urljoin(PYTHON_FIXTURES_URL, "packages/"), python_egg_filename)
    python_file = download_python_file(python_egg_filename, python_egg_url)

    body = {"relative_path": python_egg_filename, "file": python_file}
    response = python_bindings.ContentPackagesApi.create(**body)
    task = monitor_task(response.task)
    content = python_bindings.ContentPackagesApi.read(task.created_resources[0])

    python_package_data = {
        "filename": "setuptools-80.9.0.tar.gz",
        "provides_extras": '["test", "doc", "ssl", "certs",'
        ' "core", "check", "cover", "enabler", "type"]',
        "dynamic": '["license-file"]',
        "license_expression": "MIT",
        "license_file": '["LICENSE"]',
    }
    for k, v in python_package_data.items():
        assert getattr(content, k) == v


def get_package_url(package, filename):
    with PyPISimple() as client:
        page = client.get_project_page(package)
        for package in page.packages:
            if package.filename == filename:
                return package.url
        raise ValueError(f"Package {filename} not found")


@pytest.mark.parallel
def test_upload_metadata_23_spec(python_content_factory):
    """Test that packages using metadata spec 2.3 can be uploaded to pulp."""
    filename = "urllib3-2.2.2-py3-none-any.whl"
    url = get_package_url("urllib3", filename)
    content = python_content_factory(filename, url=url)
    assert content.metadata_version == "2.3"


@pytest.mark.parallel
def test_upload_requires_python(python_content_factory):
    filename = "pip-24.3.1-py3-none-any.whl"
    url = get_package_url("pip", filename)
    content = python_content_factory(filename, url=url)
    assert content.requires_python == ">=3.8"


@pytest.mark.parallel
def test_upload_metadata_24_spec(python_content_factory):
    """Test that packages using metadata spec 2.4 can be uploaded to pulp."""
    filename = "setuptools-80.9.0.tar.gz"
    url = get_package_url("setuptools", filename)
    content = python_content_factory(filename, url=url)
    assert content.metadata_version == "2.4"
    assert content.license_expression == "MIT"
    assert content.license_file == '["LICENSE"]'


@pytest.mark.parallel
def test_package_creation_with_metadata(
    pulp_content_url,
    python_content_factory,
    python_distribution_factory,
    python_repo,
):
    """
    Test that the creation of a Python wheel package creates a metadata artifact.
    """
    python_content_factory(
        repository=python_repo, relative_path=PYTHON_WHEEL_FILENAME, url=PYTHON_WHEEL_URL
    )
    distro = python_distribution_factory(repository=python_repo)

    # Test that metadata is accessible
    ensure_metadata(
        pulp_content_url, distro.base_path, PYTHON_WHEEL_FILENAME, "shelf-reader", "0.1"
    )


@pytest.mark.parallel
def test_disallow_package_substitution(
    monitor_task,
    python_bindings,
    python_repo_factory,
):
    """
    When allow_package_substitution=False, any new repository version that would substitute
    existing content (same filename, different sha256) is rejected. This applies to both
    content uploads and repository modify operations. Re-adding content with a matching
    sha256 succeeds idempotently.
    """
    repo = python_repo_factory(allow_package_substitution=False)
    msg1 = "Found duplicate packages being added with the same filename but different checksums."
    msg2 = "To allow this, set 'allow_package_substitution' to True on the repository."

    # First upload succeeds
    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file_url": PYTHON_EGG_URL}
    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
    task = monitor_task(response.task)
    content = python_bindings.ContentPackagesApi.read(task.created_resources[-1])
    assert content.filename == PYTHON_EGG_FILENAME

    # Re-upload same artifact with same filename — should succeed (idempotent)
    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
    task = monitor_task(response.task)
    assert content.pulp_href in task.created_resources
    repo = python_bindings.RepositoriesPythonApi.read(repo.pulp_href)
    assert repo.latest_version_href.endswith("/1/")

    # Upload a different artifact with the same filename — should be rejected
    second_filename = "pip-26.0.1.tar.gz"
    second_url = get_package_url("pip", second_filename)
    content_body2 = {"relative_path": PYTHON_EGG_FILENAME, "file_url": second_url}
    with pytest.raises(PulpTaskError) as exc:
        response = python_bindings.ContentPackagesApi.create(
            repository=repo.pulp_href, **content_body2
        )
        monitor_task(response.task)
    assert "[PLPY0003]" in exc.value.task.error["description"]
    assert msg1 in exc.value.task.error["description"]
    assert msg2 in exc.value.task.error["description"]

    # Also create the conflicting content without a repo, then try to add via modify
    response = python_bindings.ContentPackagesApi.create(**content_body2)
    task = monitor_task(response.task)
    content2 = python_bindings.ContentPackagesApi.read(task.created_resources[0])
    # When body only contains add_content_units, the request will be rejected
    body = {"add_content_units": [content2.pulp_href]}
    with pytest.raises(python_bindings.ApiException) as exc:
        python_bindings.RepositoriesPythonApi.modify(repo.pulp_href, body)
    assert msg1 in exc.value.body
    # Else when body contains other fields, the check will be delegated to the task
    body = {"add_content_units": [content2.pulp_href], "base_version": repo.latest_version_href}
    with pytest.raises(PulpTaskError) as exc:
        monitor_task(python_bindings.RepositoriesPythonApi.modify(repo.pulp_href, body).task)
    assert "[PLPY0003]" in exc.value.task.error["description"]
    assert msg1 in exc.value.task.error["description"]
    assert msg2 in exc.value.task.error["description"]

    # Verify the repository still has only the original content
    repo = python_bindings.RepositoriesPythonApi.read(repo.pulp_href)
    assert repo.latest_version_href.endswith("/1/")
    # Check that you can remove the conflicting content and add the new content
    body = {"remove_content_units": [content.pulp_href], "add_content_units": [content2.pulp_href]}
    response = python_bindings.RepositoriesPythonApi.modify(repo.pulp_href, body)
    task = monitor_task(response.task)
    repo = python_bindings.RepositoriesPythonApi.read(repo.pulp_href)
    assert repo.latest_version_href.endswith("/2/")


@pytest.mark.parallel
def test_package_substitution_allowed_by_default(
    monitor_task,
    python_bindings,
    python_repo_factory,
):
    """
    By default (allow_package_substitution=True), uploading a file with the same filename
    but different sha256 replaces the existing content in the repository.
    """
    repo = python_repo_factory()
    assert repo.allow_package_substitution is True

    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file_url": PYTHON_EGG_URL}
    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
    task = monitor_task(response.task)
    content1 = python_bindings.ContentPackagesApi.read(task.created_resources[-1])

    # Upload a different artifact with the same filename — should succeed and replace
    second_filename = "pip-26.0.tar.gz"
    second_url = get_package_url("pip", second_filename)
    content_body = {"relative_path": PYTHON_EGG_FILENAME, "file_url": second_url}
    response = python_bindings.ContentPackagesApi.create(repository=repo.pulp_href, **content_body)
    task = monitor_task(response.task)
    content2 = python_bindings.ContentPackagesApi.read(task.created_resources[-1])
    assert content2.pulp_href != content1.pulp_href

    # Verify the repo has only the new content
    repo = python_bindings.RepositoriesPythonApi.read(repo.pulp_href)
    content_list = python_bindings.ContentPackagesApi.list(
        repository_version=repo.latest_version_href
    )
    assert content_list.count == 1
    assert content_list.results[0].sha256 == content2.sha256