test_full_mirror.py

import subprocess
from hashlib import sha256
from random import sample
from urllib.parse import urljoin, urlsplit, urlunsplit

import pytest
import requests
from packaging.version import parse
from pypi_simple import ProjectPage

from pulp_python.tests.functional.constants import (
    PYPI_URL,
    PYTHON_SM_FIXTURE_RELEASES,
    PYTHON_SM_PROJECT_SPECIFIER,
    PYTHON_XS_FIXTURE_CHECKSUMS,
)


def test_pull_through_install(
    python_bindings, python_remote_factory, python_distribution_factory, delete_orphans_pre
):
    """Tests that a pull-through distro can be installed from."""
    remote = python_remote_factory(url=PYPI_URL, includes=[])
    distro = python_distribution_factory(remote=remote.pulp_href)
    PACKAGE = "pulpcore-releases"

    # Check if already installed
    stdout = subprocess.run(("pip", "list"), capture_output=True).stdout.decode("utf-8")
    if stdout.find(PACKAGE) != -1:
        subprocess.run(("pip", "uninstall", PACKAGE, "-y"))

    # Perform pull-through install
    host = urlsplit(distro.base_url).hostname
    url = f"{distro.base_url}simple/"
    cmd = ("pip", "install", "--no-deps", "--trusted-host", host, "-i", url, PACKAGE)
    subprocess.run(cmd, check=True)

    stdout = subprocess.run(("pip", "list"), capture_output=True).stdout.decode("utf-8")
    assert stdout.find(PACKAGE) != -1
    subprocess.run(("pip", "uninstall", PACKAGE, "-y"))
    content = python_bindings.ContentPackagesApi.list(name=PACKAGE)
    assert content.count == 1


@pytest.mark.parallel
def test_pull_through_simple(python_remote_factory, python_distribution_factory, pulp_content_url):
    """Tests that the simple page is properly modified when requesting a pull-through."""
    remote = python_remote_factory(url=PYPI_URL, includes=["shelf-reader"])
    distro = python_distribution_factory(remote=remote.pulp_href)

    url = f"{distro.base_url}simple/shelf-reader/"
    project_page = ProjectPage.from_response(requests.get(url), "shelf-reader")

    assert len(project_page.packages) == 2
    for package in project_page.packages:
        assert package.filename in PYTHON_XS_FIXTURE_CHECKSUMS
        relative_path = f"{distro.base_path}/{package.filename}?redirect="
        assert urljoin(pulp_content_url, relative_path) in package.url
        assert PYTHON_XS_FIXTURE_CHECKSUMS[package.filename] == package.digests["sha256"]


@pytest.mark.parallel
@pytest.mark.parametrize("media_type", ["application/vnd.pypi.simple.v1+json", "text/html"])
def test_pull_through_simple_media_types(
    media_type, python_remote_factory, python_distribution_factory
):
    """Tests pull-through with different media types (JSON and HTML)."""
    remote = python_remote_factory(url=PYPI_URL, includes=["shelf-reader"])
    distro = python_distribution_factory(remote=remote.pulp_href)

    url = f"{distro.base_url}simple/shelf-reader/"
    headers = {"Accept": media_type}
    response = requests.get(url, headers=headers)

    assert response.status_code == 200
    assert media_type in response.headers["Content-Type"]
    assert "X-PyPI-Last-Serial" in response.headers


@pytest.mark.parallel
def test_pull_through_filter(python_remote_factory, python_distribution_factory):
    """Tests that pull-through respects the includes/excludes filter on the remote."""
    remote = python_remote_factory(url=PYPI_URL, includes=["shelf-reader"])
    distro = python_distribution_factory(remote=remote.pulp_href)

    r = requests.get(f"{distro.base_url}simple/pulpcore/")
    assert r.status_code == 404
    assert r.text == "pulpcore does not exist."

    r = requests.get(f"{distro.base_url}simple/shelf-reader/")
    assert r.status_code == 200

    # Test complex include specifiers
    remote = python_remote_factory(includes=PYTHON_SM_PROJECT_SPECIFIER)
    distro = python_distribution_factory(remote=remote.pulp_href)
    for package, releases in PYTHON_SM_FIXTURE_RELEASES.items():
        url = f"{distro.base_url}simple/{package}/"
        project_page = ProjectPage.from_response(requests.get(url), package)
        packages = {p.filename for p in project_page.packages if not parse(p.version).is_prerelease}
        assert packages == set(releases)

    # Test exclude logic
    remote = python_remote_factory(includes=[], excludes=["django"])
    distro = python_distribution_factory(remote=remote.pulp_href)

    r = requests.get(f"{distro.base_url}simple/django/")
    assert r.status_code == 404
    assert r.text == "django does not exist."

    r = requests.get(f"{distro.base_url}simple/pulpcore/")
    assert r.status_code == 404
    assert r.text == "pulpcore does not exist."

    r = requests.get(f"{distro.base_url}simple/shelf-reader/")
    assert r.status_code == 200


@pytest.mark.parallel
def test_pull_through_with_repo(
    python_repo_factory,
    python_remote_factory,
    python_distribution_factory,
    python_bindings,
    pulpcore_bindings,
    monitor_task,
    has_pulp_plugin,
):
    """Tests that content is saved to repository."""
    remote = python_remote_factory(includes=[])
    repo = python_repo_factory()
    distro = python_distribution_factory(repository=repo.pulp_href, remote=remote.pulp_href)

    # Perform a download of aiohttp to ensure it's saved to the repo
    url = urljoin(distro.base_url, "simple/aiohttp/")
    project_page = ProjectPage.from_response(requests.get(url), "aiohttp")
    for package in sample(project_page.packages, 3):
        assert "?redirect=" in package.url
        r = requests.get(package.url)
        assert r.status_code == 200

    if has_pulp_plugin("core", max="3.73"):
        pytest.skip("Pull-through repository save added in 3.74")

    tasks = pulpcore_bindings.TasksApi.list(reserved_resources=repo.prn)
    assert tasks.count == 3

    for task in tasks.results:
        monitor_task(task.pulp_href)

    repo = python_bindings.RepositoriesPythonApi.read(repo.pulp_href)
    assert repo.latest_version_href[-2] == "3"
    content = python_bindings.ContentPackagesApi.list(repository_version=repo.latest_version_href)
    assert content.count == 3

    # Check that getting content that is already present doesn't trigger a task
    r = requests.get(package.url)
    assert r.status_code == 200
    tasks = pulpcore_bindings.TasksApi.list(reserved_resources=repo.prn)
    assert tasks.count == 3


@pytest.mark.parallel
def test_pull_through_local_only(
    python_remote_factory, python_distribution_factory, python_repo_with_sync
):
    """Tests that pull-through checks the repository if the package is not present on the remote."""
    remote = python_remote_factory(url=PYPI_URL, includes=["pulpcore"])
    repo = python_repo_with_sync(remote=remote)
    remote2 = python_remote_factory(includes=[])  # Fixtures does not have pulpcore
    distro = python_distribution_factory(repository=repo.pulp_href, remote=remote2.pulp_href)

    url = f"{distro.base_url}simple/pulpcore/"
    r = requests.get(url)
    assert r.status_code == 200
    assert "?redirect=" not in r.text

    url = f"{distro.base_url}simple/shelf-reader/"
    r = requests.get(url)
    assert r.status_code == 200
    assert "?redirect=" in r.text

    url = f"{distro.base_url}simple/pulp_python/"
    r = requests.get(url)
    assert r.status_code == 404
    assert r.text == "pulp-python does not exist."


@pytest.mark.parallel
def test_pull_through_filtering_bad_names(python_remote_factory, python_distribution_factory):
    """Tests that pull-through handles packages with invalid names gracefully."""
    # ipython version 0.13.X has improper named bdist_wininst, e.g ipython-0.13.py2-win-amd64.exe
    # pypi-simple expects the platform to go before the pyversion (for .exe), so when parsed the
    # version and package type will be None.
    remote = python_remote_factory(url=PYPI_URL, includes=["ipython"])
    distro = python_distribution_factory(remote=remote.pulp_href)

    url = f"{distro.base_url}simple/ipython/"
    response = requests.get(url)

    assert response.status_code == 200

    project_page = ProjectPage.from_response(response, "ipython")
    # Should have no packages with None version (they get filtered out)
    assert len(project_page.packages) > 0
    assert all(package.version is not None for package in project_page.packages)


@pytest.mark.parallel
def test_pull_through_metadata(python_remote_factory, python_distribution_factory):
    """
    Tests that metadata is correctly served when using pull-through.

    So when requesting the metadata url according to PEP 658 you should just need to add .metadata
    to the end of the url path. Since pull-through includes a redirect query parameter we need to
    test adding .metadata to the end of the url path vs adding it to the end of redirect query.
    """
    remote = python_remote_factory(includes=["pytz"])
    distro = python_distribution_factory(remote=remote.pulp_href)

    url = f"{distro.base_url}simple/pytz/"
    project_page = ProjectPage.from_response(requests.get(url), "pytz")
    filename1 = "pytz-2023.2-py2.py3-none-any.whl"
    filename2 = "pytz-2023.3-py2.py3-none-any.whl"
    package1 = next(p for p in project_page.packages if p.filename == filename1)
    package2 = next(p for p in project_page.packages if p.filename == filename2)
    assert package1.has_metadata
    assert package2.has_metadata

    # The correct way to get the metadata url: add to path (uv does this)
    parts1 = urlsplit(package1.url)
    url1 = urlunsplit((parts1[0], parts1[1], parts1[2] + ".metadata", parts1[3], parts1[4]))
    r = requests.get(url1)
    assert r.status_code == 200
    assert sha256(r.content).hexdigest() == package1.metadata_digests["sha256"]

    # The incorrect way to get the metadata url: add to end of string (pip does this)
    url2 = package2.url + ".metadata"
    r = requests.get(url2)
    assert r.status_code == 200
    assert sha256(r.content).hexdigest() == package2.metadata_digests["sha256"]


@pytest.mark.parallel
def test_pull_through_metadata_with_repo(
    python_repo_factory,
    python_remote_factory,
    python_distribution_factory,
    pulpcore_bindings,
):
    """Tests that metadata is correctly saved when using pull-through with a repository."""
    remote = python_remote_factory(url=PYPI_URL, includes=["pip"])
    repo = python_repo_factory()
    distro = python_distribution_factory(repository=repo.pulp_href, remote=remote.pulp_href)

    pip_url = f"{distro.base_url}simple/pip/"
    project_page = ProjectPage.from_response(requests.get(pip_url), "pip")
    filename = "pip-26.0.1-py3-none-any.whl"
    package = next(p for p in project_page.packages if p.filename == filename)
    assert package.has_metadata
    assert "?redirect=" in package.url

    # Retrieve the metadata and assert the content was not saved to the repository
    parts = urlsplit(package.url)
    url = urlunsplit((parts[0], parts[1], parts[2] + ".metadata", parts[3], parts[4]))
    r = requests.get(url)
    assert r.status_code == 200
    assert sha256(r.content).hexdigest() == package.metadata_digests["sha256"]
    project_page = ProjectPage.from_response(requests.get(pip_url), "pip")
    package = next(p for p in project_page.packages if p.filename == filename)
    assert package.has_metadata
    assert "?redirect=" in package.url

    # Now retrieve the package and assert the content was saved with metadata
    r = requests.get(package.url)
    assert r.status_code == 200
    pa = pulpcore_bindings.ArtifactsApi.list(sha256=package.digests["sha256"])
    assert pa.count == 1
    ma = pulpcore_bindings.ArtifactsApi.list(sha256=package.metadata_digests["sha256"])
    assert ma.count == 1

    # Check the simple page is updated to point to the local repository
    project_page = ProjectPage.from_response(requests.get(pip_url), "pip")
    package = next(p for p in project_page.packages if p.filename == filename)
    assert "?redirect=" not in package.url
    r = requests.get(package.metadata_url)
    assert r.status_code == 200
    assert sha256(r.content).hexdigest() == package.metadata_digests["sha256"]