Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 14 additions & 12 deletions pulp_python/pytest_plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,8 @@

from pulp_python.tests.functional.constants import (
PYTHON_EGG_FILENAME,
PYTHON_EGG_URL,
PYTHON_FIXTURE_URL,
PYTHON_URL,
PYTHON_WHEEL_FILENAME,
PYTHON_WHEEL_URL,
PYTHON_XS_PROJECT_SPECIFIER,
)

Expand Down Expand Up @@ -239,12 +236,17 @@ def get_href(item):

@pytest.fixture(scope="session")
def python_package_dist_directory(tmp_path_factory, http_get):
"""Creates a temp dir to hold package distros for uploading."""
dist_dir = tmp_path_factory.mktemp("dist")
egg_file = dist_dir / PYTHON_EGG_FILENAME
wheel_file = dist_dir / PYTHON_WHEEL_FILENAME
with open(egg_file, "wb") as f:
f.write(http_get(PYTHON_EGG_URL))
with open(wheel_file, "wb") as f:
f.write(http_get(PYTHON_WHEEL_URL))
yield dist_dir, egg_file, wheel_file
"""Returns a factory that downloads packages into a temp directory."""

def _download(*urls):
dist_dir = tmp_path_factory.mktemp("dist")
paths = []
for url in urls:
filename = url.rsplit("/", 1)[-1]
path = dist_dir / filename
with open(path, "wb") as f:
f.write(http_get(url))
paths.append(path)
return (dist_dir, *paths)

return _download
30 changes: 14 additions & 16 deletions pulp_python/tests/functional/api/test_attestations.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
import json
import shutil
import subprocess
from pathlib import Path
from urllib.parse import urljoin

import pytest
Expand All @@ -10,6 +8,8 @@

from pulpcore.tests.functional.utils import PulpTaskError

from pulp_python.tests.functional.constants import PYTHON_FIXTURES_URL


@pytest.fixture(scope="session")
def twine_package():
Expand Down Expand Up @@ -207,24 +207,22 @@ def test_attestation_twine_upload(
python_content_summary,
python_empty_repo_distro,
python_package_dist_directory,
http_get,
monitor_task,
):
"""Tests that packages with attestations can be properly uploaded through Twine."""
packages_url = urljoin(PYTHON_FIXTURES_URL, "packages/")
filenames = ("twine-6.2.0.tar.gz", "twine-6.2.0-py3-none-any.whl")
dist_dir = python_package_dist_directory(*(urljoin(packages_url, f) for f in filenames))[0]

for filename in filenames:
provenance = json.loads(http_get(urljoin(packages_url, f"{filename}.provenance.json")))
attestation = provenance["attestation_bundles"][0]["attestations"][0]
with open(dist_dir / f"{filename}.publish.attestation", "w") as f:
json.dump(attestation, f)

repo, distro = python_empty_repo_distro()
url = urljoin(distro.base_url, "legacy/")
dist_dir, _, _ = python_package_dist_directory

# Copy attestation files from test assets to dist_dir
assets_dir = Path(__file__).parent.parent / "assets"
attestation_files = [
"shelf-reader-0.1.tar.gz.publish.attestation",
"shelf_reader-0.1-py2-none-any.whl.publish.attestation",
]
for attestation_file in attestation_files:
src = assets_dir / attestation_file
dst = dist_dir / attestation_file
shutil.copy2(src, dst)

username, password = "admin", "password"
subprocess.run(
(
Expand All @@ -245,7 +243,7 @@ def test_attestation_twine_upload(
tasks = pulpcore_bindings.TasksApi.list(reserved_resources=repo.pulp_href).results
for task in reversed(tasks):
t = monitor_task(task.pulp_href)
repo_ver_href = t.created_resources[0]
repo_ver_href = [r for r in t.created_resources if "versions" in r][0]

assert repo_ver_href.endswith("versions/2/")
summary = python_content_summary(repository_version=repo_ver_href)
Expand Down
20 changes: 14 additions & 6 deletions pulp_python/tests/functional/api/test_pypi_apis.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,13 @@
PYPI_SIMPLE_V1_JSON,
PYTHON_EGG_FILENAME,
PYTHON_EGG_SHA256,
PYTHON_EGG_URL,
PYTHON_FIXTURES_URL,
PYTHON_MD_PROJECT_SPECIFIER,
PYTHON_MD_PYPI_SUMMARY,
PYTHON_WHEEL_FILENAME,
PYTHON_WHEEL_SHA256,
PYTHON_WHEEL_URL,
SHELF_PYTHON_JSON,
TWINE_WHEEL_FILENAME,
TWINE_WHEEL_URL,
Expand Down Expand Up @@ -70,7 +73,7 @@ def test_package_upload(
):
"""Tests that packages can be uploaded."""
repo, distro = python_empty_repo_distro()
dist_dir, egg_file, wheel_file = python_package_dist_directory
dist_dir, egg_file, wheel_file = python_package_dist_directory(PYTHON_EGG_URL, PYTHON_WHEEL_URL)
url = urljoin(distro.base_url, "legacy/")
response = requests.post(
url,
Expand Down Expand Up @@ -100,7 +103,7 @@ def test_package_upload_session(
"""Tests that multiple uploads will be broken up into multiple tasks."""
repo, distro = python_empty_repo_distro()
url = urljoin(distro.base_url, "legacy/")
dist_dir, egg_file, wheel_file = python_package_dist_directory
dist_dir, egg_file, wheel_file = python_package_dist_directory(PYTHON_EGG_URL, PYTHON_WHEEL_URL)
session = requests.Session()
response = session.post(
url,
Expand Down Expand Up @@ -130,7 +133,7 @@ def test_package_upload_simple(
"""Tests that the package upload endpoint exposed at `/simple/` works."""
repo, distro = python_empty_repo_distro()
url = urljoin(distro.base_url, "simple/")
dist_dir, egg_file, wheel_file = python_package_dist_directory
dist_dir, egg_file, wheel_file = python_package_dist_directory(PYTHON_EGG_URL, PYTHON_WHEEL_URL)
response = requests.post(
url,
data={"sha256_digest": PYTHON_EGG_SHA256},
Expand All @@ -156,7 +159,7 @@ def test_package_upload_with_metadata(
"""
repo, distro = python_empty_repo_distro()
url = urljoin(distro.base_url, "simple/")
dist_dir, egg_file, wheel_file = python_package_dist_directory
dist_dir, egg_file, wheel_file = python_package_dist_directory(PYTHON_EGG_URL, PYTHON_WHEEL_URL)
response = requests.post(
url,
data={"sha256_digest": PYTHON_WHEEL_SHA256},
Expand All @@ -183,9 +186,14 @@ def test_twine_upload(
monitor_task,
):
"""Tests that packages can be properly uploaded through Twine."""
packages_url = urljoin(PYTHON_FIXTURES_URL, "packages/")
dist_dir, _, _ = python_package_dist_directory(
urljoin(packages_url, "pytz-2023.3.tar.gz"),
urljoin(packages_url, "pytz-2023.3-py2.py3-none-any.whl"),
)

repo, distro = python_empty_repo_distro()
url = urljoin(distro.base_url, "legacy/")
dist_dir, _, _ = python_package_dist_directory
username, password = "admin", "password"
subprocess.run(
(
Expand All @@ -205,7 +213,7 @@ def test_twine_upload(
tasks = pulpcore_bindings.TasksApi.list(reserved_resources=repo.pulp_href).results
for task in reversed(tasks):
t = monitor_task(task.pulp_href)
repo_ver_href = t.created_resources[-1]
repo_ver_href = [r for r in t.created_resources if "versions" in r][0]
summary = python_content_summary(repository_version=repo_ver_href)
assert summary.present["python.python"]["count"] == 2

Expand Down
6 changes: 3 additions & 3 deletions pulp_python/tests/functional/api/test_upload.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ def test_synchronous_package_upload_with_metadata(
def test_legacy_upload_invalid_protocol_version(
python_empty_repo_distro, python_package_dist_directory
):
_, egg_file, _ = python_package_dist_directory
_, egg_file, _ = python_package_dist_directory(PYTHON_EGG_URL, PYTHON_WHEEL_URL)
_, distro = python_empty_repo_distro()
url = urljoin(distro.base_url, "legacy/")
with open(egg_file, "rb") as f:
Expand All @@ -108,7 +108,7 @@ def test_legacy_upload_invalid_protocol_version(

@pytest.mark.parallel
def test_legacy_upload_invalid_filetype(python_empty_repo_distro, python_package_dist_directory):
_, egg_file, wheel_file = python_package_dist_directory
_, egg_file, wheel_file = python_package_dist_directory(PYTHON_EGG_URL, PYTHON_WHEEL_URL)
_, distro = python_empty_repo_distro()
url = urljoin(distro.base_url, "legacy/")
with open(egg_file, "rb") as f:
Expand Down Expand Up @@ -140,7 +140,7 @@ def test_legacy_upload_invalid_filetype(python_empty_repo_distro, python_package
def test_legacy_upload_invalid_metadata_version(
python_empty_repo_distro, python_package_dist_directory
):
_, egg_file, _ = python_package_dist_directory
_, egg_file, _ = python_package_dist_directory(PYTHON_EGG_URL, PYTHON_WHEEL_URL)
_, distro = python_empty_repo_distro()
url = urljoin(distro.base_url, "legacy/")
with open(egg_file, "rb") as f:
Expand Down

This file was deleted.

Loading
Loading