From acc1ac58c9d1e971cbcb4b9b726935d58bb02829 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 12:56:15 -0800 Subject: [PATCH 01/13] Add Pixi to Conda-lock --- tests/test_pixi_to_conda_lock.py | 294 +++++++++++++++++++++++ unidep/pixi_to_conda_lock.py | 385 +++++++++++++++++++++++++++++++ 2 files changed, 679 insertions(+) create mode 100644 tests/test_pixi_to_conda_lock.py create mode 100755 unidep/pixi_to_conda_lock.py diff --git a/tests/test_pixi_to_conda_lock.py b/tests/test_pixi_to_conda_lock.py new file mode 100644 index 00000000..7c75e24a --- /dev/null +++ b/tests/test_pixi_to_conda_lock.py @@ -0,0 +1,294 @@ +"""Tests for the pixi_to_conda_lock.py script.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, mock_open, patch + +import pytest + +import unidep.pixi_to_conda_lock as ptcl + + +@pytest.fixture +def sample_pixi_lock() -> dict[str, Any]: + """Sample pixi.lock data for testing.""" + return { + "version": 6, + "environments": { + "default": { + "channels": [{"url": "https://conda.anaconda.org/conda-forge/"}], + "indexes": ["https://pypi.org/simple"], + "packages": { + "osx-arm64": [ + { + "conda": "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda", + }, + { + "pypi": "https://files.pythonhosted.org/packages/04/27/8739697a1d77f972feee90b844786b893217a133941477570d161de2750f/numthreads-0.5.0-py3-none-any.whl", + }, + ], + }, + }, + }, + "packages": [ + { + "conda": "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda", + "sha256": "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a", + "md5": "9d0ae3f3e43c192a992827c0abffe284", + "depends": {"bzip2": ">=1.0.8,<2.0a0", "libexpat": ">=2.6.4,<3.0a0"}, + }, + { + "pypi": "https://files.pythonhosted.org/packages/04/27/8739697a1d77f972feee90b844786b893217a133941477570d161de2750f/numthreads-0.5.0-py3-none-any.whl", + "name": "numthreads", + "version": "0.5.0", + "sha256": "e56e83cbccef103901e678aa014d64b203cdb1b3a3be7cdedb2516ef62ec8fa1", + }, + ], + } + + +@pytest.fixture +def sample_repodata() -> dict[str, Any]: + """Sample repodata for testing.""" + return { + "repo1": { + "info": {"subdir": "osx-arm64"}, + "packages": { + "python-3.13.2-hfd29fff_1_cp313t.conda": { + "name": "python", + "version": "3.13.2", + "build": "hfd29fff_1_cp313t", + "build_number": 1, + "depends": ["bzip2 >=1.0.8,<2.0a0", "libexpat >=2.6.4,<3.0a0"], + "md5": "9d0ae3f3e43c192a992827c0abffe284", + "sha256": "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a", + }, + }, + }, + } + + +def test_read_yaml_file() -> None: + """Test reading a YAML file.""" + mock_yaml_content = """ + key1: value1 + key2: value2 + """ + with patch("builtins.open", mock_open(read_data=mock_yaml_content)): + result = ptcl.read_yaml_file(Path("test.yaml")) + assert result == {"key1": "value1", "key2": "value2"} + + +def test_write_yaml_file() -> None: + """Test writing a YAML file.""" + data = {"key1": "value1", "key2": "value2"} + mock_file = mock_open() + with patch("builtins.open", mock_file): + ptcl.write_yaml_file(Path("test.yaml"), data) + mock_file.assert_called_once_with(Path("test.yaml"), "w") + mock_file().write.assert_called() + + +def test_find_repodata_cache_dir() -> None: + """Test finding the repodata cache directory.""" + with patch("pathlib.Path.exists") as mock_exists, patch( + "pathlib.Path.is_dir", + ) as mock_is_dir: + # Test when directory exists + mock_exists.return_value = True + mock_is_dir.return_value = True + result = ptcl.find_repodata_cache_dir() + assert result is not None + + # Test when directory doesn't exist + mock_exists.return_value = False + result = ptcl.find_repodata_cache_dir() + assert result is None + + +def test_load_json_file() -> None: + """Test loading a JSON file.""" + mock_json_content = '{"key1": "value1", "key2": "value2"}' + with patch("builtins.open", mock_open(read_data=mock_json_content)): + result = ptcl.load_json_file(Path("test.json")) + assert result == {"key1": "value1", "key2": "value2"} + + +def test_load_repodata_files() -> None: + """Test loading repodata files.""" + mock_dir = MagicMock() + mock_file1 = MagicMock() + mock_file1.name = "file1.json" + mock_file1.stem = "file1" + mock_file2 = MagicMock() + mock_file2.name = "file2.info.json" + + mock_dir.glob.return_value = [mock_file1, mock_file2] + + with patch("unidep.pixi_to_conda_lock.load_json_file") as mock_load: + mock_load.return_value = {"key": "value"} + result = ptcl.load_repodata_files(mock_dir) + + assert "file1" in result + assert "file2" not in result + assert result["file1"] == {"key": "value"} + + +def test_extract_filename_from_url() -> None: + """Test extracting filename from URL.""" + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + result = ptcl.extract_filename_from_url(url) + assert result == "python-3.13.2-hfd29fff_1_cp313t.conda" + + +def test_find_package_in_repodata(sample_repodata: dict[str, Any]) -> None: + """Test finding a package in repodata.""" + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + result = ptcl.find_package_in_repodata(sample_repodata, url) + assert result is not None + assert result["name"] == "python" + assert result["version"] == "3.13.2" + + # Test package not found + url_not_found = "https://conda.anaconda.org/conda-forge/osx-arm64/nonexistent-1.0.0-abc123.conda" + result_not_found = ptcl.find_package_in_repodata(sample_repodata, url_not_found) + assert result_not_found is None + + +def test_extract_platform_from_url() -> None: + """Test extracting platform from URL.""" + # Test different platforms + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/noarch/pkg-1.0.0.conda", + ) + == "noarch" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/osx-arm64/pkg-1.0.0.conda", + ) + == "osx-arm64" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/osx-64/pkg-1.0.0.conda", + ) + == "osx-64" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/linux-64/pkg-1.0.0.conda", + ) + == "linux-64" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/linux-aarch64/pkg-1.0.0.conda", + ) + == "linux-aarch64" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/win-64/pkg-1.0.0.conda", + ) + == "win-64" + ) + assert ( + ptcl.extract_platform_from_url( + "https://conda.anaconda.org/conda-forge/unknown/pkg-1.0.0.conda", + ) + == "unknown" + ) + + +def test_extract_name_version_from_url() -> None: + """Test extracting name and version from URL.""" + # Test standard package + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + name, version = ptcl.extract_name_version_from_url(url) + assert name == "python" + assert version == "3.13.2" + + # Test package with tar.bz2 extension + url_tar = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.tar.bz2" + name_tar, version_tar = ptcl.extract_name_version_from_url(url_tar) + assert name_tar == "python" + assert version_tar == "3.13.2" + + # Test package with no version + url_no_version = "https://conda.anaconda.org/conda-forge/osx-arm64/python.conda" + name_no_version, version_no_version = ptcl.extract_name_version_from_url( + url_no_version, + ) + assert name_no_version == "python" + assert version_no_version == "" + + +def test_parse_dependencies_from_repodata() -> None: + """Test parsing dependencies from repodata.""" + depends_list = ["python >=3.8", "numpy", "pandas >=1.0.0,<2.0.0"] + result = ptcl.parse_dependencies_from_repodata(depends_list) + assert result == {"python": ">=3.8", "numpy": "", "pandas": ">=1.0.0,<2.0.0"} + + +def test_create_conda_package_entry() -> None: + """Test creating a conda package entry.""" + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + repodata_info = { + "name": "python", + "version": "3.13.2", + "build": "hfd29fff_1_cp313t", + "build_number": 1, + "depends": ["bzip2 >=1.0.8,<2.0a0", "libexpat >=2.6.4,<3.0a0"], + "md5": "9d0ae3f3e43c192a992827c0abffe284", + "sha256": "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a", + } + + result = ptcl.create_conda_package_entry(url, repodata_info) + + assert result["name"] == "python" + assert result["version"] == "3.13.2" + assert result["manager"] == "conda" + assert result["platform"] == "osx-arm64" + assert result["dependencies"] == { + "bzip2": ">=1.0.8,<2.0a0", + "libexpat": ">=2.6.4,<3.0a0", + } + assert result["url"] == url + assert result["hash"]["md5"] == "9d0ae3f3e43c192a992827c0abffe284" + assert ( + result["hash"]["sha256"] + == "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a" + ) + assert result["build"] == "hfd29fff_1_cp313t" + assert result["build_number"] == 1 + + +def test_create_conda_package_entry_fallback() -> None: + """Test creating a conda package entry using fallback.""" + url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" + package_info = { + "depends": {"bzip2": ">=1.0.8,<2.0a0", "libexpat": ">=2.6.4,<3.0a0"}, + "md5": "9d0ae3f3e43c192a992827c0abffe284", + "sha256": "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a", + } + + result = ptcl.create_conda_package_entry_fallback(url, package_info) + + assert result["name"] == "python" + assert result["version"] == "3.13.2" + assert result["manager"] == "conda" + assert result["platform"] == "osx-arm64" + assert result["dependencies"] == { + "bzip2": ">=1.0.8,<2.0a0", + "libexpat": ">=2.6.4,<3.0a0", + } + assert result["url"] == url + assert result["hash"]["md5"] == "9d0ae3f3e43c192a992827c0abffe284" + assert ( + result["hash"]["sha256"] + == "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a" + ) diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py new file mode 100755 index 00000000..83365014 --- /dev/null +++ b/unidep/pixi_to_conda_lock.py @@ -0,0 +1,385 @@ +#!/usr/bin/env python3 +"""Convert a pixi.lock file to a conda-lock.yml file using repodata. + +This script reads a pixi.lock file and generates a conda-lock.yml file with the same +package information, using repodata to extract accurate package metadata. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +import yaml + + +def read_yaml_file(file_path: Path) -> dict[str, Any]: + """Read a YAML file and return its contents as a dictionary.""" + with open(file_path) as f: # noqa: PTH123 + return yaml.safe_load(f) + + +def write_yaml_file(file_path: Path, data: dict[str, Any]) -> None: + """Write data to a YAML file.""" + with open(file_path, "w") as f: # noqa: PTH123 + yaml.dump(data, f, sort_keys=False) + + +def find_repodata_cache_dir() -> Path | None: + """Find the repodata cache directory based on common locations.""" + # Try to find the cache directory in common locations + possible_paths = [ + Path.home() / "Library" / "Caches" / "rattler" / "cache" / "repodata", # macOS + Path.home() / ".cache" / "rattler" / "cache" / "repodata", # Linux + Path.home() / "AppData" / "Local" / "rattler" / "cache" / "repodata", # Windows + ] + + for path in possible_paths: + if path.exists() and path.is_dir(): + return path + + return None + + +def load_json_file(file_path: Path) -> dict[str, Any]: + """Load a JSON file and return its contents as a dictionary.""" + with open(file_path) as f: # noqa: PTH123 + return json.load(f) + + +def load_repodata_files(repodata_dir: Path) -> dict[str, dict[str, Any]]: + """Load all repodata files from the cache directory.""" + repodata = {} + + # Load all .json files (not .info.json) + for file_path in repodata_dir.glob("*.json"): + if not file_path.name.endswith(".info.json"): + try: + data = load_json_file(file_path) + repodata[file_path.stem] = data + except Exception as e: # noqa: BLE001 + print(f"Warning: Failed to load {file_path}: {e}") + + return repodata + + +def extract_filename_from_url(url: str) -> str: + """Extract the filename from a URL.""" + return url.split("/")[-1] + + +def find_package_in_repodata( + repodata: dict[str, dict[str, Any]], + package_url: str, +) -> dict[str, Any] | None: + """Find a package in repodata based on its URL.""" + filename = extract_filename_from_url(package_url) + + # Check all repodata files + for repo_data in repodata.values(): + # Check in packages + if "packages" in repo_data and filename in repo_data["packages"]: + return repo_data["packages"][filename] + + # Check in packages.conda (for newer conda formats) + if "packages.conda" in repo_data and filename in repo_data["packages.conda"]: + return repo_data["packages.conda"][filename] + + return None + + +def extract_platform_from_url(url: str) -> str: # noqa: PLR0911 + """Extract platform information from a conda package URL.""" + if "/noarch/" in url: + return "noarch" + if "/osx-arm64/" in url: + return "osx-arm64" + if "/osx-64/" in url: + return "osx-64" + if "/linux-64/" in url: + return "linux-64" + if "/linux-aarch64/" in url: + return "linux-aarch64" + if "/win-64/" in url: + return "win-64" + # Default fallback + return "unknown" + + +def extract_name_version_from_url(url: str) -> tuple[str, str]: + """Extract package name and version from a conda package URL as a fallback.""" + filename = extract_filename_from_url(url) + + # Remove file extension (.conda or .tar.bz2) + if filename.endswith(".conda"): + filename_no_ext = filename[:-6] + elif filename.endswith(".tar.bz2"): + filename_no_ext = filename[:-8] + else: + filename_no_ext = filename + + # Split by hyphens to separate name, version, and build + parts = filename_no_ext.split("-") + + # For simplicity in the fallback, assume the first part is the name + # and the second part is the version + name = parts[0] + version = parts[1] if len(parts) > 1 else "" + + return name, version + + +def parse_dependencies_from_repodata(depends_list: list[str]) -> dict[str, str]: + """Parse dependencies from repodata format to conda-lock format.""" + dependencies = {} + for dep in depends_list: + parts = dep.split() + if len(parts) > 1: + dependencies[parts[0]] = " ".join(parts[1:]) + else: + dependencies[dep] = "" + return dependencies + + +def create_conda_package_entry( + url: str, + repodata_info: dict[str, Any], +) -> dict[str, Any]: + """Create a conda package entry for conda-lock.yml from repodata.""" + platform = extract_platform_from_url(url) + + package_entry = { + "name": repodata_info["name"], + "version": repodata_info["version"], + "manager": "conda", + "platform": platform, + "dependencies": parse_dependencies_from_repodata( + repodata_info.get("depends", []), + ), + "url": url, + "hash": { + "md5": repodata_info.get("md5", ""), + "sha256": repodata_info.get("sha256", ""), + }, + "category": "main", + "optional": False, + } + + # Add build information if available + if "build" in repodata_info: + package_entry["build"] = repodata_info["build"] + + # Add build number if available + if "build_number" in repodata_info: + package_entry["build_number"] = repodata_info["build_number"] + + return package_entry + + +def create_conda_package_entry_fallback( + url: str, + package_info: dict[str, Any], +) -> dict[str, Any]: + """Create a conda package entry for conda-lock.yml using URL parsing as fallback.""" + platform = extract_platform_from_url(url) + name, version = extract_name_version_from_url(url) + + return { + "name": name, + "version": version, + "manager": "conda", + "platform": platform, + "dependencies": dict(package_info.get("depends", {}).items()), + "url": url, + "hash": { + "md5": package_info.get("md5", ""), + "sha256": package_info.get("sha256", ""), + }, + "category": "main", + "optional": False, + } + + +def create_pypi_package_entry( + platform: str, + package_info: dict[str, Any], +) -> dict[str, Any]: + """Create a PyPI package entry for conda-lock.yml.""" + url = package_info["pypi"] + + return { + "name": package_info.get("name", ""), + "version": package_info.get("version", ""), + "manager": "pip", + "platform": platform, + "dependencies": {}, # PyPI dependencies are handled differently + "url": url, + "hash": { + "sha256": package_info.get("sha256", ""), + }, + "category": "main", + "optional": False, + } + + +def extract_platforms_from_pixi(pixi_data: Any) -> list[str]: + """Extract platform information from pixi.lock data.""" + platforms = [] + for env_data in pixi_data.get("environments", {}).values(): + for platform in env_data.get("packages", {}): + if platform not in platforms and platform != "noarch": + platforms.append(platform) + return platforms + + +def extract_channels_from_pixi(pixi_data: dict[str, Any]) -> list[dict[str, str]]: + """Extract channel information from pixi.lock data.""" + return [ + {"url": channel["url"].replace("https://conda.anaconda.org/", "")} + for channel in pixi_data.get("environments", {}) + .get("default", {}) + .get("channels", []) + ] + + +def create_conda_lock_metadata( + platforms: list[str], + channels: list[dict[str, str]], +) -> dict[str, Any]: + """Create metadata section for conda-lock.yml.""" + return { + "content_hash": { + platform: "generated-from-pixi-lock" for platform in platforms + }, + "channels": channels, + "platforms": platforms, + "sources": ["converted-from-pixi.lock"], + } + + +def process_conda_packages( + pixi_data: dict[str, Any], + repodata: dict[str, dict[str, Any]], +) -> list[dict[str, Any]]: + """Process conda packages from pixi.lock and convert to conda-lock format.""" + package_entries = [] + + for package_info in pixi_data.get("packages", []): + if "conda:" in package_info: + url = package_info["conda"] + + # Try to find package in repodata + repodata_info = find_package_in_repodata(repodata, url) + + if repodata_info: + # Use the information from repodata + package_entry = create_conda_package_entry( + url, + repodata_info, + ) + else: + # Fallback to parsing the URL if repodata doesn't have the package + package_entry = create_conda_package_entry_fallback(url, package_info) + + package_entries.append(package_entry) + + return package_entries + + +def process_pypi_packages( + pixi_data: dict[str, Any], + platforms: list[str], +) -> list[dict[str, Any]]: + """Process PyPI packages from pixi.lock and convert to conda-lock format.""" + package_entries = [] + + for package_info in pixi_data.get("packages", []): + if "pypi:" in package_info: + for platform in platforms: + package_entry = create_pypi_package_entry(platform, package_info) + package_entries.append(package_entry) + + return package_entries + + +def convert_pixi_to_conda_lock( + pixi_data: dict[str, Any], + repodata: dict[str, Any], +) -> dict[str, Any]: + """Convert pixi.lock data structure to conda-lock.yml format using repodata.""" + # Extract platforms and channels + platforms = extract_platforms_from_pixi(pixi_data) + channels = extract_channels_from_pixi(pixi_data) + + # Create basic conda-lock structure + conda_lock_data = { + "version": 1, + "metadata": create_conda_lock_metadata(platforms, channels), + "package": [], + } + + # Process conda packages + conda_packages = process_conda_packages(pixi_data, repodata) + conda_lock_data["package"].extend(conda_packages) # type: ignore[attr-defined] + + # Process PyPI packages + pypi_packages = process_pypi_packages(pixi_data, platforms) + conda_lock_data["package"].extend(pypi_packages) # type: ignore[attr-defined] + + return conda_lock_data + + +def main() -> int: + """Main function to convert pixi.lock to conda-lock.yml.""" + parser = argparse.ArgumentParser(description="Convert pixi.lock to conda-lock.yml") + parser.add_argument("pixi_lock", type=Path, help="Path to pixi.lock file") + parser.add_argument( + "--output", + "-o", + type=Path, + default=Path("conda-lock.yml"), + help="Output conda-lock.yml file path", + ) + parser.add_argument( + "--repodata-dir", + type=Path, + help="Path to repodata cache directory", + ) + + args = parser.parse_args() + + if not args.pixi_lock.exists(): + print(f"Error: {args.pixi_lock} does not exist") + return 1 + + # Find repodata cache directory + repodata_dir = args.repodata_dir + if repodata_dir is None: + repodata_dir = find_repodata_cache_dir() + if repodata_dir is None: + print( + "Warning: Could not find repodata cache directory. Using fallback URL parsing.", # noqa: E501 + ) + repodata = {} + else: + print(f"Using repodata from: {repodata_dir}") + repodata = load_repodata_files(repodata_dir) + else: + if not repodata_dir.exists(): + print(f"Error: Specified repodata directory {repodata_dir} does not exist") + return 1 + repodata = load_repodata_files(repodata_dir) + + pixi_data = read_yaml_file(args.pixi_lock) + conda_lock_data = convert_pixi_to_conda_lock(pixi_data, repodata) + write_yaml_file(args.output, conda_lock_data) + + print(f"Successfully converted {args.pixi_lock} to {args.output}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) From 53fc2daa951ad743822de220796c83b4fa156dd3 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 13:09:04 -0800 Subject: [PATCH 02/13] Logging --- unidep/pixi_to_conda_lock.py | 266 ++++++++++++++++++++++++++--------- 1 file changed, 201 insertions(+), 65 deletions(-) diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index 83365014..9c798b7b 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -9,6 +9,7 @@ import argparse import json +import logging import sys from pathlib import Path from typing import Any @@ -16,20 +17,41 @@ import yaml +def setup_logging(verbose: bool = False) -> None: # noqa: FBT001, FBT002 + """Set up logging configuration. + + Args: + verbose: Whether to enable debug logging + + """ + log_level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig( + level=log_level, + format="%(asctime)s - %(levelname)s - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + def read_yaml_file(file_path: Path) -> dict[str, Any]: """Read a YAML file and return its contents as a dictionary.""" + logging.debug("Reading YAML file: %s", file_path) with open(file_path) as f: # noqa: PTH123 - return yaml.safe_load(f) + data = yaml.safe_load(f) + logging.debug("Successfully read YAML file: %s", file_path) + return data def write_yaml_file(file_path: Path, data: dict[str, Any]) -> None: """Write data to a YAML file.""" + logging.debug("Writing YAML file: %s", file_path) with open(file_path, "w") as f: # noqa: PTH123 yaml.dump(data, f, sort_keys=False) + logging.debug("Successfully wrote YAML file: %s", file_path) def find_repodata_cache_dir() -> Path | None: """Find the repodata cache directory based on common locations.""" + logging.debug("Searching for repodata cache directory") # Try to find the cache directory in common locations possible_paths = [ Path.home() / "Library" / "Caches" / "rattler" / "cache" / "repodata", # macOS @@ -38,37 +60,52 @@ def find_repodata_cache_dir() -> Path | None: ] for path in possible_paths: + logging.debug("Checking path: %s", path) if path.exists() and path.is_dir(): + logging.debug("Found repodata cache directory: %s", path) return path + logging.debug("No repodata cache directory found") return None def load_json_file(file_path: Path) -> dict[str, Any]: """Load a JSON file and return its contents as a dictionary.""" + logging.debug("Loading JSON file: %s", file_path) with open(file_path) as f: # noqa: PTH123 - return json.load(f) + data = json.load(f) + logging.debug("Successfully loaded JSON file: %s", file_path) + return data def load_repodata_files(repodata_dir: Path) -> dict[str, dict[str, Any]]: """Load all repodata files from the cache directory.""" + logging.info("Loading repodata files from: %s", repodata_dir) repodata = {} # Load all .json files (not .info.json) - for file_path in repodata_dir.glob("*.json"): + json_files = list(repodata_dir.glob("*.json")) + logging.debug("Found %d JSON files in repodata directory", len(json_files)) + + for file_path in json_files: if not file_path.name.endswith(".info.json"): try: + logging.debug("Loading repodata file: %s", file_path.name) data = load_json_file(file_path) repodata[file_path.stem] = data + logging.debug("Successfully loaded repodata file: %s", file_path.name) except Exception as e: # noqa: BLE001 - print(f"Warning: Failed to load {file_path}: {e}") + logging.warning("Failed to load %s: %s", file_path, e) + logging.info("Loaded %d repodata files", len(repodata)) return repodata def extract_filename_from_url(url: str) -> str: """Extract the filename from a URL.""" - return url.split("/")[-1] + filename = url.split("/")[-1] + logging.debug("Extracted filename '%s' from URL: %s", filename, url) + return filename def find_package_in_repodata( @@ -76,41 +113,55 @@ def find_package_in_repodata( package_url: str, ) -> dict[str, Any] | None: """Find a package in repodata based on its URL.""" + logging.debug("Searching for package in repodata: %s", package_url) filename = extract_filename_from_url(package_url) # Check all repodata files - for repo_data in repodata.values(): + for repo_name, repo_data in repodata.items(): # Check in packages if "packages" in repo_data and filename in repo_data["packages"]: + logging.debug("Found package '%s' in repository '%s'", filename, repo_name) return repo_data["packages"][filename] # Check in packages.conda (for newer conda formats) if "packages.conda" in repo_data and filename in repo_data["packages.conda"]: + logging.debug( + "Found package '%s' in repository '%s' (packages.conda)", + filename, + repo_name, + ) return repo_data["packages.conda"][filename] + logging.debug("Package not found in repo") return None -def extract_platform_from_url(url: str) -> str: # noqa: PLR0911 +def extract_platform_from_url(url: str) -> str: """Extract platform information from a conda package URL.""" + logging.debug("Extracting platform from URL: %s", url) if "/noarch/" in url: - return "noarch" - if "/osx-arm64/" in url: - return "osx-arm64" - if "/osx-64/" in url: - return "osx-64" - if "/linux-64/" in url: - return "linux-64" - if "/linux-aarch64/" in url: - return "linux-aarch64" - if "/win-64/" in url: - return "win-64" - # Default fallback - return "unknown" + platform = "noarch" + elif "/osx-arm64/" in url: + platform = "osx-arm64" + elif "/osx-64/" in url: + platform = "osx-64" + elif "/linux-64/" in url: + platform = "linux-64" + elif "/linux-aarch64/" in url: + platform = "linux-aarch64" + elif "/win-64/" in url: + platform = "win-64" + else: + # Default fallback + platform = "unknown" + + logging.debug("Extracted platform: %s", platform) + return platform def extract_name_version_from_url(url: str) -> tuple[str, str]: """Extract package name and version from a conda package URL as a fallback.""" + logging.debug("Extracting name and version from URL: %s", url) filename = extract_filename_from_url(url) # Remove file extension (.conda or .tar.bz2) @@ -129,11 +180,13 @@ def extract_name_version_from_url(url: str) -> tuple[str, str]: name = parts[0] version = parts[1] if len(parts) > 1 else "" + logging.debug("Extracted name: %s, version: %s", name, version) return name, version def parse_dependencies_from_repodata(depends_list: list[str]) -> dict[str, str]: """Parse dependencies from repodata format to conda-lock format.""" + logging.debug("Parsing dependencies from repolist") dependencies = {} for dep in depends_list: parts = dep.split() @@ -141,6 +194,7 @@ def parse_dependencies_from_repodata(depends_list: list[str]) -> dict[str, str]: dependencies[parts[0]] = " ".join(parts[1:]) else: dependencies[dep] = "" + logging.debug("Parsed dependencies: %s", dependencies) return dependencies @@ -149,6 +203,7 @@ def create_conda_package_entry( repodata_info: dict[str, Any], ) -> dict[str, Any]: """Create a conda package entry for conda-lock.yml from repodata.""" + logging.debug("Creating conda package entry from repodata for: %s", url) platform = extract_platform_from_url(url) package_entry = { @@ -168,14 +223,11 @@ def create_conda_package_entry( "optional": False, } - # Add build information if available - if "build" in repodata_info: - package_entry["build"] = repodata_info["build"] - - # Add build number if available - if "build_number" in repodata_info: - package_entry["build_number"] = repodata_info["build_number"] - + logging.debug( + "Created conda package entry: %s v%s", + package_entry["name"], + package_entry["version"], + ) return package_entry @@ -184,10 +236,11 @@ def create_conda_package_entry_fallback( package_info: dict[str, Any], ) -> dict[str, Any]: """Create a conda package entry for conda-lock.yml using URL parsing as fallback.""" + logging.debug("Creating conda package entry using fallback for: %s", url) platform = extract_platform_from_url(url) name, version = extract_name_version_from_url(url) - return { + package_entry = { "name": name, "version": version, "manager": "conda", @@ -202,6 +255,9 @@ def create_conda_package_entry_fallback( "optional": False, } + logging.debug("Created conda package entry (fallback): %s v%s", name, version) + return package_entry + def create_pypi_package_entry( platform: str, @@ -209,8 +265,9 @@ def create_pypi_package_entry( ) -> dict[str, Any]: """Create a PyPI package entry for conda-lock.yml.""" url = package_info["pypi"] + logging.debug("Creating PyPI package entry for: %s (platform: %s)", url, platform) - return { + package_entry = { "name": package_info.get("name", ""), "version": package_info.get("version", ""), "manager": "pip", @@ -224,33 +281,59 @@ def create_pypi_package_entry( "optional": False, } + logging.debug( + "Created PyPI package entry: %s v%s", + package_entry["name"], + package_entry["version"], + ) + return package_entry + def extract_platforms_from_pixi(pixi_data: Any) -> list[str]: """Extract platform information from pixi.lock data.""" + logging.debug("Extracting platforms from pixi.lock data") platforms = [] - for env_data in pixi_data.get("environments", {}).values(): + for env_name, env_data in pixi_data.get("environments", {}).items(): + logging.debug("Processing environment: %s", env_name) for platform in env_data.get("packages", {}): if platform not in platforms and platform != "noarch": platforms.append(platform) + logging.debug("Added platform: %s", platform) + + logging.info("Extracted platforms: %s", platforms) return platforms def extract_channels_from_pixi(pixi_data: dict[str, Any]) -> list[dict[str, str]]: """Extract channel information from pixi.lock data.""" - return [ - {"url": channel["url"].replace("https://conda.anaconda.org/", "")} + logging.debug("Extracting channels from pixi.lock data") + channels = [ + { + "url": channel["url"] + .replace("https://conda.anaconda.org/", "") + .rstrip("/"), + "used_env_vars": [], + } for channel in pixi_data.get("environments", {}) .get("default", {}) .get("channels", []) ] + logging.info( + "Extracted %d channels: %s", + len(channels), + [c["url"] for c in channels], + ) + return channels + def create_conda_lock_metadata( platforms: list[str], channels: list[dict[str, str]], ) -> dict[str, Any]: """Create metadata section for conda-lock.yml.""" - return { + logging.debug("Creating conda-lock metadata") + metadata = { "content_hash": { platform: "generated-from-pixi-lock" for platform in platforms }, @@ -258,6 +341,8 @@ def create_conda_lock_metadata( "platforms": platforms, "sources": ["converted-from-pixi.lock"], } + logging.debug("Created conda-lock metadata with %d platforms", len(platforms)) + return metadata def process_conda_packages( @@ -265,27 +350,33 @@ def process_conda_packages( repodata: dict[str, dict[str, Any]], ) -> list[dict[str, Any]]: """Process conda packages from pixi.lock and convert to conda-lock format.""" + logging.info("Processing conda packages from pixi.lock") package_entries = [] + conda_packages = [p for p in pixi_data.get("packages", []) if "conda" in p] + logging.debug("Found %d conda packages to process", len(conda_packages)) + + for package_info in conda_packages: + url = package_info["conda"] + logging.debug("Processing conda package: %s", url) + + # Try to find package in repodata + repodata_info = find_package_in_repodata(repodata, url) + + if repodata_info: + # Use the information from repodata + logging.debug("Using repodata information for package") + package_entry = create_conda_package_entry( + url, + repodata_info, + ) + else: + # Fallback to parsing the URL if repodata doesn't have the package + logging.debug("Repodata not found, using fallback method") + package_entry = create_conda_package_entry_fallback(url, package_info) - for package_info in pixi_data.get("packages", []): - if "conda:" in package_info: - url = package_info["conda"] - - # Try to find package in repodata - repodata_info = find_package_in_repodata(repodata, url) - - if repodata_info: - # Use the information from repodata - package_entry = create_conda_package_entry( - url, - repodata_info, - ) - else: - # Fallback to parsing the URL if repodata doesn't have the package - package_entry = create_conda_package_entry_fallback(url, package_info) - - package_entries.append(package_entry) + package_entries.append(package_entry) + logging.info("Processed %d conda packages", len(package_entries)) return package_entries @@ -294,22 +385,37 @@ def process_pypi_packages( platforms: list[str], ) -> list[dict[str, Any]]: """Process PyPI packages from pixi.lock and convert to conda-lock format.""" + logging.info("Processing PyPI packages from pixi.lock") package_entries = [] + pypi_packages = [p for p in pixi_data.get("packages", []) if "pypi" in p] + logging.debug("Found %d PyPI packages to process", len(pypi_packages)) + + for package_info in pypi_packages: + logging.debug( + "Processing PyPI package: %s v%s", + package_info.get("name", "unknown"), + package_info.get("version", "unknown"), + ) + + for platform in platforms: + logging.debug("Creating entry for platform: %s", platform) + package_entry = create_pypi_package_entry(platform, package_info) + package_entries.append(package_entry) - for package_info in pixi_data.get("packages", []): - if "pypi:" in package_info: - for platform in platforms: - package_entry = create_pypi_package_entry(platform, package_info) - package_entries.append(package_entry) - + logging.info( + "Processed %d PyPI package entries (across all platforms)", + len(package_entries), + ) return package_entries def convert_pixi_to_conda_lock( pixi_data: dict[str, Any], - repodata: dict[str, Any], + repodata: dict[str, dict[str, Any]], ) -> dict[str, Any]: """Convert pixi.lock data structure to conda-lock.yml format using repodata.""" + logging.info("Converting pixi.lock to conda-lock.yml format") + # Extract platforms and channels platforms = extract_platforms_from_pixi(pixi_data) channels = extract_channels_from_pixi(pixi_data) @@ -322,13 +428,21 @@ def convert_pixi_to_conda_lock( } # Process conda packages + logging.info("Processing conda packages") conda_packages = process_conda_packages(pixi_data, repodata) conda_lock_data["package"].extend(conda_packages) # type: ignore[attr-defined] + logging.info("Added %d conda packages to conda-lock data", len(conda_packages)) # Process PyPI packages + logging.info("Processing PyPI packages") pypi_packages = process_pypi_packages(pixi_data, platforms) conda_lock_data["package"].extend(pypi_packages) # type: ignore[attr-defined] + logging.info("Added %d PyPI package entries to conda-lock data", len(pypi_packages)) + logging.info( + "Conversion complete - conda-lock data contains %d package entries", + len(conda_lock_data["package"]), # type: ignore[arg-type] + ) # type: ignore[attr-defined] return conda_lock_data @@ -348,11 +462,24 @@ def main() -> int: type=Path, help="Path to repodata cache directory", ) + parser.add_argument( + "--verbose", + "-v", + action="store_true", + help="Enable verbose logging", + ) args = parser.parse_args() + # Set up logging + setup_logging(args.verbose) + + logging.info("Starting pixi.lock to conda-lock.yml conversion") + logging.info("Input file: %s", args.pixi_lock) + logging.info("Output file: %s", args.output) + if not args.pixi_lock.exists(): - print(f"Error: {args.pixi_lock} does not exist") + logging.error("Error: %s does not exist", args.pixi_lock) return 1 # Find repodata cache directory @@ -360,24 +487,33 @@ def main() -> int: if repodata_dir is None: repodata_dir = find_repodata_cache_dir() if repodata_dir is None: - print( - "Warning: Could not find repodata cache directory. Using fallback URL parsing.", # noqa: E501 + logging.warning( + "Could not find repodata cache directory. Using fallback URL parsing.", ) repodata = {} else: - print(f"Using repodata from: {repodata_dir}") + logging.info("Using repodata from: %s", repodata_dir) repodata = load_repodata_files(repodata_dir) else: if not repodata_dir.exists(): - print(f"Error: Specified repodata directory {repodata_dir} does not exist") + logging.error( + "Specified repodata directory %s does not exist", + repodata_dir, + ) return 1 + logging.info("Using specified repodata directory: %s", repodata_dir) repodata = load_repodata_files(repodata_dir) + logging.info("Reading pixi.lock file") pixi_data = read_yaml_file(args.pixi_lock) + + logging.info("Converting pixi.lock data to conda-lock format") conda_lock_data = convert_pixi_to_conda_lock(pixi_data, repodata) + + logging.info("Writing conda-lock.yml file") write_yaml_file(args.output, conda_lock_data) - print(f"Successfully converted {args.pixi_lock} to {args.output}") + logging.info("Successfully converted %s to %s", args.pixi_lock, args.output) return 0 From affe53fb03ef7491e2e3c980fafddb30b79b60aa Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 13:25:11 -0800 Subject: [PATCH 03/13] . --- tests/test_pixi_to_conda_lock.py | 2 -- unidep/pixi_to_conda_lock.py | 9 +++------ 2 files changed, 3 insertions(+), 8 deletions(-) diff --git a/tests/test_pixi_to_conda_lock.py b/tests/test_pixi_to_conda_lock.py index 7c75e24a..0642b184 100644 --- a/tests/test_pixi_to_conda_lock.py +++ b/tests/test_pixi_to_conda_lock.py @@ -263,8 +263,6 @@ def test_create_conda_package_entry() -> None: result["hash"]["sha256"] == "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a" ) - assert result["build"] == "hfd29fff_1_cp313t" - assert result["build_number"] == 1 def test_create_conda_package_entry_fallback() -> None: diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index 9c798b7b..48eaafd3 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -216,8 +216,8 @@ def create_conda_package_entry( ), "url": url, "hash": { - "md5": repodata_info.get("md5", ""), - "sha256": repodata_info.get("sha256", ""), + "md5": repodata_info["md5"], + "sha256": repodata_info["sha256"], }, "category": "main", "optional": False, @@ -365,10 +365,7 @@ def process_conda_packages( if repodata_info: # Use the information from repodata logging.debug("Using repodata information for package") - package_entry = create_conda_package_entry( - url, - repodata_info, - ) + package_entry = create_conda_package_entry(url, repodata_info) else: # Fallback to parsing the URL if repodata doesn't have the package logging.debug("Repodata not found, using fallback method") From c0f984ac1a363260aac25845e64c0e83fd9d8d73 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 13:30:10 -0800 Subject: [PATCH 04/13] pixi info --- tests/test_pixi_to_conda_lock.py | 27 +++++++++++++++++---------- unidep/pixi_to_conda_lock.py | 32 ++++++++++++++++---------------- 2 files changed, 33 insertions(+), 26 deletions(-) diff --git a/tests/test_pixi_to_conda_lock.py b/tests/test_pixi_to_conda_lock.py index 0642b184..b2516268 100644 --- a/tests/test_pixi_to_conda_lock.py +++ b/tests/test_pixi_to_conda_lock.py @@ -93,19 +93,26 @@ def test_write_yaml_file() -> None: def test_find_repodata_cache_dir() -> None: """Test finding the repodata cache directory.""" - with patch("pathlib.Path.exists") as mock_exists, patch( - "pathlib.Path.is_dir", - ) as mock_is_dir: - # Test when directory exists - mock_exists.return_value = True - mock_is_dir.return_value = True + # Simulate a valid repodata directory exists + with ( + patch("pathlib.Path.exists", return_value=True), + patch("pathlib.Path.is_dir", return_value=True), + patch("subprocess.check_output", return_value='{"cache_dir": "/dummy/path"}'), + patch("json.loads", return_value={"cache_dir": "/dummy/path"}), + # Create a dummy Path object that behaves like it exists + patch("unidep.pixi_to_conda_lock.Path", wraps=Path), + ): result = ptcl.find_repodata_cache_dir() assert result is not None - # Test when directory doesn't exist - mock_exists.return_value = False - result = ptcl.find_repodata_cache_dir() - assert result is None + # Simulate the repodata directory does not exist, and expect a ValueError + with ( + patch("pathlib.Path.exists", return_value=False), + patch("subprocess.check_output", return_value='{"cache_dir": "/dummy/path"}'), + patch("json.loads", return_value={"cache_dir": "/dummy/path"}), + pytest.raises(ValueError, match="Could not find repodata cache directory"), + ): + ptcl.find_repodata_cache_dir() def test_load_json_file() -> None: diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index 48eaafd3..621b8d3f 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -10,6 +10,7 @@ import argparse import json import logging +import subprocess import sys from pathlib import Path from typing import Any @@ -50,23 +51,22 @@ def write_yaml_file(file_path: Path, data: dict[str, Any]) -> None: def find_repodata_cache_dir() -> Path | None: - """Find the repodata cache directory based on common locations.""" - logging.debug("Searching for repodata cache directory") - # Try to find the cache directory in common locations - possible_paths = [ - Path.home() / "Library" / "Caches" / "rattler" / "cache" / "repodata", # macOS - Path.home() / ".cache" / "rattler" / "cache" / "repodata", # Linux - Path.home() / "AppData" / "Local" / "rattler" / "cache" / "repodata", # Windows - ] - - for path in possible_paths: - logging.debug("Checking path: %s", path) - if path.exists() and path.is_dir(): - logging.debug("Found repodata cache directory: %s", path) - return path + """Find the repodata cache directory using 'pixi info --json' output. - logging.debug("No repodata cache directory found") - return None + This function runs 'pixi info --json' and extract the 'cache_dir' + field, appending 'repodata' to it. + """ + cmd = ["pixi", "info", "--json"] + result = subprocess.check_output(cmd, text=True) + info = json.loads(result) + cache_dir = info.get("cache_dir") + if cache_dir: + repodata_path = Path(cache_dir) / "repodata" + logging.debug("Using cache_dir from pixi info: %s", repodata_path) + if repodata_path.exists() and repodata_path.is_dir(): + return repodata_path + msg = "Could not find repodata cache directory" + raise ValueError(msg) def load_json_file(file_path: Path) -> dict[str, Any]: From bdb3793a87160aaace5aa6c1f18ee882dc467ff6 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 14:22:26 -0800 Subject: [PATCH 05/13] . --- unidep/pixi_to_conda_lock.py | 33 +++++++++++++++------------------ 1 file changed, 15 insertions(+), 18 deletions(-) diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index 621b8d3f..c5e7c52b 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -293,7 +293,8 @@ def extract_platforms_from_pixi(pixi_data: Any) -> list[str]: """Extract platform information from pixi.lock data.""" logging.debug("Extracting platforms from pixi.lock data") platforms = [] - for env_name, env_data in pixi_data.get("environments", {}).items(): + environments = pixi_data.get("environments", {}) + for env_name, env_data in environments.items(): logging.debug("Processing environment: %s", env_name) for platform in env_data.get("packages", {}): if platform not in platforms and platform != "noarch": @@ -304,19 +305,20 @@ def extract_platforms_from_pixi(pixi_data: Any) -> list[str]: return platforms +def _channel_url_to_name(url: str) -> str: + """Convert a channel URL to a channel name.""" + return url.replace("https://conda.anaconda.org/", "").rstrip("/") + + def extract_channels_from_pixi(pixi_data: dict[str, Any]) -> list[dict[str, str]]: """Extract channel information from pixi.lock data.""" logging.debug("Extracting channels from pixi.lock data") + channels_data = ( + pixi_data.get("environments", {}).get("default", {}).get("channels", []) + ) channels = [ - { - "url": channel["url"] - .replace("https://conda.anaconda.org/", "") - .rstrip("/"), - "used_env_vars": [], - } - for channel in pixi_data.get("environments", {}) - .get("default", {}) - .get("channels", []) + {"url": _channel_url_to_name(channel["url"]), "used_env_vars": []} + for channel in channels_data ] logging.info( @@ -368,6 +370,7 @@ def process_conda_packages( package_entry = create_conda_package_entry(url, repodata_info) else: # Fallback to parsing the URL if repodata doesn't have the package + # TODO: Is this needed? logging.debug("Repodata not found, using fallback method") package_entry = create_conda_package_entry_fallback(url, package_info) @@ -483,14 +486,8 @@ def main() -> int: repodata_dir = args.repodata_dir if repodata_dir is None: repodata_dir = find_repodata_cache_dir() - if repodata_dir is None: - logging.warning( - "Could not find repodata cache directory. Using fallback URL parsing.", - ) - repodata = {} - else: - logging.info("Using repodata from: %s", repodata_dir) - repodata = load_repodata_files(repodata_dir) + logging.info("Using repodata from: %s", repodata_dir) + repodata = load_repodata_files(repodata_dir) else: if not repodata_dir.exists(): logging.error( From bc2323e210c3e8c2d5bd483ca9974a28e654f685 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 15:00:43 -0800 Subject: [PATCH 06/13] fix noarch --- tests/test_pixi_to_conda_lock.py | 67 ++++++++++++++++++++++++++++++++ unidep/pixi_to_conda_lock.py | 22 ++++++++--- 2 files changed, 83 insertions(+), 6 deletions(-) diff --git a/tests/test_pixi_to_conda_lock.py b/tests/test_pixi_to_conda_lock.py index b2516268..b27776aa 100644 --- a/tests/test_pixi_to_conda_lock.py +++ b/tests/test_pixi_to_conda_lock.py @@ -297,3 +297,70 @@ def test_create_conda_package_entry_fallback() -> None: result["hash"]["sha256"] == "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a" ) + + +def test_noarch_package_expansion(sample_pixi_lock: dict[str, Any]) -> None: + """Test that a noarch package is expanded into entries for each platform.""" + # Modify sample_pixi_lock to include a noarch package and specific platforms + sample_pixi_lock["packages"] = [ + { + "conda": "https://conda.anaconda.org/conda-forge/noarch/cached-property-1.5.2-hd8ed1ab_1.tar.bz2", + }, + ] + sample_pixi_lock["environments"] = { + "default": { + "channels": [{"url": "https://conda.anaconda.org/conda-forge/"}], + "indexes": ["https://pypi.org/simple"], + # Define two platforms for testing + "packages": {"linux-64": [], "osx-arm64": []}, + }, + } + + # Create a sample repodata that contains info for the noarch package. + repodata = { + "repo1": { + "info": {"subdir": "noarch"}, + "packages": { + "cached-property-1.5.2-hd8ed1ab_1.tar.bz2": { + "name": "cached-property", + "version": "1.5.2", + "build": "hd8ed1ab_1", + "build_number": 1, + "depends": ["cached_property >=1.5.2,<1.5.3.0a0"], + "md5": "9b347a7ec10940d3f7941ff6c460b551", + "sha256": "561e6660f26c35d137ee150187d89767c988413c978e1b712d53f27ddf70ea17", + }, + }, + }, + } + + # Import the module under test. + from unidep import pixi_to_conda_lock as ptcl + + # Process the conda packages. + result = ptcl.process_conda_packages(sample_pixi_lock, repodata) + + # Expect an entry per platform (linux-64 and osx-arm64) + assert len(result) == 2, "Expected two package entries, one per platform" + + # Check that each entry has the expected properties. + for entry in result: + assert entry["name"] == "cached-property" + assert entry["version"] == "1.5.2" + assert entry["manager"] == "conda" + # Even though the URL is noarch, the entry should have the platform set to the target environment. + assert ( + entry["url"] + == "https://conda.anaconda.org/conda-forge/noarch/cached-property-1.5.2-hd8ed1ab_1.tar.bz2" + ) + assert entry["hash"]["md5"] == "9b347a7ec10940d3f7941ff6c460b551" + assert ( + entry["hash"]["sha256"] + == "561e6660f26c35d137ee150187d89767c988413c978e1b712d53f27ddf70ea17" + ) + + # Verify that the packages were duplicated for each of the two platforms. + platforms = {entry["platform"] for entry in result} + assert platforms == {"linux-64", "osx-arm64"}, ( + "Expected platforms to be linux-64 and osx-arm64" + ) diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index c5e7c52b..3f748512 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -310,7 +310,9 @@ def _channel_url_to_name(url: str) -> str: return url.replace("https://conda.anaconda.org/", "").rstrip("/") -def extract_channels_from_pixi(pixi_data: dict[str, Any]) -> list[dict[str, str]]: +def extract_channels_from_pixi( + pixi_data: dict[str, Any], +) -> list[dict[str, Any]]: """Extract channel information from pixi.lock data.""" logging.debug("Extracting channels from pixi.lock data") channels_data = ( @@ -355,6 +357,7 @@ def process_conda_packages( logging.info("Processing conda packages from pixi.lock") package_entries = [] conda_packages = [p for p in pixi_data.get("packages", []) if "conda" in p] + platforms = extract_platforms_from_pixi(pixi_data) logging.debug("Found %d conda packages to process", len(conda_packages)) for package_info in conda_packages: @@ -364,17 +367,24 @@ def process_conda_packages( # Try to find package in repodata repodata_info = find_package_in_repodata(repodata, url) + # Create a base package entry, either using repodata or fallback. if repodata_info: # Use the information from repodata logging.debug("Using repodata information for package") - package_entry = create_conda_package_entry(url, repodata_info) + base_entry = create_conda_package_entry(url, repodata_info) else: # Fallback to parsing the URL if repodata doesn't have the package - # TODO: Is this needed? logging.debug("Repodata not found, using fallback method") - package_entry = create_conda_package_entry_fallback(url, package_info) - - package_entries.append(package_entry) + base_entry = create_conda_package_entry_fallback(url, package_info) + + # If the package is noarch, replicate it for each platform. + if "noarch" in url: + for plat in platforms: + entry = base_entry.copy() + entry["platform"] = plat + package_entries.append(entry) + else: + package_entries.append(base_entry) logging.info("Processed %d conda packages", len(package_entries)) return package_entries From 556573b65435bbdf7c2b80c40679a58b59f5ac3f Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 15:03:22 -0800 Subject: [PATCH 07/13] _validate_pip_in_conda_packages --- unidep/pixi_to_conda_lock.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index 3f748512..a05c4ebd 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -446,6 +446,9 @@ def convert_pixi_to_conda_lock( # Process PyPI packages logging.info("Processing PyPI packages") pypi_packages = process_pypi_packages(pixi_data, platforms) + if pypi_packages: + _validate_pip_in_conda_packages(conda_packages) + conda_lock_data["package"].extend(pypi_packages) # type: ignore[attr-defined] logging.info("Added %d PyPI package entries to conda-lock data", len(pypi_packages)) @@ -456,6 +459,19 @@ def convert_pixi_to_conda_lock( return conda_lock_data +def _validate_pip_in_conda_packages(conda_packages: list[dict[str, Any]]) -> None: + pip_included = any( + pkg.get("name") == "pip" and pkg.get("manager") == "conda" + for pkg in conda_packages + ) + if not pip_included: + msg = ( + "PyPI packages are present but no pip package found in conda packages. " + "Please ensure that pip is included in your pixi.lock file." + ) + raise ValueError(msg) + + def main() -> int: """Main function to convert pixi.lock to conda-lock.yml.""" parser = argparse.ArgumentParser(description="Convert pixi.lock to conda-lock.yml") From 13a872df13297c8d6eb2ee24e81719e32f08d02b Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 15:06:26 -0800 Subject: [PATCH 08/13] test --- tests/test_pixi_to_conda_lock.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/tests/test_pixi_to_conda_lock.py b/tests/test_pixi_to_conda_lock.py index b27776aa..f00e0dc3 100644 --- a/tests/test_pixi_to_conda_lock.py +++ b/tests/test_pixi_to_conda_lock.py @@ -364,3 +364,35 @@ def test_noarch_package_expansion(sample_pixi_lock: dict[str, Any]) -> None: assert platforms == {"linux-64", "osx-arm64"}, ( "Expected platforms to be linux-64 and osx-arm64" ) + + +def test_missing_pip_exception() -> None: + """Test that convert_pixi_to_conda_lock raises a ValueError + when there are PyPI packages but no pip package in conda packages. + """ # noqa: D205 + # Create a pixi_data sample with a PyPI package and no pip package. + pixi_data = { + "environments": { + "default": { + "channels": [{"url": "https://conda.anaconda.org/conda-forge/"}], + # Define two target platforms. + "packages": {"linux-64": [], "osx-arm64": []}, + }, + }, + "packages": [ + { + # Only a PyPI package entry, no conda package for pip. + "pypi": "https://files.pythonhosted.org/packages/example/somepypi-1.0.0-py3-none-any.whl", + "name": "somepypi", + "version": "1.0.0", + }, + ], + } + # For this test, repodata can be empty since it's only used for conda packages. + repodata: dict[str, dict[str, Any]] = {} + + with pytest.raises( + ValueError, + match="PyPI packages are present but no pip package found in conda packages", + ): + ptcl.convert_pixi_to_conda_lock(pixi_data, repodata) From f7034c78d832ef279b462d0066ca6949fd956e56 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 15:10:39 -0800 Subject: [PATCH 09/13] cov --- tests/test_pixi_to_conda_lock.py | 4 +--- unidep/pixi_to_conda_lock.py | 13 ++----------- 2 files changed, 3 insertions(+), 14 deletions(-) diff --git a/tests/test_pixi_to_conda_lock.py b/tests/test_pixi_to_conda_lock.py index f00e0dc3..5024e351 100644 --- a/tests/test_pixi_to_conda_lock.py +++ b/tests/test_pixi_to_conda_lock.py @@ -203,12 +203,10 @@ def test_extract_platform_from_url() -> None: ) == "win-64" ) - assert ( + with pytest.raises(ValueError, match="Unknown platform"): ptcl.extract_platform_from_url( "https://conda.anaconda.org/conda-forge/unknown/pkg-1.0.0.conda", ) - == "unknown" - ) def test_extract_name_version_from_url() -> None: diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index a05c4ebd..885f6fd8 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -123,15 +123,6 @@ def find_package_in_repodata( logging.debug("Found package '%s' in repository '%s'", filename, repo_name) return repo_data["packages"][filename] - # Check in packages.conda (for newer conda formats) - if "packages.conda" in repo_data and filename in repo_data["packages.conda"]: - logging.debug( - "Found package '%s' in repository '%s' (packages.conda)", - filename, - repo_name, - ) - return repo_data["packages.conda"][filename] - logging.debug("Package not found in repo") return None @@ -152,8 +143,8 @@ def extract_platform_from_url(url: str) -> str: elif "/win-64/" in url: platform = "win-64" else: - # Default fallback - platform = "unknown" + msg = f"Unknown platform in URL: {url}" + raise ValueError(msg) logging.debug("Extracted platform: %s", platform) return platform From b726cbaeaf0f569240c2b78db248ba532a991156 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 16:22:49 -0800 Subject: [PATCH 10/13] first fallback --- tests/test_pixi_to_conda_lock.py | 14 +++++++------- unidep/pixi_to_conda_lock.py | 15 ++++++--------- 2 files changed, 13 insertions(+), 16 deletions(-) diff --git a/tests/test_pixi_to_conda_lock.py b/tests/test_pixi_to_conda_lock.py index 5024e351..68f91cd4 100644 --- a/tests/test_pixi_to_conda_lock.py +++ b/tests/test_pixi_to_conda_lock.py @@ -223,13 +223,13 @@ def test_extract_name_version_from_url() -> None: assert name_tar == "python" assert version_tar == "3.13.2" - # Test package with no version - url_no_version = "https://conda.anaconda.org/conda-forge/osx-arm64/python.conda" - name_no_version, version_no_version = ptcl.extract_name_version_from_url( - url_no_version, + # Test with dash + url_with_dash = "https://conda.anaconda.org/conda-forge/osx-arm64/ca-certificates-2025.1.31-hf0a4a13_0.conda" + name_with_dash, version_with_dash = ptcl.extract_name_version_from_url( + url_with_dash, ) - assert name_no_version == "python" - assert version_no_version == "" + assert name_with_dash == "ca-certificates" + assert version_with_dash == "2025.1.31" def test_parse_dependencies_from_repodata() -> None: @@ -274,7 +274,7 @@ def test_create_conda_package_entry_fallback() -> None: """Test creating a conda package entry using fallback.""" url = "https://conda.anaconda.org/conda-forge/osx-arm64/python-3.13.2-hfd29fff_1_cp313t.conda" package_info = { - "depends": {"bzip2": ">=1.0.8,<2.0a0", "libexpat": ">=2.6.4,<3.0a0"}, + "depends": ["bzip2 >=1.0.8,<2.0a0", "libexpat >=2.6.4,<3.0a0"], "md5": "9d0ae3f3e43c192a992827c0abffe284", "sha256": "a64466b8f65b77604c3c87092c65d9e51e7db44b11eaa6c469894f0b88b1af5a", } diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index 885f6fd8..a253335a 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -164,12 +164,7 @@ def extract_name_version_from_url(url: str) -> tuple[str, str]: filename_no_ext = filename # Split by hyphens to separate name, version, and build - parts = filename_no_ext.split("-") - - # For simplicity in the fallback, assume the first part is the name - # and the second part is the version - name = parts[0] - version = parts[1] if len(parts) > 1 else "" + name, version, _build_string = filename_no_ext.rsplit("-", 2) logging.debug("Extracted name: %s, version: %s", name, version) return name, version @@ -230,13 +225,15 @@ def create_conda_package_entry_fallback( logging.debug("Creating conda package entry using fallback for: %s", url) platform = extract_platform_from_url(url) name, version = extract_name_version_from_url(url) - + print(package_info) package_entry = { "name": name, "version": version, "manager": "conda", "platform": platform, - "dependencies": dict(package_info.get("depends", {}).items()), + "dependencies": parse_dependencies_from_repodata( + package_info.get("depends", []), + ), "url": url, "hash": { "md5": package_info.get("md5", ""), @@ -365,7 +362,7 @@ def process_conda_packages( base_entry = create_conda_package_entry(url, repodata_info) else: # Fallback to parsing the URL if repodata doesn't have the package - logging.debug("Repodata not found, using fallback method") + logging.warning("Repodata not found, using fallback method") base_entry = create_conda_package_entry_fallback(url, package_info) # If the package is noarch, replicate it for each platform. From 2604b39793b5f4a0561e71a3a81e9f532f15fe03 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 16:30:04 -0800 Subject: [PATCH 11/13] rich --- unidep/pixi_to_conda_lock.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index a253335a..465c59e3 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -25,11 +25,19 @@ def setup_logging(verbose: bool = False) -> None: # noqa: FBT001, FBT002 verbose: Whether to enable debug logging """ + try: + from rich.logging import RichHandler + + handlers = [RichHandler(rich_tracebacks=True)] + except ImportError: + handlers = [logging.StreamHandler()] + log_level = logging.DEBUG if verbose else logging.INFO logging.basicConfig( level=log_level, format="%(asctime)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", + handlers=handlers, ) From 46334cff63528f8b87e379d78d5f97b1b6cfaa40 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 16:33:46 -0800 Subject: [PATCH 12/13] revert --- unidep/pixi_to_conda_lock.py | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/unidep/pixi_to_conda_lock.py b/unidep/pixi_to_conda_lock.py index 465c59e3..651c7d09 100755 --- a/unidep/pixi_to_conda_lock.py +++ b/unidep/pixi_to_conda_lock.py @@ -126,11 +126,24 @@ def find_package_in_repodata( # Check all repodata files for repo_name, repo_data in repodata.items(): - # Check in packages + # Check in packages (.tar.bz2) if "packages" in repo_data and filename in repo_data["packages"]: - logging.debug("Found package '%s' in repository '%s'", filename, repo_name) + logging.debug( + "🔍 Found package '%s' in repository '%s'", + filename, + repo_name, + ) return repo_data["packages"][filename] + # Check in packages.conda (.conda) + if "packages.conda" in repo_data and filename in repo_data["packages.conda"]: + logging.debug( + "🔍 Found package '%s' in repository '%s' (packages.conda)", + filename, + repo_name, + ) + return repo_data["packages.conda"][filename] + logging.debug("Package not found in repo") return None @@ -366,11 +379,11 @@ def process_conda_packages( # Create a base package entry, either using repodata or fallback. if repodata_info: # Use the information from repodata - logging.debug("Using repodata information for package") + logging.debug("✅ Using repodata information for package") base_entry = create_conda_package_entry(url, repodata_info) else: # Fallback to parsing the URL if repodata doesn't have the package - logging.warning("Repodata not found, using fallback method") + logging.warning("❌ Repodata not found, using fallback method") base_entry = create_conda_package_entry_fallback(url, package_info) # If the package is noarch, replicate it for each platform. From 80926d77bbdeff127dc987323d95b4f6f27664a9 Mon Sep 17 00:00:00 2001 From: Bas Nijholt Date: Tue, 25 Feb 2025 16:35:58 -0800 Subject: [PATCH 13/13] cov --- tests/test_pixi_to_conda_lock.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_pixi_to_conda_lock.py b/tests/test_pixi_to_conda_lock.py index 68f91cd4..afcbefb4 100644 --- a/tests/test_pixi_to_conda_lock.py +++ b/tests/test_pixi_to_conda_lock.py @@ -55,7 +55,7 @@ def sample_repodata() -> dict[str, Any]: return { "repo1": { "info": {"subdir": "osx-arm64"}, - "packages": { + "packages.conda": { "python-3.13.2-hfd29fff_1_cp313t.conda": { "name": "python", "version": "3.13.2",