faucet_cli/commands/
schema.rs

1//! `faucet schema` — print the JSON Schema for a connector's config.
2
3use crate::cli::{SchemaArgs, SchemaTarget};
4use crate::error::CliResult;
5use crate::registry::{sink_schema, source_schema};
6use crate::transforms::transform_schema;
7
8/// Execute the `schema` subcommand.
9pub async fn run(args: SchemaArgs) -> CliResult<()> {
10    let schema = match args.target {
11        SchemaTarget::Source { name } => source_schema(&name)?,
12        SchemaTarget::Sink { name } => sink_schema(&name)?,
13        SchemaTarget::Transform { name } => transform_schema(&name)?,
14        SchemaTarget::Dlq => {
15            let dlq_schema = faucet_core::schema_for!(crate::config::DlqSpec);
16            serde_json::to_value(dlq_schema)
17                .unwrap_or_else(|_| serde_json::json!({"type": "object"}))
18        }
19        SchemaTarget::Replication => {
20            let s = faucet_core::schema_for!(crate::replication::spec::ReplicationSpec);
21            serde_json::to_value(s).unwrap_or_else(|_| serde_json::json!({"type": "object"}))
22        }
23        SchemaTarget::Execution => {
24            let s = faucet_core::schema_for!(crate::config::ExecutionSpec);
25            serde_json::to_value(s).unwrap_or_else(|_| serde_json::json!({"type": "object"}))
26        }
27        SchemaTarget::Resilience => {
28            let s = faucet_core::schema_for!(crate::config::ResilienceSpec);
29            serde_json::to_value(s).unwrap_or_else(|_| serde_json::json!({"type": "object"}))
30        }
31        #[cfg(feature = "quality")]
32        SchemaTarget::Quality => {
33            let quality_schema = faucet_core::schema_for!(faucet_core::QualitySpec);
34            serde_json::to_value(quality_schema)
35                .unwrap_or_else(|_| serde_json::json!({"type": "object"}))
36        }
37        #[cfg(feature = "schedule")]
38        SchemaTarget::Schedule => {
39            let s = faucet_core::schema_for!(crate::schedule::spec::ScheduleSpec);
40            serde_json::to_value(s).unwrap_or_else(|_| serde_json::json!({"type": "object"}))
41        }
42        #[cfg(feature = "lineage")]
43        SchemaTarget::Lineage => lineage_schema(),
44        #[cfg(feature = "triggers")]
45        SchemaTarget::Triggers => {
46            let s = faucet_core::schema_for!(crate::serve::triggers::spec::TriggersFile);
47            serde_json::to_value(s).unwrap_or_else(|_| serde_json::json!({"type": "object"}))
48        }
49        SchemaTarget::Secrets => serde_json::json!({
50            "title": "Secrets-manager interpolation grammar",
51            "schemes": {
52                "vault":    { "syntax": "${vault:<path>[#field]}", "auth": ["VAULT_ADDR", "VAULT_TOKEN", "VAULT_NAMESPACE (optional)"] },
53                "aws-sm":   { "syntax": "${aws-sm:<name-or-ARN>[#field]}", "auth": ["aws-config default credential chain"] },
54                "gcp-sm":   { "syntax": "${gcp-sm:projects/<p>/secrets/<s>/versions/<v>}", "auth": ["Application Default Credentials"] },
55                "azure-kv": { "syntax": "${azure-kv:<vault>/<secret>[/<version>]}", "auth": ["AZURE_* env / managed identity / az login"] }
56            },
57            "notes": [
58                "#field parses the secret as JSON and extracts one key (vault, aws-sm).",
59                "Resolved at config load; fetched concurrently and de-duplicated; never persisted.",
60                "Build with --features secrets (or per-backend secrets-vault / secrets-aws-sm / ...)."
61            ]
62        }),
63    };
64    let body = serde_json::to_string_pretty(&schema).unwrap_or_else(|_| schema.to_string());
65    println!("{body}");
66    Ok(())
67}
68
69/// JSON Schema for the `lineage:` config block (`faucet schema lineage`).
70#[cfg(feature = "lineage")]
71pub fn lineage_schema() -> serde_json::Value {
72    serde_json::to_value(faucet_lineage::schemars_schema())
73        .unwrap_or_else(|_| serde_json::json!({"type": "object"}))
74}
75
76#[cfg(test)]
77mod tests {
78    use crate::cli::{SchemaArgs, SchemaTarget};
79
80    #[cfg(feature = "lineage")]
81    #[test]
82    fn schema_lineage_returns_object_schema() {
83        let v = super::lineage_schema();
84        assert_eq!(v["type"], "object");
85        assert!(v["properties"].get("transport").is_some());
86        assert!(v["properties"].get("namespace").is_some());
87    }
88
89    #[tokio::test]
90    async fn schema_replication_target_ok() {
91        // Covers the `SchemaTarget::Replication` arm — it serializes the
92        // ReplicationSpec JSON Schema to stdout and returns Ok.
93        let r = super::run(SchemaArgs {
94            target: SchemaTarget::Replication,
95        })
96        .await;
97        assert!(r.is_ok(), "{r:?}");
98    }
99
100    #[tokio::test]
101    async fn schema_execution_target_ok() {
102        let r = super::run(SchemaArgs {
103            target: SchemaTarget::Execution,
104        })
105        .await;
106        assert!(r.is_ok(), "{r:?}");
107    }
108
109    #[test]
110    fn execution_schema_includes_adaptive_batch_size() {
111        let schema = faucet_core::schema_for!(crate::config::ExecutionSpec);
112        let value = serde_json::to_value(schema).expect("execution schema serializes");
113        assert!(value["properties"].get("adaptive_batch_size").is_some());
114    }
115
116    #[tokio::test]
117    async fn schema_resilience_target_ok() {
118        let r = super::run(SchemaArgs {
119            target: SchemaTarget::Resilience,
120        })
121        .await;
122        assert!(r.is_ok(), "{r:?}");
123    }
124
125    #[test]
126    fn schema_resilience_emits_json_schema() {
127        // Mirrors `faucet schema resilience`: the serialized ResilienceSpec
128        // schema must expose the retry `max_attempts` knob and the
129        // `circuit_breaker` sub-block.
130        let schema = faucet_core::schema_for!(crate::config::ResilienceSpec);
131        let out = serde_json::to_string(&schema).expect("resilience schema serializes");
132        assert!(out.contains("max_attempts"), "{out}");
133        assert!(out.contains("circuit_breaker"), "{out}");
134    }
135}
faucet_cli/commands/schema.rs

faucet_cli/commands/
schema.rs
