968cb00477
When resolving dependencies like certifi@2025.10.5, the bare version string "2025.10.5" was being rejected as an invalid SpecifierSet and falling back to wildcard, which fetched the latest version instead. Now bare versions starting with a digit are automatically prefixed with "==" to create an exact match constraint.
301 lines
10 KiB
Python
301 lines
10 KiB
Python
"""Unit tests for registry client functionality."""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
import httpx
|
|
from packaging.specifiers import SpecifierSet
|
|
|
|
from app.registry_client import (
|
|
PyPIRegistryClient,
|
|
VersionInfo,
|
|
FetchResult,
|
|
get_registry_client,
|
|
)
|
|
|
|
|
|
class TestPyPIRegistryClient:
|
|
"""Tests for PyPI registry client."""
|
|
|
|
@pytest.fixture
|
|
def mock_http_client(self):
|
|
"""Create a mock async HTTP client."""
|
|
return AsyncMock(spec=httpx.AsyncClient)
|
|
|
|
@pytest.fixture
|
|
def client(self, mock_http_client):
|
|
"""Create a PyPI registry client with mocked HTTP."""
|
|
return PyPIRegistryClient(
|
|
http_client=mock_http_client,
|
|
upstream_sources=[],
|
|
pypi_api_url="https://pypi.org/pypi",
|
|
)
|
|
|
|
def test_source_type(self, client):
|
|
"""Test source_type returns 'pypi'."""
|
|
assert client.source_type == "pypi"
|
|
|
|
def test_normalize_package_name(self, client):
|
|
"""Test package name normalization per PEP 503."""
|
|
assert client._normalize_package_name("My_Package") == "my-package"
|
|
assert client._normalize_package_name("my.package") == "my-package"
|
|
assert client._normalize_package_name("my-package") == "my-package"
|
|
assert client._normalize_package_name("MY-PACKAGE") == "my-package"
|
|
assert client._normalize_package_name("my__package") == "my-package"
|
|
assert client._normalize_package_name("my..package") == "my-package"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_available_versions_success(self, client, mock_http_client):
|
|
"""Test fetching available versions from PyPI."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"releases": {
|
|
"1.0.0": [{"packagetype": "bdist_wheel"}],
|
|
"1.1.0": [{"packagetype": "bdist_wheel"}],
|
|
"2.0.0": [{"packagetype": "bdist_wheel"}],
|
|
}
|
|
}
|
|
mock_http_client.get.return_value = mock_response
|
|
|
|
versions = await client.get_available_versions("test-package")
|
|
|
|
assert "1.0.0" in versions
|
|
assert "1.1.0" in versions
|
|
assert "2.0.0" in versions
|
|
mock_http_client.get.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_available_versions_empty(self, client, mock_http_client):
|
|
"""Test handling package with no releases."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"releases": {}}
|
|
mock_http_client.get.return_value = mock_response
|
|
|
|
versions = await client.get_available_versions("empty-package")
|
|
|
|
assert versions == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_available_versions_404(self, client, mock_http_client):
|
|
"""Test handling non-existent package."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 404
|
|
mock_http_client.get.return_value = mock_response
|
|
|
|
versions = await client.get_available_versions("nonexistent")
|
|
|
|
assert versions == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_constraint_wildcard(self, client, mock_http_client):
|
|
"""Test resolving wildcard constraint returns latest."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"info": {"version": "2.0.0"},
|
|
"releases": {
|
|
"1.0.0": [
|
|
{
|
|
"packagetype": "bdist_wheel",
|
|
"url": "https://files.pythonhosted.org/test-1.0.0.whl",
|
|
"filename": "test-1.0.0.whl",
|
|
"digests": {"sha256": "abc123"},
|
|
"size": 1000,
|
|
}
|
|
],
|
|
"2.0.0": [
|
|
{
|
|
"packagetype": "bdist_wheel",
|
|
"url": "https://files.pythonhosted.org/test-2.0.0.whl",
|
|
"filename": "test-2.0.0.whl",
|
|
"digests": {"sha256": "def456"},
|
|
"size": 2000,
|
|
}
|
|
],
|
|
},
|
|
}
|
|
mock_http_client.get.return_value = mock_response
|
|
|
|
result = await client.resolve_constraint("test-package", "*")
|
|
|
|
assert result is not None
|
|
assert result.version == "2.0.0"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_constraint_specific_version(self, client, mock_http_client):
|
|
"""Test resolving specific version constraint."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"releases": {
|
|
"1.0.0": [
|
|
{
|
|
"packagetype": "bdist_wheel",
|
|
"url": "https://files.pythonhosted.org/test-1.0.0.whl",
|
|
"filename": "test-1.0.0.whl",
|
|
"digests": {"sha256": "abc123"},
|
|
"size": 1000,
|
|
}
|
|
],
|
|
"2.0.0": [
|
|
{
|
|
"packagetype": "bdist_wheel",
|
|
"url": "https://files.pythonhosted.org/test-2.0.0.whl",
|
|
"filename": "test-2.0.0.whl",
|
|
}
|
|
],
|
|
},
|
|
}
|
|
mock_http_client.get.return_value = mock_response
|
|
|
|
result = await client.resolve_constraint("test-package", ">=1.0.0,<2.0.0")
|
|
|
|
assert result is not None
|
|
assert result.version == "1.0.0"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_constraint_no_match(self, client, mock_http_client):
|
|
"""Test resolving constraint with no matching version."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"releases": {
|
|
"1.0.0": [
|
|
{
|
|
"packagetype": "bdist_wheel",
|
|
"url": "https://files.pythonhosted.org/test-1.0.0.whl",
|
|
"filename": "test-1.0.0.whl",
|
|
}
|
|
],
|
|
},
|
|
}
|
|
mock_http_client.get.return_value = mock_response
|
|
|
|
result = await client.resolve_constraint("test-package", ">=5.0.0")
|
|
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_constraint_bare_version(self, client, mock_http_client):
|
|
"""Test resolving bare version string as exact match."""
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"info": {"version": "2.0.0"},
|
|
"releases": {
|
|
"1.0.0": [
|
|
{
|
|
"packagetype": "bdist_wheel",
|
|
"url": "https://files.pythonhosted.org/test-1.0.0.whl",
|
|
"filename": "test-1.0.0.whl",
|
|
"digests": {"sha256": "abc123"},
|
|
"size": 1000,
|
|
}
|
|
],
|
|
"2.0.0": [
|
|
{
|
|
"packagetype": "bdist_wheel",
|
|
"url": "https://files.pythonhosted.org/test-2.0.0.whl",
|
|
"filename": "test-2.0.0.whl",
|
|
"digests": {"sha256": "def456"},
|
|
"size": 2000,
|
|
}
|
|
],
|
|
},
|
|
}
|
|
mock_http_client.get.return_value = mock_response
|
|
|
|
# Bare version "1.0.0" should resolve to exactly 1.0.0, not latest
|
|
result = await client.resolve_constraint("test-package", "1.0.0")
|
|
|
|
assert result is not None
|
|
assert result.version == "1.0.0"
|
|
|
|
|
|
class TestVersionInfo:
|
|
"""Tests for VersionInfo dataclass."""
|
|
|
|
def test_create_version_info(self):
|
|
"""Test creating VersionInfo with all fields."""
|
|
info = VersionInfo(
|
|
version="1.0.0",
|
|
download_url="https://example.com/pkg-1.0.0.whl",
|
|
filename="pkg-1.0.0.whl",
|
|
sha256="abc123",
|
|
size=5000,
|
|
content_type="application/zip",
|
|
)
|
|
assert info.version == "1.0.0"
|
|
assert info.download_url == "https://example.com/pkg-1.0.0.whl"
|
|
assert info.filename == "pkg-1.0.0.whl"
|
|
assert info.sha256 == "abc123"
|
|
assert info.size == 5000
|
|
|
|
def test_create_version_info_minimal(self):
|
|
"""Test creating VersionInfo with only required fields."""
|
|
info = VersionInfo(
|
|
version="1.0.0",
|
|
download_url="https://example.com/pkg.whl",
|
|
filename="pkg.whl",
|
|
)
|
|
assert info.sha256 is None
|
|
assert info.size is None
|
|
|
|
|
|
class TestFetchResult:
|
|
"""Tests for FetchResult dataclass."""
|
|
|
|
def test_create_fetch_result(self):
|
|
"""Test creating FetchResult."""
|
|
result = FetchResult(
|
|
artifact_id="abc123def456",
|
|
size=10000,
|
|
version="2.0.0",
|
|
filename="pkg-2.0.0.whl",
|
|
already_cached=True,
|
|
)
|
|
assert result.artifact_id == "abc123def456"
|
|
assert result.size == 10000
|
|
assert result.version == "2.0.0"
|
|
assert result.already_cached is True
|
|
|
|
def test_fetch_result_default_not_cached(self):
|
|
"""Test FetchResult defaults to not cached."""
|
|
result = FetchResult(
|
|
artifact_id="xyz",
|
|
size=100,
|
|
version="1.0.0",
|
|
filename="pkg.whl",
|
|
)
|
|
assert result.already_cached is False
|
|
|
|
|
|
class TestGetRegistryClient:
|
|
"""Tests for registry client factory function."""
|
|
|
|
def test_get_pypi_client(self):
|
|
"""Test getting PyPI client."""
|
|
mock_client = MagicMock()
|
|
mock_sources = []
|
|
|
|
client = get_registry_client("pypi", mock_client, mock_sources)
|
|
|
|
assert isinstance(client, PyPIRegistryClient)
|
|
|
|
def test_get_unsupported_client(self):
|
|
"""Test getting unsupported registry type returns None."""
|
|
mock_client = MagicMock()
|
|
|
|
client = get_registry_client("npm", mock_client, [])
|
|
|
|
assert client is None
|
|
|
|
def test_get_unknown_client(self):
|
|
"""Test getting unknown registry type returns None."""
|
|
mock_client = MagicMock()
|
|
|
|
client = get_registry_client("unknown", mock_client, [])
|
|
|
|
assert client is None
|