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 deps_core::error::{DepsError, Result};
use deps_core::lockfile::{
    LockFileProvider, ResolvedPackage, ResolvedPackages, ResolvedSource,
    locate_lockfile_for_manifest, read_lockfile_content,
};
use std::path::{Path, PathBuf};
use toml_span::value::Table;
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"];
}

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

    fn parse_lockfile<'a>(
        &'a self,
        lockfile_path: &'a Path,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<ResolvedPackages>> + Send + 'a>>
    {
        Box::pin(async move {
            tracing::debug!("Parsing lock file: {}", lockfile_path.display());

            let content = read_lockfile_content(lockfile_path, "lock file").await?;

            if let Err(depth) =
                deps_core::check_toml_nesting_depth(&content, deps_core::MAX_TOML_NESTING_DEPTH)
            {
                return Err(DepsError::ParseError {
                    file_type: "Python lock file".into(),
                    source: Box::new(std::io::Error::other(format!(
                        "array/table nesting depth {depth} exceeds maximum of {}",
                        deps_core::MAX_TOML_NESTING_DEPTH
                    ))),
                });
            }

            let doc = toml_span::parse(&content).map_err(|e| DepsError::ParseError {
                file_type: "Python lock file".into(),
                source: Box::new(std::io::Error::other(e.to_string())),
            })?;

            let mut packages = ResolvedPackages::new();

            // [[package]] in TOML is an array of tables; toml-span represents it as an array of values
            let Some(package_array) = doc
                .as_table()
                .and_then(|t| t.get("package"))
                .and_then(|v| v.as_array())
            else {
                tracing::warn!("Lock file missing [[package]] array of tables");
                return Ok(packages);
            };

            for item in package_array {
                let Some(table) = item.as_table() else {
                    continue;
                };

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

                let Some(version) = table.get("version").and_then(|v| 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 (PEP 503: case/`_`/`.`-insensitive)
                let normalized_name = crate::name::normalize(name);
                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: &Table<'_>) -> ResolvedSource {
    let Some(source_val) = table.get("source") else {
        // No source field = PyPI registry (poetry default)
        return ResolvedSource::Registry {
            url: "https://pypi.org/simple".to_string(),
            checksum: String::new(),
        };
    };

    // In toml-span, both inline tables and regular tables are represented as Table
    if let Some(source_table) = source_val.as_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(),
            };
        }

        // 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: &Table<'_>) -> Vec<String> {
    let Some(deps_val) = table.get("dependencies") else {
        return vec![];
    };

    // uv format: dependencies as array of inline tables [{ name = "certifi" }]
    if let Some(deps_array) = deps_val.as_array() {
        return deps_array
            .iter()
            .filter_map(|item| {
                // uv format: { name = "certifi" }
                if let Some(dep_table) = item.as_table() {
                    return dep_table
                        .get("name")
                        .and_then(|v| v.as_str())
                        .map(String::from);
                }

                // Simple string format (fallback)
                item.as_str().map(String::from)
            })
            .collect();
    }

    // Poetry format: dependencies as table with package names as keys
    if let Some(deps_table) = deps_val.as_table() {
        return deps_table.keys().map(|key| key.name.to_string()).collect();
    }

    vec![]
}

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

    use std::assert_matches;

    #[tokio::test]
    async fn test_parse_lockfile_rejects_excessive_nesting() {
        // Well past MAX_TOML_NESTING_DEPTH (64) but far below the depth
        // that would actually overflow the stack, so the guard is what's
        // being exercised here, not the crash itself.
        let lockfile_content = format!("a = {}1{}", "[".repeat(300), "]".repeat(300));

        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_matches!(
            result,
            Err(DepsError::ParseError { file_type, .. }) if file_type == "Python lock file"
        );
    }

    #[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 via PEP 503 (case/`_`/`.` collapsed to `-`)
        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 via PEP 503 (case/`_`/`.` collapsed to `-`)
        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 via PEP 503 (case/`_`/`.` collapsed to `-`)
        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 via PEP 503 (case/`_`/`.` collapsed to `-`)
        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"
        );
    }
}