deps_pypi/

lockfile.rs

//! Poetry/uv lock file parsing.
//!
//! Parses `poetry.lock` and `uv.lock` files to extract resolved dependency
//! versions. Both formats use TOML with `[[package]]` sections.
//!
//! # Lock File Formats
//!
//! ## Poetry
//!
//! ```toml
//! # This file is automatically generated by poetry.
//! [[package]]
//! name = "requests"
//! version = "2.31.0"
//! description = "Python HTTP for Humans."
//!
//! [package.dependencies]
//! certifi = ">=2017.4.17"
//! charset-normalizer = ">=2,<4"
//!
//! [metadata]
//! lock-version = "2.0"
//! python-versions = "^3.9"
//! ```
//!
//! ## uv
//!
//! ```toml
//! version = 1
//!
//! [[package]]
//! name = "requests"
//! version = "2.31.0"
//! source = { registry = "https://pypi.org/simple" }
//! dependencies = [
//!     { name = "certifi" },
//!     { name = "charset-normalizer" },
//! ]
//! ```

use async_trait::async_trait;
use deps_core::error::{DepsError, Result};
use deps_core::lockfile::{
    LockFileProvider, ResolvedPackage, ResolvedPackages, ResolvedSource,
    locate_lockfile_for_manifest,
};
use std::path::{Path, PathBuf};
use toml_edit::DocumentMut;
use tower_lsp_server::ls_types::Uri;

/// PyPI lock file parser.
///
/// Implements lock file parsing for Python package managers (Poetry, uv).
/// Supports both `poetry.lock` and `uv.lock` formats.
///
/// # Lock File Location
///
/// The parser searches for lock files in the following order:
/// 1. `poetry.lock` in the same directory as `pyproject.toml`
/// 2. `uv.lock` in the same directory as `pyproject.toml`
///
/// Poetry takes priority because it's more established.
///
/// # Examples
///
/// ```no_run
/// use deps_pypi::lockfile::PypiLockParser;
/// use deps_core::lockfile::LockFileProvider;
/// use tower_lsp_server::ls_types::Uri;
///
/// # async fn example() -> deps_core::error::Result<()> {
/// let parser = PypiLockParser;
/// let manifest_uri = Uri::from_file_path("/path/to/pyproject.toml").unwrap();
///
/// if let Some(lockfile_path) = parser.locate_lockfile(&manifest_uri) {
///     let resolved = parser.parse_lockfile(&lockfile_path).await?;
///     println!("Found {} resolved packages", resolved.len());
/// }
/// # Ok(())
/// # }
/// ```
pub struct PypiLockParser;

impl PypiLockParser {
    /// Lock file names for PyPI ecosystem (poetry.lock, uv.lock).
    const LOCKFILE_NAMES: &'static [&'static str] = &["poetry.lock", "uv.lock"];
}

#[async_trait]
impl LockFileProvider for PypiLockParser {
    fn locate_lockfile(&self, manifest_uri: &Uri) -> Option<PathBuf> {
        locate_lockfile_for_manifest(manifest_uri, Self::LOCKFILE_NAMES)
    }

    async fn parse_lockfile(&self, lockfile_path: &Path) -> Result<ResolvedPackages> {
        tracing::debug!("Parsing lock file: {}", lockfile_path.display());

        let content = tokio::fs::read_to_string(lockfile_path)
            .await
            .map_err(|e| DepsError::ParseError {
                file_type: format!("lock file at {}", lockfile_path.display()),
                source: Box::new(e),
            })?;

        let doc: DocumentMut = content.parse().map_err(|e| DepsError::ParseError {
            file_type: "Python lock file".into(),
            source: Box::new(e),
        })?;

        let mut packages = ResolvedPackages::new();

        let Some(package_array) = doc
            .get("package")
            .and_then(|v: &toml_edit::Item| v.as_array_of_tables())
        else {
            tracing::warn!("Lock file missing [[package]] array of tables");
            return Ok(packages);
        };

        for table in package_array {
            // Extract required fields
            let Some(name) = table.get("name").and_then(|v: &toml_edit::Item| v.as_str()) else {
                tracing::warn!("Package missing name field");
                continue;
            };

            let Some(version) = table
                .get("version")
                .and_then(|v: &toml_edit::Item| v.as_str())
            else {
                tracing::warn!("Package '{}' missing version field", name);
                continue;
            };

            // Parse source (format varies between poetry and uv)
            let source = parse_pypi_source(table);

            // Parse dependencies (format varies between poetry and uv)
            let dependencies = parse_pypi_dependencies(table);

            // Normalize name for consistent lookup (PyPI names are case-insensitive, - == _)
            let normalized_name = name.to_lowercase().replace('-', "_");
            packages.insert(ResolvedPackage {
                name: normalized_name,
                version: version.to_string(),
                source,
                dependencies,
            });
        }

        tracing::info!(
            "Parsed lock file: {} packages from {}",
            packages.len(),
            lockfile_path.display()
        );

        Ok(packages)
    }
}

/// Parses source information from package table.
///
/// Handles both Poetry and uv source formats:
///
/// # Poetry Format
///
/// - No `source` field → PyPI registry (default)
/// - `source.type = "git"` with `source.url` and `source.resolved_reference`
/// - `source.type = "directory"` or `source.type = "file"` with `source.url`
///
/// # uv Format
///
/// - `source.registry = "https://pypi.org/simple"` → Registry
/// - `source.git = "https://github.com/..."` → Git
/// - `source.path = "..."` → Path
fn parse_pypi_source(table: &toml_edit::Table) -> ResolvedSource {
    let Some(source_item) = table.get("source") else {
        // No source field = PyPI registry (poetry default)
        return ResolvedSource::Registry {
            url: "https://pypi.org/simple".to_string(),
            checksum: String::new(),
        };
    };

    // Handle inline table format (uv style)
    if let Some(source_table) = source_item.as_inline_table() {
        // uv: source = { registry = "https://pypi.org/simple" }
        if let Some(registry) = source_table.get("registry").and_then(|v| v.as_str()) {
            return ResolvedSource::Registry {
                url: registry.to_string(),
                checksum: String::new(),
            };
        }

        // uv: source = { git = "https://github.com/..." }
        if let Some(git_url) = source_table.get("git").and_then(|v| v.as_str()) {
            let rev = source_table
                .get("rev")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string();

            return ResolvedSource::Git {
                url: git_url.to_string(),
                rev,
            };
        }

        // uv: source = { path = "..." }
        if let Some(path) = source_table.get("path").and_then(|v| v.as_str()) {
            return ResolvedSource::Path {
                path: path.to_string(),
            };
        }
    }

    // Handle table format (poetry style)
    if let Some(source_table) = source_item.as_table() {
        // poetry: [package.source] type = "git"
        if let Some(source_type) = source_table.get("type").and_then(|v| v.as_str()) {
            match source_type {
                "git" => {
                    let url = source_table
                        .get("url")
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .to_string();

                    let rev = source_table
                        .get("resolved_reference")
                        .or_else(|| source_table.get("reference"))
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .to_string();

                    return ResolvedSource::Git { url, rev };
                }
                "directory" | "file" => {
                    let path = source_table
                        .get("url")
                        .and_then(|v| v.as_str())
                        .unwrap_or("")
                        .to_string();

                    return ResolvedSource::Path { path };
                }
                _ => {}
            }
        }
    }

    // Default to PyPI registry
    ResolvedSource::Registry {
        url: "https://pypi.org/simple".to_string(),
        checksum: String::new(),
    }
}

/// Parses dependencies from package table.
///
/// Handles both Poetry and uv dependency formats:
///
/// # Poetry Format
///
/// ```toml
/// [package.dependencies]
/// certifi = ">=2017.4.17"
/// charset-normalizer = ">=2,<4"
/// ```
///
/// # uv Format
///
/// ```toml
/// dependencies = [
///     { name = "certifi" },
///     { name = "charset-normalizer" },
/// ]
/// ```
fn parse_pypi_dependencies(table: &toml_edit::Table) -> Vec<String> {
    // Try uv format first (dependencies array)
    if let Some(deps_value) = table.get("dependencies")
        && let Some(deps_array) = deps_value.as_array()
    {
        return deps_array
            .iter()
            .filter_map(|item| {
                // uv format: { name = "certifi" }
                if let Some(dep_table) = item.as_inline_table()
                    && let Some(name) = dep_table.get("name").and_then(|v| v.as_str())
                {
                    return Some(name.to_string());
                }

                // Simple string format (fallback)
                if let Some(s) = item.as_str() {
                    return Some(s.to_string());
                }

                None
            })
            .collect();
    }

    // Try poetry format (package.dependencies table)
    if let Some(deps_item) = table.get("dependencies")
        && let Some(deps_table) = deps_item.as_table()
    {
        return deps_table
            .iter()
            .map(|(name, _)| name.to_string())
            .collect();
    }

    vec![]
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_parse_simple_poetry_lock() {
        let lockfile_content = r#"
# This file is automatically generated by poetry.
[[package]]
name = "requests"
version = "2.31.0"
description = "Python HTTP for Humans."

[package.dependencies]
certifi = ">=2017.4.17"
charset-normalizer = ">=2,<4"

[[package]]
name = "certifi"
version = "2023.7.22"
description = "Python package for providing Mozilla's CA Bundle."

[metadata]
lock-version = "2.0"
python-versions = "^3.9"
"#;

        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("poetry.lock");
        std::fs::write(&lockfile_path, lockfile_content).unwrap();

        let parser = PypiLockParser;
        let resolved = parser.parse_lockfile(&lockfile_path).await.unwrap();

        assert_eq!(resolved.len(), 2);
        assert_eq!(resolved.get_version("requests"), Some("2.31.0"));
        assert_eq!(resolved.get_version("certifi"), Some("2023.7.22"));

        let requests_pkg = resolved.get("requests").unwrap();
        assert_eq!(requests_pkg.dependencies.len(), 2);
        assert!(requests_pkg.dependencies.contains(&"certifi".to_string()));
        assert!(
            requests_pkg
                .dependencies
                .contains(&"charset-normalizer".to_string())
        );

        // Verify it's a registry source
        match &requests_pkg.source {
            ResolvedSource::Registry { url, .. } => {
                assert_eq!(url, "https://pypi.org/simple");
            }
            _ => panic!("Expected Registry source"),
        }
    }

    #[tokio::test]
    async fn test_parse_uv_lock() {
        let lockfile_content = r#"
version = 1

[[package]]
name = "requests"
version = "2.31.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
    { name = "certifi" },
    { name = "charset-normalizer" },
]

[[package]]
name = "certifi"
version = "2023.7.22"
source = { registry = "https://pypi.org/simple" }
"#;

        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("uv.lock");
        std::fs::write(&lockfile_path, lockfile_content).unwrap();

        let parser = PypiLockParser;
        let resolved = parser.parse_lockfile(&lockfile_path).await.unwrap();

        assert_eq!(resolved.len(), 2);
        assert_eq!(resolved.get_version("requests"), Some("2.31.0"));
        assert_eq!(resolved.get_version("certifi"), Some("2023.7.22"));

        let requests_pkg = resolved.get("requests").unwrap();
        assert_eq!(requests_pkg.dependencies.len(), 2);
        assert!(requests_pkg.dependencies.contains(&"certifi".to_string()));

        match &requests_pkg.source {
            ResolvedSource::Registry { url, .. } => {
                assert_eq!(url, "https://pypi.org/simple");
            }
            _ => panic!("Expected Registry source"),
        }
    }

    #[tokio::test]
    async fn test_parse_poetry_lock_with_git() {
        let lockfile_content = r#"
[[package]]
name = "my-git-dep"
version = "0.1.0"
description = "Git dependency"

[package.source]
type = "git"
url = "https://github.com/user/repo"
resolved_reference = "abc123def456"
"#;

        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("poetry.lock");
        std::fs::write(&lockfile_path, lockfile_content).unwrap();

        let parser = PypiLockParser;
        let resolved = parser.parse_lockfile(&lockfile_path).await.unwrap();

        // Names are normalized: - → _
        assert_eq!(resolved.len(), 1);
        let pkg = resolved.get("my_git_dep").unwrap();
        assert_eq!(pkg.version, "0.1.0");

        match &pkg.source {
            ResolvedSource::Git { url, rev } => {
                assert_eq!(url, "https://github.com/user/repo");
                assert_eq!(rev, "abc123def456");
            }
            _ => panic!("Expected Git source"),
        }
    }

    #[tokio::test]
    async fn test_parse_uv_lock_with_git() {
        let lockfile_content = r#"
version = 1

[[package]]
name = "my-git-dep"
version = "0.1.0"
source = { git = "https://github.com/user/repo", rev = "abc123" }
"#;

        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("uv.lock");
        std::fs::write(&lockfile_path, lockfile_content).unwrap();

        let parser = PypiLockParser;
        let resolved = parser.parse_lockfile(&lockfile_path).await.unwrap();

        // Names are normalized: - → _
        assert_eq!(resolved.len(), 1);
        let pkg = resolved.get("my_git_dep").unwrap();

        match &pkg.source {
            ResolvedSource::Git { url, rev } => {
                assert_eq!(url, "https://github.com/user/repo");
                assert_eq!(rev, "abc123");
            }
            _ => panic!("Expected Git source"),
        }
    }

    #[tokio::test]
    async fn test_parse_poetry_lock_with_path() {
        let lockfile_content = r#"
[[package]]
name = "my-local-dep"
version = "0.1.0"

[package.source]
type = "directory"
url = "../local-package"
"#;

        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("poetry.lock");
        std::fs::write(&lockfile_path, lockfile_content).unwrap();

        let parser = PypiLockParser;
        let resolved = parser.parse_lockfile(&lockfile_path).await.unwrap();

        // Names are normalized: - → _
        assert_eq!(resolved.len(), 1);
        let pkg = resolved.get("my_local_dep").unwrap();

        match &pkg.source {
            ResolvedSource::Path { path } => {
                assert_eq!(path, "../local-package");
            }
            _ => panic!("Expected Path source"),
        }
    }

    #[tokio::test]
    async fn test_parse_uv_lock_with_path() {
        let lockfile_content = r#"
version = 1

[[package]]
name = "my-local-dep"
version = "0.1.0"
source = { path = "../local-package" }
"#;

        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("uv.lock");
        std::fs::write(&lockfile_path, lockfile_content).unwrap();

        let parser = PypiLockParser;
        let resolved = parser.parse_lockfile(&lockfile_path).await.unwrap();

        // Names are normalized: - → _
        assert_eq!(resolved.len(), 1);
        let pkg = resolved.get("my_local_dep").unwrap();

        match &pkg.source {
            ResolvedSource::Path { path } => {
                assert_eq!(path, "../local-package");
            }
            _ => panic!("Expected Path source"),
        }
    }

    #[tokio::test]
    async fn test_parse_empty_lock_file() {
        let lockfile_content = r"
version = 1
";

        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("poetry.lock");
        std::fs::write(&lockfile_path, lockfile_content).unwrap();

        let parser = PypiLockParser;
        let resolved = parser.parse_lockfile(&lockfile_path).await.unwrap();

        assert_eq!(resolved.len(), 0);
        assert!(resolved.is_empty());
    }

    #[tokio::test]
    async fn test_parse_malformed_toml() {
        let lockfile_content = "not valid toml {{{";

        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("poetry.lock");
        std::fs::write(&lockfile_path, lockfile_content).unwrap();

        let parser = PypiLockParser;
        let result = parser.parse_lockfile(&lockfile_path).await;

        assert!(result.is_err());
    }

    #[test]
    fn test_locate_lockfile_poetry_priority() {
        let temp_dir = tempfile::tempdir().unwrap();
        let manifest_path = temp_dir.path().join("pyproject.toml");
        let poetry_lock = temp_dir.path().join("poetry.lock");
        let uv_lock = temp_dir.path().join("uv.lock");

        std::fs::write(&manifest_path, "[project]\nname = \"test\"").unwrap();
        std::fs::write(&poetry_lock, "# poetry.lock").unwrap();
        std::fs::write(&uv_lock, "# uv.lock").unwrap();

        let manifest_uri = Uri::from_file_path(&manifest_path).unwrap();
        let parser = PypiLockParser;

        let located = parser.locate_lockfile(&manifest_uri);
        assert!(located.is_some());
        assert_eq!(
            located.unwrap(),
            poetry_lock,
            "poetry.lock should take priority over uv.lock"
        );
    }

    #[test]
    fn test_locate_lockfile_uv_fallback() {
        let temp_dir = tempfile::tempdir().unwrap();
        let manifest_path = temp_dir.path().join("pyproject.toml");
        let uv_lock = temp_dir.path().join("uv.lock");

        std::fs::write(&manifest_path, "[project]\nname = \"test\"").unwrap();
        std::fs::write(&uv_lock, "# uv.lock").unwrap();

        let manifest_uri = Uri::from_file_path(&manifest_path).unwrap();
        let parser = PypiLockParser;

        let located = parser.locate_lockfile(&manifest_uri);
        assert!(located.is_some());
        assert_eq!(located.unwrap(), uv_lock);
    }

    #[test]
    fn test_locate_lockfile_not_found() {
        let temp_dir = tempfile::tempdir().unwrap();
        let manifest_path = temp_dir.path().join("pyproject.toml");
        std::fs::write(&manifest_path, "[project]\nname = \"test\"").unwrap();

        let manifest_uri = Uri::from_file_path(&manifest_path).unwrap();
        let parser = PypiLockParser;

        let located = parser.locate_lockfile(&manifest_uri);
        assert!(located.is_none());
    }

    #[tokio::test]
    async fn test_parse_poetry_lock_missing_fields() {
        let lockfile_content = r#"
[[package]]
name = "valid-package"
version = "1.0.0"

[[package]]
# Missing name field
version = "2.0.0"

[[package]]
name = "missing-version"
# Missing version field
"#;

        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("poetry.lock");
        std::fs::write(&lockfile_path, lockfile_content).unwrap();

        let parser = PypiLockParser;
        let resolved = parser.parse_lockfile(&lockfile_path).await.unwrap();

        // Should only parse valid package (names are normalized: - → _)
        assert_eq!(resolved.len(), 1);
        assert_eq!(resolved.get_version("valid_package"), Some("1.0.0"));
        assert!(resolved.get("missing_version").is_none());
    }

    #[test]
    fn test_is_lockfile_stale_not_modified() {
        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("poetry.lock");
        std::fs::write(&lockfile_path, "version = 1").unwrap();

        let mtime = std::fs::metadata(&lockfile_path)
            .unwrap()
            .modified()
            .unwrap();
        let parser = PypiLockParser;

        assert!(
            !parser.is_lockfile_stale(&lockfile_path, mtime),
            "Lock file should not be stale when mtime matches"
        );
    }

    #[test]
    fn test_is_lockfile_stale_modified() {
        let temp_dir = tempfile::tempdir().unwrap();
        let lockfile_path = temp_dir.path().join("poetry.lock");
        std::fs::write(&lockfile_path, "version = 1").unwrap();

        let old_time = std::time::UNIX_EPOCH;
        let parser = PypiLockParser;

        assert!(
            parser.is_lockfile_stale(&lockfile_path, old_time),
            "Lock file should be stale when last_modified is old"
        );
    }

    #[test]
    fn test_is_lockfile_stale_deleted() {
        let parser = PypiLockParser;
        let non_existent = std::path::Path::new("/nonexistent/poetry.lock");

        assert!(
            parser.is_lockfile_stale(non_existent, std::time::SystemTime::now()),
            "Non-existent lock file should be considered stale"
        );
    }
}