rusticity_core/
config.rs

1use anyhow::Result;
2
3/// ECR Public is a global service only available in us-east-1.
4pub const ECR_PUBLIC_REGION: &str = "us-east-1";
5
6#[derive(Clone, Debug)]
7pub struct AwsConfig {
8    pub region: String,
9    pub account_id: String,
10    pub role_arn: String,
11    pub region_auto_detected: bool,
12    /// Resolved SDK config — reused by all client factories to avoid re-running credential chain.
13    sdk_config: aws_config::SdkConfig,
14}
15
16impl AwsConfig {
17    pub async fn new(region: Option<String>) -> Result<Self> {
18        Self::new_with_timeout(region, std::time::Duration::from_secs(10)).await
19    }
20
21    pub async fn new_with_timeout(
22        region: Option<String>,
23        timeout: std::time::Duration,
24    ) -> Result<Self> {
25        // Check for region early to avoid IMDS timeout
26        if region.is_none()
27            && std::env::var("AWS_REGION").is_err()
28            && std::env::var("AWS_DEFAULT_REGION").is_err()
29        {
30            return Err(anyhow::anyhow!("Missing Region"));
31        }
32
33        let mut config_loader = aws_config::defaults(aws_config::BehaviorVersion::latest());
34
35        // Use profile from AWS_PROFILE env var if set
36        let profile_to_use = std::env::var("AWS_PROFILE").ok();
37        if let Some(ref profile) = profile_to_use {
38            if !profile.is_empty() {
39                tracing::info!("Using AWS profile: {}", profile);
40                config_loader = config_loader.profile_name(profile);
41            }
42        }
43
44        if let Some(r) = &region {
45            config_loader = config_loader.region(aws_config::Region::new(r.clone()));
46        }
47
48        // Load config once — stored and reused for all client factories.
49        let sdk_config = tokio::time::timeout(timeout, config_loader.load())
50            .await
51            .map_err(|_| anyhow::anyhow!("Timeout loading AWS config"))?;
52
53        // Double-check region is set
54        if sdk_config.region().is_none() {
55            return Err(anyhow::anyhow!("Missing Region"));
56        }
57
58        // Try to get identity with timeout
59        let (account_id, role_arn) =
60            match tokio::time::timeout(timeout, Self::try_get_identity(&sdk_config)).await {
61                Ok(Ok((acc, role))) => {
62                    tracing::info!("Loaded identity: account={}, role={}", acc, role);
63                    (acc, role)
64                }
65                Ok(Err(e)) => {
66                    tracing::error!("Failed to get identity: {}", e);
67                    return Err(e);
68                }
69                Err(_) => return Err(anyhow::anyhow!("Timeout getting AWS identity")),
70            };
71
72        let (region_str, auto_detected) = match sdk_config.region() {
73            Some(r) => (r.as_ref().to_string(), false),
74            None => {
75                let fastest = Self::find_fastest_region().await?;
76                (fastest, true)
77            }
78        };
79
80        Ok(Self {
81            region: region_str,
82            account_id,
83            role_arn,
84            region_auto_detected: auto_detected,
85            sdk_config,
86        })
87    }
88
89    async fn try_get_identity(config: &aws_config::SdkConfig) -> Result<(String, String)> {
90        let sts_client = aws_sdk_sts::Client::new(config);
91        let identity = sts_client.get_caller_identity().send().await?;
92        let account_id = identity.account().unwrap_or("").to_string();
93        let role_arn = identity.arn().unwrap_or("").to_string();
94        Ok((account_id, role_arn))
95    }
96
97    async fn find_fastest_region() -> Result<String> {
98        use std::time::Instant;
99
100        let regions = [
101            "us-east-1",
102            "us-east-2",
103            "us-west-1",
104            "us-west-2",
105            "af-south-1",
106            "ap-east-1",
107            "ap-south-1",
108            "ap-south-2",
109            "ap-northeast-1",
110            "ap-northeast-2",
111            "ap-northeast-3",
112            "ap-southeast-1",
113            "ap-southeast-2",
114            "ap-southeast-3",
115            "ap-southeast-4",
116            "ca-central-1",
117            "ca-west-1",
118            "eu-central-1",
119            "eu-central-2",
120            "eu-west-1",
121            "eu-west-2",
122            "eu-west-3",
123            "eu-north-1",
124            "eu-south-1",
125            "eu-south-2",
126            "il-central-1",
127            "me-central-1",
128            "me-south-1",
129            "sa-east-1",
130        ];
131
132        let mut tasks = Vec::new();
133
134        for &region in &regions {
135            let region_name = region.to_string();
136            tasks.push(tokio::spawn(async move {
137                let config = aws_config::defaults(aws_config::BehaviorVersion::latest())
138                    .region(aws_config::Region::new(region_name.clone()))
139                    .load()
140                    .await;
141                let s3 = aws_sdk_s3::Client::new(&config);
142                let start = Instant::now();
143                match tokio::time::timeout(
144                    std::time::Duration::from_secs(2),
145                    s3.list_buckets().send(),
146                )
147                .await
148                {
149                    Ok(Ok(_)) => Some((region_name, start.elapsed())),
150                    _ => Some((region_name, std::time::Duration::from_secs(9999))),
151                }
152            }));
153        }
154
155        let results = futures::future::join_all(tasks).await;
156        let mut latencies: Vec<(String, std::time::Duration)> = results
157            .into_iter()
158            .filter_map(|r| r.ok().flatten())
159            .collect();
160
161        latencies.sort_by_key(|(_, d)| *d);
162
163        latencies
164            .first()
165            .map(|(r, _)| r.clone())
166            .ok_or_else(|| anyhow::anyhow!("Could not determine fastest region"))
167    }
168
169    pub fn dummy(region: Option<String>) -> Self {
170        // Build a minimal SdkConfig with just the region — no credential resolution.
171        let region_str = region.unwrap_or_default();
172        let sdk_config = aws_config::SdkConfig::builder()
173            .behavior_version(aws_config::BehaviorVersion::latest())
174            .region(aws_config::Region::new(region_str.clone()))
175            .build();
176        Self {
177            region: region_str,
178            account_id: "".to_string(),
179            role_arn: "".to_string(),
180            region_auto_detected: false,
181            sdk_config,
182        }
183    }
184
185    /// Test-only constructor: build an `AwsConfig` from a pre-built `SdkConfig`
186    /// (e.g. with a mock HTTP client) and a region/account. The `sdk_config`
187    /// replaces the one from `dummy`, allowing tests outside this module to
188    /// inject mock transport.
189    #[cfg(test)]
190    pub(crate) fn from_sdk_config(
191        sdk_config: aws_config::SdkConfig,
192        region: &str,
193        account_id: &str,
194    ) -> Self {
195        Self {
196            region: region.to_string(),
197            account_id: account_id.to_string(),
198            role_arn: String::new(),
199            region_auto_detected: false,
200            sdk_config,
201        }
202    }
203
204    pub async fn get_account_for_profile(profile: &str, region: &str) -> Result<String> {
205        let config = aws_config::defaults(aws_config::BehaviorVersion::latest())
206            .profile_name(profile)
207            .region(aws_config::Region::new(region.to_string()))
208            .load()
209            .await;
210
211        let sts_client = aws_sdk_sts::Client::new(&config);
212        let identity = sts_client.get_caller_identity().send().await?;
213        Ok(identity.account().unwrap_or("").to_string())
214    }
215
216    // ── Client factories ──────────────────────────────────────────────────────
217    // All factories reuse self.sdk_config — credentials are already resolved,
218    // so these are instantaneous (no network calls).
219
220    pub fn s3_client(&self) -> aws_sdk_s3::Client {
221        aws_sdk_s3::Client::new(&self.sdk_config)
222    }
223
224    pub async fn s3_client_with_region(&self, region: &str) -> aws_sdk_s3::Client {
225        // Needs a different region — build a new config with the same credentials.
226        let config = self
227            .sdk_config
228            .to_builder()
229            .region(aws_config::Region::new(region.to_string()))
230            .build();
231        aws_sdk_s3::Client::new(&config)
232    }
233
234    pub fn cloudformation_client(&self) -> aws_sdk_cloudformation::Client {
235        aws_sdk_cloudformation::Client::new(&self.sdk_config)
236    }
237
238    pub fn cloudtrail_client(&self) -> aws_sdk_cloudtrail::Client {
239        aws_sdk_cloudtrail::Client::new(&self.sdk_config)
240    }
241
242    pub fn lambda_client(&self) -> aws_sdk_lambda::Client {
243        aws_sdk_lambda::Client::new(&self.sdk_config)
244    }
245
246    pub fn iam_client(&self) -> aws_sdk_iam::Client {
247        aws_sdk_iam::Client::new(&self.sdk_config)
248    }
249
250    pub fn ecr_client(&self) -> aws_sdk_ecr::Client {
251        aws_sdk_ecr::Client::new(&self.sdk_config)
252    }
253
254    pub fn kms_client(&self) -> aws_sdk_kms::Client {
255        aws_sdk_kms::Client::new(&self.sdk_config)
256    }
257
258    pub fn efs_client(&self) -> aws_sdk_efs::Client {
259        aws_sdk_efs::Client::new(&self.sdk_config)
260    }
261
262    pub fn dynamodb_client(&self) -> aws_sdk_dynamodb::Client {
263        aws_sdk_dynamodb::Client::new(&self.sdk_config)
264    }
265
266    pub fn resource_explorer_client(&self) -> aws_sdk_resourceexplorer2::Client {
267        aws_sdk_resourceexplorer2::Client::new(&self.sdk_config)
268    }
269
270    pub fn rds_client(&self) -> aws_sdk_rds::Client {
271        aws_sdk_rds::Client::new(&self.sdk_config)
272    }
273
274    pub fn fsx_client(&self) -> aws_sdk_fsx::Client {
275        aws_sdk_fsx::Client::new(&self.sdk_config)
276    }
277
278    pub fn ecr_public_client(&self) -> aws_sdk_ecrpublic::Client {
279        // ECR Public is a global service only available in us-east-1.
280        // Reuse stored credentials but override region.
281        let config = aws_config::SdkConfig::builder()
282            .behavior_version(aws_config::BehaviorVersion::latest())
283            .region(aws_config::Region::new(ECR_PUBLIC_REGION))
284            .credentials_provider(
285                self.sdk_config
286                    .credentials_provider()
287                    .expect("credentials must be set")
288                    .clone(),
289            )
290            .build();
291        aws_sdk_ecrpublic::Client::new(&config)
292    }
293
294    pub fn cloudwatch_client(&self) -> aws_sdk_cloudwatch::Client {
295        aws_sdk_cloudwatch::Client::new(&self.sdk_config)
296    }
297
298    pub fn sqs_client(&self) -> aws_sdk_sqs::Client {
299        aws_sdk_sqs::Client::new(&self.sdk_config)
300    }
301
302    pub fn sns_client(&self) -> aws_sdk_sns::Client {
303        aws_sdk_sns::Client::new(&self.sdk_config)
304    }
305
306    pub fn cloudwatch_logs_client(&self) -> aws_sdk_cloudwatchlogs::Client {
307        aws_sdk_cloudwatchlogs::Client::new(&self.sdk_config)
308    }
309
310    pub fn pipes_client(&self) -> aws_sdk_pipes::Client {
311        aws_sdk_pipes::Client::new(&self.sdk_config)
312    }
313
314    pub fn ec2_client(&self) -> aws_sdk_ec2::Client {
315        aws_sdk_ec2::Client::new(&self.sdk_config)
316    }
317
318    pub fn apigateway_client(&self) -> aws_sdk_apigateway::Client {
319        aws_sdk_apigateway::Client::new(&self.sdk_config)
320    }
321
322    pub fn apigatewayv2_client(&self) -> aws_sdk_apigatewayv2::Client {
323        aws_sdk_apigatewayv2::Client::new(&self.sdk_config)
324    }
325
326    pub fn alarms_client(&self) -> aws_sdk_cloudwatch::Client {
327        aws_sdk_cloudwatch::Client::new(&self.sdk_config)
328    }
329}
330
331#[cfg(test)]
332mod tests {
333    use super::*;
334
335    #[test]
336    fn test_dummy_config_with_region() {
337        let region = "us-west-2";
338        let config = AwsConfig::dummy(Some(region.to_string()));
339        assert_eq!(config.region, region);
340        assert_eq!(config.account_id, "");
341        assert!(!config.region_auto_detected);
342    }
343
344    #[test]
345    fn test_dummy_config_without_region() {
346        let config = AwsConfig::dummy(None);
347        assert_eq!(config.region, "");
348        assert_eq!(config.account_id, "");
349        assert!(!config.region_auto_detected);
350    }
351
352    #[tokio::test]
353    async fn test_new_fails_without_credentials() {
354        // Clear AWS env vars to simulate no credentials
355        std::env::remove_var("AWS_ACCESS_KEY_ID");
356        std::env::remove_var("AWS_SECRET_ACCESS_KEY");
357        std::env::remove_var("AWS_SESSION_TOKEN");
358        std::env::set_var("AWS_PROFILE", "nonexistent-profile-test");
359
360        let result = AwsConfig::new(Some("us-east-1".to_string())).await;
361
362        // Should fail with credentials error before attempting region detection
363        assert!(result.is_err());
364        let err_str = format!("{}", result.unwrap_err());
365        // Can be credentials error or dispatch failure (both indicate auth issues)
366        assert!(
367            err_str.contains("credentials")
368                || err_str.contains("profile")
369                || err_str.contains("dispatch"),
370            "Expected auth error, got: {}",
371            err_str
372        );
373    }
374
375    #[tokio::test]
376    async fn test_timeout_is_configurable() {
377        std::env::remove_var("AWS_ACCESS_KEY_ID");
378        std::env::remove_var("AWS_SECRET_ACCESS_KEY");
379        std::env::set_var("AWS_PROFILE", "nonexistent-profile-test");
380
381        // Test with very short timeout
382        let result = AwsConfig::new_with_timeout(
383            Some("us-east-1".to_string()),
384            std::time::Duration::from_millis(100),
385        )
386        .await;
387
388        assert!(result.is_err());
389    }
390
391    #[test]
392    fn test_dummy_config_preserves_values() {
393        let region = "eu-west-1";
394        let config = AwsConfig::dummy(Some(region.to_string()));
395        assert_eq!(config.region, region);
396        assert_eq!(config.account_id, "");
397        assert_eq!(config.role_arn, "");
398        assert!(!config.region_auto_detected);
399    }
400
401    #[test]
402    fn test_ecr_public_region_is_us_east_1() {
403        // ECR Public is a global service — must always use us-east-1 regardless of user region.
404        assert_eq!(ECR_PUBLIC_REGION, "us-east-1");
405    }
406
407    #[test]
408    fn test_client_factories_are_sync() {
409        // All client factories (except s3_client_with_region) must be sync —
410        // they should not do any async work after credentials are resolved.
411        let config = AwsConfig::dummy(Some("us-east-1".to_string()));
412        let _ = config.s3_client();
413        let _ = config.ecr_client();
414        let _ = config.cloudwatch_client();
415        let _ = config.lambda_client();
416        let _ = config.iam_client();
417        let _ = config.sqs_client();
418        let _ = config.ec2_client();
419    }
420}
421
422#[cfg(test)]
423mod pagination_tests {
424    use super::*;
425    use crate::dynamodb::DynamoDbClient;
426    use crate::rds::{InventoryOptions, RdsClient};
427    use crate::VpcClient;
428    use aws_smithy_http_client::test_util::{ReplayEvent, StaticReplayClient};
429    use aws_smithy_types::body::SdkBody;
430
431    fn replay(uri: &str, request: String, status: u16, response: String) -> ReplayEvent {
432        ReplayEvent::new(
433            http::Request::builder()
434                .method("POST")
435                .uri(uri)
436                .body(SdkBody::from(request))
437                .unwrap(),
438            http::Response::builder()
439                .status(status)
440                .body(SdkBody::from(response))
441                .unwrap(),
442        )
443    }
444
445    fn config(http_client: &StaticReplayClient) -> AwsConfig {
446        let sdk_config = aws_config::SdkConfig::builder()
447            .behavior_version(aws_config::BehaviorVersion::latest())
448            .region(aws_config::Region::new("us-east-1"))
449            .credentials_provider(aws_sdk_iam::config::SharedCredentialsProvider::new(
450                aws_sdk_iam::config::Credentials::new(
451                    "test-key",
452                    "test-secret",
453                    None,
454                    None,
455                    "test",
456                ),
457            ))
458            .retry_config(aws_config::retry::RetryConfig::disabled())
459            .http_client(http_client.clone())
460            .build();
461        AwsConfig {
462            sdk_config,
463            ..AwsConfig::dummy(Some("us-east-1".to_string()))
464        }
465    }
466
467    #[tokio::test]
468    async fn s3_object_preview_uses_bounded_range_in_the_bucket_region_and_handles_empty_objects() {
469        use aws_smithy_http_client::test_util::infallible_client_fn;
470        use std::sync::{Arc, Mutex};
471        let calls = Arc::new(Mutex::new(Vec::new()));
472        let capture = calls.clone();
473        let http = infallible_client_fn(move |request| {
474            assert_eq!(
475                request.uri().host(),
476                Some("fixture-bucket.s3.us-west-2.amazonaws.com")
477            );
478            capture.lock().unwrap().push((
479                request.method().to_string(),
480                request.uri().path().to_string(),
481            ));
482            if request.method() == http::Method::HEAD {
483                assert_eq!(request.uri().path(), "/empty.txt");
484                return http::Response::builder()
485                    .status(200)
486                    .header("content-length", "0")
487                    .header("content-type", "text/plain")
488                    .body(vec![])
489                    .unwrap();
490            }
491            assert_eq!(
492                request.headers()["range"],
493                format!("bytes=0-{}", crate::s3::object::PREVIEW_LIMIT)
494            );
495            if request.uri().path() == "/empty.txt" {
496                http::Response::builder()
497                    .status(416)
498                    .body(b"<Error><Code>InvalidRange</Code></Error>".to_vec())
499                    .unwrap()
500            } else {
501                assert_eq!(request.uri().path(), "/prefix/hello%20world.json");
502                http::Response::builder()
503                    .status(206)
504                    .header("content-type", "application/json")
505                    .body(br#"{"enabled":true}"#.to_vec())
506                    .unwrap()
507            }
508        });
509        let mut config = config(&StaticReplayClient::new(vec![]));
510        config.sdk_config = config.sdk_config.to_builder().http_client(http).build();
511        let client = crate::S3Client::new(config);
512        let preview = client
513            .preview_object("fixture-bucket", "us-west-2", "prefix/hello world.json")
514            .await
515            .unwrap();
516        assert_eq!(preview.format, crate::s3::object::TextFormat::Json);
517        assert!(preview.text.contains("\"enabled\": true"));
518        assert!(!preview.truncated);
519        let empty = client
520            .preview_object("fixture-bucket", "us-west-2", "empty.txt")
521            .await
522            .unwrap();
523        assert!(empty.text.is_empty());
524        assert_eq!(calls.lock().unwrap().len(), 3);
525    }
526
527    #[tokio::test]
528    async fn s3_object_download_preserves_binary_bytes_and_existing_files_without_more_requests() {
529        use aws_smithy_http_client::test_util::infallible_client_fn;
530        use std::sync::{
531            atomic::{AtomicUsize, Ordering},
532            Arc,
533        };
534        let calls = Arc::new(AtomicUsize::new(0));
535        let capture = calls.clone();
536        let http = infallible_client_fn(move |request| {
537            capture.fetch_add(1, Ordering::Relaxed);
538            assert_eq!(request.method(), http::Method::GET);
539            assert_eq!(request.uri().path(), "/prefix/object.bin");
540            assert!(request.headers().get("range").is_none());
541            http::Response::builder()
542                .status(200)
543                .header("content-length", "5")
544                .body(vec![0, 1, 2, 254, 255])
545                .unwrap()
546        });
547        let mut config = config(&StaticReplayClient::new(vec![]));
548        config.sdk_config = config.sdk_config.to_builder().http_client(http).build();
549        let client = crate::S3Client::new(config);
550        let directory = std::env::temp_dir().join(format!(
551            "rusticity-s3-wire-{}-{}",
552            std::process::id(),
553            std::time::SystemTime::now()
554                .duration_since(std::time::UNIX_EPOCH)
555                .unwrap()
556                .as_nanos()
557        ));
558        std::fs::create_dir(&directory).unwrap();
559        let path = client
560            .download_object(
561                "fixture-bucket",
562                "us-east-1",
563                "prefix/object.bin",
564                &directory,
565            )
566            .await
567            .unwrap();
568        assert_eq!(path, directory.join("object.bin"));
569        assert_eq!(std::fs::read(&path).unwrap(), [0, 1, 2, 254, 255]);
570        assert!(client
571            .download_object(
572                "fixture-bucket",
573                "us-east-1",
574                "prefix/object.bin",
575                &directory
576            )
577            .await
578            .is_err());
579        assert_eq!(calls.load(Ordering::Relaxed), 1);
580        assert_eq!(std::fs::read_dir(&directory).unwrap().count(), 1);
581        std::fs::remove_dir_all(directory).unwrap();
582    }
583
584    #[tokio::test]
585    async fn linked_log_group_lookup_requests_exact_identifier_and_keeps_metadata() {
586        let name = "/aws/lambda/GrantFn";
587        let uri = "https://logs.us-east-1.amazonaws.com/";
588        let request = serde_json::json!({"logGroupIdentifiers": [name]}).to_string();
589        let http = StaticReplayClient::new(vec![replay(uri, request, 200,
590            r#"{"logGroups":[{"logGroupName":"/aws/lambda/GrantFn","storedBytes":13529,"retentionInDays":30,"creationTime":1700000000000,"logGroupClass":"STANDARD","arn":"arn:aws:logs:us-east-1:123456789012:log-group:/aws/lambda/GrantFn:*","logGroupArn":"arn:aws:logs:us-east-1:123456789012:log-group:/aws/lambda/GrantFn","deletionProtectionEnabled":true}]}"#.into())]);
591        let client = crate::CloudWatchClient::new(config(&http)).await.unwrap();
592        let row = client.get_log_group(name).await.unwrap().unwrap();
593        assert_eq!(row.name, name);
594        assert_eq!(row.stored_bytes, Some(13529));
595        assert_eq!(row.retention_days, Some(30));
596        assert_eq!(row.log_class.as_deref(), Some("STANDARD"));
597        assert!(row.creation_time.is_some());
598        assert!(row.arn.as_deref().unwrap().ends_with(":*"));
599        assert!(row.log_group_arn.as_deref().unwrap().ends_with(name));
600        assert_eq!(row.deletion_protection_enabled, Some(true));
601        http.assert_requests_match(&[]);
602    }
603
604    #[tokio::test]
605    async fn linked_log_group_lookup_does_not_accept_neighbors_or_hide_access_errors() {
606        for (status, response) in [
607            (200, r#"{"logGroups":[]}"#),
608            (
609                200,
610                r#"{"logGroups":[{"logGroupName":"/aws/lambda/GrantFn-old"}]}"#,
611            ),
612            (
613                400,
614                r#"{"__type":"AccessDeniedException","message":"denied"}"#,
615            ),
616        ] {
617            let http = StaticReplayClient::new(vec![replay(
618                "https://logs.us-east-1.amazonaws.com/",
619                r#"{"logGroupIdentifiers":["/aws/lambda/GrantFn"]}"#.into(),
620                status,
621                response.into(),
622            )]);
623            let client = crate::CloudWatchClient::new(config(&http)).await.unwrap();
624            let result = client.get_log_group("/aws/lambda/GrantFn").await;
625            if status == 200 {
626                assert!(result.unwrap().is_none());
627            } else {
628                assert!(result.is_err());
629            }
630            http.assert_requests_match(&[]);
631        }
632        let http = StaticReplayClient::new(vec![]);
633        let client = crate::CloudWatchClient::new(config(&http)).await.unwrap();
634        assert!(client.get_log_group("").await.is_err());
635        http.assert_requests_match(&[]);
636    }
637
638    #[tokio::test]
639    async fn metric_reader_callers_keep_service_specific_requests_and_propagate_errors() {
640        use aws_smithy_http_client::test_util::infallible_client_fn;
641        use std::sync::{Arc, Mutex};
642        let requests = Arc::new(Mutex::new(Vec::new()));
643        let capture = requests.clone();
644        let http = infallible_client_fn(move |request| {
645            capture
646                .lock()
647                .unwrap()
648                .push(request.body().bytes().unwrap().to_vec());
649            // CloudWatch's CBOR response: an empty map represents no datapoints.
650            http::Response::builder()
651                .status(200)
652                .body(vec![0xa0u8])
653                .unwrap()
654        });
655        let mut config = config(&StaticReplayClient::new(vec![]));
656        config.sdk_config = config.sdk_config.to_builder().http_client(http).build();
657        let lambda = crate::LambdaClient::new(config.clone());
658        let ec2 = crate::Ec2Client::new(config.clone());
659        let sqs = crate::SqsClient::new(config.clone());
660        macro_rules! check {
661            ($call:expr, $namespace:literal, $name:literal, $stat:literal) => {{
662                assert!($call.await.unwrap().is_empty());
663                let guard = requests.lock().unwrap();
664                let body = guard.last().unwrap();
665                for token in [$namespace, $name, $stat] {
666                    assert!(
667                        body.windows(token.len()).any(|b| b == token.as_bytes()),
668                        "Missing {token}"
669                    );
670                }
671            }};
672        }
673        check!(
674            lambda.get_invocations_metric("fn", Some("fn:alias")),
675            "AWS/Lambda",
676            "Invocations",
677            "Sum"
678        );
679        assert!(requests.lock().unwrap()[0]
680            .windows(8)
681            .any(|b| b == b"fn:alias"));
682        check!(
683            lambda.get_duration_metric("fn", aws_sdk_cloudwatch::types::Statistic::Minimum),
684            "AWS/Lambda",
685            "Duration",
686            "Minimum"
687        );
688        check!(
689            lambda.get_errors_metric("fn"),
690            "AWS/Lambda",
691            "Errors",
692            "Sum"
693        );
694        check!(
695            lambda.get_throttles_metric("fn"),
696            "AWS/Lambda",
697            "Throttles",
698            "Sum"
699        );
700        check!(
701            lambda.get_concurrent_executions_metric("fn"),
702            "AWS/Lambda",
703            "ConcurrentExecutions",
704            "Maximum"
705        );
706        check!(
707            lambda.get_recursive_invocations_dropped_metric("fn"),
708            "AWS/Lambda",
709            "RecursiveInvocationsDropped",
710            "Sum"
711        );
712        check!(
713            lambda.get_async_event_age_metric("fn", aws_sdk_cloudwatch::types::Statistic::Maximum),
714            "AWS/Lambda",
715            "AsyncEventAge",
716            "Maximum"
717        );
718        check!(
719            lambda.get_async_events_received_metric("fn"),
720            "AWS/Lambda",
721            "AsyncEventsReceived",
722            "Sum"
723        );
724        check!(
725            lambda.get_async_events_dropped_metric("fn"),
726            "AWS/Lambda",
727            "AsyncEventsDropped",
728            "Sum"
729        );
730        check!(
731            lambda.get_destination_delivery_failures_metric("fn"),
732            "AWS/Lambda",
733            "DestinationDeliveryFailures",
734            "Sum"
735        );
736        check!(
737            lambda.get_dead_letter_errors_metric("fn"),
738            "AWS/Lambda",
739            "DeadLetterErrors",
740            "Sum"
741        );
742        check!(
743            lambda.get_iterator_age_metric("fn"),
744            "AWS/Lambda",
745            "IteratorAge",
746            "Maximum"
747        );
748        check!(
749            ec2.get_cpu_metrics("i-test"),
750            "AWS/EC2",
751            "CPUUtilization",
752            "Average"
753        );
754        check!(
755            ec2.get_network_in_metrics("i-test"),
756            "AWS/EC2",
757            "NetworkIn",
758            "Average"
759        );
760        check!(
761            ec2.get_network_out_metrics("i-test"),
762            "AWS/EC2",
763            "NetworkOut",
764            "Average"
765        );
766        check!(
767            ec2.get_network_packets_in_metrics("i-test"),
768            "AWS/EC2",
769            "NetworkPacketsIn",
770            "Average"
771        );
772        check!(
773            ec2.get_network_packets_out_metrics("i-test"),
774            "AWS/EC2",
775            "NetworkPacketsOut",
776            "Average"
777        );
778        check!(
779            ec2.get_metadata_no_token_metrics("i-test"),
780            "AWS/EC2",
781            "MetadataNoToken",
782            "Average"
783        );
784        check!(
785            sqs.get_queue_metrics("queue"),
786            "AWS/SQS",
787            "ApproximateAgeOfOldestMessage",
788            "Maximum"
789        );
790        check!(
791            sqs.get_queue_delayed_metrics("queue"),
792            "AWS/SQS",
793            "ApproximateNumberOfMessagesDelayed",
794            "Average"
795        );
796        check!(
797            sqs.get_queue_not_visible_metrics("queue"),
798            "AWS/SQS",
799            "ApproximateNumberOfMessagesNotVisible",
800            "Average"
801        );
802        check!(
803            sqs.get_queue_visible_metrics("queue"),
804            "AWS/SQS",
805            "ApproximateNumberOfMessagesVisible",
806            "Average"
807        );
808        check!(
809            sqs.get_queue_empty_receives_metrics("queue"),
810            "AWS/SQS",
811            "NumberOfEmptyReceives",
812            "Sum"
813        );
814        check!(
815            sqs.get_queue_messages_deleted_metrics("queue"),
816            "AWS/SQS",
817            "NumberOfMessagesDeleted",
818            "Sum"
819        );
820        check!(
821            sqs.get_queue_messages_received_metrics("queue"),
822            "AWS/SQS",
823            "NumberOfMessagesReceived",
824            "Sum"
825        );
826        check!(
827            sqs.get_queue_messages_sent_metrics("queue"),
828            "AWS/SQS",
829            "NumberOfMessagesSent",
830            "Sum"
831        );
832        check!(
833            sqs.get_queue_sent_message_size_metrics("queue"),
834            "AWS/SQS",
835            "SentMessageSize",
836            "Average"
837        );
838        let alarms = crate::AlarmsClient::new(config.clone());
839        check!(
840            alarms.get_metric_statistics(crate::MetricStatsRequest {
841                namespace: "Custom/Test",
842                metric_name: "Samples",
843                statistic: "SampleCount",
844                period: 120,
845                dimensions: &[("Key".into(), "Value".into())],
846                start_time: chrono::DateTime::from_timestamp(100, 0).unwrap(),
847                end_time: chrono::DateTime::from_timestamp(1000, 0).unwrap(),
848            }),
849            "Custom/Test",
850            "Samples",
851            "SampleCount"
852        );
853        assert_eq!(requests.lock().unwrap().len(), 28);
854        config.sdk_config = config
855            .sdk_config
856            .to_builder()
857            .http_client(infallible_client_fn(|_| {
858                http::Response::builder()
859                    .status(403)
860                    .body(vec![0xa0u8])
861                    .unwrap()
862            }))
863            .build();
864        assert!(crate::LambdaClient::new(config)
865            .get_errors_metric("fn")
866            .await
867            .is_err());
868    }
869
870    #[tokio::test]
871    async fn vpc_full_properties_paginate_related_inventories_and_survive_denied_reads() {
872        use aws_smithy_http_client::test_util::infallible_client_fn;
873        use std::sync::{Arc, Mutex};
874        for deny_routes in [false, true] {
875            let requests = Arc::new(Mutex::new(Vec::new()));
876            let captured = requests.clone();
877            // Dispatch by operation: related inventories are fetched concurrently.
878            let http = infallible_client_fn(move |request| {
879                let body = std::str::from_utf8(request.body().bytes().unwrap())
880                    .unwrap()
881                    .to_string();
882                captured.lock().unwrap().push(body.clone());
883                let (status, response) = if body.contains("Action=DescribeVpcs&") {
884                    (200, "<DescribeVpcsResponse><vpcSet><item><vpcId>vpc-test</vpcId><state>available</state></item></vpcSet></DescribeVpcsResponse>")
885                } else if body.contains("Action=DescribeRouteTables&") {
886                    assert!(body.contains("association.main"));
887                    assert!(body.contains("true"));
888                    if deny_routes {
889                        (403, "<Response><Errors><Error><Code>UnauthorizedOperation</Code><Message>denied</Message></Error></Errors></Response>")
890                    } else if body.contains("NextToken=route2") {
891                        (200, "<DescribeRouteTablesResponse><routeTableSet><item><vpcId>vpc-test</vpcId><routeTableId>rtb-main</routeTableId><associationSet><item><main>true</main></item></associationSet></item></routeTableSet></DescribeRouteTablesResponse>")
892                    } else {
893                        (200, "<DescribeRouteTablesResponse><routeTableSet/><nextToken>route2</nextToken></DescribeRouteTablesResponse>")
894                    }
895                } else if body.contains("Action=DescribeNetworkAcls&") {
896                    assert!(body.contains("default"));
897                    assert!(body.contains("true"));
898                    if body.contains("NextToken=acl2") {
899                        (200, "<DescribeNetworkAclsResponse><networkAclSet><item><vpcId>vpc-test</vpcId><networkAclId>acl-main</networkAclId><default>true</default></item></networkAclSet></DescribeNetworkAclsResponse>")
900                    } else {
901                        (200, "<DescribeNetworkAclsResponse><networkAclSet/><nextToken>acl2</nextToken></DescribeNetworkAclsResponse>")
902                    }
903                } else {
904                    panic!("Unexpected operation: {body}");
905                };
906                http::Response::builder()
907                    .status(status)
908                    .body(response)
909                    .unwrap()
910            });
911            let mut config = config(&StaticReplayClient::new(vec![]));
912            config.sdk_config = config.sdk_config.to_builder().http_client(http).build();
913            let rows = VpcClient::new(config)
914                .list_vpcs_with_properties()
915                .await
916                .unwrap();
917            assert_eq!(rows.len(), 1);
918            assert_eq!(rows[0].vpc_id, "vpc-test");
919            assert_eq!(
920                rows[0].main_route_table,
921                if deny_routes {
922                    "Unavailable"
923                } else {
924                    "rtb-main"
925                }
926            );
927            assert_eq!(rows[0].main_network_acl, "acl-main");
928            assert_eq!(
929                requests.lock().unwrap().len(),
930                if deny_routes { 4 } else { 5 }
931            );
932        }
933    }
934
935    #[tokio::test]
936    async fn vpc_inventory_follows_tokens_and_preserves_owner_metadata() {
937        let uri = "https://ec2.us-east-1.amazonaws.com/";
938        let request = "Action=DescribeVpcs&Version=2016-11-15";
939        let http = StaticReplayClient::new(vec![
940            replay(uri, request.into(), 200,
941                "<DescribeVpcsResponse><vpcSet><item><vpcId>vpc-002</vpcId><state>available</state>\
942                <cidrBlock>172.31.0.0/16</cidrBlock><ownerId>123456789012</ownerId><isDefault>true</isDefault>\
943                <instanceTenancy>default</instanceTenancy><tagSet><item><key>Name</key><value>default-vpc</value></item></tagSet>\
944                </item></vpcSet><nextToken>page2</nextToken></DescribeVpcsResponse>".into()),
945            replay(uri, format!("{request}&NextToken=page2"), 200,
946                "<DescribeVpcsResponse><vpcSet><item><vpcId>vpc-001</vpcId><state>pending</state>\
947                <ownerId>999999999999</ownerId><isDefault>false</isDefault></item></vpcSet></DescribeVpcsResponse>".into()),
948        ]);
949        let rows = VpcClient::new(config(&http)).list_vpcs().await.unwrap();
950        assert_eq!(
951            rows.iter().map(|v| v.vpc_id.as_str()).collect::<Vec<_>>(),
952            ["vpc-001", "vpc-002"]
953        );
954        assert_eq!(
955            rows[0].arn,
956            "arn:aws:ec2:us-east-1:999999999999:vpc/vpc-001"
957        );
958        assert_eq!(rows[1].name, "default-vpc");
959        assert_eq!(rows[1].is_default, Some(true));
960        assert_eq!(rows[1].ipv4_cidrs, "172.31.0.0/16");
961        http.assert_requests_match(&[]);
962    }
963
964    #[tokio::test]
965    async fn vpc_inventory_empty_and_duplicate_tokens_terminate() {
966        let uri = "https://ec2.us-east-1.amazonaws.com/";
967        let request = "Action=DescribeVpcs&Version=2016-11-15";
968        for repeated in [false, true] {
969            let response = format!(
970                "<DescribeVpcsResponse><vpcSet/>{}</DescribeVpcsResponse>",
971                if repeated {
972                    "<nextToken>same</nextToken>"
973                } else {
974                    ""
975                }
976            );
977            let mut events = vec![replay(uri, request.into(), 200, response.clone())];
978            if repeated {
979                events.push(replay(
980                    uri,
981                    format!("{request}&NextToken=same"),
982                    200,
983                    response,
984                ));
985            }
986            let http = StaticReplayClient::new(events);
987            assert!(VpcClient::new(config(&http))
988                .list_vpcs()
989                .await
990                .unwrap()
991                .is_empty());
992            http.assert_requests_match(&[]);
993        }
994    }
995
996    #[tokio::test]
997    async fn vpc_inventory_does_not_report_success_on_failed_later_page() {
998        let uri = "https://ec2.us-east-1.amazonaws.com/";
999        let request = "Action=DescribeVpcs&Version=2016-11-15";
1000        let http = StaticReplayClient::new(vec![
1001            replay(uri, request.into(), 200, "<DescribeVpcsResponse><vpcSet><item><vpcId>vpc-first</vpcId></item></vpcSet><nextToken>page2</nextToken></DescribeVpcsResponse>".into()),
1002            replay(uri, format!("{request}&NextToken=page2"), 403, "<Response><Errors><Error><Code>UnauthorizedOperation</Code><Message>denied</Message></Error></Errors></Response>".into()),
1003        ]);
1004        assert!(VpcClient::new(config(&http)).list_vpcs().await.is_err());
1005        http.assert_requests_match(&[]);
1006    }
1007
1008    #[tokio::test]
1009    async fn finder_indexed_search_streams_pages_and_filters_unsupported_resources() {
1010        use crate::finder::{search, SearchCache, SearchQuery, SearchUpdate};
1011        use std::sync::Arc;
1012        use tokio::sync::{mpsc, Mutex};
1013        let query = "rusticity* resourcetype:dynamodb:table region:us-east-1,global";
1014        let request = serde_json::json!({"QueryString": query, "MaxResults": 100});
1015        let mut next = request.clone();
1016        next["NextToken"] = "page2".into();
1017        let uri = "https://resource-explorer-2.us-east-1.amazonaws.com/Search";
1018        let http = StaticReplayClient::new(vec![
1019            replay(uri, request.to_string(), 200, serde_json::json!({
1020                "Resources": [{"Arn": "arn:aws:dynamodb:us-east-1:123456789012:table/rusticity-orders", "Region": "us-east-1",
1021                    "ResourceType": "dynamodb:table", "Service": "dynamodb", "OwningAccountId": "123456789012"}],
1022                "NextToken": "page2", "Count": {"Complete": false, "TotalResources": 1000}
1023            }).to_string()),
1024            replay(uri, next.to_string(), 200, serde_json::json!({
1025                "Resources": [
1026                    {"Arn": "arn:aws:dynamodb:us-east-1:123456789012:table/rusticity-events", "Region": "us-east-1", "ResourceType": "dynamodb:table", "OwningAccountId": "123456789012"},
1027                    {"Arn": "arn:aws:rds:us-east-1:123456789012:cluster/rusticity", "Region": "us-east-1", "ResourceType": "rds:cluster", "OwningAccountId": "123456789012"},
1028                    {"Arn": "arn:aws:dynamodb:us-east-1:999999999999:table/rusticity-other-account", "Region": "us-east-1", "ResourceType": "dynamodb:table", "OwningAccountId": "999999999999"}
1029                ], "Count": {"Complete": true, "TotalResources": 2}
1030            }).to_string()),
1031        ]);
1032        let mut config = config(&http);
1033        config.account_id = "123456789012".into();
1034        let (tx, mut rx) = mpsc::channel(8);
1035        search(
1036            config,
1037            SearchQuery::parse("@ddb rusticity").unwrap(),
1038            Arc::new(Mutex::new(SearchCache::default())),
1039            tx,
1040        )
1041        .await;
1042        let mut rows = Vec::new();
1043        let mut batches = 0;
1044        let mut finished = false;
1045        while let Some(update) = rx.recv().await {
1046            match update {
1047                SearchUpdate::Results(batch) => {
1048                    batches += 1;
1049                    rows.extend(batch);
1050                }
1051                SearchUpdate::Finished { limited } => {
1052                    assert!(limited);
1053                    finished = true;
1054                }
1055                SearchUpdate::Warning(warning) => panic!("Unexpected warning: {warning}"),
1056                _ => {}
1057            }
1058        }
1059        assert!(finished);
1060        assert_eq!(batches, 2);
1061        assert_eq!(
1062            rows.iter().map(|r| r.id.as_str()).collect::<Vec<_>>(),
1063            ["rusticity-orders", "rusticity-events"]
1064        );
1065        http.assert_requests_match(&[]);
1066    }
1067
1068    #[tokio::test]
1069    async fn finder_missing_view_falls_back_to_paginated_inventory_and_caches_it() {
1070        use crate::finder::{search, SearchCache, SearchQuery, SearchUpdate};
1071        use std::sync::Arc;
1072        use tokio::sync::{mpsc, Mutex};
1073        let request = serde_json::json!({"QueryString": "rusticity* resourcetype:dynamodb:table region:us-east-1,global", "MaxResults": 100});
1074        let http = StaticReplayClient::new(vec![
1075            replay("https://resource-explorer-2.us-east-1.amazonaws.com/Search", request.to_string(), 401, r#"{"message":"No default view"}"#.into()),
1076            replay("https://dynamodb.us-east-1.amazonaws.com/", "{}".into(), 200, r#"{"TableNames":["rusticity-orders"],"LastEvaluatedTableName":"rusticity-orders"}"#.into()),
1077            replay("https://dynamodb.us-east-1.amazonaws.com/", r#"{"ExclusiveStartTableName":"rusticity-orders"}"#.into(), 200, r#"{"TableNames":["rusticity-events","unrelated"]}"#.into()),
1078        ]);
1079        let cache = Arc::new(Mutex::new(SearchCache::default()));
1080        for _ in 0..2 {
1081            let (tx, mut rx) = mpsc::channel(8);
1082            search(
1083                config(&http),
1084                SearchQuery::parse("@dynamodb rusticity").unwrap(),
1085                cache.clone(),
1086                tx,
1087            )
1088            .await;
1089            let mut rows = Vec::new();
1090            while let Some(update) = rx.recv().await {
1091                if let SearchUpdate::Results(batch) = update {
1092                    rows.extend(batch);
1093                }
1094            }
1095            assert_eq!(rows.len(), 2);
1096        }
1097        // The second query uses cached inventory; neither provider makes another request.
1098        assert_eq!(http.actual_requests().count(), 3);
1099        http.assert_requests_match(&[]);
1100    }
1101
1102    #[tokio::test]
1103    async fn rds_optional_reads_are_gated_and_fail_without_hiding_inventory() {
1104        let uri = "https://rds.us-east-1.amazonaws.com/";
1105        let inventory = "<DescribeDBInstancesResponse><DescribeDBInstancesResult><DBInstances><DBInstance>\
1106            <DBInstanceIdentifier>orders</DBInstanceIdentifier><DBInstanceStatus>available</DBInstanceStatus>\
1107            <DBInstanceArn>arn:aws:rds:us-east-1:123456789012:db:orders</DBInstanceArn>\
1108            </DBInstance></DBInstances></DescribeDBInstancesResult></DescribeDBInstancesResponse>";
1109        let denied = "<ErrorResponse><Error><Type>Sender</Type><Code>AccessDenied</Code><Message>denied</Message></Error></ErrorResponse>";
1110        for enabled in [false, true] {
1111            let mut events = vec![replay(
1112                uri,
1113                "Action=DescribeDBInstances&Version=2014-10-31".into(),
1114                200,
1115                inventory.into(),
1116            )];
1117            if enabled {
1118                events.push(replay(
1119                    uri,
1120                    "Action=DescribeDBRecommendations&Version=2014-10-31".into(),
1121                    403,
1122                    denied.into(),
1123                ));
1124                events.push(replay(
1125                    uri,
1126                    "Action=DescribePendingMaintenanceActions&Version=2014-10-31".into(),
1127                    403,
1128                    denied.into(),
1129                ));
1130                // CloudWatch uses a different wire protocol; an HTTP failure is enough
1131                // to verify the optional metric error path, not its serialization.
1132                events.push(replay(
1133                    "https://monitoring.us-east-1.amazonaws.com/",
1134                    String::new(),
1135                    403,
1136                    String::new(),
1137                ));
1138            }
1139            let http = StaticReplayClient::new(events);
1140            let rows = RdsClient::new(config(&http))
1141                .list_instances_with_options(InventoryOptions {
1142                    recommendations: enabled,
1143                    maintenance: enabled,
1144                    cpu: enabled,
1145                    current_activity: enabled,
1146                    ..Default::default()
1147                })
1148                .await
1149                .unwrap();
1150            assert_eq!(rows.len(), 1);
1151            assert_eq!(rows[0].identifier, "orders");
1152            assert_eq!(rows[0].status, "available");
1153            let expected = if enabled { "Unavailable" } else { "" };
1154            assert_eq!(rows[0].recommendations, expected);
1155            assert_eq!(rows[0].maintenance, expected);
1156            assert_eq!(rows[0].cpu, expected);
1157            assert_eq!(rows[0].current_activity, expected);
1158            assert_eq!(http.actual_requests().count(), if enabled { 4 } else { 1 });
1159            if !enabled {
1160                http.assert_requests_match(&[]);
1161            }
1162        }
1163    }
1164
1165    #[tokio::test]
1166    async fn rds_inventory_follows_markers_maps_fields_and_sorts_identifiers() {
1167        let uri = "https://rds.us-east-1.amazonaws.com/";
1168        let body = "Action=DescribeDBInstances&Version=2014-10-31";
1169        let http = StaticReplayClient::new(vec![
1170            replay(uri, body.into(), 200,
1171                "<DescribeDBInstancesResponse><DescribeDBInstancesResult><DBInstances><DBInstance>\
1172                <DBInstanceIdentifier>zeta</DBInstanceIdentifier><DBInstanceStatus>available</DBInstanceStatus>\
1173                <Engine>postgres</Engine><EngineVersion>17.6</EngineVersion><DBInstanceClass>db.t4g.micro</DBInstanceClass>\
1174                <AvailabilityZone>us-east-1a</AvailabilityZone><DBInstanceArn>arn:aws:rds:us-east-1:123456789012:db:zeta</DBInstanceArn>\
1175                <Endpoint><Address>zeta.example.invalid</Address><Port>5432</Port></Endpoint>\
1176                </DBInstance></DBInstances><Marker>page2</Marker></DescribeDBInstancesResult></DescribeDBInstancesResponse>".into()),
1177            replay(uri, format!("{body}&Marker=page2"), 200,
1178                "<DescribeDBInstancesResponse><DescribeDBInstancesResult><DBInstances><DBInstance>\
1179                <DBInstanceIdentifier>alpha</DBInstanceIdentifier><DBInstanceStatus>creating</DBInstanceStatus>\
1180                </DBInstance></DBInstances></DescribeDBInstancesResult></DescribeDBInstancesResponse>".into()),
1181        ]);
1182        let rows = RdsClient::new(config(&http))
1183            .list_instances()
1184            .await
1185            .unwrap();
1186        assert_eq!(
1187            rows.iter()
1188                .map(|i| i.identifier.as_str())
1189                .collect::<Vec<_>>(),
1190            ["alpha", "zeta"]
1191        );
1192        assert!(rows[0].endpoint.is_empty());
1193        assert_eq!(rows[1].endpoint, "zeta.example.invalid");
1194        assert_eq!(rows[1].port, Some(5432));
1195        assert_eq!(rows[1].engine_version, "17.6");
1196        assert_eq!(rows[1].region, "us-east-1");
1197        http.assert_requests_match(&[]);
1198    }
1199
1200    #[tokio::test]
1201    async fn rds_inventory_empty_and_repeated_markers_terminate() {
1202        let uri = "https://rds.us-east-1.amazonaws.com/";
1203        let body = "Action=DescribeDBInstances&Version=2014-10-31";
1204        for repeat in [false, true] {
1205            let response = format!(
1206                "<DescribeDBInstancesResponse><DescribeDBInstancesResult><DBInstances/>{}\
1207                </DescribeDBInstancesResult></DescribeDBInstancesResponse>",
1208                if repeat { "<Marker>same</Marker>" } else { "" }
1209            );
1210            let mut events = vec![replay(uri, body.into(), 200, response.clone())];
1211            if repeat {
1212                events.push(replay(uri, format!("{body}&Marker=same"), 200, response));
1213            }
1214            let http = StaticReplayClient::new(events);
1215            let rows = RdsClient::new(config(&http))
1216                .list_instances()
1217                .await
1218                .unwrap();
1219            assert!(rows.is_empty());
1220            http.assert_requests_match(&[]);
1221        }
1222    }
1223
1224    #[tokio::test]
1225    async fn rds_inventory_propagates_later_page_failure_without_partial_success() {
1226        let uri = "https://rds.us-east-1.amazonaws.com/";
1227        let body = "Action=DescribeDBInstances&Version=2014-10-31";
1228        let http = StaticReplayClient::new(vec![
1229            replay(uri, body.into(), 200,
1230                "<DescribeDBInstancesResponse><DescribeDBInstancesResult><DBInstances><DBInstance>\
1231                <DBInstanceIdentifier>first</DBInstanceIdentifier></DBInstance></DBInstances><Marker>page2</Marker>\
1232                </DescribeDBInstancesResult></DescribeDBInstancesResponse>".into()),
1233            replay(uri, format!("{body}&Marker=page2"), 403,
1234                "<ErrorResponse><Error><Type>Sender</Type><Code>AccessDenied</Code><Message>denied</Message></Error></ErrorResponse>".into()),
1235        ]);
1236        assert!(RdsClient::new(config(&http))
1237            .list_instances()
1238            .await
1239            .is_err());
1240        http.assert_requests_match(&[]);
1241    }
1242
1243    #[tokio::test]
1244    async fn dynamodb_scan_preserves_typed_cursor_and_requests_one_page_at_a_time() {
1245        let uri = "https://dynamodb.us-east-1.amazonaws.com/";
1246        let key = serde_json::json!({
1247            "pk": {"S": "customer"},
1248            "sk": {"N": "123456789012345678901234567890"},
1249            "binary": {"B": "AAEC/w=="},
1250        });
1251        let request = serde_json::json!({
1252            "TableName": "orders", "IndexName": "by-customer", "Limit": 100,
1253            "ProjectionExpression": "#p0, #p1",
1254            "ExpressionAttributeNames": {"#p0": "pk", "#p1": "sk"},
1255        });
1256        let mut next_request = request.clone();
1257        next_request["ExclusiveStartKey"] = key.clone();
1258        let http = StaticReplayClient::new(vec![
1259            replay(
1260                uri,
1261                request.to_string(),
1262                200,
1263                serde_json::json!({
1264                    "Items": [{"pk": {"S": "first"}}], "LastEvaluatedKey": key,
1265                })
1266                .to_string(),
1267            ),
1268            replay(
1269                uri,
1270                next_request.to_string(),
1271                200,
1272                r#"{"Items": [], "LastEvaluatedKey": {}}"#.into(),
1273            ),
1274        ]);
1275        let client = DynamoDbClient::new(config(&http));
1276        let projection = vec!["pk".into(), "sk".into()];
1277        let page = client
1278            .scan_items("orders", Some("by-customer"), 100, Some(&projection), None)
1279            .await
1280            .unwrap();
1281        assert_eq!(http.actual_requests().count(), 1);
1282        assert_eq!(page.items[0].get("pk"), "first");
1283        let cursor = page.next_cursor.unwrap();
1284        assert_eq!(
1285            cursor["sk"].as_n().unwrap(),
1286            "123456789012345678901234567890"
1287        );
1288        assert_eq!(cursor["binary"].as_b().unwrap().as_ref(), &[0, 1, 2, 255]);
1289        let page = client
1290            .scan_items(
1291                "orders",
1292                Some("by-customer"),
1293                100,
1294                Some(&projection),
1295                Some(cursor),
1296            )
1297            .await
1298            .unwrap();
1299        assert!(page.items.is_empty());
1300        assert!(page.next_cursor.is_none());
1301        http.assert_requests_match(&[]);
1302    }
1303
1304    #[tokio::test]
1305    async fn dynamodb_query_continues_after_an_empty_page_without_changing_conditions() {
1306        let uri = "https://dynamodb.us-east-1.amazonaws.com/";
1307        let key = serde_json::json!({"customer_id": {"N": "42"}, "sk": {"S": "last"}});
1308        let request = serde_json::json!({
1309            "TableName": "orders", "IndexName": "by-customer", "Limit": 100,
1310            "KeyConditionExpression": "#pk = :pk",
1311            "ExpressionAttributeNames": {"#pk": "customer_id", "#p0": "customer_id", "#p1": "sk"},
1312            "ExpressionAttributeValues": {":pk": {"N": "42"}},
1313            "ProjectionExpression": "#p0, #p1",
1314        });
1315        let mut next_request = request.clone();
1316        next_request["ExclusiveStartKey"] = key.clone();
1317        let http = StaticReplayClient::new(vec![
1318            replay(
1319                uri,
1320                request.to_string(),
1321                200,
1322                serde_json::json!({"Items": [], "LastEvaluatedKey": key}).to_string(),
1323            ),
1324            replay(
1325                uri,
1326                next_request.to_string(),
1327                200,
1328                r#"{"Items": [{"customer_id": {"N": "42"}, "sk": {"S": "next"}}]}"#.into(),
1329            ),
1330        ]);
1331        let client = DynamoDbClient::new(config(&http));
1332        let projection = vec!["customer_id".into(), "sk".into()];
1333        let page = client
1334            .query_items(
1335                "orders",
1336                Some("by-customer"),
1337                "customer_id",
1338                "42",
1339                "N",
1340                100,
1341                Some(&projection),
1342                None,
1343            )
1344            .await
1345            .unwrap();
1346        assert!(page.items.is_empty());
1347        assert!(page.next_cursor.is_some());
1348        assert_eq!(http.actual_requests().count(), 1);
1349        let page = client
1350            .query_items(
1351                "orders",
1352                Some("by-customer"),
1353                "customer_id",
1354                "42",
1355                "N",
1356                100,
1357                Some(&projection),
1358                page.next_cursor,
1359            )
1360            .await
1361            .unwrap();
1362        assert_eq!(page.items[0].get("sk"), "next");
1363        assert!(page.next_cursor.is_none());
1364        http.assert_requests_match(&[]);
1365    }
1366
1367    #[tokio::test]
1368    async fn iam_paginators_follow_markers_and_collect_items() {
1369        for (operation, collection, name, id) in [
1370            ("ListUsers", "Users", "UserName", "UserId"),
1371            ("ListRoles", "Roles", "RoleName", "RoleId"),
1372            ("ListGroups", "Groups", "GroupName", "GroupId"),
1373        ] {
1374            let page = |item: &str, more: &str| {
1375                format!(
1376                    "<{operation}Response><{operation}Result><{collection}>\
1377                 <member><Path>/</Path><{name}>{item}</{name}><{id}>{item}</{id}>\
1378                 <Arn>arn:aws:iam::123456789012:test/{item}</Arn>\
1379                 <CreateDate>2026-01-01T00:00:00Z</CreateDate></member>\
1380                 </{collection}>{more}</{operation}Result></{operation}Response>"
1381                )
1382            };
1383            let body = format!("Action={operation}&Version=2010-05-08");
1384            let http = StaticReplayClient::new(vec![
1385                replay(
1386                    "https://iam.amazonaws.com/",
1387                    body.clone(),
1388                    200,
1389                    page(
1390                        "first",
1391                        "<IsTruncated>true</IsTruncated><Marker>page2</Marker>",
1392                    ),
1393                ),
1394                replay(
1395                    "https://iam.amazonaws.com/",
1396                    format!("{body}&Marker=page2"),
1397                    200,
1398                    page("second", "<IsTruncated>false</IsTruncated>"),
1399                ),
1400            ]);
1401            let client = crate::iam::IamClient::new(config(&http));
1402            let names: Vec<String> = match operation {
1403                "ListUsers" => client
1404                    .list_users()
1405                    .await
1406                    .unwrap()
1407                    .into_iter()
1408                    .map(|u| u.user_name)
1409                    .collect(),
1410                "ListRoles" => client
1411                    .list_roles()
1412                    .await
1413                    .unwrap()
1414                    .into_iter()
1415                    .map(|r| r.role_name)
1416                    .collect(),
1417                _ => client
1418                    .list_groups()
1419                    .await
1420                    .unwrap()
1421                    .into_iter()
1422                    .map(|g| g.group_name)
1423                    .collect(),
1424            };
1425            assert_eq!(names, ["first", "second"]);
1426            http.assert_requests_match(&[]);
1427        }
1428    }
1429
1430    #[tokio::test]
1431    async fn iam_paginator_stops_on_duplicate_marker_even_for_empty_pages() {
1432        let response = "<ListUsersResponse><ListUsersResult><Users/>\
1433            <IsTruncated>true</IsTruncated><Marker>same</Marker>\
1434            </ListUsersResult></ListUsersResponse>";
1435        let body = "Action=ListUsers&Version=2010-05-08";
1436        let http = StaticReplayClient::new(vec![
1437            replay(
1438                "https://iam.amazonaws.com/",
1439                body.into(),
1440                200,
1441                response.into(),
1442            ),
1443            replay(
1444                "https://iam.amazonaws.com/",
1445                format!("{body}&Marker=same"),
1446                200,
1447                response.into(),
1448            ),
1449        ]);
1450        let client = crate::iam::IamClient::new(config(&http));
1451        assert!(client.list_users().await.unwrap().is_empty());
1452        http.assert_requests_match(&[]);
1453    }
1454
1455    #[tokio::test]
1456    async fn iam_paginator_propagates_errors() {
1457        let http = StaticReplayClient::new(vec![replay(
1458            "https://iam.amazonaws.com/",
1459            "Action=ListUsers&Version=2010-05-08".into(),
1460            403,
1461            "<ErrorResponse><Error><Type>Sender</Type><Code>AccessDenied</Code>\
1462             <Message>denied</Message></Error></ErrorResponse>"
1463                .into(),
1464        )]);
1465        let client = crate::iam::IamClient::new(config(&http));
1466        assert!(client
1467            .list_users()
1468            .await
1469            .unwrap_err()
1470            .starts_with("Failed to list users:"));
1471        http.assert_requests_match(&[]);
1472    }
1473
1474    #[tokio::test]
1475    async fn log_stream_paginator_keeps_order_and_stops_at_existing_limit() {
1476        let page = |start: usize, token: &str| {
1477            serde_json::json!({
1478                "logStreams": (start..start + 50).map(|i| serde_json::json!({
1479                    "logStreamName": format!("stream-{i}"), "creationTime": 0,
1480                    "lastEventTimestamp": 1000
1481                })).collect::<Vec<_>>(),
1482                "nextToken": token,
1483            })
1484            .to_string()
1485        };
1486        let uri = "https://logs.us-east-1.amazonaws.com/";
1487        let http = StaticReplayClient::new(vec![
1488            replay(uri, r#"{"logGroupName":"group","orderBy":"LastEventTime","descending":true}"#.into(), 200, page(0, "page2")),
1489            replay(uri, r#"{"logGroupName":"group","orderBy":"LastEventTime","descending":true,"nextToken":"page2"}"#.into(), 200, page(50, "page3")),
1490        ]);
1491        let client = crate::cw::CloudWatchClient::new(config(&http))
1492            .await
1493            .unwrap();
1494        let streams = client.list_log_streams("group").await.unwrap();
1495        assert_eq!(streams.len(), 100);
1496        assert_eq!(streams[0].name, "stream-0");
1497        assert_eq!(streams[99].name, "stream-99");
1498        assert_eq!(streams[0].last_event_time.unwrap().timestamp(), 1);
1499        // A third request would fail the replay, even though page2 has a token.
1500        http.assert_requests_match(&[]);
1501    }
1502}
rusticity_core/config.rs

rusticity_core/
config.rs
