use serde::Deserialize;
use std::collections::{BTreeMap, HashSet};
use std::ffi::OsString;
use std::io;
use std::path::PathBuf;
use tokio::process::Command;
use crate::core::analyzers::python_manager::Capability;
use crate::core::error::{ErrorCode, Result, UpkeepError};
use crate::core::pep440;
use crate::core::python::{
normalize_package_name, PythonDependencyScope, PythonMarker, PythonOutdatedPackage,
PythonOutdatedReport, PythonUnavailableReason, PythonUpdateCounts, PythonUpdateType,
};
pub const POETRY_BIN_ENV: &str = "UPKEEP_POETRY_BIN";
const CAPABILITY_PROBE_VALUE: &str = "cargo-upkeep-capability-probe";
const UPGRADE_HINT: &str =
"upgrade Poetry (`poetry self update`, or through whichever package manager installed it)";
const SHOW_ARGS: [&str; 3] = ["show", "--latest", "--format"];
const TOP_LEVEL_ARGS: [&str; 4] = ["show", "--latest", "--top-level", "--format"];
pub const SECURITY_UNSUPPORTED_DETAIL: &str =
"Poetry has no vulnerability scanner: `poetry check` validates pyproject.toml against the \
lockfile and is not a scan. No install closes this gap — run a dedicated scanner and gate on \
that instead.";
pub struct Poetry {
binary: OsString,
project_root: PathBuf,
version: Option<String>,
}
impl Poetry {
pub async fn detect(project_root: PathBuf) -> Result<Self> {
let binary = std::env::var_os(POETRY_BIN_ENV).unwrap_or_else(|| OsString::from("poetry"));
let output = Command::new(&binary)
.arg("--version")
.current_dir(&project_root)
.output()
.await
.map_err(|err| match err.kind() {
io::ErrorKind::NotFound => UpkeepError::message(
ErrorCode::MissingTool,
"no supported Python manager could be detected: this project is a Poetry \
project, but poetry is not installed or not on PATH; see \
https://python-poetry.org/docs/#installation",
),
_ => UpkeepError::context(
ErrorCode::ExternalCommand,
"failed to execute poetry --version",
err,
),
})?;
Ok(Self {
binary,
project_root,
version: parse_version(&String::from_utf8_lossy(&output.stdout)),
})
}
pub fn version(&self) -> Option<&str> {
self.version.as_deref()
}
async fn run(&self, args: &[&str]) -> Result<std::process::Output> {
Command::new(&self.binary)
.args(args)
.current_dir(&self.project_root)
.env("POETRY_VIRTUALENVS_CREATE", "false")
.output()
.await
.map_err(|err| {
UpkeepError::context(
ErrorCode::ExternalCommand,
format!("failed to execute poetry {}", args.join(" ")),
err,
)
})
}
pub async fn probe_outdated(&self) -> Capability {
let mut argv: Vec<&str> = SHOW_ARGS.to_vec();
argv.push(CAPABILITY_PROBE_VALUE);
let output = match self.run(&argv).await {
Ok(output) => output,
Err(err) => {
return Capability::Unavailable {
reason: PythonUnavailableReason::Failed,
detail: err.to_string(),
}
}
};
probe_capability(
&String::from_utf8_lossy(&output.stderr),
output.status.success(),
&SHOW_ARGS,
)
}
pub async fn outdated(&self) -> Result<(PythonOutdatedReport, Vec<String>)> {
let all = self.show(&SHOW_ARGS).await?;
let mut warnings = Vec::new();
let top_level = match self.show(&TOP_LEVEL_ARGS).await {
Ok(packages) => Some(packages),
Err(err) => {
warnings.push(format!(
"`poetry show --top-level` did not answer, so no package could be shown to be \
a direct dependency and every `scope` is reported as `unknown`: {err}"
));
None
}
};
Ok((normalize_show(&all, top_level.as_deref()), warnings))
}
async fn show(&self, args: &[&str]) -> Result<Vec<PoetryPackage>> {
let mut argv: Vec<&str> = args.to_vec();
argv.push("json");
let output = self.run(&argv).await?;
let stdout = String::from_utf8_lossy(&output.stdout);
if stdout.trim().is_empty() {
return Err(external_failure(
&format!("poetry {}", argv.join(" ")),
&output,
));
}
serde_json::from_str(&stdout).map_err(|err| {
UpkeepError::context(
ErrorCode::InvalidData,
format!(
"poetry {} did not produce the expected JSON",
argv.join(" ")
),
err,
)
})
}
}
fn external_failure(command: &str, output: &std::process::Output) -> UpkeepError {
let stderr = String::from_utf8_lossy(&output.stderr);
UpkeepError::message(
ErrorCode::ExternalCommand,
match poetry_message(&stderr) {
Some(message) => format!("{command} failed: {message}"),
None => format!("{command} failed with no output"),
},
)
}
const VIRTUALENV_NOTICE: &str = "Skipping virtualenv creation";
fn poetry_message(stderr: &str) -> Option<String> {
let interesting = || {
stderr
.lines()
.map(str::trim)
.filter(|line| !line.is_empty() && !line.starts_with(VIRTUALENV_NOTICE))
};
interesting()
.find(|line| line.starts_with("Error:"))
.map(str::to_string)
.or_else(|| {
let lines: Vec<&str> = interesting().collect();
(!lines.is_empty()).then(|| lines.join(" "))
})
}
fn parse_version(stdout: &str) -> Option<String> {
let line = stdout.lines().next()?.trim();
let inside = line.strip_prefix("Poetry (version ")?.strip_suffix(')')?;
let version = inside.trim();
(!version.is_empty()).then(|| version.to_string())
}
pub fn probe_capability(stderr: &str, succeeded: bool, args: &[&str]) -> Capability {
if let Some(formats) = supported_formats(stderr) {
return if formats.iter().any(|format| format == "json") {
Capability::Available
} else {
Capability::Unavailable {
reason: PythonUnavailableReason::NotInstalled,
detail: format!(
"`poetry show --format` accepts only {} on this Poetry; {UPGRADE_HINT} for \
machine-readable output",
formats.join(", "),
),
}
};
}
for flag in args.iter().filter(|arg| arg.starts_with("--")) {
if is_unknown_option(stderr, flag) {
return Capability::Unavailable {
reason: PythonUnavailableReason::NotInstalled,
detail: format!(
"`poetry show` on this Poetry does not accept `{flag}`; {UPGRADE_HINT}"
),
};
}
}
if is_unknown_command(stderr, "show") {
return Capability::Unavailable {
reason: PythonUnavailableReason::NotInstalled,
detail: format!("this Poetry has no `show` command; {UPGRADE_HINT}"),
};
}
let detail = if succeeded {
format!(
"could not establish whether `poetry show` supports JSON output: Poetry accepted the \
probe value `{CAPABILITY_PROBE_VALUE}` instead of rejecting it"
)
} else {
format!(
"could not establish whether `poetry show` supports JSON output: {}",
poetry_message(stderr).unwrap_or_else(|| "poetry produced no output".to_string())
)
};
Capability::Unavailable {
reason: PythonUnavailableReason::Failed,
detail,
}
}
fn supported_formats(stderr: &str) -> Option<Vec<String>> {
const PREFIX: &str = "invalid output format. supported formats are:";
let collapsed = stderr
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
.to_lowercase();
let list = collapsed.get(collapsed.find(PREFIX)? + PREFIX.len()..)?;
let end = list.find('.').unwrap_or(list.len());
Some(
list.get(..end)?
.split(',')
.map(|format| format.trim().trim_matches(['`', '\'', '"']).to_string())
.filter(|format| !format.is_empty())
.collect(),
)
}
fn is_unknown_option(stderr: &str, flag: &str) -> bool {
let needle = format!("the option \"{}\" does not exist", flag.to_lowercase());
collapsed_lowercase(stderr).contains(&needle)
}
fn is_unknown_command(stderr: &str, command: &str) -> bool {
let needle = format!(
"the requested command {} does not exist",
command.to_lowercase()
);
collapsed_lowercase(stderr).contains(&needle)
}
fn collapsed_lowercase(stderr: &str) -> String {
stderr
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
.to_lowercase()
}
#[derive(Debug, Deserialize)]
pub struct PoetryPackage {
name: String,
version: Option<String>,
#[serde(default)]
latest_version: Option<String>,
}
pub fn normalize_show(
all: &[PoetryPackage],
top_level: Option<&[PoetryPackage]>,
) -> PythonOutdatedReport {
let top_level = top_level.filter(|packages| !packages.is_empty() || all.is_empty());
let direct: Option<HashSet<String>> = top_level.map(|packages| {
packages
.iter()
.map(|package| normalize_package_name(&package.name))
.collect()
});
let mut settled: BTreeMap<String, (&str, &str)> = BTreeMap::new();
for package in all {
let (Some(current), Some(latest)) = (
package.version.as_deref(),
package.latest_version.as_deref(),
) else {
continue;
};
settled.insert(normalize_package_name(&package.name), (current, latest));
}
let mut counts = PythonUpdateCounts {
epoch: 0,
major: 0,
minor: 0,
patch: 0,
qualifier: 0,
unclassified: 0,
};
let mut packages = Vec::new();
for (name, (current, latest)) in &settled {
if pep440::is_same_version(current, latest) {
continue;
}
let update_type = pep440::classify(current, latest);
match update_type {
PythonUpdateType::Epoch => counts.epoch += 1,
PythonUpdateType::Major => counts.major += 1,
PythonUpdateType::Minor => counts.minor += 1,
PythonUpdateType::Patch => counts.patch += 1,
PythonUpdateType::Qualifier => counts.qualifier += 1,
PythonUpdateType::Unclassified => counts.unclassified += 1,
}
packages.push(PythonOutdatedPackage {
name: name.clone(),
current: (*current).to_string(),
latest: (*latest).to_string(),
update_type,
scope: match &direct {
Some(direct) if direct.contains(name) => PythonDependencyScope::Direct,
Some(_) => PythonDependencyScope::Transitive,
None => PythonDependencyScope::Unknown,
},
groups: None,
extras: None,
marker: PythonMarker::NotReported,
});
}
PythonOutdatedReport {
checked: settled.len(),
outdated: packages.len(),
counts,
packages,
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::Path;
fn fixture(name: &str) -> Vec<PoetryPackage> {
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("tests")
.join("fixtures")
.join("poetry")
.join(name);
let contents = std::fs::read_to_string(&path)
.unwrap_or_else(|err| panic!("missing poetry fixture {}: {err}", path.display()));
serde_json::from_str(&contents)
.unwrap_or_else(|err| panic!("parse poetry fixture {}: {err}", path.display()))
}
fn all() -> Vec<PoetryPackage> {
fixture("show-latest.json")
}
fn top_level() -> Vec<PoetryPackage> {
fixture("show-latest-top-level.json")
}
fn report() -> PythonOutdatedReport {
normalize_show(&all(), Some(&top_level()))
}
fn package<'a>(report: &'a PythonOutdatedReport, name: &str) -> &'a PythonOutdatedPackage {
report
.packages
.iter()
.find(|package| package.name == name)
.unwrap_or_else(|| panic!("{name} missing from the outdated report"))
}
#[test]
fn unreported_attributes_are_null_and_never_empty() {
let report = report();
assert!(
!report.packages.is_empty(),
"fixture premise: there are entries to inspect"
);
for package in &report.packages {
assert_eq!(
package.groups, None,
"{}: Poetry reports no groups, so `[]` would claim it has none",
package.name
);
assert_eq!(
package.extras, None,
"{}: Poetry reports no extras, so `[]` would claim it has none",
package.name
);
assert_eq!(
package.marker,
PythonMarker::NotReported,
"{}: `absent` would claim Poetry reports markers and this one has none",
package.name
);
}
let serialized = serde_json::to_value(&report.packages[0]).expect("serialize");
assert!(serialized["groups"].is_null());
assert!(serialized["extras"].is_null());
assert_ne!(serialized["groups"], serde_json::json!([]));
assert_eq!(
serialized["marker"],
serde_json::json!({"status": "not_reported"})
);
}
#[test]
fn checked_counts_every_package_not_only_the_outdated_ones() {
let report = report();
assert_eq!(report.checked, 16, "the fixture lists sixteen packages");
assert_eq!(report.outdated, 10);
assert!(
report.checked > report.outdated,
"an --outdated-only listing would make these equal and the denominator useless"
);
for current in [
"certifi",
"itsdangerous",
"jinja2",
"pyparsing",
"pysocks",
"six",
] {
assert!(
!report
.packages
.iter()
.any(|package| package.name == current),
"{current} is at its latest version in the fixture"
);
}
}
#[test]
fn an_empty_top_level_listing_leaves_scope_unknown() {
let report = normalize_show(&all(), Some(&[]));
assert!(
!report.packages.is_empty(),
"fixture premise: packages were settled"
);
for package in &report.packages {
assert_eq!(
package.scope,
PythonDependencyScope::Unknown,
"{}: an unanswered listing is not evidence of transitivity",
package.name
);
}
}
#[test]
fn scope_comes_from_the_top_level_listing() {
let report = report();
for direct in [
"requests",
"flask",
"pyyaml",
"click",
"packaging",
"markupsafe",
] {
assert_eq!(
package(&report, direct).scope,
PythonDependencyScope::Direct,
"{direct} is declared by the project"
);
}
for transitive in ["urllib3", "idna", "chardet", "werkzeug"] {
assert_eq!(
package(&report, transitive).scope,
PythonDependencyScope::Transitive,
"{transitive} is only reached through another package"
);
}
assert!(
top_level().iter().any(|package| package.name == "six"),
"fixture premise: --top-level lists a direct dependency that is not outdated, which \
is what proves it lists direct dependencies rather than fewer outdated ones"
);
}
#[test]
fn a_missing_top_level_listing_reports_unknown_scope() {
let report = normalize_show(&all(), None);
assert!(!report.packages.is_empty(), "fixture premise");
assert!(
report
.packages
.iter()
.all(|package| package.scope == PythonDependencyScope::Unknown),
"an entry that cannot be established as direct must not be filed as transitive"
);
}
#[test]
fn summaries_agree_with_their_entries() {
let report = report();
let counts = &report.counts;
assert_eq!(report.outdated, report.packages.len());
assert_eq!(
counts.epoch
+ counts.major
+ counts.minor
+ counts.patch
+ counts.qualifier
+ counts.unclassified,
report.outdated
);
assert!(report.checked >= report.outdated);
}
#[test]
fn update_types_classify_the_captured_versions() {
let report = report();
for major in ["chardet", "flask", "urllib3"] {
assert_eq!(package(&report, major).update_type, PythonUpdateType::Major);
}
for minor in ["click", "requests"] {
assert_eq!(package(&report, minor).update_type, PythonUpdateType::Minor);
}
assert_eq!(report.counts.major, 8);
assert_eq!(report.counts.minor, 2);
assert_eq!(report.counts.unclassified, 0);
}
#[test]
fn versions_that_differ_only_in_spelling_are_not_outdated() {
let packages: Vec<PoetryPackage> = serde_json::from_str(
r#"[{"name":"Zope.Interface","installed_status":"installed","version":"1.0","latest_version":"1.0.0"},
{"name":"real","installed_status":"installed","version":"1.0","latest_version":"1.1"}]"#,
)
.expect("parse");
let report = normalize_show(&packages, None);
assert_eq!(report.checked, 2, "both had their freshness settled");
assert_eq!(report.outdated, 1);
assert_eq!(report.packages[0].name, "real");
}
#[test]
fn a_latest_below_current_is_unclassified_rather_than_an_upgrade() {
let packages: Vec<PoetryPackage> = serde_json::from_str(
r#"[{"name":"pre","installed_status":"installed","version":"2.0.0rc1","latest_version":"1.9.0"}]"#,
)
.expect("parse");
let report = normalize_show(&packages, None);
assert_eq!(
report.packages[0].update_type,
PythonUpdateType::Unclassified,
"calling a downgrade a `major` update would advertise it as an upgrade"
);
assert_eq!(report.counts.unclassified, 1);
assert_eq!(report.counts.major, 0);
}
#[test]
fn a_package_without_a_latest_version_is_not_counted_as_checked() {
let packages: Vec<PoetryPackage> = serde_json::from_str(
r#"[{"name":"answered","installed_status":"installed","version":"1.0","latest_version":"2.0"},
{"name":"unanswered","installed_status":"installed","version":"1.0"}]"#,
)
.expect("parse");
let report = normalize_show(&packages, None);
assert_eq!(
report.checked, 1,
"only one package's freshness was settled"
);
assert_eq!(report.outdated, 1);
}
#[test]
fn package_names_are_normalized_on_both_sides() {
let all: Vec<PoetryPackage> = serde_json::from_str(
r#"[{"name":"Zope.Interface","installed_status":"installed","version":"5.0","latest_version":"6.0"}]"#,
)
.expect("parse");
let top: Vec<PoetryPackage> = serde_json::from_str(
r#"[{"name":"zope_interface","installed_status":"installed","version":"5.0","latest_version":"6.0"}]"#,
)
.expect("parse");
let report = normalize_show(&all, Some(&top));
assert_eq!(report.packages[0].name, "zope-interface");
assert_eq!(report.packages[0].scope, PythonDependencyScope::Direct);
}
const CURRENT_FORMAT_PROBE: &str =
"Error: Invalid output format. Supported formats are: json, text.\n";
const UNKNOWN_OPTION_PROBE: &str = "\nThe option \"--format\" does not exist\n";
fn unavailable(capability: Capability) -> (PythonUnavailableReason, String) {
match capability {
Capability::Available => panic!("expected the capability to be unavailable"),
Capability::Unavailable { reason, detail } => (reason, detail),
}
}
#[test]
fn current_poetry_advertises_json() {
assert!(matches!(
probe_capability(CURRENT_FORMAT_PROBE, false, &SHOW_ARGS),
Capability::Available
));
}
#[test]
fn a_poetry_without_the_format_flag_reports_an_upgrade_hint() {
let (reason, detail) =
unavailable(probe_capability(UNKNOWN_OPTION_PROBE, false, &SHOW_ARGS));
assert_eq!(reason, PythonUnavailableReason::NotInstalled);
assert!(detail.contains("--format"), "{detail}");
assert!(detail.contains("poetry self update"), "{detail}");
}
#[test]
fn a_rejected_flag_is_reported_against_the_flag_poetry_named() {
let (reason, detail) = unavailable(probe_capability(
"\nThe option \"--latest\" does not exist\n",
false,
&SHOW_ARGS,
));
assert_eq!(reason, PythonUnavailableReason::NotInstalled);
assert!(detail.contains("--latest"), "{detail}");
}
#[test]
fn a_format_flag_without_json_is_a_capability_gap() {
let (reason, detail) = unavailable(probe_capability(
"Error: Invalid output format. Supported formats are: text.\n",
false,
&SHOW_ARGS,
));
assert_eq!(reason, PythonUnavailableReason::NotInstalled);
assert!(detail.contains("accepts only text"), "{detail}");
}
#[test]
fn a_rejected_command_is_not_read_as_a_rejected_flag() {
let (reason, detail) = unavailable(probe_capability(
"The requested command show does not exist.\n\nDocumentation: \
https://python-poetry.org/docs/cli/\n",
false,
&SHOW_ARGS,
));
assert_eq!(reason, PythonUnavailableReason::NotInstalled);
assert!(
detail.contains("no `show` command"),
"a missing command must not be reported as a missing flag: {detail}"
);
}
#[test]
fn a_wrapped_message_is_still_read() {
assert!(matches!(
probe_capability(
"Error: Invalid output\n format. Supported formats\n are: json, text.\n",
false,
&SHOW_ARGS
),
Capability::Available
));
let (reason, _) = unavailable(probe_capability(
"The option\n \"--format\" does not\n exist\n",
false,
&SHOW_ARGS,
));
assert_eq!(reason, PythonUnavailableReason::NotInstalled);
}
#[test]
fn an_accepted_probe_value_is_inconclusive() {
let (reason, detail) = unavailable(probe_capability("", true, &SHOW_ARGS));
assert_eq!(reason, PythonUnavailableReason::Failed);
assert!(detail.contains("accepted the probe value"), "{detail}");
}
#[test]
fn an_unrecognized_failure_is_inconclusive_not_a_missing_tool() {
let (reason, detail) = unavailable(probe_capability(
"Error: poetry.lock not found. Run `poetry lock` to create it.\n",
false,
&SHOW_ARGS,
));
assert_eq!(
reason,
PythonUnavailableReason::Failed,
"a broken project is not an old Poetry"
);
assert!(detail.contains("poetry.lock not found"), "{detail}");
}
#[test]
fn our_own_virtualenv_notice_is_never_reported_as_the_failure() {
const REAL: &str = "Skipping virtualenv creation, as specified in config file.\nError: \
poetry.lock not found. Run `poetry lock` to create it.\n";
assert_eq!(
poetry_message(REAL),
Some("Error: poetry.lock not found. Run `poetry lock` to create it.".to_string()),
"the actionable line is Poetry's, not ours"
);
let (_, detail) = unavailable(probe_capability(REAL, false, &SHOW_ARGS));
assert!(
detail.contains("poetry.lock not found"),
"the probe detail must carry Poetry's complaint: {detail}"
);
assert!(
!detail.contains(VIRTUALENV_NOTICE),
"the probe detail must not blame our own suppression: {detail}"
);
}
#[test]
fn the_reported_message_prefers_poetrys_own_error_line() {
assert_eq!(
poetry_message("Loading configuration\nError: something broke\ntrailing\n"),
Some("Error: something broke".to_string())
);
assert_eq!(
poetry_message("\n The option \"--format\" does not exist\n"),
Some("The option \"--format\" does not exist".to_string())
);
assert_eq!(
poetry_message(
"Skipping virtualenv creation, as specified in config file.\n\n\
The lock file does not have a metadata entry.\n\
Regenerate the lock file with the `poetry lock` command.\n"
),
Some(
"The lock file does not have a metadata entry. Regenerate the lock file \
with the `poetry lock` command."
.to_string()
),
);
assert_eq!(poetry_message(""), None);
assert_eq!(
poetry_message("Skipping virtualenv creation, as specified in config file.\n"),
None,
"a stderr holding nothing but our own notice said nothing at all"
);
}
#[test]
fn version_is_parsed_from_poetry_version_output() {
assert_eq!(
parse_version("Poetry (version 2.4.2)\n").as_deref(),
Some("2.4.2")
);
assert_eq!(
parse_version("Poetry (version 1.8.3)").as_deref(),
Some("1.8.3")
);
assert_eq!(parse_version("something else\n"), None);
assert_eq!(parse_version("Poetry (version )\n"), None);
assert_eq!(parse_version(""), None);
}
}
#[cfg(test)]
mod stderr_parsing_tests {
use super::*;
#[test]
fn a_non_ascii_prefix_does_not_break_the_format_list() {
let stderr = "İ Error: Invalid output format. Supported formats are:\u{e9}json, text.";
assert_eq!(
supported_formats(stderr),
Some(vec!["\u{e9}json".to_string(), "text".to_string()]),
"the list must be read from the same string the offset was found in"
);
}
#[test]
fn a_non_ascii_prefix_does_not_corrupt_the_format_names() {
let stderr = "İ Error: Invalid output format. Supported formats are:json, text.";
let formats = supported_formats(stderr).expect("the list is still found");
assert!(
formats.iter().any(|format| format == "json"),
"json must survive verbatim; got {formats:?}"
);
}
#[test]
fn an_unrelated_stderr_has_no_format_list() {
assert_eq!(supported_formats("Error: something else entirely"), None);
}
}