1use std::io::Read;
21use std::path::Path;
22
23use anyhow::{Context, Result};
24use k8s_openapi::api::core::v1::ConfigMap;
25use kube::api::{ApiResource, DynamicObject, ListParams, PostParams};
26use kube::core::GroupVersionKind;
27use kube::{Api, Client, ResourceExt};
28use serde_json::{Map, Value, json};
29
30use crate::{read_manifest_file, validate_manifest};
31
32const GROUP: &str = "pgroles.io";
37const VERSION: &str = "v1alpha1";
38
39fn candidate_api(client: Client, namespace: &str) -> Api<DynamicObject> {
40 let gvk = GroupVersionKind::gvk(GROUP, VERSION, "PostgresPolicyCandidate");
41 let resource = ApiResource::from_gvk_with_plural(&gvk, "postgrespolicycandidates");
42 Api::namespaced_with(client, namespace, &resource)
43}
44
45fn plan_api(client: Client, namespace: &str) -> Api<DynamicObject> {
46 let gvk = GroupVersionKind::gvk(GROUP, VERSION, "PostgresPolicyPlan");
47 let resource = ApiResource::from_gvk_with_plural(&gvk, "postgrespolicyplans");
48 Api::namespaced_with(client, namespace, &resource)
49}
50
51fn policy_api(client: Client, namespace: &str) -> Api<DynamicObject> {
52 let gvk = GroupVersionKind::gvk(GROUP, VERSION, "PostgresPolicy");
53 let resource = ApiResource::from_gvk_with_plural(&gvk, "postgrespolicies");
54 Api::namespaced_with(client, namespace, &resource)
55}
56
57pub const CONTENT_KEYS: &[&str] = &[
66 "default_owner",
67 "default_privileges",
68 "grants",
69 "memberships",
70 "profiles",
71 "reconciliation_mode",
72 "retirements",
73 "roles",
74 "schemas",
75];
76
77pub const POLICY_EXECUTION_KEYS: &[&str] =
81 &["connection", "interval", "mode", "suspend", "approval"];
82
83pub fn extract_candidate_content(yaml: &str) -> Result<Value> {
98 let document: serde_yaml::Value =
99 serde_yaml::from_str(yaml).context("failed to parse manifest YAML")?;
100 let body = candidate_content_body(&document)?;
101
102 let mapping = body.as_mapping().ok_or_else(|| {
103 anyhow::anyhow!("policy content must be a YAML mapping of content keys (roles, grants, …)")
104 })?;
105
106 let mut content = Map::new();
107 let mut unsupported = Vec::new();
108
109 for (key, value) in mapping {
110 let Some(key) = key.as_str() else {
111 anyhow::bail!("policy content keys must be strings");
112 };
113
114 if POLICY_EXECUTION_KEYS.contains(&key) {
115 continue;
117 }
118 if !CONTENT_KEYS.contains(&key) {
119 unsupported.push(key.to_string());
120 continue;
121 }
122 if value.is_null() {
123 anyhow::bail!(
124 "policy content key `{key}` has no value; remove it or give it a value \
125 (a candidate stores exactly what it is given)"
126 );
127 }
128
129 let json = serde_json::to_value(value)
130 .with_context(|| format!("failed to convert policy content key `{key}` to JSON"))?;
131 content.insert(key.to_string(), json);
132 }
133
134 if !unsupported.is_empty() {
135 anyhow::bail!(
136 "manifest key(s) {} cannot be carried by a PostgresPolicyCandidate; \
137 `spec.content` accepts only: {}",
138 unsupported
139 .iter()
140 .map(|key| format!("`{key}`"))
141 .collect::<Vec<_>>()
142 .join(", "),
143 CONTENT_KEYS.join(", "),
144 );
145 }
146
147 if content.is_empty() {
148 anyhow::bail!(
149 "manifest declares no policy content; a candidate proposing nothing has \
150 nothing to review"
151 );
152 }
153
154 Ok(Value::Object(content))
155}
156
157fn candidate_content_body(document: &serde_yaml::Value) -> Result<&serde_yaml::Value> {
159 let Some(map) = document.as_mapping() else {
160 anyhow::bail!("manifest must be a YAML mapping");
161 };
162
163 let get = |key: &str| map.get(serde_yaml::Value::String(key.to_string()));
164
165 let is_cr = get("apiVersion").is_some() && get("spec").is_some();
166 if !is_cr {
167 return Ok(document);
168 }
169
170 let spec = get("spec").expect("checked above");
171 let kind = get("kind").and_then(serde_yaml::Value::as_str);
172
173 if kind == Some("PostgresPolicyCandidate") {
174 return spec
175 .as_mapping()
176 .and_then(|spec| spec.get(serde_yaml::Value::String("content".to_string())))
177 .ok_or_else(|| {
178 anyhow::anyhow!("PostgresPolicyCandidate manifest has no `spec.content`")
179 });
180 }
181
182 Ok(spec)
183}
184
185pub fn build_candidate_object(policy: &str, replaces: Option<&str>, content: Value) -> Value {
191 let mut spec = Map::new();
192 spec.insert("policyRef".to_string(), json!({ "name": policy }));
193 if let Some(replaces) = replaces {
194 spec.insert("replaces".to_string(), Value::String(replaces.to_string()));
195 }
196 spec.insert("content".to_string(), content);
197
198 json!({
199 "apiVersion": format!("{GROUP}/{VERSION}"),
200 "kind": "PostgresPolicyCandidate",
201 "metadata": { "generateName": generate_name_prefix(policy) },
202 "spec": Value::Object(spec),
203 })
204}
205
206pub fn generate_name_prefix(policy: &str) -> String {
209 const MAX_GENERATE_NAME_PREFIX: usize = 248;
210
211 let mut prefix = format!("{policy}-");
212 if prefix.len() > MAX_GENERATE_NAME_PREFIX {
213 prefix.truncate(MAX_GENERATE_NAME_PREFIX);
214 }
215 prefix
216}
217
218#[derive(Debug, Clone, PartialEq, Eq)]
224pub struct Condition {
225 pub status: String,
226 pub reason: Option<String>,
227 pub message: Option<String>,
228}
229
230impl Condition {
231 pub fn summary(&self) -> String {
233 match &self.reason {
234 Some(reason) => format!("{}/{}", self.status, reason),
235 None => self.status.clone(),
236 }
237 }
238}
239
240pub fn find_condition(object: &Value, condition_type: &str) -> Option<Condition> {
242 object
243 .get("status")?
244 .get("conditions")?
245 .as_array()?
246 .iter()
247 .find(|condition| condition.get("type").and_then(Value::as_str) == Some(condition_type))
248 .map(|condition| Condition {
249 status: condition
250 .get("status")
251 .and_then(Value::as_str)
252 .unwrap_or("Unknown")
253 .to_string(),
254 reason: condition
255 .get("reason")
256 .and_then(Value::as_str)
257 .map(str::to_string),
258 message: condition
259 .get("message")
260 .and_then(Value::as_str)
261 .map(str::to_string),
262 })
263}
264
265pub fn string_at(object: &Value, path: &str) -> Option<String> {
267 let mut current = object;
268 for segment in path.split('.') {
269 current = current.get(segment)?;
270 }
271 current.as_str().map(str::to_string)
272}
273
274pub fn abbreviate_digest(digest: &str) -> String {
277 const KEEP: usize = 12;
278
279 fn head(value: &str) -> Option<&str> {
285 value
286 .char_indices()
287 .nth(KEEP)
288 .map(|(boundary, _)| &value[..boundary])
289 }
290
291 match digest.split_once(':') {
292 Some((algorithm, hex)) => match head(hex) {
293 Some(head) => format!("{algorithm}:{head}…"),
294 None => digest.to_string(),
295 },
296 None => match head(digest) {
297 Some(head) => format!("{head}…"),
298 None => digest.to_string(),
299 },
300 }
301}
302
303#[derive(Debug, Clone, PartialEq, Eq)]
309pub struct CandidateRow {
310 pub name: String,
311 pub phase: String,
312 pub digest: String,
313 pub plan: String,
314 pub ready: String,
315 pub superseded: String,
316 pub promoted: String,
317}
318
319const NONE_CELL: &str = "-";
322
323fn cell(value: Option<String>) -> String {
324 value.unwrap_or_else(|| NONE_CELL.to_string())
325}
326
327pub fn candidate_row(candidate: &Value, name: &str) -> CandidateRow {
329 CandidateRow {
330 name: name.to_string(),
331 phase: cell(string_at(candidate, "status.phase")),
332 digest: cell(string_at(candidate, "status.contentDigest").map(|d| abbreviate_digest(&d))),
333 plan: cell(string_at(candidate, "status.planRef.name")),
334 ready: cell(find_condition(candidate, "Ready").map(|c| c.summary())),
335 superseded: cell(find_condition(candidate, "Superseded").map(|c| c.summary())),
336 promoted: cell(find_condition(candidate, "Promoted").map(|c| c.summary())),
337 }
338}
339
340pub fn format_candidate_table(rows: &[CandidateRow]) -> String {
343 let headers = [
344 "NAME",
345 "PHASE",
346 "DIGEST",
347 "PLAN",
348 "READY",
349 "SUPERSEDED",
350 "PROMOTED",
351 ];
352 let columns: Vec<[&str; 7]> = rows
353 .iter()
354 .map(|row| {
355 [
356 row.name.as_str(),
357 row.phase.as_str(),
358 row.digest.as_str(),
359 row.plan.as_str(),
360 row.ready.as_str(),
361 row.superseded.as_str(),
362 row.promoted.as_str(),
363 ]
364 })
365 .collect();
366
367 let mut widths: Vec<usize> = headers
368 .iter()
369 .map(|header| header.chars().count())
370 .collect();
371 for row in &columns {
372 for (index, value) in row.iter().enumerate() {
373 widths[index] = widths[index].max(value.chars().count());
374 }
375 }
376
377 let mut output = String::new();
378 let mut push_row = |values: &[&str]| {
379 let mut line = String::new();
380 for (index, value) in values.iter().enumerate() {
381 if index + 1 == values.len() {
382 line.push_str(value);
383 } else {
384 let pad = widths[index].saturating_sub(value.chars().count()) + 2;
385 line.push_str(value);
386 line.push_str(&" ".repeat(pad));
387 }
388 }
389 output.push_str(line.trim_end());
390 output.push('\n');
391 };
392
393 push_row(&headers);
394 for row in &columns {
395 push_row(row);
396 }
397 output
398}
399
400#[derive(Debug, Clone, PartialEq, Eq)]
406pub enum PlanSqlSource {
407 Inline(String),
409 ConfigMap {
411 name: String,
412 key: String,
413 gzip: bool,
414 },
415}
416
417pub fn select_plan_sql(plan: &Value) -> Result<PlanSqlSource> {
424 let status = plan.get("status");
425 let inline = status
426 .and_then(|status| status.get("sqlInline"))
427 .and_then(Value::as_str)
428 .filter(|sql| !sql.is_empty());
429 let truncated = status
430 .and_then(|status| status.get("sqlTruncated"))
431 .and_then(Value::as_bool)
432 .unwrap_or(false);
433 let sql_ref = status.and_then(|status| status.get("sqlRef"));
434
435 if let Some(inline) = inline
436 && !truncated
437 {
438 return Ok(PlanSqlSource::Inline(inline.to_string()));
439 }
440
441 if let Some(sql_ref) = sql_ref.filter(|value| !value.is_null()) {
442 let name = sql_ref.get("name").and_then(Value::as_str);
443 let key = sql_ref.get("key").and_then(Value::as_str);
444 match (name, key) {
445 (Some(name), Some(key)) if !name.is_empty() && !key.is_empty() => {
446 let gzip = sql_ref.get("compression").and_then(Value::as_str) == Some("gzip");
447 return Ok(PlanSqlSource::ConfigMap {
448 name: name.to_string(),
449 key: key.to_string(),
450 gzip,
451 });
452 }
453 _ => anyhow::bail!(
454 "plan has an incomplete status.sqlRef (name={:?}, key={:?}); \
455 the stored SQL cannot be located",
456 name.unwrap_or_default(),
457 key.unwrap_or_default(),
458 ),
459 }
460 }
461
462 if inline.is_some() && truncated {
463 anyhow::bail!(
464 "plan stores only a truncated SQL preview (status.sqlTruncated=true) and no \
465 status.sqlRef ConfigMap, so the full reviewed SQL is not recoverable from the \
466 cluster; read the operator log for the plan, or reduce the size of the change"
467 );
468 }
469
470 anyhow::bail!(
471 "plan has no SQL recorded yet (neither status.sqlInline nor status.sqlRef); \
472 it has probably not finished computing — check `pgroles candidate status`"
473 )
474}
475
476pub fn decode_configmap_sql(configmap: &ConfigMap, key: &str, gzip: bool) -> Result<String> {
482 let name = configmap.metadata.name.as_deref().unwrap_or("<unnamed>");
483
484 if gzip {
485 let bytes = configmap
486 .binary_data
487 .as_ref()
488 .and_then(|data| data.get(key))
489 .ok_or_else(|| {
490 anyhow::anyhow!(
491 "ConfigMap {name} has no binaryData key `{key}`; \
492 the plan's SQL artifact is missing or was pruned"
493 )
494 })?;
495
496 let mut sql = String::new();
497 flate2::read::GzDecoder::new(bytes.0.as_slice())
498 .read_to_string(&mut sql)
499 .with_context(|| format!("failed to decompress SQL from ConfigMap {name}/{key}"))?;
500 return Ok(sql);
501 }
502
503 configmap
504 .data
505 .as_ref()
506 .and_then(|data| data.get(key))
507 .cloned()
508 .ok_or_else(|| {
509 anyhow::anyhow!(
510 "ConfigMap {name} has no data key `{key}`; \
511 the plan's SQL artifact is missing or was pruned"
512 )
513 })
514}
515
516pub fn format_candidate_status(
526 name: &str,
527 namespace: &str,
528 candidate: &Value,
529 plan_name: Option<&str>,
530 plan: Option<&Value>,
531) -> String {
532 let mut output = String::new();
533
534 output.push_str(&format!("Candidate: {name}\n"));
535 output.push_str(&format!("Namespace: {namespace}\n"));
536 output.push_str(&format!(
537 "Policy: {}\n",
538 cell(string_at(candidate, "spec.policyRef.name"))
539 ));
540 if let Some(replaces) = string_at(candidate, "spec.replaces") {
541 output.push_str(&format!("Replaces: {replaces}\n"));
542 }
543 if let Some(secret) = string_at(candidate, "spec.target.connectionRef.secretName") {
544 let key = cell(string_at(candidate, "spec.target.connectionRef.key"));
545 output.push_str(&format!(
546 "Target: override — Secret {secret} key {key} (a preview, never a cutover)\n"
547 ));
548 }
549 output.push_str(&format!(
550 "Phase: {}\n",
551 cell(string_at(candidate, "status.phase"))
552 ));
553 output.push_str(&format!(
554 "Digest: {}\n",
555 cell(string_at(candidate, "status.contentDigest"))
556 ));
557
558 output.push_str("\nConditions:\n");
559 let mut any_condition = false;
560 for condition_type in ["Ready", "Superseded", "Promoted"] {
561 if let Some(condition) = find_condition(candidate, condition_type) {
562 any_condition = true;
563 output.push_str(&format!(
564 " {condition_type}={} ({})\n",
565 condition.status,
566 condition.reason.as_deref().unwrap_or("no reason recorded"),
567 ));
568 if let Some(message) = &condition.message {
569 output.push_str(&format!(" {message}\n"));
570 }
571 }
572 }
573 if !any_condition {
574 output.push_str(" none recorded yet — the operator has not observed this candidate\n");
575 }
576
577 match (plan_name, plan) {
578 (None, _) => {
579 output.push_str(
580 "\nPlan: none yet.\n \
581 The operator publishes a plan on the parent policy's next reconcile. \
582 A candidate blocked by its parent reports Ready=False, \
583 reason=BlockedByActivePolicy above.\n",
584 );
585 }
586 (Some(plan_name), None) => {
587 output.push_str(&format!(
588 "\nPlan: {plan_name} (NOT READABLE)\n \
589 The candidate names this plan but it could not be read — it may have been \
590 pruned by plan retention.\n",
591 ));
592 }
593 (Some(plan_name), Some(plan)) => {
594 output.push_str(&format!("\nPlan: {plan_name}\n"));
595 output.push_str(&format!(
596 " Phase: {}\n",
597 cell(string_at(plan, "status.phase"))
598 ));
599 output.push_str(&format!(
600 " Decision: {}\n",
601 format_plan_decision(plan)
602 ));
603 output.push_str(&format!(
604 " Change digest: {}\n",
605 cell(string_at(plan, "status.changeDigest"))
606 ));
607 output.push_str(&format!(
608 " Computed at: {}\n",
609 cell(string_at(plan, "status.computedAt"))
610 ));
611 output.push_str(&format!(
612 " Staleness: {}\n",
613 format_plan_staleness(plan)
614 ));
615 output.push_str(&format!(
616 " Base pinned: {}\n",
617 cell(
618 string_at(plan, "spec.origin.baseContentDigest")
619 .map(|digest| abbreviate_digest(&digest))
620 )
621 ));
622 if let Some(error) = string_at(plan, "status.lastError") {
623 output.push_str(&format!(" Last error: {error}\n"));
624 }
625 }
626 }
627
628 output.push_str(&format!(
629 "\nPromotion: {}\n",
630 format_promotion_outcome(candidate)
631 ));
632 output
633}
634
635pub fn format_plan_decision(plan: &Value) -> String {
637 let decided_by = string_at(plan, "status.decidedBy.username");
638
639 for (condition_type, verb) in [("Approved", "approved"), ("Denied", "denied")] {
640 if let Some(condition) = find_condition(plan, condition_type)
641 && condition.status == "True"
642 {
643 return match decided_by {
644 Some(username) => format!("{verb} by {username}"),
645 None => format!("{verb}, but no decidedBy identity is recorded"),
648 };
649 }
650 }
651
652 "none recorded — awaiting review (decide with kubectl, see the plan-approval docs)".to_string()
653}
654
655pub fn format_plan_staleness(plan: &Value) -> String {
657 if let Some(condition) = find_condition(plan, "Superseded")
658 && condition.status == "True"
659 {
660 let reason = condition.reason.as_deref().unwrap_or("no reason recorded");
661 return format!("STALE — superseded ({reason}); the candidate is replanned from scratch");
662 }
663
664 match string_at(plan, "status.revalidatedAt") {
665 Some(at) => format!("current — last revalidated {at}"),
666 None => "current — not revalidated since it was computed".to_string(),
667 }
668}
669
670pub fn format_promotion_outcome(candidate: &Value) -> String {
672 match find_condition(candidate, "Promoted") {
673 Some(condition) if condition.status == "True" => format!(
674 "promoted and executed ({})",
675 condition.reason.as_deref().unwrap_or("Promoted")
676 ),
677 Some(condition) => format!(
678 "did NOT complete ({}): {}",
679 condition.reason.as_deref().unwrap_or("no reason recorded"),
680 condition
681 .message
682 .as_deref()
683 .unwrap_or("no message recorded on the condition"),
684 ),
685 None => "not promoted — the content has not been merged into the policy".to_string(),
686 }
687}
688
689pub async fn cmd_create(
695 policy: &str,
696 file: &Path,
697 replaces: Option<&str>,
698 namespace: &str,
699) -> Result<()> {
700 let yaml = read_manifest_file(file)?;
701 let content = extract_candidate_content(&yaml)?;
702
703 let content_yaml =
707 serde_yaml::to_string(&content).context("failed to re-serialize policy content")?;
708 let validated = validate_manifest(&content_yaml)
709 .with_context(|| format!("policy content in {} is not valid", file.display()))?;
710
711 let client = Client::try_default()
712 .await
713 .context("failed to create Kubernetes client")?;
714 let candidates = candidate_api(client.clone(), namespace);
715
716 policy_api(client, namespace)
719 .get(policy)
720 .await
721 .with_context(|| {
722 format!("failed to read postgrespolicy/{policy} in namespace {namespace}")
723 })?;
724
725 if let Some(replaces) = replaces {
726 let replaced = candidates.get(replaces).await.with_context(|| {
727 format!(
728 "failed to read the candidate named by --replaces \
729 (postgrespolicycandidate/{replaces} in namespace {namespace})"
730 )
731 })?;
732 let replaced_policy = string_at(&replaced.data, "spec.policyRef.name");
733 if replaced_policy.as_deref() != Some(policy) {
734 anyhow::bail!(
735 "candidate {replaces} proposes content for policy {}, not {policy}; \
736 supersession is only meaningful within one policy",
737 replaced_policy.as_deref().unwrap_or("<unknown>"),
738 );
739 }
740 }
741
742 let manifest = build_candidate_object(policy, replaces, content);
743 let mut object: DynamicObject =
744 serde_json::from_value(manifest).context("failed to build the candidate object")?;
745 object.metadata.name = None;
747
748 let created = candidates
749 .create(&PostParams::default(), &object)
750 .await
751 .with_context(|| {
752 format!("failed to create PostgresPolicyCandidate for policy {policy} in {namespace}")
753 })?;
754
755 let name = created.name_any();
756 println!("Created postgrespolicycandidate/{name} in namespace {namespace}.");
757 println!(
758 " policy: {policy}, {} role(s), {} grant(s), {} membership(s) after expansion",
759 validated.expanded.roles.len(),
760 validated.expanded.grants.len(),
761 validated.expanded.memberships.len(),
762 );
763 if let Some(replaces) = replaces {
764 println!(" supersedes: {replaces}");
765 }
766 println!(" review with: pgroles candidate status {name} -n {namespace}");
767
768 Ok(())
769}
770
771pub async fn cmd_list(policy: &str, namespace: &str) -> Result<()> {
773 let client = Client::try_default()
774 .await
775 .context("failed to create Kubernetes client")?;
776
777 policy_api(client.clone(), namespace)
780 .get(policy)
781 .await
782 .with_context(|| {
783 format!("failed to read postgrespolicy/{policy} in namespace {namespace}")
784 })?;
785
786 let candidates = candidate_api(client, namespace)
787 .list(&ListParams::default())
788 .await
789 .with_context(|| format!("failed to list candidates in namespace {namespace}"))?;
790
791 let mut rows: Vec<CandidateRow> = candidates
792 .items
793 .iter()
794 .filter(|candidate| {
795 string_at(&candidate.data, "spec.policyRef.name").as_deref() == Some(policy)
796 })
797 .map(|candidate| candidate_row(&candidate.data, &candidate.name_any()))
798 .collect();
799 rows.sort_by(|a, b| a.name.cmp(&b.name));
800
801 if rows.is_empty() {
802 println!("No candidates for postgrespolicy/{policy} in namespace {namespace}.");
803 return Ok(());
804 }
805
806 print!("{}", format_candidate_table(&rows));
807 Ok(())
808}
809
810pub async fn cmd_status(name: &str, namespace: &str) -> Result<()> {
812 let client = Client::try_default()
813 .await
814 .context("failed to create Kubernetes client")?;
815 let candidate = candidate_api(client.clone(), namespace)
816 .get(name)
817 .await
818 .with_context(|| {
819 format!("failed to read postgrespolicycandidate/{name} in namespace {namespace}")
820 })?;
821
822 let plan_name = string_at(&candidate.data, "status.planRef.name");
823 let plan = match &plan_name {
824 Some(plan_name) => plan_api(client, namespace).get(plan_name).await.ok(),
825 None => None,
826 };
827
828 print!(
829 "{}",
830 format_candidate_status(
831 name,
832 namespace,
833 &candidate.data,
834 plan_name.as_deref(),
835 plan.as_ref().map(|plan| &plan.data),
836 )
837 );
838
839 Ok(())
840}
841
842pub async fn cmd_diff(name: &str, namespace: &str) -> Result<()> {
844 let client = Client::try_default()
845 .await
846 .context("failed to create Kubernetes client")?;
847 let candidate = candidate_api(client.clone(), namespace)
848 .get(name)
849 .await
850 .with_context(|| {
851 format!("failed to read postgrespolicycandidate/{name} in namespace {namespace}")
852 })?;
853
854 let plan_name = string_at(&candidate.data, "status.planRef.name").ok_or_else(|| {
855 let phase = string_at(&candidate.data, "status.phase").unwrap_or_else(|| "?".to_string());
856 anyhow::anyhow!(
857 "candidate {name} has no plan yet (phase {phase}); there is nothing to diff. \
858 Run `pgroles candidate status {name} -n {namespace}` to see why"
859 )
860 })?;
861
862 let plan = plan_api(client.clone(), namespace)
863 .get(&plan_name)
864 .await
865 .with_context(|| {
866 format!("failed to read postgrespolicyplan/{plan_name} named by candidate {name}")
867 })?;
868
869 let sql = match select_plan_sql(&plan.data)
870 .with_context(|| format!("cannot show the SQL for plan {plan_name}"))?
871 {
872 PlanSqlSource::Inline(sql) => sql,
873 PlanSqlSource::ConfigMap {
874 name: configmap_name,
875 key,
876 gzip,
877 } => {
878 let configmaps: Api<ConfigMap> = Api::namespaced(client, namespace);
879 let configmap = configmaps.get(&configmap_name).await.with_context(|| {
880 format!(
881 "failed to read ConfigMap {configmap_name} holding the SQL for plan {plan_name}"
882 )
883 })?;
884 decode_configmap_sql(&configmap, &key, gzip)?
885 }
886 };
887
888 if sql.trim().is_empty() {
889 anyhow::bail!(
890 "plan {plan_name} recorded an empty SQL preview; that is not the same as a plan \
891 with no changes, which would report Ready=True, reason=NoEffects on the candidate"
892 );
893 }
894
895 eprintln!("-- candidate {name}, plan {plan_name}, namespace {namespace}");
897 eprintln!(
898 "-- this is what approving the plan would execute (passwords redacted by the operator)"
899 );
900 print!("{sql}");
901 if !sql.ends_with('\n') {
902 println!();
903 }
904
905 Ok(())
906}
907
908#[cfg(test)]
909mod tests {
910 use super::*;
911 use k8s_openapi::ByteString;
912 use std::collections::BTreeMap;
913 use std::io::Write;
914
915 fn gzip(input: &str) -> Vec<u8> {
916 let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
917 encoder.write_all(input.as_bytes()).expect("gzip write");
918 encoder.finish().expect("gzip finish")
919 }
920
921 #[test]
926 fn extract_content_takes_a_bare_manifest_verbatim() {
927 let content = extract_candidate_content(
928 r#"
929default_owner: app_owner
930roles:
931 - name: reporting_reader
932 login: true
933"#,
934 )
935 .expect("bare manifest should extract");
936
937 assert_eq!(content["default_owner"], json!("app_owner"));
938 assert_eq!(content["roles"][0]["name"], json!("reporting_reader"));
939 }
940
941 #[test]
942 fn extract_content_drops_execution_fields_from_a_policy_cr() {
943 let content = extract_candidate_content(
944 r#"
945apiVersion: pgroles.io/v1alpha1
946kind: PostgresPolicy
947metadata:
948 name: orders
949spec:
950 connection:
951 secretRef:
952 name: postgres-credentials
953 interval: "5m"
954 mode: apply
955 suspend: false
956 approval: manual
957 reconciliation_mode: authoritative
958 roles:
959 - name: reporting_reader
960 login: true
961"#,
962 )
963 .expect("policy CR should extract");
964
965 let object = content.as_object().expect("content is an object");
966 assert_eq!(
967 object.keys().collect::<Vec<_>>(),
968 vec!["reconciliation_mode", "roles"]
969 );
970 }
971
972 #[test]
973 fn extract_content_unwraps_a_candidate_cr() {
974 let content = extract_candidate_content(
975 r#"
976apiVersion: pgroles.io/v1alpha1
977kind: PostgresPolicyCandidate
978metadata:
979 generateName: orders-change-
980spec:
981 policyRef:
982 name: orders
983 content:
984 roles:
985 - name: reporting_reader
986 login: true
987"#,
988 )
989 .expect("candidate CR should extract");
990
991 assert_eq!(content["roles"][0]["name"], json!("reporting_reader"));
992 }
993
994 #[test]
995 fn extract_content_rejects_keys_a_candidate_cannot_carry() {
996 let error = extract_candidate_content(
997 r#"
998auth_providers:
999 - type: cloud_sql_iam
1000roles:
1001 - name: reporting_reader
1002"#,
1003 )
1004 .expect_err("auth_providers has no candidate counterpart");
1005
1006 let message = error.to_string();
1007 assert!(message.contains("auth_providers"), "unexpected: {message}");
1008 assert!(message.contains("spec.content"), "unexpected: {message}");
1009 }
1010
1011 #[test]
1012 fn extract_content_rejects_an_empty_manifest() {
1013 let error = extract_candidate_content("mode: apply\n")
1014 .expect_err("a candidate proposing nothing should fail");
1015
1016 assert!(
1017 error.to_string().contains("no policy content"),
1018 "unexpected: {error}"
1019 );
1020 }
1021
1022 #[test]
1023 fn extract_content_rejects_a_valueless_content_key() {
1024 let error =
1025 extract_candidate_content("roles:\n").expect_err("a null content key should fail");
1026
1027 assert!(error.to_string().contains("`roles`"), "unexpected: {error}");
1028 }
1029
1030 #[test]
1035 fn build_candidate_object_uses_generate_name_and_omits_absent_replaces() {
1036 let object = build_candidate_object("orders", None, json!({ "roles": [] }));
1037
1038 assert_eq!(object["metadata"]["generateName"], json!("orders-"));
1039 assert_eq!(object["metadata"].get("name"), None);
1040 assert_eq!(object["spec"]["policyRef"]["name"], json!("orders"));
1041 assert_eq!(object["spec"].get("replaces"), None);
1042 assert_eq!(object["kind"], json!("PostgresPolicyCandidate"));
1043 }
1044
1045 #[test]
1046 fn build_candidate_object_records_replaces_when_given() {
1047 let object = build_candidate_object("orders", Some("orders-x7k2p"), json!({ "roles": [] }));
1048
1049 assert_eq!(object["spec"]["replaces"], json!("orders-x7k2p"));
1050 }
1051
1052 #[test]
1053 fn generate_name_prefix_leaves_room_for_the_server_suffix() {
1054 let long_policy = "a".repeat(253);
1055 let prefix = generate_name_prefix(&long_policy);
1056
1057 assert_eq!(prefix.len(), 248);
1058 assert!(prefix.len() + 5 <= 253);
1059 }
1060
1061 #[test]
1066 fn abbreviate_digest_keeps_the_algorithm_prefix() {
1067 assert_eq!(
1068 abbreviate_digest("sha256:0123456789abcdef0123456789abcdef"),
1069 "sha256:0123456789ab…"
1070 );
1071 assert_eq!(abbreviate_digest("short"), "short");
1072 }
1073
1074 #[test]
1075 fn abbreviate_digest_does_not_panic_on_a_malformed_multibyte_digest() {
1076 assert_eq!(
1080 abbreviate_digest("sha256:ααααααααααααααα"),
1081 "sha256:αααααααααααα…"
1082 );
1083 assert_eq!(abbreviate_digest("ααααααααααααααα"), "αααααααααααα…");
1084 assert_eq!(abbreviate_digest("αααααααααααα"), "αααααααααααα");
1086 }
1087
1088 fn planned_candidate() -> Value {
1089 json!({
1090 "spec": { "policyRef": { "name": "orders" } },
1091 "status": {
1092 "phase": "Planned",
1093 "contentDigest": "sha256:0123456789abcdef0123456789abcdef",
1094 "planRef": { "name": "orders-change-plan-9f21c4" },
1095 "conditions": [
1096 { "type": "Ready", "status": "True", "reason": "Planned",
1097 "message": "a current plan exists" }
1098 ]
1099 }
1100 })
1101 }
1102
1103 #[test]
1104 fn candidate_row_fills_absent_cells_with_a_placeholder() {
1105 let row = candidate_row(&planned_candidate(), "orders-change-x7k2p");
1106
1107 assert_eq!(row.phase, "Planned");
1108 assert_eq!(row.digest, "sha256:0123456789ab…");
1109 assert_eq!(row.plan, "orders-change-plan-9f21c4");
1110 assert_eq!(row.ready, "True/Planned");
1111 assert_eq!(row.superseded, NONE_CELL);
1113 assert_eq!(row.promoted, NONE_CELL);
1114 }
1115
1116 #[test]
1117 fn candidate_row_of_an_unobserved_candidate_is_all_placeholders() {
1118 let row = candidate_row(&json!({ "spec": {} }), "orders-change-new");
1119
1120 assert_eq!(row.phase, NONE_CELL);
1121 assert_eq!(row.digest, NONE_CELL);
1122 assert_eq!(row.plan, NONE_CELL);
1123 assert_eq!(row.ready, NONE_CELL);
1124 }
1125
1126 #[test]
1127 fn format_candidate_table_aligns_columns_under_headers() {
1128 let rows = vec![candidate_row(&planned_candidate(), "orders-change-x7k2p")];
1129 let table = format_candidate_table(&rows);
1130 let mut lines = table.lines();
1131
1132 let header = lines.next().expect("header row");
1133 let row = lines.next().expect("data row");
1134 assert!(header.starts_with("NAME"));
1135 assert!(row.starts_with("orders-change-x7k2p"));
1136 assert_eq!(
1137 header.find("PHASE"),
1138 row.find("Planned"),
1139 "PHASE column is misaligned:\n{table}"
1140 );
1141 assert!(lines.next().is_none(), "unexpected extra rows:\n{table}");
1142 }
1143
1144 #[test]
1145 fn format_plan_decision_names_the_decider() {
1146 let plan = json!({
1147 "status": {
1148 "decidedBy": { "username": "e2e-reviewer" },
1149 "conditions": [{ "type": "Approved", "status": "True", "reason": "Approved" }]
1150 }
1151 });
1152
1153 assert_eq!(format_plan_decision(&plan), "approved by e2e-reviewer");
1154 }
1155
1156 #[test]
1157 fn format_plan_decision_reports_an_undecided_plan_explicitly() {
1158 let plan = json!({ "status": { "phase": "Pending", "conditions": [] } });
1159
1160 assert!(format_plan_decision(&plan).starts_with("none recorded"));
1161 }
1162
1163 #[test]
1164 fn format_plan_staleness_names_the_supersede_reason() {
1165 let plan = json!({
1166 "status": {
1167 "conditions": [
1168 { "type": "Superseded", "status": "True", "reason": "EffectsChanged" }
1169 ]
1170 }
1171 });
1172
1173 let rendered = format_plan_staleness(&plan);
1174 assert!(rendered.contains("STALE"), "unexpected: {rendered}");
1175 assert!(
1176 rendered.contains("EffectsChanged"),
1177 "unexpected: {rendered}"
1178 );
1179 }
1180
1181 #[test]
1182 fn format_promotion_outcome_distinguishes_incomplete_promotion() {
1183 let candidate = json!({
1184 "status": {
1185 "conditions": [{
1186 "type": "Promoted",
1187 "status": "False",
1188 "reason": "PromotionDigestMismatch",
1189 "message": "the merged spec is not being enforced"
1190 }]
1191 }
1192 });
1193
1194 let rendered = format_promotion_outcome(&candidate);
1195 assert!(
1196 rendered.contains("did NOT complete"),
1197 "unexpected: {rendered}"
1198 );
1199 assert!(
1200 rendered.contains("PromotionDigestMismatch"),
1201 "unexpected: {rendered}"
1202 );
1203 }
1204
1205 #[test]
1206 fn format_candidate_status_explains_a_candidate_with_no_plan() {
1207 let candidate = json!({
1208 "spec": { "policyRef": { "name": "orders" } },
1209 "status": { "phase": "Pending" }
1210 });
1211
1212 let rendered = format_candidate_status("orders-new", "default", &candidate, None, None);
1213 assert!(rendered.contains("Plan: none yet"), "{rendered}");
1214 assert!(rendered.contains("none recorded yet"), "{rendered}");
1215 assert!(rendered.contains("not promoted"), "{rendered}");
1216 }
1217
1218 #[test]
1219 fn format_candidate_status_renders_the_plan_decision_and_target_override() {
1220 let mut candidate = planned_candidate();
1221 candidate["spec"]["target"] = json!({
1222 "connectionRef": { "secretName": "orders-new-postgres", "key": "url" }
1223 });
1224 let plan = json!({
1225 "spec": { "origin": { "baseContentDigest": "sha256:fedcba9876543210" } },
1226 "status": {
1227 "phase": "Approved",
1228 "changeDigest": "sha256:aaaa",
1229 "computedAt": "2026-08-17T00:00:00Z",
1230 "decidedBy": { "username": "reviewer" },
1231 "conditions": [{ "type": "Approved", "status": "True", "reason": "Approved" }]
1232 }
1233 });
1234
1235 let rendered = format_candidate_status(
1236 "orders-change-x7k2p",
1237 "default",
1238 &candidate,
1239 Some("orders-change-plan-9f21c4"),
1240 Some(&plan),
1241 );
1242
1243 assert!(rendered.contains("approved by reviewer"), "{rendered}");
1244 assert!(rendered.contains("orders-new-postgres"), "{rendered}");
1245 assert!(
1246 rendered.contains("Base pinned: sha256:fedcba987654"),
1247 "{rendered}"
1248 );
1249 }
1250
1251 #[test]
1256 fn select_plan_sql_prefers_a_complete_inline_preview() {
1257 let plan = json!({ "status": { "sqlInline": "CREATE ROLE a;\n" } });
1258
1259 assert_eq!(
1260 select_plan_sql(&plan).expect("inline SQL should be selected"),
1261 PlanSqlSource::Inline("CREATE ROLE a;\n".to_string())
1262 );
1263 }
1264
1265 #[test]
1266 fn select_plan_sql_falls_back_to_the_configmap_when_inline_is_truncated() {
1267 let plan = json!({
1268 "status": {
1269 "sqlInline": "CREATE ROLE a; -- truncated",
1270 "sqlTruncated": true,
1271 "sqlRef": { "name": "plan-sql", "key": "plan.sql.gz", "compression": "gzip" }
1272 }
1273 });
1274
1275 assert_eq!(
1276 select_plan_sql(&plan).expect("sqlRef should win over a truncated preview"),
1277 PlanSqlSource::ConfigMap {
1278 name: "plan-sql".to_string(),
1279 key: "plan.sql.gz".to_string(),
1280 gzip: true,
1281 }
1282 );
1283 }
1284
1285 #[test]
1286 fn select_plan_sql_reads_an_uncompressed_legacy_configmap() {
1287 let plan = json!({
1288 "status": { "sqlRef": { "name": "plan-sql", "key": "plan.sql" } }
1289 });
1290
1291 assert_eq!(
1292 select_plan_sql(&plan).expect("legacy sqlRef should be selected"),
1293 PlanSqlSource::ConfigMap {
1294 name: "plan-sql".to_string(),
1295 key: "plan.sql".to_string(),
1296 gzip: false,
1297 }
1298 );
1299 }
1300
1301 #[test]
1302 fn select_plan_sql_refuses_a_truncated_preview_with_no_configmap() {
1303 let plan = json!({
1304 "status": { "sqlInline": "CREATE ROLE a; -- trunc", "sqlTruncated": true }
1305 });
1306
1307 let error = select_plan_sql(&plan).expect_err("a fragment must not be shown as the plan");
1308 assert!(
1309 error.to_string().contains("truncated"),
1310 "unexpected: {error}"
1311 );
1312 }
1313
1314 #[test]
1315 fn select_plan_sql_refuses_a_plan_with_no_sql_at_all() {
1316 let error = select_plan_sql(&json!({ "status": { "phase": "Pending" } }))
1317 .expect_err("a plan with no SQL must not render as an empty diff");
1318
1319 assert!(
1320 error.to_string().contains("no SQL recorded"),
1321 "unexpected: {error}"
1322 );
1323 }
1324
1325 #[test]
1326 fn select_plan_sql_refuses_a_half_populated_sql_ref() {
1327 let plan = json!({ "status": { "sqlRef": { "name": "plan-sql" } } });
1328
1329 let error = select_plan_sql(&plan).expect_err("an incomplete sqlRef must fail loudly");
1330 assert!(
1331 error.to_string().contains("incomplete"),
1332 "unexpected: {error}"
1333 );
1334 }
1335
1336 #[test]
1341 fn decode_configmap_sql_inflates_gzipped_binary_data() {
1342 let configmap = ConfigMap {
1343 binary_data: Some(BTreeMap::from([(
1344 "plan.sql.gz".to_string(),
1345 ByteString(gzip("CREATE ROLE reporting_reader;\n")),
1346 )])),
1347 ..Default::default()
1348 };
1349
1350 assert_eq!(
1351 decode_configmap_sql(&configmap, "plan.sql.gz", true).expect("should inflate"),
1352 "CREATE ROLE reporting_reader;\n"
1353 );
1354 }
1355
1356 #[test]
1357 fn decode_configmap_sql_reads_uncompressed_data() {
1358 let configmap = ConfigMap {
1359 data: Some(BTreeMap::from([(
1360 "plan.sql".to_string(),
1361 "CREATE ROLE a;\n".to_string(),
1362 )])),
1363 ..Default::default()
1364 };
1365
1366 assert_eq!(
1367 decode_configmap_sql(&configmap, "plan.sql", false).expect("should read"),
1368 "CREATE ROLE a;\n"
1369 );
1370 }
1371
1372 #[test]
1373 fn decode_configmap_sql_fails_when_the_key_is_missing() {
1374 let configmap = ConfigMap::default();
1375
1376 let error = decode_configmap_sql(&configmap, "plan.sql.gz", true)
1377 .expect_err("a missing artifact must not read as empty SQL");
1378 assert!(
1379 error.to_string().contains("plan.sql.gz"),
1380 "unexpected: {error}"
1381 );
1382 }
1383}