use serde::Deserialize;
use serde_json::Value;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
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, PythonSecurityReport, PythonSecuritySummary, PythonSeverity,
PythonUnavailableReason, PythonUpdateCounts, PythonUpdateType, PythonVulnerability,
};
pub const UV_BIN_ENV: &str = "UPKEEP_UV_BIN";
const CAPABILITY_PROBE_VALUE: &str = "cargo-upkeep-capability-probe";
const UPGRADE_HINT: &str =
"upgrade uv (`uv self update`, or through whichever package manager installed it)";
const TREE_ARGS: [&str; 4] = ["tree", "--outdated", "--frozen", "--format"];
const AUDIT_ARGS: [&str; 3] = ["audit", "--frozen", "--output-format"];
pub struct Uv {
binary: OsString,
project_root: PathBuf,
version: Option<String>,
}
impl Uv {
pub async fn detect(project_root: PathBuf) -> Result<Self> {
let binary = std::env::var_os(UV_BIN_ENV).unwrap_or_else(|| OsString::from("uv"));
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: uv is not installed or not on \
PATH; see https://docs.astral.sh/uv/getting-started/installation/",
),
_ => UpkeepError::context(
ErrorCode::ExternalCommand,
"failed to execute uv --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)
.output()
.await
.map_err(|err| {
UpkeepError::context(
ErrorCode::ExternalCommand,
format!("failed to execute uv {}", args.join(" ")),
err,
)
})
}
pub async fn probe_outdated(&self) -> Capability {
self.probe(&TREE_ARGS, "tree", "--format").await
}
pub async fn probe_security(&self) -> Capability {
self.probe(&AUDIT_ARGS, "audit", "--output-format").await
}
async fn probe(&self, args: &[&str], subcommand: &str, format_flag: &str) -> Capability {
let mut argv: Vec<&str> = 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(),
args,
subcommand,
format_flag,
)
}
pub async fn outdated(&self) -> Result<(PythonOutdatedReport, ScopeIndex)> {
let mut argv: Vec<&str> = TREE_ARGS.to_vec();
argv.push("json");
let output = self.run(&argv).await?;
let stdout = String::from_utf8_lossy(&output.stdout);
if !output.status.success() && stdout.trim().is_empty() {
return Err(external_failure("uv tree --outdated", &output));
}
let tree: UvTree = serde_json::from_str(&stdout).map_err(|err| {
UpkeepError::context(
ErrorCode::InvalidData,
"uv tree --format json did not produce the expected JSON",
err,
)
})?;
Ok(normalize_tree_with_scopes(&tree))
}
pub async fn security(
&self,
scopes: &ScopeIndex,
) -> Result<(PythonSecurityReport, Vec<String>)> {
let mut argv: Vec<&str> = AUDIT_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("uv audit", &output));
}
let audit: UvAudit = serde_json::from_str(&stdout).map_err(|err| {
UpkeepError::context(
ErrorCode::InvalidData,
"uv audit --output-format json did not produce the expected JSON",
err,
)
})?;
Ok(normalize_audit(&audit, scopes))
}
}
fn external_failure(command: &str, output: &std::process::Output) -> UpkeepError {
let stderr = String::from_utf8_lossy(&output.stderr);
let message = stderr.trim();
UpkeepError::message(
ErrorCode::ExternalCommand,
if message.is_empty() {
format!("{command} failed with no output")
} else {
format!("{command} failed: {message}")
},
)
}
fn parse_version(stdout: &str) -> Option<String> {
let mut tokens = stdout.lines().next()?.split_whitespace();
if tokens.next()? != "uv" {
return None;
}
tokens.next().map(str::to_string)
}
pub fn probe_capability(
stderr: &str,
succeeded: bool,
args: &[&str],
subcommand: &str,
format_flag: &str,
) -> Capability {
if let Some(values) = possible_values(stderr, format_flag) {
return if values.iter().any(|value| value == "json") {
Capability::Available
} else {
Capability::Unavailable {
reason: PythonUnavailableReason::NotInstalled,
detail: format!(
"`uv {subcommand} {format_flag}` accepts only {} on this uv; {UPGRADE_HINT} \
for machine-readable output",
values.join(", "),
),
}
};
}
if is_unrecognized_subcommand(stderr, subcommand) {
return Capability::Unavailable {
reason: PythonUnavailableReason::NotInstalled,
detail: format!("this uv has no `{subcommand}` subcommand; {UPGRADE_HINT}"),
};
}
for flag in args.iter().filter(|arg| arg.starts_with("--")) {
if crate::core::analyzers::external_tool::is_unknown_flag(stderr, flag) {
return Capability::Unavailable {
reason: PythonUnavailableReason::NotInstalled,
detail: format!(
"`uv {subcommand}` on this uv does not accept `{flag}`; {UPGRADE_HINT}"
),
};
}
}
let detail = if succeeded {
format!(
"could not establish whether `uv {subcommand}` supports JSON output: uv accepted the \
probe value `{CAPABILITY_PROBE_VALUE}` instead of rejecting it"
)
} else {
format!(
"could not establish whether `uv {subcommand}` supports JSON output: {}",
first_line(stderr)
)
};
Capability::Unavailable {
reason: PythonUnavailableReason::Failed,
detail,
}
}
fn possible_values(stderr: &str, format_flag: &str) -> Option<Vec<String>> {
let collapsed = stderr.split_whitespace().collect::<Vec<_>>().join(" ");
let rejection = collapsed.find(&format!("invalid value '{CAPABILITY_PROBE_VALUE}'"))?;
let after_rejection = &collapsed[rejection..];
if !after_rejection.contains(format_flag) {
return None;
}
let start = after_rejection.find("[possible values:")? + "[possible values:".len();
let list = &after_rejection[start..];
let end = list.find(']')?;
Some(
list[..end]
.split(',')
.map(|value| value.trim().trim_matches(['\'', '"']).to_string())
.filter(|value| !value.is_empty())
.collect(),
)
}
fn is_unrecognized_subcommand(stderr: &str, subcommand: &str) -> bool {
const PATTERN: &str = "unrecognized subcommand";
stderr.lines().any(|line| {
let lower = line.to_lowercase();
lower
.find(PATTERN)
.is_some_and(|start| lower[start + PATTERN.len()..].contains(subcommand))
})
}
fn first_line(stderr: &str) -> String {
stderr
.lines()
.map(str::trim)
.find(|line| !line.is_empty())
.unwrap_or("uv produced no output")
.to_string()
}
#[derive(Debug, Default, Deserialize)]
pub struct UvTree {
#[serde(default)]
roots: Vec<UvRef>,
#[serde(default)]
resolution: BTreeMap<String, UvNode>,
}
#[derive(Debug, Deserialize)]
struct UvRef {
id: String,
}
#[derive(Debug, Deserialize)]
struct UvNamedRef {
name: String,
id: String,
}
#[derive(Debug, Deserialize)]
struct UvNode {
#[serde(default)]
name: Option<String>,
#[serde(default)]
version: Option<String>,
#[serde(default)]
latest_version: Option<String>,
#[serde(default)]
source: Option<Value>,
#[serde(default)]
kind: Option<Value>,
#[serde(default)]
dependencies: Vec<UvRef>,
#[serde(default)]
optional_dependencies: Vec<UvNamedRef>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum NodeKind {
Package,
Workspace,
Group(String),
Extra(String),
Unrecognized,
}
fn node_kind(kind: Option<&Value>) -> NodeKind {
match kind {
Some(Value::String(label)) if label == "package" => NodeKind::Package,
Some(Value::String(label)) if label == "workspace" => NodeKind::Workspace,
Some(Value::Object(map)) => {
if let Some(Value::String(name)) = map.get("group") {
NodeKind::Group(name.clone())
} else if let Some(Value::String(name)) = map.get("extra") {
NodeKind::Extra(name.clone())
} else {
NodeKind::Unrecognized
}
}
_ => NodeKind::Unrecognized,
}
}
fn is_registry(source: Option<&Value>) -> bool {
match source {
Some(Value::Object(map)) => map.contains_key("registry"),
Some(Value::String(text)) => text.starts_with("registry+"),
_ => false,
}
}
#[derive(Debug, Default)]
pub struct ScopeIndex {
scopes: HashMap<String, PythonDependencyScope>,
}
impl ScopeIndex {
pub fn get(&self, name: &str) -> PythonDependencyScope {
self.scopes
.get(&normalize_package_name(name))
.copied()
.unwrap_or(PythonDependencyScope::Unknown)
}
}
struct TreeGraph<'a> {
tree: &'a UvTree,
kinds: HashMap<&'a str, NodeKind>,
roots: HashSet<&'a str>,
alias_base: HashMap<&'a str, &'a str>,
extras: HashMap<&'a str, BTreeSet<String>>,
groups: HashMap<&'a str, BTreeSet<String>>,
direct: HashSet<&'a str>,
reachable: HashSet<&'a str>,
}
impl<'a> TreeGraph<'a> {
fn build(tree: &'a UvTree) -> Self {
let kinds: HashMap<&str, NodeKind> = tree
.resolution
.iter()
.map(|(id, node)| (id.as_str(), node_kind(node.kind.as_ref())))
.collect();
let roots: HashSet<&str> = tree.roots.iter().map(|root| root.id.as_str()).collect();
let mut alias_base = HashMap::new();
let mut extras: HashMap<&str, BTreeSet<String>> = HashMap::new();
for (id, node) in &tree.resolution {
for optional in &node.optional_dependencies {
if !tree.resolution.contains_key(&optional.id)
|| roots.contains(optional.id.as_str())
{
continue;
}
alias_base.insert(optional.id.as_str(), id.as_str());
extras
.entry(id.as_str())
.or_default()
.insert(optional.name.clone());
}
}
let mut graph = Self {
tree,
kinds,
roots,
alias_base,
extras,
groups: HashMap::new(),
direct: HashSet::new(),
reachable: HashSet::new(),
};
graph.walk_sections();
graph
}
fn resolve(&self, id: &'a str) -> &'a str {
let mut current = id;
for _ in 0..self.tree.resolution.len() {
match self.alias_base.get(current) {
Some(base) => current = base,
None => break,
}
}
current
}
fn effective_dependencies(&self, id: &'a str) -> Vec<&'a str> {
let mut targets = Vec::new();
let mut push = |node: &'a UvNode| {
for dependency in &node.dependencies {
let resolved = self.resolve(dependency.id.as_str());
if resolved != id {
targets.push(resolved);
}
}
};
if let Some(node) = self.tree.resolution.get(id) {
push(node);
}
for (alias, base) in &self.alias_base {
if *base == id {
if let Some(node) = self.tree.resolution.get(*alias) {
push(node);
}
}
}
targets
}
fn walk_sections(&mut self) {
for root in self.tree.roots.iter().map(|root| root.id.as_str()) {
let Some(label) = self.section_label(root) else {
continue;
};
let mut queue: VecDeque<&str> = VecDeque::new();
let mut seen: HashSet<&str> = HashSet::new();
for target in self.effective_dependencies(root) {
if self.roots.contains(target) {
continue;
}
self.direct.insert(target);
if seen.insert(target) {
queue.push_back(target);
}
}
while let Some(current) = queue.pop_front() {
self.reachable.insert(current);
self.groups
.entry(current)
.or_default()
.insert(label.clone());
for target in self.effective_dependencies(current) {
if self.roots.contains(target)
|| self.kinds.get(target) == Some(&NodeKind::Workspace)
{
continue;
}
if seen.insert(target) {
queue.push_back(target);
}
}
}
}
}
fn section_label(&self, root: &str) -> Option<String> {
match self.kinds.get(root)? {
NodeKind::Package => Some("main".to_string()),
NodeKind::Group(name) => Some(name.clone()),
NodeKind::Extra(name) => Some(name.clone()),
NodeKind::Workspace | NodeKind::Unrecognized => None,
}
}
fn scope(&self, id: &str) -> PythonDependencyScope {
if self.direct.contains(id) {
PythonDependencyScope::Direct
} else if self.reachable.contains(id) {
PythonDependencyScope::Transitive
} else {
PythonDependencyScope::Unknown
}
}
}
pub fn normalize_tree_with_scopes(tree: &UvTree) -> (PythonOutdatedReport, ScopeIndex) {
let graph = TreeGraph::build(tree);
let mut checked = 0usize;
let mut counts = PythonUpdateCounts {
epoch: 0,
major: 0,
minor: 0,
patch: 0,
qualifier: 0,
unclassified: 0,
};
let mut packages = Vec::new();
let mut scopes = HashMap::new();
let sections_known = !tree.roots.is_empty();
for (id, node) in &tree.resolution {
if graph.kinds.get(id.as_str()) != Some(&NodeKind::Package)
|| !is_registry(node.source.as_ref())
{
continue;
}
let (Some(name), Some(current)) = (node.name.as_deref(), node.version.as_deref()) else {
continue;
};
checked += 1;
let normalized_name = normalize_package_name(name);
scopes.insert(normalized_name.clone(), graph.scope(id));
let Some(latest) = node.latest_version.as_deref() else {
continue;
};
let update_type = pep440::classify(current, latest);
if pep440::is_same_version(current, latest) {
continue;
}
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: normalized_name,
current: current.to_string(),
latest: latest.to_string(),
update_type,
scope: graph.scope(id),
groups: sections_known.then(|| {
graph
.groups
.get(id.as_str())
.map(|labels| labels.iter().cloned().collect())
.unwrap_or_default()
}),
extras: Some(
graph
.extras
.get(id.as_str())
.map(|names| names.iter().cloned().collect())
.unwrap_or_default(),
),
marker: PythonMarker::NotReported,
});
}
packages.sort_by(|left, right| left.name.cmp(&right.name));
(
PythonOutdatedReport {
checked,
outdated: packages.len(),
counts,
packages,
},
ScopeIndex { scopes },
)
}
#[derive(Debug, Default, Deserialize)]
pub struct UvAudit {
#[serde(default)]
vulnerabilities: Vec<UvVulnerability>,
#[serde(default)]
adverse_statuses: Vec<Value>,
}
#[derive(Debug, Deserialize)]
struct UvVulnerability {
id: String,
#[serde(default)]
aliases: Option<Vec<String>>,
#[serde(default)]
summary: Option<String>,
#[serde(default)]
fix_versions: Option<Vec<String>>,
dependency: UvAuditDependency,
}
#[derive(Debug, Deserialize)]
struct UvAuditDependency {
name: String,
version: String,
}
pub fn normalize_audit(
audit: &UvAudit,
scopes: &ScopeIndex,
) -> (PythonSecurityReport, Vec<String>) {
let findings: Vec<PythonVulnerability> = audit
.vulnerabilities
.iter()
.map(|vulnerability| PythonVulnerability {
id: vulnerability.id.clone(),
aliases: vulnerability.aliases.clone(),
package: normalize_package_name(&vulnerability.dependency.name),
installed_version: vulnerability.dependency.version.clone(),
severity: PythonSeverity::Unknown,
title: vulnerability
.summary
.as_deref()
.map(str::trim)
.filter(|summary| !summary.is_empty())
.map(str::to_string),
scope: scopes.get(&vulnerability.dependency.name),
fixed_versions: vulnerability.fix_versions.clone(),
})
.collect();
let mut warnings = Vec::new();
if !findings.is_empty() {
warnings.push(
"uv audit publishes no severity, so every finding is reported as `unknown`; an \
`unknown` severity satisfies every --fail-on-vulnerability threshold"
.to_string(),
);
}
if !audit.adverse_statuses.is_empty() {
warnings.push(format!(
"uv audit reported {} package(s) with an adverse status (such as a yanked release); \
the schema carries vulnerabilities only, so run `uv audit` directly for those",
audit.adverse_statuses.len()
));
}
let summary = PythonSecuritySummary {
critical: 0,
high: 0,
moderate: 0,
low: 0,
unknown: findings.len(),
total: findings.len(),
};
(PythonSecurityReport { summary, findings }, warnings)
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::Path;
fn fixture(name: &str) -> String {
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("tests")
.join("fixtures")
.join("uv")
.join(name);
std::fs::read_to_string(&path)
.unwrap_or_else(|err| panic!("missing uv fixture {}: {err}", path.display()))
}
fn tree_fixture() -> UvTree {
serde_json::from_str(&fixture("tree-outdated.json")).expect("parse tree fixture")
}
fn audit_fixture() -> UvAudit {
serde_json::from_str(&fixture("audit.json")).expect("parse audit fixture")
}
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 checked_counts_packages_not_graph_nodes() {
let (report, _) = normalize_tree_with_scopes(&tree_fixture());
assert_eq!(
report.checked, 12,
"expected the twelve registry packages, not the seventeen resolution nodes"
);
assert_eq!(
report
.packages
.iter()
.filter(|package| package.name == "requests")
.count(),
1,
"`requests` and `requests[socks]` are one package"
);
assert!(
!report
.packages
.iter()
.any(|package| package.name == "demo-app"),
"the editable workspace member has no registry version to be behind"
);
}
#[test]
fn summaries_agree_with_their_entries() {
let (report, _) = normalize_tree_with_scopes(&tree_fixture());
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 packages_without_a_latest_version_are_not_outdated() {
let (report, _) = normalize_tree_with_scopes(&tree_fixture());
for current in ["certifi", "pyparsing", "pysocks"] {
assert!(
!report
.packages
.iter()
.any(|package| package.name == current),
"{current} carries no latest_version in the fixture and is up to date"
);
}
assert_eq!(report.outdated, 9);
}
#[test]
fn scope_follows_the_dependency_graph() {
let (report, scopes) = normalize_tree_with_scopes(&tree_fixture());
for direct in ["requests", "jinja2", "pyyaml", "click", "packaging"] {
assert_eq!(
package(&report, direct).scope,
PythonDependencyScope::Direct,
"{direct} is declared by the project"
);
}
for transitive in ["urllib3", "idna", "chardet", "markupsafe"] {
assert_eq!(
package(&report, transitive).scope,
PythonDependencyScope::Transitive,
"{transitive} is only reached through another package"
);
}
assert_eq!(scopes.get("pysocks"), PythonDependencyScope::Transitive);
}
#[test]
fn no_roots_reports_sections_as_unknown_rather_than_empty() {
let mut tree = tree_fixture();
tree.roots.clear();
let (report, _) = normalize_tree_with_scopes(&tree);
assert!(
!report.packages.is_empty(),
"fixture premise: packages are still resolved without roots"
);
for package in &report.packages {
assert_eq!(
package.groups, None,
"{}: sections are unknown, not empty",
package.name
);
}
}
#[test]
fn groups_name_the_sections_that_reach_a_package() {
let (report, _) = normalize_tree_with_scopes(&tree_fixture());
assert_eq!(
package(&report, "jinja2").groups.as_deref(),
Some(&["main".to_string()][..])
);
assert_eq!(
package(&report, "click").groups.as_deref(),
Some(&["dev".to_string()][..])
);
assert_eq!(
package(&report, "packaging").groups.as_deref(),
Some(&["extra-feature".to_string()][..])
);
assert_eq!(
package(&report, "markupsafe").groups.as_deref(),
Some(&["main".to_string()][..]),
"markupsafe is reached through jinja2, not through the docs group"
);
}
#[test]
fn extras_are_sourced_from_the_alias_nodes() {
let (report, _) = normalize_tree_with_scopes(&tree_fixture());
assert_eq!(
package(&report, "requests").extras.as_deref(),
Some(&["socks".to_string()][..])
);
assert_eq!(
package(&report, "jinja2").extras.as_deref(),
Some(&[][..]),
"reported-and-empty, not null: uv does report extras"
);
}
#[test]
fn markers_are_reported_as_unavailable_rather_than_absent() {
let (report, _) = normalize_tree_with_scopes(&tree_fixture());
assert!(report
.packages
.iter()
.all(|package| package.marker == PythonMarker::NotReported));
}
#[test]
fn update_types_classify_the_captured_versions() {
let (report, _) = normalize_tree_with_scopes(&tree_fixture());
for major in ["jinja2", "urllib3", "click"] {
assert_eq!(package(&report, major).update_type, PythonUpdateType::Major);
}
assert_eq!(report.counts.major, 8);
assert_eq!(report.counts.unclassified, 0);
}
#[test]
fn audit_findings_carry_no_severity() {
let (_, scopes) = normalize_tree_with_scopes(&tree_fixture());
let (report, warnings) = normalize_audit(&audit_fixture(), &scopes);
assert_eq!(report.summary.total, 6);
assert_eq!(report.summary.unknown, 6);
assert_eq!(
report.summary.critical
+ report.summary.high
+ report.summary.moderate
+ report.summary.low,
0,
"uv publishes no severity, so no finding may land in a graded bucket"
);
assert!(report
.findings
.iter()
.all(|finding| finding.severity == PythonSeverity::Unknown));
assert!(
warnings
.iter()
.any(|warning| warning.contains("no severity")),
"a payload of nothing but `unknown` needs the reason on it: {warnings:?}"
);
}
#[test]
fn audit_scope_comes_from_the_tree_when_there_is_one() {
let (_, scopes) = normalize_tree_with_scopes(&tree_fixture());
let audit = audit_fixture();
let (joined, _) = normalize_audit(&audit, &scopes);
let scope_of = |package: &str| {
joined
.findings
.iter()
.find(|finding| finding.package == package)
.unwrap_or_else(|| panic!("no finding for {package}"))
.scope
};
assert_eq!(scope_of("jinja2"), PythonDependencyScope::Direct);
assert_eq!(scope_of("urllib3"), PythonDependencyScope::Transitive);
assert_eq!(scope_of("click"), PythonDependencyScope::Direct);
let (unjoined, _) = normalize_audit(&audit, &ScopeIndex::default());
assert!(unjoined
.findings
.iter()
.all(|finding| finding.scope == PythonDependencyScope::Unknown));
}
#[test]
fn audit_set_valued_fields_pass_through_unreported_as_null() {
let (_, scopes) = normalize_tree_with_scopes(&tree_fixture());
let (report, _) = normalize_audit(&audit_fixture(), &scopes);
let finding = report
.findings
.iter()
.find(|finding| finding.id == "PYSEC-2026-2132")
.expect("the click advisory");
assert_eq!(
finding.title, None,
"uv reports a null summary for this one"
);
assert_eq!(finding.aliases.as_ref().map(Vec::len), Some(2));
assert_eq!(
finding.fixed_versions.as_deref(),
Some(&["8.3.3".to_string()][..])
);
let empty: UvAudit = serde_json::from_str(
r#"{"vulnerabilities":[{"id":"X","dependency":{"name":"pkg","version":"1.0"}}]}"#,
)
.expect("parse minimal audit");
let (report, _) = normalize_audit(&empty, &ScopeIndex::default());
assert_eq!(report.findings[0].aliases, None);
assert_eq!(report.findings[0].fixed_versions, None);
}
#[test]
fn a_clean_audit_reports_zero_findings_and_no_severity_warning() {
let clean: UvAudit = serde_json::from_str(
r#"{"schema":{"version":"preview"},"summary":{"audited_packages":3,"vulnerabilities":0,"adverse_statuses":0},"vulnerabilities":[],"adverse_statuses":[]}"#,
)
.expect("parse clean audit");
let (report, warnings) = normalize_audit(&clean, &ScopeIndex::default());
assert_eq!(report.summary.total, 0);
assert!(report.findings.is_empty());
assert!(
warnings.is_empty(),
"a run with no findings has no severity caveat to give: {warnings:?}"
);
}
#[test]
fn adverse_statuses_are_surfaced_as_a_warning() {
let audit: UvAudit = serde_json::from_str(
r#"{"vulnerabilities":[],"adverse_statuses":[{"kind":"yanked"},{"kind":"yanked"}]}"#,
)
.expect("parse audit with adverse statuses");
let (_, warnings) = normalize_audit(&audit, &ScopeIndex::default());
assert_eq!(warnings.len(), 1);
assert!(
warnings[0].contains('2'),
"unexpected warning: {warnings:?}"
);
}
const CURRENT_TREE_PROBE: &str = "error: invalid value 'cargo-upkeep-capability-probe' for '--format <FORMAT>'\n [possible values: text, json]\n\nFor more information, try '--help'.\n";
const CURRENT_AUDIT_PROBE: &str = "error: invalid value 'cargo-upkeep-capability-probe' for '--output-format <OUTPUT_FORMAT>'\n [possible values: text, json, sarif]\n\nFor more information, try '--help'.\n";
const LEGACY_AUDIT_PROBE: &str =
"error: unrecognized subcommand 'audit'\n\nUsage: uv [OPTIONS] <COMMAND>\n\nFor more information, try '--help'.\n";
const LEGACY_TREE_PROBE: &str =
"error: unexpected argument '--format' found\n\nUsage: uv tree [OPTIONS]\n\nFor more information, try '--help'.\n";
fn outdated_probe(stderr: &str, succeeded: bool) -> Capability {
probe_capability(stderr, succeeded, &TREE_ARGS, "tree", "--format")
}
fn security_probe(stderr: &str, succeeded: bool) -> Capability {
probe_capability(stderr, succeeded, &AUDIT_ARGS, "audit", "--output-format")
}
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_uv_advertises_json_for_both_capabilities() {
assert!(matches!(
outdated_probe(CURRENT_TREE_PROBE, false),
Capability::Available
));
assert!(matches!(
security_probe(CURRENT_AUDIT_PROBE, false),
Capability::Available
));
}
#[test]
fn legacy_uv_reports_missing_capabilities_with_an_upgrade_hint() {
let (reason, detail) = unavailable(security_probe(LEGACY_AUDIT_PROBE, false));
assert_eq!(reason, PythonUnavailableReason::NotInstalled);
assert!(detail.contains("no `audit` subcommand"), "{detail}");
assert!(detail.contains("uv self update"), "{detail}");
let (reason, detail) = unavailable(outdated_probe(LEGACY_TREE_PROBE, false));
assert_eq!(reason, PythonUnavailableReason::NotInstalled);
assert!(detail.contains("--format"), "{detail}");
assert!(detail.contains("uv self update"), "{detail}");
}
#[test]
fn a_format_flag_without_json_is_a_capability_gap() {
let stderr = "error: invalid value 'cargo-upkeep-capability-probe' for '--output-format <OUTPUT_FORMAT>'\n [possible values: text]\n";
let (reason, detail) = unavailable(security_probe(stderr, false));
assert_eq!(reason, PythonUnavailableReason::NotInstalled);
assert!(detail.contains("accepts only text"), "{detail}");
}
#[test]
fn a_wrapped_possible_values_list_is_still_read() {
let stderr = "error: invalid value\n 'cargo-upkeep-capability-probe' for\n '--output-format <OUTPUT_FORMAT>'\n [possible values: text,\n json, sarif]\n";
assert!(matches!(
security_probe(stderr, false),
Capability::Available
));
}
#[test]
fn a_possible_values_list_for_another_flag_is_not_an_answer() {
let stderr = "error: invalid value 'x' for '--service-format <SERVICE_FORMAT>'\n [possible values: osv, json]\n";
let (reason, _) = unavailable(security_probe(stderr, false));
assert_eq!(
reason,
PythonUnavailableReason::Failed,
"a list about a different flag establishes nothing"
);
}
#[test]
fn an_accepted_probe_value_is_inconclusive() {
let (reason, detail) = unavailable(security_probe("", true));
assert_eq!(reason, PythonUnavailableReason::Failed);
assert!(detail.contains("accepted the probe value"), "{detail}");
}
#[test]
fn version_is_parsed_from_uv_version_output() {
assert_eq!(
parse_version("uv 0.12.8 (68209e5c6 2026-08-31 aarch64-apple-darwin)\n").as_deref(),
Some("0.12.8")
);
assert_eq!(
parse_version("uv 0.7.11 (90a4416ab 2025-06-04)\n").as_deref(),
Some("0.7.11")
);
assert_eq!(parse_version("something else\n"), None);
assert_eq!(parse_version(""), None);
}
#[test]
fn an_unrecognized_node_kind_is_not_a_parse_failure() {
let tree: UvTree = serde_json::from_str(
r#"{"roots":[],"resolution":{
"a==1.0@registry+https://pypi.org/simple":{"name":"a","version":"1.0","kind":{"tomorrow":"x"},"source":{"registry":{"url":"https://pypi.org/simple"}}},
"b==1.0@registry+https://pypi.org/simple":{"name":"b","version":"1.0","latest_version":"2.0","kind":"package","source":{"registry":{"url":"https://pypi.org/simple"}}}
}}"#,
)
.expect("an unknown kind must still deserialize");
let (report, _) = normalize_tree_with_scopes(&tree);
assert_eq!(report.checked, 1, "only the recognized package is counted");
assert_eq!(report.outdated, 1);
assert_eq!(report.packages[0].name, "b");
}
#[test]
fn package_names_are_normalized_on_both_sides() {
let tree: UvTree = serde_json::from_str(
r#"{"roots":[],"resolution":{
"zope.interface==5.0@registry+https://pypi.org/simple":{"name":"Zope.Interface","version":"5.0","latest_version":"6.0","kind":"package","source":{"registry":{"url":"https://pypi.org/simple"}}}
}}"#,
)
.expect("parse tree");
let (report, scopes) = normalize_tree_with_scopes(&tree);
assert_eq!(report.packages[0].name, "zope-interface");
let audit: UvAudit = serde_json::from_str(
r#"{"vulnerabilities":[{"id":"X","dependency":{"name":"Zope_Interface","version":"5.0"}}]}"#,
)
.expect("parse audit");
let (security, _) = normalize_audit(&audit, &scopes);
assert_eq!(security.findings[0].package, "zope-interface");
}
}