use serde::Serialize;
use std::fmt;
pub fn normalize_package_name(name: &str) -> String {
let mut normalized = String::with_capacity(name.len());
let mut previous_was_separator = false;
for character in name.chars() {
if matches!(character, '-' | '_' | '.') {
if !previous_was_separator {
normalized.push('-');
}
previous_was_separator = true;
} else {
normalized.extend(character.to_lowercase());
previous_was_separator = false;
}
}
normalized
}
pub const PYTHON_SCHEMA_VERSION: u32 = 1;
#[derive(Debug, Serialize)]
pub struct PythonOutput {
pub schema_version: u32,
pub manager: PythonManager,
pub complete: bool,
pub capabilities: Vec<PythonCapabilityCoverage>,
pub unavailable: Vec<PythonUnavailableCapability>,
pub outdated: Option<PythonOutdatedReport>,
pub security: Option<PythonSecurityReport>,
pub warnings: Vec<String>,
}
#[derive(Debug, Serialize)]
pub struct PythonManager {
pub name: PythonManagerName,
pub version: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PythonManagerName {
Uv,
#[allow(dead_code)]
Poetry,
#[allow(dead_code)]
PipTools,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PythonCapability {
Outdated,
Security,
}
#[derive(Debug, Serialize)]
pub struct PythonCapabilityCoverage {
pub name: PythonCapability,
pub measured: bool,
}
#[derive(Debug, Serialize)]
pub struct PythonUnavailableCapability {
pub name: PythonCapability,
pub reason: PythonUnavailableReason,
pub detail: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PythonUnavailableReason {
NotInstalled,
Failed,
#[allow(dead_code)]
Unsupported,
}
#[derive(Debug, Serialize)]
pub struct PythonOutdatedReport {
pub checked: usize,
pub outdated: usize,
pub counts: PythonUpdateCounts,
pub packages: Vec<PythonOutdatedPackage>,
}
#[derive(Debug, Serialize)]
pub struct PythonUpdateCounts {
pub epoch: usize,
pub major: usize,
pub minor: usize,
pub patch: usize,
pub qualifier: usize,
pub unclassified: usize,
}
#[derive(Debug, Serialize)]
pub struct PythonOutdatedPackage {
pub name: String,
pub current: String,
pub latest: String,
pub update_type: PythonUpdateType,
pub scope: PythonDependencyScope,
pub groups: Option<Vec<String>>,
pub extras: Option<Vec<String>>,
pub marker: PythonMarker,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PythonUpdateType {
Epoch,
Major,
Minor,
Patch,
Qualifier,
Unclassified,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PythonDependencyScope {
Direct,
Transitive,
Unknown,
}
#[derive(Debug, Serialize)]
pub struct PythonSecurityReport {
pub summary: PythonSecuritySummary,
pub findings: Vec<PythonVulnerability>,
}
#[derive(Debug, Serialize)]
pub struct PythonSecuritySummary {
pub critical: usize,
pub high: usize,
pub moderate: usize,
pub low: usize,
pub unknown: usize,
pub total: usize,
}
#[derive(Debug, Serialize)]
pub struct PythonVulnerability {
pub id: String,
pub aliases: Option<Vec<String>>,
pub package: String,
pub installed_version: String,
pub severity: PythonSeverity,
pub title: Option<String>,
pub scope: PythonDependencyScope,
pub fixed_versions: Option<Vec<String>>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PythonSeverity {
Critical,
#[allow(dead_code)]
High,
#[allow(dead_code)]
Moderate,
#[allow(dead_code)]
Low,
Unknown,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "status", content = "expression", rename_all = "snake_case")]
pub enum PythonMarker {
NotReported,
#[allow(dead_code)]
Absent,
#[allow(dead_code)]
Reported(String),
}
impl fmt::Display for PythonOutput {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
writeln!(
f,
"Manager: {} {}",
self.manager.name,
self.manager
.version
.as_deref()
.unwrap_or("(unknown version)")
)?;
writeln!(
f,
"Coverage: {}",
if self.complete {
"complete".to_string()
} else {
format!(
"incomplete ({} of {} capabilities measured)",
self.capabilities
.iter()
.filter(|capability| capability.measured)
.count(),
self.capabilities.len()
)
}
)?;
for gap in &self.unavailable {
writeln!(
f,
" {} not measured ({}): {}",
gap.name, gap.reason, gap.detail
)?;
}
match &self.outdated {
Some(report) => {
writeln!(
f,
"\nOutdated: {} of {} checked",
report.outdated, report.checked
)?;
if report.outdated > 0 {
let counts = &report.counts;
writeln!(
f,
" epoch {} | major {} | minor {} | patch {} | qualifier {} | unclassified {}",
counts.epoch,
counts.major,
counts.minor,
counts.patch,
counts.qualifier,
counts.unclassified
)?;
}
for package in &report.packages {
writeln!(
f,
" {} {} -> {} ({}, {})",
package.name,
package.current,
package.latest,
package.update_type,
package.scope
)?;
}
}
None => writeln!(f, "\nOutdated: not measured")?,
}
match &self.security {
Some(report) => {
let summary = &report.summary;
writeln!(f, "\nVulnerabilities: {}", summary.total)?;
if summary.total > 0 {
writeln!(
f,
" critical {} | high {} | moderate {} | low {} | unknown {}",
summary.critical,
summary.high,
summary.moderate,
summary.low,
summary.unknown
)?;
}
for finding in &report.findings {
writeln!(
f,
" [{}] {} {} — {}",
finding.severity,
finding.package,
finding.installed_version,
finding.title.as_deref().unwrap_or(finding.id.as_str())
)?;
}
}
None => writeln!(f, "\nVulnerabilities: not measured")?,
}
for warning in &self.warnings {
writeln!(f, "\nwarning: {warning}")?;
}
Ok(())
}
}
impl fmt::Display for PythonManagerName {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
PythonManagerName::Uv => "uv",
PythonManagerName::Poetry => "poetry",
PythonManagerName::PipTools => "pip-tools",
})
}
}
impl fmt::Display for PythonCapability {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
PythonCapability::Outdated => "outdated",
PythonCapability::Security => "security",
})
}
}
impl fmt::Display for PythonUnavailableReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
PythonUnavailableReason::NotInstalled => "not_installed",
PythonUnavailableReason::Failed => "failed",
PythonUnavailableReason::Unsupported => "unsupported",
})
}
}
impl fmt::Display for PythonUpdateType {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
PythonUpdateType::Epoch => "epoch",
PythonUpdateType::Major => "major",
PythonUpdateType::Minor => "minor",
PythonUpdateType::Patch => "patch",
PythonUpdateType::Qualifier => "qualifier",
PythonUpdateType::Unclassified => "unclassified",
})
}
}
impl fmt::Display for PythonDependencyScope {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
PythonDependencyScope::Direct => "direct",
PythonDependencyScope::Transitive => "transitive",
PythonDependencyScope::Unknown => "unknown",
})
}
}
impl fmt::Display for PythonSeverity {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
PythonSeverity::Critical => "critical",
PythonSeverity::High => "high",
PythonSeverity::Moderate => "moderate",
PythonSeverity::Low => "low",
PythonSeverity::Unknown => "unknown",
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::doc_examples::DocumentedExamples;
use serde_json::Value;
pub(crate) fn measured_output() -> PythonOutput {
PythonOutput {
schema_version: PYTHON_SCHEMA_VERSION,
manager: PythonManager {
name: PythonManagerName::Uv,
version: Some("0.0.0".to_string()),
},
complete: true,
capabilities: vec![
PythonCapabilityCoverage {
name: PythonCapability::Outdated,
measured: true,
},
PythonCapabilityCoverage {
name: PythonCapability::Security,
measured: true,
},
],
unavailable: Vec::new(),
outdated: Some(PythonOutdatedReport {
checked: 12,
outdated: 2,
counts: PythonUpdateCounts {
epoch: 0,
major: 1,
minor: 0,
patch: 0,
qualifier: 1,
unclassified: 0,
},
packages: vec![
PythonOutdatedPackage {
name: "example-http".to_string(),
current: "1.4.2".to_string(),
latest: "2.0.0".to_string(),
update_type: PythonUpdateType::Major,
scope: PythonDependencyScope::Direct,
groups: Some(vec!["main".to_string()]),
extras: Some(vec!["socks".to_string()]),
marker: PythonMarker::Reported("python_version >= '3.10'".to_string()),
},
PythonOutdatedPackage {
name: "example-parser".to_string(),
current: "0.9.0".to_string(),
latest: "0.9.0.post1".to_string(),
update_type: PythonUpdateType::Qualifier,
scope: PythonDependencyScope::Transitive,
groups: Some(Vec::new()),
extras: None,
marker: PythonMarker::Absent,
},
],
}),
security: Some(PythonSecurityReport {
summary: PythonSecuritySummary {
critical: 0,
high: 1,
moderate: 0,
low: 0,
unknown: 1,
total: 2,
},
findings: vec![
PythonVulnerability {
id: "GHSA-0000-0000-0000".to_string(),
aliases: Some(vec!["CVE-0000-00000".to_string()]),
package: "example-http".to_string(),
installed_version: "1.4.2".to_string(),
severity: PythonSeverity::High,
title: Some("Example advisory".to_string()),
scope: PythonDependencyScope::Direct,
fixed_versions: Some(vec!["1.4.3".to_string(), "2.0.0".to_string()]),
},
PythonVulnerability {
id: "PYSEC-0000-0000".to_string(),
aliases: None,
package: "example-parser".to_string(),
installed_version: "0.9.0".to_string(),
severity: PythonSeverity::Unknown,
title: None,
scope: PythonDependencyScope::Transitive,
fixed_versions: None,
},
],
}),
warnings: vec![
"uv documents its own JSON output as unstable; this report is normalized into \
cargo-upkeep schema_version 1"
.to_string(),
],
}
}
pub(crate) fn capability_gap_output() -> PythonOutput {
PythonOutput {
schema_version: PYTHON_SCHEMA_VERSION,
manager: PythonManager {
name: PythonManagerName::Poetry,
version: None,
},
complete: false,
capabilities: vec![
PythonCapabilityCoverage {
name: PythonCapability::Outdated,
measured: true,
},
PythonCapabilityCoverage {
name: PythonCapability::Security,
measured: false,
},
],
unavailable: vec![PythonUnavailableCapability {
name: PythonCapability::Security,
reason: PythonUnavailableReason::Unsupported,
detail: "the detected manager reports no vulnerability data; run a dedicated \
scanner and gate on that instead"
.to_string(),
}],
outdated: Some(PythonOutdatedReport {
checked: 4,
outdated: 0,
counts: PythonUpdateCounts {
epoch: 0,
major: 0,
minor: 0,
patch: 0,
qualifier: 0,
unclassified: 0,
},
packages: Vec::new(),
}),
security: None,
warnings: Vec::new(),
}
}
#[test]
fn documented_json_examples_match_output_contract() {
let documentation = DocumentedExamples::load("docs/python-schema.md");
for (example, output) in [
("python", measured_output()),
("python-capability-gap", capability_gap_output()),
] {
assert_eq!(
documentation.example(example),
crate::core::doc_examples::serialized_value(&output),
"{example}: docs/python-schema.md JSON example drifted from its serialized \
output contract"
);
}
}
#[test]
fn schema_version_is_emitted_verbatim() {
let value = crate::core::doc_examples::serialized_value(&measured_output());
assert_eq!(value["schema_version"], Value::from(PYTHON_SCHEMA_VERSION));
}
#[test]
fn unmeasured_capability_serializes_as_explicit_null() {
let value = crate::core::doc_examples::serialized_value(&capability_gap_output());
assert!(
value.as_object().expect("object").contains_key("security"),
"an unmeasured capability must stay present as null, not be omitted"
);
assert_eq!(value["security"], Value::Null);
assert_eq!(value["complete"], Value::Bool(false));
assert_eq!(value["unavailable"][0]["name"], Value::from("security"));
assert_eq!(
value["unavailable"][0]["reason"],
Value::from("unsupported")
);
assert_eq!(value["outdated"]["outdated"], Value::from(0));
assert_eq!(value["outdated"]["packages"], Value::Array(Vec::new()));
}
#[test]
fn set_valued_fields_distinguish_empty_from_unreported() {
let reported_empty: Option<Vec<String>> = Some(Vec::new());
let not_reported: Option<Vec<String>> = None;
assert_eq!(
serde_json::to_value(&reported_empty).unwrap(),
Value::Array(Vec::new())
);
assert_eq!(serde_json::to_value(¬_reported).unwrap(), Value::Null);
assert_ne!(
serde_json::to_value(&reported_empty).unwrap(),
serde_json::to_value(¬_reported).unwrap()
);
}
#[test]
fn unreported_attributes_stay_present_as_null() {
let value = crate::core::doc_examples::serialized_value(&measured_output());
let package = &value["outdated"]["packages"][1];
for field in ["groups", "extras"] {
assert!(
package
.as_object()
.expect("package object")
.contains_key(field),
"{field} must stay present even when unreported"
);
}
assert_eq!(package["extras"], Value::Null, "unreported reads as null");
assert_eq!(
package["groups"],
Value::Array(Vec::new()),
"reported-and-empty reads as [], not null"
);
let finding = &value["security"]["findings"][1];
assert!(
finding
.as_object()
.expect("finding object")
.contains_key("fixed_versions"),
"fixed_versions must stay present even when unreported"
);
assert_eq!(finding["fixed_versions"], Value::Null);
}
#[test]
fn python_marker_has_three_distinct_states() {
assert_eq!(
serde_json::to_value(PythonMarker::Reported("sys_platform == 'win32'".into())).unwrap(),
serde_json::json!({
"status": "reported",
"expression": "sys_platform == 'win32'"
})
);
assert_eq!(
serde_json::to_value(PythonMarker::Absent).unwrap(),
serde_json::json!({ "status": "absent" })
);
assert_eq!(
serde_json::to_value(PythonMarker::NotReported).unwrap(),
serde_json::json!({ "status": "not_reported" })
);
}
#[test]
fn summaries_agree_with_the_entries_they_summarize() {
let output = measured_output();
let outdated = output.outdated.as_ref().expect("outdated measured");
let counts = &outdated.counts;
let bucketed = counts.epoch
+ counts.major
+ counts.minor
+ counts.patch
+ counts.qualifier
+ counts.unclassified;
assert_eq!(
outdated.outdated,
outdated.packages.len(),
"`outdated` counts the entries in `packages`; it is never truncated"
);
assert_eq!(
bucketed, outdated.outdated,
"every entry lands in one bucket"
);
assert!(
outdated.checked >= outdated.outdated,
"an outdated package was necessarily checked"
);
let security = output.security.as_ref().expect("security measured");
let summary = &security.summary;
assert_eq!(
summary.total,
security.findings.len(),
"`total` counts findings, not deduplicated advisories"
);
assert_eq!(
summary.critical + summary.high + summary.moderate + summary.low + summary.unknown,
summary.total,
"every finding lands in exactly one severity bucket"
);
}
#[test]
fn enum_labels_are_pinned() {
for (value, label) in [
(
serde_json::to_value(PythonManagerName::PipTools),
"pip_tools",
),
(serde_json::to_value(PythonUpdateType::Epoch), "epoch"),
(serde_json::to_value(PythonUpdateType::Major), "major"),
(serde_json::to_value(PythonUpdateType::Minor), "minor"),
(serde_json::to_value(PythonUpdateType::Patch), "patch"),
(
serde_json::to_value(PythonUpdateType::Qualifier),
"qualifier",
),
(
serde_json::to_value(PythonUpdateType::Unclassified),
"unclassified",
),
(
serde_json::to_value(PythonDependencyScope::Unknown),
"unknown",
),
(serde_json::to_value(PythonSeverity::Critical), "critical"),
(serde_json::to_value(PythonSeverity::High), "high"),
(serde_json::to_value(PythonSeverity::Moderate), "moderate"),
(serde_json::to_value(PythonSeverity::Low), "low"),
(serde_json::to_value(PythonSeverity::Unknown), "unknown"),
(
serde_json::to_value(PythonUnavailableReason::NotInstalled),
"not_installed",
),
(
serde_json::to_value(PythonUnavailableReason::Unsupported),
"unsupported",
),
(serde_json::to_value(PythonCapability::Security), "security"),
] {
assert_eq!(value.unwrap(), Value::String(label.to_string()));
}
}
#[test]
fn unavailable_reason_labels_match_the_rust_side() {
use crate::core::output::UnavailableReason;
assert_eq!(
serde_json::to_value(PythonUnavailableReason::NotInstalled).unwrap(),
serde_json::to_value(UnavailableReason::NotInstalled).unwrap()
);
assert_eq!(
serde_json::to_value(PythonUnavailableReason::Failed).unwrap(),
serde_json::to_value(UnavailableReason::Failed).unwrap()
);
}
}