00fb2729e4
The test was checking for the wrong URL pattern. When urljoin resolves ../../packages/ab/cd/... relative to /api/pypi/pypi-remote/simple/requests/, it correctly produces /api/pypi/pypi-remote/packages/ab/cd/... (not /api/pypi/packages/...).
138 lines
5.8 KiB
Python
138 lines
5.8 KiB
Python
"""Integration tests for PyPI transparent proxy."""
|
|
|
|
import os
|
|
import pytest
|
|
import httpx
|
|
|
|
|
|
def get_base_url():
|
|
"""Get the base URL for the Orchard server from environment."""
|
|
return os.environ.get("ORCHARD_TEST_URL", "http://localhost:8080")
|
|
|
|
|
|
class TestPyPIProxyEndpoints:
|
|
"""Tests for PyPI proxy endpoints.
|
|
|
|
These endpoints are public (no auth required) since pip needs to use them.
|
|
"""
|
|
|
|
@pytest.mark.integration
|
|
def test_pypi_simple_index(self):
|
|
"""Test that /pypi/simple/ returns HTML response."""
|
|
with httpx.Client(base_url=get_base_url(), timeout=30.0) as client:
|
|
response = client.get("/pypi/simple/")
|
|
# Returns 200 if sources configured, 503 if not
|
|
assert response.status_code in (200, 503)
|
|
if response.status_code == 200:
|
|
assert "text/html" in response.headers.get("content-type", "")
|
|
else:
|
|
assert "No PyPI upstream sources configured" in response.json()["detail"]
|
|
|
|
@pytest.mark.integration
|
|
def test_pypi_package_endpoint(self):
|
|
"""Test that /pypi/simple/{package}/ returns appropriate response."""
|
|
with httpx.Client(base_url=get_base_url(), timeout=30.0) as client:
|
|
response = client.get("/pypi/simple/requests/")
|
|
# Returns 200 if sources configured and package found,
|
|
# 404 if package not found, 503 if no sources
|
|
assert response.status_code in (200, 404, 503)
|
|
if response.status_code == 200:
|
|
assert "text/html" in response.headers.get("content-type", "")
|
|
elif response.status_code == 404:
|
|
assert "not found" in response.json()["detail"].lower()
|
|
else: # 503
|
|
assert "No PyPI upstream sources configured" in response.json()["detail"]
|
|
|
|
@pytest.mark.integration
|
|
def test_pypi_download_missing_upstream_param(self):
|
|
"""Test that /pypi/simple/{package}/{filename} requires upstream param."""
|
|
with httpx.Client(base_url=get_base_url(), timeout=30.0) as client:
|
|
response = client.get("/pypi/simple/requests/requests-2.31.0.tar.gz")
|
|
assert response.status_code == 400
|
|
assert "upstream" in response.json()["detail"].lower()
|
|
|
|
|
|
class TestPyPILinkRewriting:
|
|
"""Tests for URL rewriting in PyPI proxy responses."""
|
|
|
|
def test_rewrite_package_links(self):
|
|
"""Test that download links are rewritten to go through proxy."""
|
|
from app.pypi_proxy import _rewrite_package_links
|
|
|
|
html = '''
|
|
<html>
|
|
<body>
|
|
<a href="https://files.pythonhosted.org/packages/ab/cd/requests-2.31.0.tar.gz#sha256=abc123">requests-2.31.0.tar.gz</a>
|
|
<a href="https://files.pythonhosted.org/packages/ef/gh/requests-2.31.0-py3-none-any.whl#sha256=def456">requests-2.31.0-py3-none-any.whl</a>
|
|
</body>
|
|
</html>
|
|
'''
|
|
|
|
# upstream_base_url is used to resolve relative URLs (not needed here since URLs are absolute)
|
|
result = _rewrite_package_links(
|
|
html,
|
|
"http://localhost:8080",
|
|
"requests",
|
|
"https://pypi.org/simple/requests/"
|
|
)
|
|
|
|
# Links should be rewritten to go through our proxy
|
|
assert "/pypi/simple/requests/requests-2.31.0.tar.gz?upstream=" in result
|
|
assert "/pypi/simple/requests/requests-2.31.0-py3-none-any.whl?upstream=" in result
|
|
# Original URLs should be encoded in upstream param
|
|
assert "files.pythonhosted.org" in result
|
|
# Hash fragments should be preserved
|
|
assert "#sha256=abc123" in result
|
|
assert "#sha256=def456" in result
|
|
|
|
def test_rewrite_relative_links(self):
|
|
"""Test that relative URLs are resolved to absolute URLs."""
|
|
from app.pypi_proxy import _rewrite_package_links
|
|
|
|
# Artifactory-style relative URLs
|
|
html = '''
|
|
<html>
|
|
<body>
|
|
<a href="../../packages/ab/cd/requests-2.31.0.tar.gz#sha256=abc123">requests-2.31.0.tar.gz</a>
|
|
</body>
|
|
</html>
|
|
'''
|
|
|
|
result = _rewrite_package_links(
|
|
html,
|
|
"https://orchard.example.com",
|
|
"requests",
|
|
"https://artifactory.example.com/api/pypi/pypi-remote/simple/requests/"
|
|
)
|
|
|
|
# The relative URL should be resolved to absolute
|
|
# ../../packages/ab/cd/... from /api/pypi/pypi-remote/simple/requests/ resolves to /api/pypi/pypi-remote/packages/ab/cd/...
|
|
assert "upstream=https%3A%2F%2Fartifactory.example.com%2Fapi%2Fpypi%2Fpypi-remote%2Fpackages" in result
|
|
# Hash fragment should be preserved
|
|
assert "#sha256=abc123" in result
|
|
|
|
|
|
class TestPyPIPackageNormalization:
|
|
"""Tests for PyPI package name normalization."""
|
|
|
|
@pytest.mark.integration
|
|
def test_package_name_normalized(self):
|
|
"""Test that package names are normalized per PEP 503.
|
|
|
|
Different capitalizations/separators should all be valid paths.
|
|
The endpoint normalizes to lowercase with hyphens before lookup.
|
|
"""
|
|
with httpx.Client(base_url=get_base_url(), timeout=30.0) as client:
|
|
# Test various name formats - all should be valid endpoint paths
|
|
for package_name in ["Requests", "some_package", "some-package"]:
|
|
response = client.get(f"/pypi/simple/{package_name}/")
|
|
# 200 = found, 404 = not found, 503 = no sources configured
|
|
assert response.status_code in (200, 404, 503), \
|
|
f"Unexpected status {response.status_code} for {package_name}"
|
|
|
|
# Verify response is appropriate for the status code
|
|
if response.status_code == 200:
|
|
assert "text/html" in response.headers.get("content-type", "")
|
|
elif response.status_code == 503:
|
|
assert "No PyPI upstream sources configured" in response.json()["detail"]
|