Files
windmill/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs

194 lines
6.5 KiB
Rust

//! Parser-parity guard: `parse_pipeline_annotations` (Rust, drives deploy)
//! and `parsePipelineAnnotations` (TS, drives the live graph preview —
//! frontend/src/lib/components/assets/AssetGraph/parsePipelineAnnotations.ts)
//! must stay behaviorally identical, or the graph the user previews is not
//! the graph that deploys. Both implementations run the SAME fixture corpus:
//!
//! tests/fixtures/pipeline_annotations.json
//!
//! The frontend counterpart is parsePipelineAnnotations.parity.test.ts.
//! When the annotation grammar changes, extend the corpus — a fixture that
//! passes on one side and fails on the other is exactly the drift this
//! exists to catch. Only the fields both parsers produce are compared
//! (join_mode / debounce_default are deploy-only, parsed solely in Rust).
use serde::Deserialize;
use windmill_parser::asset_parser::{
parse_pipeline_annotations, AssetKind, PartitionKind, TriggerSpec,
};
#[derive(Deserialize)]
struct Fixture {
name: String,
code: String,
expected: Expected,
}
#[derive(Deserialize)]
struct Expected {
in_pipeline: bool,
/// `kind:path`, in declaration order, deduped.
asset_triggers: Vec<String>,
native_triggers: Vec<String>,
partition: Option<ExpectedPartition>,
freshness: Option<String>,
tag: Option<String>,
retry: Option<ExpectedRetry>,
// Default-on-absent so the pre-existing fixtures (which omit it) keep
// deserializing; only fixtures exercising materialization set it.
#[serde(default)]
materialize: Option<ExpectedMaterialize>,
}
#[derive(Deserialize)]
struct ExpectedMaterialize {
target_kind: String,
target_path: String,
#[serde(default)]
manual: bool,
#[serde(default)]
append: bool,
#[serde(default)]
unique_key: Option<String>,
}
#[derive(Deserialize)]
struct ExpectedPartition {
kind: String,
#[serde(default)]
key: Option<String>,
tz: Option<String>,
format: Option<String>,
start: Option<String>,
}
#[derive(Deserialize)]
struct ExpectedRetry {
count: u32,
delay: Option<String>,
}
fn kind_str(k: AssetKind) -> &'static str {
match k {
AssetKind::S3Object => "s3object",
AssetKind::Resource => "resource",
AssetKind::Ducklake => "ducklake",
AssetKind::DataTable => "datatable",
AssetKind::Volume => "volume",
}
}
fn native_str(t: &TriggerSpec) -> Option<&'static str> {
Some(match t {
TriggerSpec::Asset { .. } => return None,
TriggerSpec::Schedule => "schedule",
TriggerSpec::Webhook => "webhook",
TriggerSpec::Email => "email",
TriggerSpec::Kafka => "kafka",
TriggerSpec::Mqtt => "mqtt",
TriggerSpec::Nats => "nats",
TriggerSpec::Postgres => "postgres",
TriggerSpec::Sqs => "sqs",
TriggerSpec::Gcp => "gcp",
TriggerSpec::DataUpload => "data_upload",
})
}
#[test]
fn pipeline_annotation_fixtures_match() {
let fixtures: Vec<Fixture> =
serde_json::from_str(include_str!("fixtures/pipeline_annotations.json"))
.expect("fixture corpus must deserialize");
assert!(!fixtures.is_empty());
for f in fixtures {
let got = parse_pipeline_annotations(&f.code);
let ctx = format!("fixture '{}'", f.name);
assert_eq!(
got.in_pipeline, f.expected.in_pipeline,
"{ctx}: in_pipeline"
);
let asset_triggers: Vec<String> = got
.triggers
.iter()
.filter_map(|t| match t {
TriggerSpec::Asset { asset_kind, path, .. } => {
Some(format!("{}:{}", kind_str(*asset_kind), path))
}
_ => None,
})
.collect();
assert_eq!(
asset_triggers, f.expected.asset_triggers,
"{ctx}: asset triggers"
);
let native: Vec<&str> = got.triggers.iter().filter_map(native_str).collect();
assert_eq!(native, f.expected.native_triggers, "{ctx}: native triggers");
match (&got.partition, &f.expected.partition) {
(None, None) => {}
(Some(p), Some(e)) => {
let (kind, key) = match &p.kind {
PartitionKind::Daily => ("daily", None),
PartitionKind::Hourly => ("hourly", None),
PartitionKind::Weekly => ("weekly", None),
PartitionKind::Monthly => ("monthly", None),
PartitionKind::Dynamic { key } => ("dynamic", Some(key.clone())),
};
assert_eq!(kind, e.kind, "{ctx}: partition kind");
assert_eq!(key, e.key, "{ctx}: partition key");
assert_eq!(p.tz, e.tz, "{ctx}: partition tz");
assert_eq!(p.format, e.format, "{ctx}: partition format");
assert_eq!(p.start, e.start, "{ctx}: partition start");
}
(got, want) => panic!(
"{ctx}: partition mismatch — got {:?}, want present={}",
got,
want.is_some()
),
}
assert_eq!(
got.freshness.as_ref().map(|fr| fr.duration.clone()),
f.expected.freshness,
"{ctx}: freshness"
);
assert_eq!(got.tag, f.expected.tag, "{ctx}: tag");
match (&got.retry, &f.expected.retry) {
(None, None) => {}
(Some(r), Some(e)) => {
assert_eq!(r.count, e.count, "{ctx}: retry count");
assert_eq!(r.delay, e.delay, "{ctx}: retry delay");
}
(got, want) => panic!(
"{ctx}: retry mismatch — got {:?}, want present={}",
got,
want.is_some()
),
}
match (&got.materialize, &f.expected.materialize) {
(None, None) => {}
(Some(m), Some(e)) => {
assert_eq!(
kind_str(m.target_kind),
e.target_kind,
"{ctx}: materialize kind"
);
assert_eq!(m.target_path, e.target_path, "{ctx}: materialize path");
assert_eq!(m.manual, e.manual, "{ctx}: materialize manual");
assert_eq!(m.append, e.append, "{ctx}: materialize append");
assert_eq!(m.unique_key, e.unique_key, "{ctx}: materialize key");
}
(got, want) => panic!(
"{ctx}: materialize mismatch — got {:?}, want present={}",
got,
want.is_some()
),
}
}
}