alien_aws_clients/aws/
rds.rs

1//! RDS client scoped to Aurora Serverless v2 (PostgreSQL) — the AWS Postgres backend.
2//!
3//! RDS is an AWS query-protocol API (form-encoded request, XML response), so this mirrors
4//! the EC2 client shape rather than the JSON DynamoDB one. Only the operations the Aurora
5//! Postgres controller needs are implemented.
6
7use crate::aws::aws_request_utils::{AwsRequestBuilderExt, AwsSignConfig};
8use crate::aws::credential_provider::AwsCredentialProvider;
9use alien_client_core::{ErrorData, Result};
10use alien_error::ContextError;
11use reqwest::{Client, Method, StatusCode};
12use serde::de::DeserializeOwned;
13use serde::Deserialize;
14use std::collections::HashMap;
15
16#[cfg(feature = "test-utils")]
17use mockall::automock;
18
19const RDS_API_VERSION: &str = "2014-10-31";
20pub const AURORA_POSTGRESQL_ENGINE: &str = "aurora-postgresql";
21pub const SERVERLESS_INSTANCE_CLASS: &str = "db.serverless";
22
23#[cfg_attr(feature = "test-utils", automock)]
24#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
25#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
26pub trait RdsApi: Send + Sync + std::fmt::Debug {
27    async fn create_db_subnet_group(&self, request: CreateDbSubnetGroupRequest) -> Result<()>;
28    async fn delete_db_subnet_group(&self, name: &str) -> Result<()>;
29    async fn create_db_cluster(&self, request: CreateDbClusterRequest) -> Result<DbCluster>;
30    async fn modify_db_cluster(&self, request: ModifyDbClusterRequest) -> Result<()>;
31    async fn delete_db_cluster(&self, request: DeleteDbClusterRequest) -> Result<()>;
32    async fn describe_db_clusters(&self, identifier: &str) -> Result<Vec<DbCluster>>;
33    async fn create_db_instance(&self, request: CreateDbInstanceRequest) -> Result<()>;
34    async fn delete_db_instance(&self, request: DeleteDbInstanceRequest) -> Result<()>;
35    async fn describe_db_instances(&self, cluster_identifier: &str) -> Result<Vec<DbInstance>>;
36}
37
38// ─────────────────────────── request types ───────────────────────────
39
40#[derive(Debug, Clone)]
41pub struct CreateDbSubnetGroupRequest {
42    pub name: String,
43    pub description: String,
44    pub subnet_ids: Vec<String>,
45    pub tags: HashMap<String, String>,
46}
47
48#[derive(Debug, Clone)]
49pub struct CreateDbClusterRequest {
50    pub identifier: String,
51    pub engine_version: String,
52    pub master_username: String,
53    pub master_user_password: String,
54    pub database_name: String,
55    pub db_subnet_group_name: String,
56    pub vpc_security_group_ids: Vec<String>,
57    /// Aurora Serverless v2 max ACU ceiling (1 ACU ≈ 2 GiB).
58    pub max_capacity: f64,
59    pub backup_retention_days: u16,
60    pub tags: HashMap<String, String>,
61}
62
63#[derive(Debug, Clone)]
64pub struct ModifyDbClusterRequest {
65    pub identifier: String,
66    /// Only set for an in-place major upgrade.
67    pub engine_version: Option<String>,
68    /// Only set for an in-place ACU (memory) resize; mirrors the create scaling ceiling.
69    pub max_capacity: Option<f64>,
70    /// Only set to re-sync the master password with the connection secret when adopting an
71    /// existing cluster. Applied immediately (never deferred to the maintenance window).
72    pub master_user_password: Option<String>,
73}
74
75#[derive(Debug, Clone)]
76pub struct DeleteDbClusterRequest {
77    pub identifier: String,
78}
79
80#[derive(Debug, Clone)]
81pub struct CreateDbInstanceRequest {
82    pub identifier: String,
83    pub cluster_identifier: String,
84    pub engine_version: String,
85}
86
87#[derive(Debug, Clone)]
88pub struct DeleteDbInstanceRequest {
89    pub identifier: String,
90}
91
92// ─────────────────────────── response types (XML) ───────────────────────────
93
94#[derive(Debug, Clone, Deserialize, PartialEq)]
95#[serde(rename_all = "PascalCase")]
96pub struct DbCluster {
97    #[serde(rename = "DBClusterIdentifier")]
98    pub identifier: String,
99    pub status: String,
100    #[serde(default)]
101    pub endpoint: Option<String>,
102    #[serde(default)]
103    pub reader_endpoint: Option<String>,
104    #[serde(default)]
105    pub port: Option<u16>,
106    #[serde(default)]
107    pub engine_version: Option<String>,
108    /// The serverless v2 scaling window the cluster reports — drives in-place ACU resize detection.
109    #[serde(default, rename = "ServerlessV2ScalingConfiguration")]
110    pub serverless_v2_scaling_configuration: Option<ServerlessV2ScalingConfiguration>,
111}
112
113/// The cluster's Serverless v2 ACU scaling window, as reported by DescribeDBClusters.
114#[derive(Debug, Clone, Deserialize, PartialEq)]
115#[serde(rename_all = "PascalCase")]
116pub struct ServerlessV2ScalingConfiguration {
117    /// The ACU ceiling (`memory` maps to this; the floor stays 0 for auto-pause).
118    pub max_capacity: f64,
119}
120
121#[derive(Debug, Clone, Deserialize, PartialEq)]
122#[serde(rename_all = "PascalCase")]
123pub struct DbInstance {
124    #[serde(rename = "DBInstanceIdentifier")]
125    pub identifier: String,
126    #[serde(rename = "DBInstanceStatus")]
127    pub status: String,
128}
129
130#[derive(Debug, Deserialize)]
131#[serde(rename_all = "PascalCase")]
132struct CreateDbClusterEnvelope {
133    #[serde(rename = "CreateDBClusterResult")]
134    result: CreateDbClusterResult,
135}
136#[derive(Debug, Deserialize)]
137#[serde(rename_all = "PascalCase")]
138struct CreateDbClusterResult {
139    #[serde(rename = "DBCluster")]
140    db_cluster: DbCluster,
141}
142
143#[derive(Debug, Deserialize)]
144#[serde(rename_all = "PascalCase")]
145struct DescribeDbClustersEnvelope {
146    #[serde(rename = "DescribeDBClustersResult")]
147    result: DescribeDbClustersResult,
148}
149#[derive(Debug, Deserialize)]
150#[serde(rename_all = "PascalCase")]
151struct DescribeDbClustersResult {
152    #[serde(rename = "DBClusters", default)]
153    db_clusters: DbClusterList,
154}
155#[derive(Debug, Default, Deserialize)]
156struct DbClusterList {
157    #[serde(rename = "DBCluster", default)]
158    members: Vec<DbCluster>,
159}
160
161#[derive(Debug, Deserialize)]
162#[serde(rename_all = "PascalCase")]
163struct DescribeDbInstancesEnvelope {
164    #[serde(rename = "DescribeDBInstancesResult")]
165    result: DescribeDbInstancesResult,
166}
167#[derive(Debug, Deserialize)]
168#[serde(rename_all = "PascalCase")]
169struct DescribeDbInstancesResult {
170    #[serde(rename = "DBInstances", default)]
171    db_instances: DbInstanceList,
172}
173#[derive(Debug, Default, Deserialize)]
174struct DbInstanceList {
175    #[serde(rename = "DBInstance", default)]
176    members: Vec<DbInstance>,
177}
178
179#[derive(Debug, Deserialize)]
180struct RdsErrorEnvelope {
181    #[serde(rename = "Error")]
182    error: RdsError,
183}
184#[derive(Debug, Deserialize)]
185#[serde(rename_all = "PascalCase")]
186struct RdsError {
187    code: String,
188    message: String,
189}
190
191// ─────────────────────────── form builders (pure, unit-tested) ───────────────────────────
192
193fn indexed_members(
194    form: &mut HashMap<String, String>,
195    prefix: &str,
196    member: &str,
197    values: &[String],
198) {
199    for (i, value) in values.iter().enumerate() {
200        form.insert(format!("{prefix}.{member}.{}", i + 1), value.clone());
201    }
202}
203
204fn tag_members(form: &mut HashMap<String, String>, tags: &HashMap<String, String>) {
205    for (i, (key, value)) in tags.iter().enumerate() {
206        form.insert(format!("Tags.Tag.{}.Key", i + 1), key.clone());
207        form.insert(format!("Tags.Tag.{}.Value", i + 1), value.clone());
208    }
209}
210
211fn base_form(action: &str) -> HashMap<String, String> {
212    HashMap::from([
213        ("Action".to_string(), action.to_string()),
214        ("Version".to_string(), RDS_API_VERSION.to_string()),
215    ])
216}
217
218fn create_db_subnet_group_form(r: &CreateDbSubnetGroupRequest) -> HashMap<String, String> {
219    let mut form = base_form("CreateDBSubnetGroup");
220    form.insert("DBSubnetGroupName".into(), r.name.clone());
221    form.insert("DBSubnetGroupDescription".into(), r.description.clone());
222    indexed_members(&mut form, "SubnetIds", "SubnetIdentifier", &r.subnet_ids);
223    tag_members(&mut form, &r.tags);
224    form
225}
226
227fn create_db_cluster_form(r: &CreateDbClusterRequest) -> HashMap<String, String> {
228    let mut form = base_form("CreateDBCluster");
229    form.insert("DBClusterIdentifier".into(), r.identifier.clone());
230    form.insert("Engine".into(), AURORA_POSTGRESQL_ENGINE.to_string());
231    form.insert("EngineVersion".into(), r.engine_version.clone());
232    form.insert("MasterUsername".into(), r.master_username.clone());
233    form.insert("MasterUserPassword".into(), r.master_user_password.clone());
234    form.insert("DatabaseName".into(), r.database_name.clone());
235    form.insert("DBSubnetGroupName".into(), r.db_subnet_group_name.clone());
236    indexed_members(
237        &mut form,
238        "VpcSecurityGroupIds",
239        "VpcSecurityGroupId",
240        &r.vpc_security_group_ids,
241    );
242    // minCapacity is always 0 — auto-pause is the point of the AWS backend, not a knob.
243    form.insert(
244        "ServerlessV2ScalingConfiguration.MinCapacity".into(),
245        "0".into(),
246    );
247    form.insert(
248        "ServerlessV2ScalingConfiguration.MaxCapacity".into(),
249        format!("{}", r.max_capacity),
250    );
251    form.insert(
252        "BackupRetentionPeriod".into(),
253        r.backup_retention_days.to_string(),
254    );
255    form.insert("StorageEncrypted".into(), "true".into());
256    tag_members(&mut form, &r.tags);
257    form
258}
259
260fn modify_db_cluster_form(r: &ModifyDbClusterRequest) -> HashMap<String, String> {
261    let mut form = base_form("ModifyDBCluster");
262    form.insert("DBClusterIdentifier".into(), r.identifier.clone());
263    form.insert("ApplyImmediately".into(), "true".into());
264    if let Some(version) = &r.engine_version {
265        form.insert("EngineVersion".into(), version.clone());
266        form.insert("AllowMajorVersionUpgrade".into(), "true".into());
267    }
268    if let Some(max_capacity) = r.max_capacity {
269        // Mirror create: the floor stays 0 (auto-pause); only the ACU ceiling moves.
270        form.insert(
271            "ServerlessV2ScalingConfiguration.MinCapacity".into(),
272            "0".into(),
273        );
274        form.insert(
275            "ServerlessV2ScalingConfiguration.MaxCapacity".into(),
276            format!("{max_capacity}"),
277        );
278    }
279    if let Some(password) = &r.master_user_password {
280        form.insert("MasterUserPassword".into(), password.clone());
281    }
282    form
283}
284
285fn create_db_instance_form(r: &CreateDbInstanceRequest) -> HashMap<String, String> {
286    let mut form = base_form("CreateDBInstance");
287    form.insert("DBInstanceIdentifier".into(), r.identifier.clone());
288    form.insert("DBClusterIdentifier".into(), r.cluster_identifier.clone());
289    form.insert("Engine".into(), AURORA_POSTGRESQL_ENGINE.to_string());
290    form.insert("EngineVersion".into(), r.engine_version.clone());
291    form.insert(
292        "DBInstanceClass".into(),
293        SERVERLESS_INSTANCE_CLASS.to_string(),
294    );
295    // Hard constraint: Postgres is never public. Pin it on the instance (the cluster has no such
296    // flag) so the guarantee holds regardless of the subnet group, matching GCP (`ipv4Enabled=false`)
297    // and Azure (`publicNetworkAccess=Disabled`) rather than relying on the subnet group's default.
298    form.insert("PubliclyAccessible".into(), "false".into());
299    form
300}
301
302fn delete_db_cluster_form(r: &DeleteDbClusterRequest) -> HashMap<String, String> {
303    let mut form = base_form("DeleteDBCluster");
304    form.insert("DBClusterIdentifier".into(), r.identifier.clone());
305    // No final snapshot — consistent with every other Alien resource and required for
306    // E2E teardown to actually tear down.
307    form.insert("SkipFinalSnapshot".into(), "true".into());
308    form
309}
310
311fn delete_db_instance_form(r: &DeleteDbInstanceRequest) -> HashMap<String, String> {
312    let mut form = base_form("DeleteDBInstance");
313    form.insert("DBInstanceIdentifier".into(), r.identifier.clone());
314    form.insert("SkipFinalSnapshot".into(), "true".into());
315    form
316}
317
318// ─────────────────────────── client ───────────────────────────
319
320#[derive(Debug, Clone)]
321pub struct RdsClient {
322    client: Client,
323    credentials: AwsCredentialProvider,
324}
325
326impl RdsClient {
327    pub fn new(client: Client, credentials: AwsCredentialProvider) -> Self {
328        Self {
329            client,
330            credentials,
331        }
332    }
333
334    fn sign_config(&self) -> AwsSignConfig {
335        AwsSignConfig {
336            service_name: "rds".into(),
337            region: self.credentials.region().to_string(),
338            credentials: self.credentials.get_credentials(),
339            signing_region: None,
340        }
341    }
342
343    fn get_base_url(&self) -> String {
344        if let Some(url) = self.credentials.get_service_endpoint_option("rds") {
345            url.to_string()
346        } else {
347            format!("https://rds.{}.amazonaws.com", self.credentials.region())
348        }
349    }
350
351    fn get_host(&self) -> String {
352        format!("rds.{}.amazonaws.com", self.credentials.region())
353    }
354
355    async fn send_form<T: DeserializeOwned + Send + 'static>(
356        &self,
357        form_data: HashMap<String, String>,
358        operation: &str,
359        resource: &str,
360    ) -> Result<T> {
361        self.credentials.ensure_fresh().await?;
362        let body = form_urlencoded::Serializer::new(String::new())
363            .extend_pairs(form_data.iter())
364            .finish();
365        let builder = self
366            .client
367            .request(Method::POST, &self.get_base_url())
368            .host(&self.get_host())
369            .content_type_form()
370            .content_sha256(&body)
371            .body(body);
372        let result =
373            crate::aws::aws_request_utils::sign_send_xml(builder, &self.sign_config()).await;
374        Self::map_result(result, operation, resource)
375    }
376
377    async fn send_form_no_body(
378        &self,
379        form_data: HashMap<String, String>,
380        operation: &str,
381        resource: &str,
382    ) -> Result<()> {
383        self.credentials.ensure_fresh().await?;
384        let body = form_urlencoded::Serializer::new(String::new())
385            .extend_pairs(form_data.iter())
386            .finish();
387        let builder = self
388            .client
389            .request(Method::POST, &self.get_base_url())
390            .host(&self.get_host())
391            .content_type_form()
392            .content_sha256(&body)
393            .body(body);
394        let result =
395            crate::aws::aws_request_utils::sign_send_no_response(builder, &self.sign_config())
396                .await;
397        Self::map_result(result, operation, resource)
398    }
399
400    fn map_result<T>(result: Result<T>, operation: &str, resource: &str) -> Result<T> {
401        // The create form-body carries MasterUserPassword; strip it from the whole chain BEFORE the
402        // non-internal head wraps below, or it survives in the source chain (head-only sanitization)
403        // and reaches durable state / external responses. See `redact_request_body`.
404        let result = alien_client_core::redact_request_body(result);
405        let Err(e) = result else {
406            return result;
407        };
408        let Some(ErrorData::HttpResponseError {
409            http_status,
410            http_response_text: Some(text),
411            ..
412        }) = &e.error
413        else {
414            return Err(e);
415        };
416        let status =
417            StatusCode::from_u16(*http_status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
418        match Self::map_rds_error(status, text, resource) {
419            Some(mapped) => Err(e.context(mapped)),
420            // No RDS-specific mapping: keep the original HttpResponseError as the source (its request
421            // body was stripped above) and attach the non-sensitive `operation` for debuggability.
422            // GenericError matches the current HttpResponseError metadata (internal=true,
423            // retryable=true), so semantics are unchanged.
424            None => Err(e.context(ErrorData::GenericError {
425                message: format!("RDS {operation} failed"),
426            })),
427        }
428    }
429
430    fn map_rds_error(status: StatusCode, body: &str, resource: &str) -> Option<ErrorData> {
431        // RDS reports "already gone" as 404 with DBClusterNotFoundFault / DBInstanceNotFoundFault;
432        // surfacing it as RemoteResourceNotFound lets the controller's delete be best-effort.
433        if let Ok(parsed) = quick_xml::de::from_str::<RdsErrorEnvelope>(body) {
434            let code = parsed.error.code.as_str();
435            if code.ends_with("NotFoundFault") {
436                return Some(ErrorData::RemoteResourceNotFound {
437                    resource_type: "RDS Resource".into(),
438                    resource_name: resource.into(),
439                });
440            }
441            // "still in use / wrong lifecycle state" and "already exists" are conflicts, not
442            // generic failures — the controller's create/delete flows key their re-entry retry
443            // (`is_conflict_or_exists`) off RemoteResourceConflict (e.g. a subnet group still
444            // attached to a deleting cluster: `InvalidDBSubnetGroupStateFault`).
445            if code.ends_with("StateFault")
446                || code.ends_with("InUseFault")
447                || code.ends_with("AlreadyExistsFault")
448            {
449                return Some(ErrorData::RemoteResourceConflict {
450                    resource_type: "RDS Resource".into(),
451                    resource_name: resource.into(),
452                    message: format!("{}: {}", code, parsed.error.message),
453                });
454            }
455            // Surface throttling/rate-limit faults as RateLimitExceeded — a non-internal,
456            // rate-limit-shaped error the executor backs off on — rather than letting them fall
457            // through to the internal-flagged GenericError catch-all below.
458            if matches!(
459                code,
460                "Throttling"
461                    | "ThrottlingException"
462                    | "RequestLimitExceeded"
463                    | "ProvisionedThroughputExceededException"
464            ) || code.contains("Throttl")
465                || code.ends_with("LimitExceeded")
466            {
467                return Some(ErrorData::RateLimitExceeded {
468                    message: format!("{}: {}", code, parsed.error.message),
469                });
470            }
471            // AWS query-protocol client faults — bad parameters, malformed requests, validation —
472            // are deterministic: retrying the same call cannot succeed. Map them to a non-retryable
473            // InvalidInput so the executor fails fast instead of retrying until timeout. (StateFault
474            // / InUseFault / NotFoundFault / AlreadyExistsFault / throttles already returned above as
475            // their own retry-appropriate kinds.)
476            if code.starts_with("Invalid")
477                || code.contains("Validation")
478                || code.contains("MalformedQuery")
479            {
480                return Some(ErrorData::InvalidInput {
481                    message: format!("{}: {}", code, parsed.error.message),
482                    field_name: None,
483                });
484            }
485            // Quota / capacity faults are operator-actionable (raise the limit) or transient
486            // (capacity frees up), not opaque internal failures: map them to non-internal kinds so
487            // the message reaches the user (QuotaExceeded) or the executor backs off
488            // (RemoteServiceUnavailable), instead of the internal-flagged GenericError below.
489            if code.ends_with("QuotaExceededFault") {
490                return Some(ErrorData::QuotaExceeded {
491                    message: format!("{}: {}", code, parsed.error.message),
492                });
493            }
494            if code.starts_with("Insufficient") || code.ends_with("CapacityFault") {
495                return Some(ErrorData::RemoteServiceUnavailable {
496                    message: format!("{}: {}", code, parsed.error.message),
497                });
498            }
499            return Some(ErrorData::GenericError {
500                message: format!("{}: {}", code, parsed.error.message),
501            });
502        }
503        match status {
504            StatusCode::NOT_FOUND => Some(ErrorData::RemoteResourceNotFound {
505                resource_type: "RDS Resource".into(),
506                resource_name: resource.into(),
507            }),
508            StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED => {
509                Some(ErrorData::RemoteAccessDenied {
510                    resource_type: "RDS Resource".into(),
511                    resource_name: resource.into(),
512                })
513            }
514            _ => None,
515        }
516    }
517}
518
519#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
520#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
521impl RdsApi for RdsClient {
522    async fn create_db_subnet_group(&self, request: CreateDbSubnetGroupRequest) -> Result<()> {
523        let name = request.name.clone();
524        self.send_form_no_body(
525            create_db_subnet_group_form(&request),
526            "CreateDBSubnetGroup",
527            &name,
528        )
529        .await
530    }
531
532    async fn delete_db_subnet_group(&self, name: &str) -> Result<()> {
533        let mut form = base_form("DeleteDBSubnetGroup");
534        form.insert("DBSubnetGroupName".into(), name.to_string());
535        self.send_form_no_body(form, "DeleteDBSubnetGroup", name)
536            .await
537    }
538
539    async fn create_db_cluster(&self, request: CreateDbClusterRequest) -> Result<DbCluster> {
540        let id = request.identifier.clone();
541        let envelope: CreateDbClusterEnvelope = self
542            .send_form(create_db_cluster_form(&request), "CreateDBCluster", &id)
543            .await?;
544        Ok(envelope.result.db_cluster)
545    }
546
547    async fn modify_db_cluster(&self, request: ModifyDbClusterRequest) -> Result<()> {
548        self.send_form_no_body(
549            modify_db_cluster_form(&request),
550            "ModifyDBCluster",
551            &request.identifier,
552        )
553        .await
554    }
555
556    async fn delete_db_cluster(&self, request: DeleteDbClusterRequest) -> Result<()> {
557        let id = request.identifier.clone();
558        self.send_form_no_body(delete_db_cluster_form(&request), "DeleteDBCluster", &id)
559            .await
560    }
561
562    async fn describe_db_clusters(&self, identifier: &str) -> Result<Vec<DbCluster>> {
563        let mut form = base_form("DescribeDBClusters");
564        form.insert("DBClusterIdentifier".into(), identifier.to_string());
565        let envelope: DescribeDbClustersEnvelope = self
566            .send_form(form, "DescribeDBClusters", identifier)
567            .await?;
568        Ok(envelope.result.db_clusters.members)
569    }
570
571    async fn create_db_instance(&self, request: CreateDbInstanceRequest) -> Result<()> {
572        let id = request.identifier.clone();
573        self.send_form_no_body(create_db_instance_form(&request), "CreateDBInstance", &id)
574            .await
575    }
576
577    async fn delete_db_instance(&self, request: DeleteDbInstanceRequest) -> Result<()> {
578        let id = request.identifier.clone();
579        self.send_form_no_body(delete_db_instance_form(&request), "DeleteDBInstance", &id)
580            .await
581    }
582
583    async fn describe_db_instances(&self, cluster_identifier: &str) -> Result<Vec<DbInstance>> {
584        let mut form = base_form("DescribeDBInstances");
585        form.insert("Filters.Filter.1.Name".into(), "db-cluster-id".into());
586        form.insert(
587            "Filters.Filter.1.Values.Value.1".into(),
588            cluster_identifier.to_string(),
589        );
590        let envelope: DescribeDbInstancesEnvelope = self
591            .send_form(form, "DescribeDBInstances", cluster_identifier)
592            .await?;
593        Ok(envelope.result.db_instances.members)
594    }
595}
596
597#[cfg(test)]
598mod tests {
599    use super::*;
600
601    fn cluster_request() -> CreateDbClusterRequest {
602        CreateDbClusterRequest {
603            identifier: "stack-db".into(),
604            engine_version: "17.4".into(),
605            master_username: "alien".into(),
606            master_user_password: "secret".into(),
607            database_name: "db".into(),
608            db_subnet_group_name: "stack-subnets".into(),
609            vpc_security_group_ids: vec!["sg-1".into(), "sg-2".into()],
610            max_capacity: 4.0,
611            backup_retention_days: 7,
612            tags: HashMap::new(),
613        }
614    }
615
616    #[test]
617    fn create_cluster_form_pins_min_capacity_zero_and_engine() {
618        let form = create_db_cluster_form(&cluster_request());
619        assert_eq!(form["Action"], "CreateDBCluster");
620        assert_eq!(form["Engine"], "aurora-postgresql");
621        assert_eq!(form["ServerlessV2ScalingConfiguration.MinCapacity"], "0");
622        assert_eq!(form["ServerlessV2ScalingConfiguration.MaxCapacity"], "4");
623        assert_eq!(form["BackupRetentionPeriod"], "7");
624        assert_eq!(form["VpcSecurityGroupIds.VpcSecurityGroupId.1"], "sg-1");
625        assert_eq!(form["VpcSecurityGroupIds.VpcSecurityGroupId.2"], "sg-2");
626    }
627
628    #[test]
629    fn modify_cluster_form_moves_acu_ceiling_and_pins_min_zero() {
630        let form = modify_db_cluster_form(&ModifyDbClusterRequest {
631            identifier: "stack-db".into(),
632            engine_version: None,
633            max_capacity: Some(8.0),
634            master_user_password: None,
635        });
636        assert_eq!(form["Action"], "ModifyDBCluster");
637        assert_eq!(form["ApplyImmediately"], "true");
638        assert_eq!(form["ServerlessV2ScalingConfiguration.MinCapacity"], "0");
639        assert_eq!(form["ServerlessV2ScalingConfiguration.MaxCapacity"], "8");
640        // A memory-only resize must not touch the engine version or the password.
641        assert!(!form.contains_key("EngineVersion"));
642        assert!(!form.contains_key("MasterUserPassword"));
643    }
644
645    #[test]
646    fn modify_cluster_form_carries_master_password_applied_immediately() {
647        let form = modify_db_cluster_form(&ModifyDbClusterRequest {
648            identifier: "stack-db".into(),
649            engine_version: None,
650            max_capacity: None,
651            master_user_password: Some("s3cret-pw".into()),
652        });
653        // A password re-sync on adopt must apply immediately, never deferred to the
654        // maintenance window (which would leave the secret and cluster diverged).
655        assert_eq!(form["ApplyImmediately"], "true");
656        assert_eq!(form["MasterUserPassword"], "s3cret-pw");
657    }
658
659    #[test]
660    fn modify_cluster_form_omits_scaling_when_no_memory_change() {
661        let form = modify_db_cluster_form(&ModifyDbClusterRequest {
662            identifier: "stack-db".into(),
663            engine_version: None,
664            max_capacity: None,
665            master_user_password: None,
666        });
667        assert!(!form.contains_key("ServerlessV2ScalingConfiguration.MaxCapacity"));
668    }
669
670    #[test]
671    fn delete_cluster_skips_final_snapshot() {
672        let form = delete_db_cluster_form(&DeleteDbClusterRequest {
673            identifier: "stack-db".into(),
674        });
675        assert_eq!(form["Action"], "DeleteDBCluster");
676        assert_eq!(form["SkipFinalSnapshot"], "true");
677    }
678
679    #[test]
680    fn create_instance_uses_serverless_class_and_is_never_public() {
681        let form = create_db_instance_form(&CreateDbInstanceRequest {
682            identifier: "stack-db-1".into(),
683            cluster_identifier: "stack-db".into(),
684            engine_version: "17.4".into(),
685        });
686        assert_eq!(form["DBInstanceClass"], "db.serverless");
687        assert_eq!(form["Engine"], "aurora-postgresql");
688        // pinned on the instance — the cluster has no such flag.
689        assert_eq!(form["PubliclyAccessible"], "false");
690    }
691
692    #[test]
693    fn map_result_strips_master_password_from_error_chain() {
694        // The create form-body carries MasterUserPassword and the transport captures it into the
695        // HttpResponseError; map_result must strip it before the non-internal head wraps the error,
696        // or it survives into durable state / status responses. (The chain-walk is covered in
697        // alien-client-core; this pins the redaction ordering at the AWS call site.)
698        use alien_error::AlienError;
699        const PW: &str = "Sup3rSecret-MasterPassword!";
700        let raw = AlienError::new(ErrorData::HttpResponseError {
701            message: "Request failed with HTTP 400".into(),
702            url: "https://rds.us-east-1.amazonaws.com/".into(),
703            http_status: 400,
704            http_request_text: Some(format!("Action=CreateDBCluster&MasterUserPassword={PW}")),
705            http_response_text: Some("<Error><Code>InvalidParameterValue</Code></Error>".into()),
706        });
707        let mapped = RdsClient::map_result::<()>(Err(raw), "CreateDBCluster", "stack-db")
708            .expect_err("an error result stays an error");
709        let json = serde_json::to_string(&mapped).expect("serialize");
710        assert!(
711            !json.contains(PW),
712            "master password leaked through map_result: {json}"
713        );
714    }
715
716    #[test]
717    fn describe_clusters_parses_xml() {
718        let xml = r#"<DescribeDBClustersResponse><DescribeDBClustersResult><DBClusters>
719            <DBCluster><DBClusterIdentifier>stack-db</DBClusterIdentifier><Status>available</Status>
720            <Endpoint>stack-db.cluster-x.us-east-1.rds.amazonaws.com</Endpoint><Port>5432</Port>
721            <EngineVersion>17.4</EngineVersion>
722            <ServerlessV2ScalingConfiguration><MinCapacity>0</MinCapacity><MaxCapacity>8</MaxCapacity></ServerlessV2ScalingConfiguration></DBCluster>
723        </DBClusters></DescribeDBClustersResult></DescribeDBClustersResponse>"#;
724        let env: DescribeDbClustersEnvelope = quick_xml::de::from_str(xml).expect("parses");
725        let clusters = env.result.db_clusters.members;
726        assert_eq!(clusters.len(), 1);
727        assert_eq!(clusters[0].identifier, "stack-db");
728        assert_eq!(clusters[0].status, "available");
729        assert_eq!(clusters[0].port, Some(5432));
730        // The serverless scaling ceiling must parse: `ready`/refresh read it to detect a day-2 memory
731        // change, and an imported cluster (max_capacity None) relies on this XML to learn its ceiling.
732        assert_eq!(
733            clusters[0]
734                .serverless_v2_scaling_configuration
735                .as_ref()
736                .map(|s| s.max_capacity),
737            Some(8.0)
738        );
739    }
740}
alien_aws_clients/aws/rds.rs

alien_aws_clients/aws/
rds.rs
