fakecloud-rds 0.21.1

use std::collections::HashMap;
use std::sync::Arc;

use bytes::Bytes;
use chrono::Utc;
use http::{HeaderMap, Method};
use parking_lot::RwLock;
use uuid::Uuid;

use super::{
    build_restored_instance, db_instance_xml, default_db_name, default_parameter_group,
    default_port_for_engine, filter_engine_versions, filter_orderable_options,
    license_model_for_engine, merge_tags, optional_i32_param, parse_tag_keys, parse_tags,
    save_snapshot_static, validate_create_request, RdsService, RdsSourceType,
};
use crate::state::{
    default_engine_versions, default_orderable_options, DbInstance, RdsSnapshot, RdsTag,
    SharedRdsState, RDS_SNAPSHOT_SCHEMA_VERSION,
};
use fakecloud_core::delivery::DeliveryBus;
use fakecloud_core::service::{AwsRequest, AwsService, AwsServiceError};
use fakecloud_persistence::{DiskSnapshotStore, SnapshotStore};
use tokio::sync::Mutex as AsyncMutex;

#[test]
fn default_port_matches_aws_for_each_engine() {
    assert_eq!(default_port_for_engine("postgres"), 5432);
    assert_eq!(default_port_for_engine("mysql"), 3306);
    assert_eq!(default_port_for_engine("mariadb"), 3306);
    assert_eq!(default_port_for_engine("oracle-ee"), 1521);
    assert_eq!(default_port_for_engine("oracle-se2"), 1521);
    assert_eq!(default_port_for_engine("sqlserver-ee"), 1433);
    assert_eq!(default_port_for_engine("sqlserver-ex"), 1433);
    assert_eq!(default_port_for_engine("db2-se"), 50000);
    assert_eq!(default_port_for_engine("db2-ae"), 50000);
}

#[test]
fn default_parameter_group_uses_engine_major_version() {
    assert_eq!(
        default_parameter_group("postgres", "16.3"),
        "default.postgres16"
    );
    assert_eq!(
        default_parameter_group("mysql", "8.0.35"),
        "default.mysql8.0"
    );
    assert_eq!(
        default_parameter_group("oracle-ee", "23.0.0"),
        "default.oracle-ee-23"
    );
    assert_eq!(
        default_parameter_group("sqlserver-ex", "16.00.4085.2.v1"),
        "default.sqlserver-ex-16"
    );
    assert_eq!(
        default_parameter_group("db2-se", "11.5.9.0.sb00000000.r1"),
        "default.db2-se-11.5"
    );
}

#[test]
fn license_model_reflects_engine_class() {
    assert_eq!(license_model_for_engine("postgres"), "postgresql-license");
    assert_eq!(license_model_for_engine("mysql"), "general-public-license");
    assert_eq!(license_model_for_engine("oracle-ee"), "license-included");
    assert_eq!(license_model_for_engine("sqlserver-se"), "license-included");
    assert_eq!(license_model_for_engine("db2-ae"), "bring-your-own-license");
}

#[test]
fn default_db_name_picks_per_engine_default() {
    assert_eq!(default_db_name("postgres"), "postgres");
    assert_eq!(default_db_name("mysql"), "mysql");
    assert_eq!(default_db_name("oracle-ee"), "ORCL");
    assert_eq!(default_db_name("sqlserver-ex"), "master");
    assert_eq!(default_db_name("db2-se"), "BLUDB");
}

#[test]
fn validate_create_request_accepts_new_engines() {
    for (engine, version, port) in [
        ("oracle-ee", "23.0.0", 1521),
        ("sqlserver-ex", "16.00.4085.2.v1", 1433),
        ("db2-se", "11.5.9.0.sb00000000.r1", 50000),
    ] {
        validate_create_request("test-db", 20, "db.t3.micro", engine, version, port)
            .expect("engine should be accepted");
    }
}

#[test]
fn validate_create_request_rejects_unsupported_engine_version() {
    let err = validate_create_request("test-db", 20, "db.t3.micro", "oracle-ee", "12.0.0", 1521)
        .expect_err("12.x is not in the supported list");
    let msg = format!("{err:?}");
    assert!(msg.contains("EngineVersion"), "unexpected: {msg}");
}

#[test]
fn filter_engine_versions_matches_requested_engine() {
    let versions = default_engine_versions();

    let filtered = filter_engine_versions(&versions, &Some("postgres".to_string()), &None, &None);

    assert_eq!(filtered.len(), 4); // All postgres versions
    assert!(filtered.iter().all(|v| v.engine == "postgres"));
}

#[test]
fn filter_orderable_options_respects_instance_class() {
    let options = default_orderable_options();

    let filtered = filter_orderable_options(
        &options,
        &Some("postgres".to_string()),
        &Some("16.3".to_string()),
        &Some("db.t3.micro".to_string()),
        &None,
        Some(true),
    );

    assert_eq!(filtered.len(), 1);
    assert_eq!(filtered[0].db_instance_class, "db.t3.micro");
}

#[test]
fn validate_create_request_rejects_unsupported_engine() {
    let error = validate_create_request("test-db", 20, "db.t3.micro", "mysql", "16.3", 5432)
        .expect_err("unsupported engine");

    assert_eq!(error.code(), "InsufficientDBInstanceCapacity");
}

#[test]
fn optional_i32_param_rejects_invalid_integer() {
    let request = request("CreateDBInstance", &[("Port", "not-a-number")]);

    let error = optional_i32_param(&request, "Port").expect_err("invalid port");

    assert_eq!(error.code(), "InvalidParameterValue");
}

#[test]
fn db_instance_xml_renders_endpoint_and_status() {
    let created_at = Utc::now();
    let instance = DbInstance {
        db_instance_identifier: "test-db".to_string(),
        db_instance_arn: "arn:aws:rds:us-east-1:123456789012:db:test-db".to_string(),
        db_instance_class: "db.t3.micro".to_string(),
        engine: "postgres".to_string(),
        engine_version: "16.3".to_string(),
        db_instance_status: "available".to_string(),
        master_username: "admin".to_string(),
        db_name: Some("appdb".to_string()),
        endpoint_address: "127.0.0.1".to_string(),
        port: 15432,
        allocated_storage: 20,
        publicly_accessible: true,
        deletion_protection: false,
        created_at,
        dbi_resource_id: format!("db-{}", Uuid::new_v4().simple()),
        master_user_password: "secret123".to_string(),
        container_id: "container".to_string(),
        host_port: 15432,
        tags: Vec::new(),
        read_replica_source_db_instance_identifier: None,
        read_replica_db_instance_identifiers: Vec::new(),
        vpc_security_group_ids: vec!["sg-12345678".to_string()],
        db_parameter_group_name: Some("default.postgres16".to_string()),
        backup_retention_period: 1,
        preferred_backup_window: "03:00-04:00".to_string(),
        preferred_maintenance_window: None,
        latest_restorable_time: Some(created_at),
        option_group_name: None,
        multi_az: false,
        pending_modified_values: None,
        availability_zone: None,
        storage_type: None,
        storage_encrypted: false,
        kms_key_id: None,
        iam_database_authentication_enabled: false,
        iops: None,
        monitoring_interval: None,
        monitoring_role_arn: None,
        performance_insights_enabled: false,
        performance_insights_kms_key_id: None,
        performance_insights_retention_period: None,
        enabled_cloudwatch_logs_exports: Vec::new(),
        ca_certificate_identifier: None,
        network_type: None,
        character_set_name: None,
        auto_minor_version_upgrade: None,
        copy_tags_to_snapshot: None,
        master_user_secret_arn: None,
        master_user_secret_kms_key_id: None,
        license_model: None,
        max_allocated_storage: None,
        multi_tenant: None,
        storage_throughput: None,
        tde_credential_arn: None,
        delete_automated_backups: None,
        db_security_groups: Vec::new(),
        domain: None,
        domain_fqdn: None,
        domain_ou: None,
        domain_iam_role_name: None,
        domain_auth_secret_arn: None,
        domain_dns_ips: Vec::new(),
        db_cluster_identifier: None,
    };

    let xml = db_instance_xml(&instance, Some("creating"));

    assert!(xml.contains("<DBInstanceIdentifier>test-db</DBInstanceIdentifier>"));
    assert!(xml.contains("<DBInstanceStatus>creating</DBInstanceStatus>"));
    assert!(xml.contains("<Address>127.0.0.1</Address><Port>15432</Port>"));
    // Fields AWS always returns and SDKs deserialize unconditionally.
    assert!(
        xml.contains("<IAMDatabaseAuthenticationEnabled>false</IAMDatabaseAuthenticationEnabled>")
    );
    assert!(xml.contains("<PerformanceInsightsEnabled>false</PerformanceInsightsEnabled>"));
    assert!(xml.contains("<EnabledCloudwatchLogsExports/>"));
    assert!(xml.contains("<ProcessorFeatures/>"));
    assert!(xml.contains("<ActivityStreamStatus>stopped</ActivityStreamStatus>"));
    assert!(xml.contains("<StorageEncrypted>false</StorageEncrypted>"));
}

#[test]
fn db_instance_xml_renders_dynamic_storage_and_kms() {
    let mut instance = make_instance_with_defaults("dyn");
    instance.availability_zone = Some("eu-west-1c".to_string());
    instance.storage_type = Some("gp3".to_string());
    instance.storage_encrypted = true;
    instance.kms_key_id = Some("arn:aws:kms:us-east-1:123456789012:key/abc".to_string());
    instance.iam_database_authentication_enabled = true;
    instance.iops = Some(3000);
    instance.monitoring_interval = Some(60);
    instance.monitoring_role_arn = Some("arn:aws:iam::123456789012:role/rds-monitor".to_string());
    instance.performance_insights_enabled = true;
    instance.performance_insights_retention_period = Some(7);
    instance.enabled_cloudwatch_logs_exports = vec!["error".to_string(), "general".to_string()];
    instance.ca_certificate_identifier = Some("rds-ca-rsa2048-g1".to_string());
    instance.network_type = Some("DUAL".to_string());
    instance.master_user_secret_arn =
        Some("arn:aws:secretsmanager:us-east-1:123:secret:rds!sec-abc".to_string());
    instance.master_user_secret_kms_key_id =
        Some("arn:aws:kms:us-east-1:123:key/aws/secretsmanager".to_string());

    let xml = db_instance_xml(&instance, None);

    assert!(xml.contains("<AvailabilityZone>eu-west-1c</AvailabilityZone>"));
    assert!(xml.contains("<StorageType>gp3</StorageType>"));
    assert!(xml.contains("<StorageEncrypted>true</StorageEncrypted>"));
    assert!(xml.contains("<KmsKeyId>arn:aws:kms:us-east-1:123456789012:key/abc</KmsKeyId>"));
    assert!(
        xml.contains("<IAMDatabaseAuthenticationEnabled>true</IAMDatabaseAuthenticationEnabled>")
    );
    assert!(xml.contains("<Iops>3000</Iops>"));
    assert!(xml.contains("<MonitoringInterval>60</MonitoringInterval>"));
    assert!(xml.contains("<EnhancedMonitoringResourceArn>arn:aws:iam::123456789012:role/rds-monitor</EnhancedMonitoringResourceArn>"));
    assert!(xml.contains("<PerformanceInsightsEnabled>true</PerformanceInsightsEnabled>"));
    assert!(
        xml.contains("<PerformanceInsightsRetentionPeriod>7</PerformanceInsightsRetentionPeriod>")
    );
    assert!(xml.contains("<EnabledCloudwatchLogsExports><member>error</member><member>general</member></EnabledCloudwatchLogsExports>"));
    assert!(xml.contains("<CACertificateIdentifier>rds-ca-rsa2048-g1</CACertificateIdentifier>"));
    assert!(xml.contains("<NetworkType>DUAL</NetworkType>"));
    assert!(xml.contains("<MasterUserSecret>"));
    assert!(xml.contains("<SecretStatus>active</SecretStatus>"));
}

#[test]
fn db_snapshot_xml_emits_extended_fields() {
    use super::db_snapshot_xml;
    let snapshot = crate::state::DbSnapshot {
        db_snapshot_identifier: "snap-1".to_string(),
        db_snapshot_arn: "arn:aws:rds:us-east-1:123:snapshot:snap-1".to_string(),
        db_instance_identifier: "src-db".to_string(),
        snapshot_create_time: Utc::now(),
        engine: "postgres".to_string(),
        engine_version: "16.3".to_string(),
        allocated_storage: 20,
        status: "available".to_string(),
        port: 5432,
        master_username: "admin".to_string(),
        db_name: Some("appdb".to_string()),
        dbi_resource_id: "db-rid".to_string(),
        snapshot_type: "manual".to_string(),
        master_user_password: "secret".to_string(),
        tags: Vec::new(),
        dump_data: Vec::new(),
        availability_zone: Some("us-east-1a".to_string()),
        vpc_id: Some("vpc-abc".to_string()),
        instance_create_time: Some(Utc::now()),
        license_model: Some("postgresql-license".to_string()),
        iops: Some(3000),
        option_group_name: Some("default:postgres-16".to_string()),
        percent_progress: Some(100),
        storage_type: Some("gp3".to_string()),
        encrypted: true,
        kms_key_id: Some("arn:aws:kms:us-east-1:123:key/abc".to_string()),
        iam_database_authentication_enabled: true,
        timezone: None,
        storage_throughput: Some(125),
    };

    let xml = db_snapshot_xml(&snapshot);

    assert!(xml.contains("<AvailabilityZone>us-east-1a</AvailabilityZone>"));
    assert!(xml.contains("<VpcId>vpc-abc</VpcId>"));
    assert!(xml.contains("<InstanceCreateTime>"));
    assert!(xml.contains("<LicenseModel>postgresql-license</LicenseModel>"));
    assert!(xml.contains("<Iops>3000</Iops>"));
    assert!(xml.contains("<OptionGroupName>default:postgres-16</OptionGroupName>"));
    assert!(xml.contains("<PercentProgress>100</PercentProgress>"));
    assert!(xml.contains("<StorageType>gp3</StorageType>"));
    assert!(xml.contains("<Encrypted>true</Encrypted>"));
    assert!(xml.contains("<KmsKeyId>arn:aws:kms:us-east-1:123:key/abc</KmsKeyId>"));
    assert!(
        xml.contains("<IAMDatabaseAuthenticationEnabled>true</IAMDatabaseAuthenticationEnabled>")
    );
    assert!(xml.contains("<StorageThroughput>125</StorageThroughput>"));
    assert!(xml.contains("<ProcessorFeatures/>"));
}

fn make_instance_with_defaults(id: &str) -> DbInstance {
    let created_at = Utc::now();
    DbInstance {
        db_instance_identifier: id.to_string(),
        db_instance_arn: format!("arn:aws:rds:us-east-1:123:db:{id}"),
        db_instance_class: "db.t3.micro".to_string(),
        engine: "postgres".to_string(),
        engine_version: "16.3".to_string(),
        db_instance_status: "available".to_string(),
        master_username: "admin".to_string(),
        db_name: None,
        endpoint_address: "127.0.0.1".to_string(),
        port: 5432,
        allocated_storage: 20,
        publicly_accessible: true,
        deletion_protection: false,
        created_at,
        dbi_resource_id: format!("db-{}", Uuid::new_v4().simple()),
        master_user_password: "p".to_string(),
        container_id: "c".to_string(),
        host_port: 0,
        tags: Vec::new(),
        read_replica_source_db_instance_identifier: None,
        read_replica_db_instance_identifiers: Vec::new(),
        vpc_security_group_ids: Vec::new(),
        db_parameter_group_name: None,
        backup_retention_period: 0,
        preferred_backup_window: String::new(),
        preferred_maintenance_window: None,
        latest_restorable_time: None,
        option_group_name: None,
        multi_az: false,
        pending_modified_values: None,
        availability_zone: None,
        storage_type: None,
        storage_encrypted: false,
        kms_key_id: None,
        iam_database_authentication_enabled: false,
        iops: None,
        monitoring_interval: None,
        monitoring_role_arn: None,
        performance_insights_enabled: false,
        performance_insights_kms_key_id: None,
        performance_insights_retention_period: None,
        enabled_cloudwatch_logs_exports: Vec::new(),
        ca_certificate_identifier: None,
        network_type: None,
        character_set_name: None,
        auto_minor_version_upgrade: None,
        copy_tags_to_snapshot: None,
        master_user_secret_arn: None,
        master_user_secret_kms_key_id: None,
        license_model: None,
        max_allocated_storage: None,
        multi_tenant: None,
        storage_throughput: None,
        tde_credential_arn: None,
        delete_automated_backups: None,
        db_security_groups: Vec::new(),
        domain: None,
        domain_fqdn: None,
        domain_ou: None,
        domain_iam_role_name: None,
        domain_auth_secret_arn: None,
        domain_dns_ips: Vec::new(),
        db_cluster_identifier: None,
    }
}

#[test]
fn parse_tags_reads_rds_query_shape() {
    let request = request(
        "AddTagsToResource",
        &[
            ("Tags.Tag.1.Key", "env"),
            ("Tags.Tag.1.Value", "dev"),
            ("Tags.Tag.2.Key", "team"),
            ("Tags.Tag.2.Value", "core"),
        ],
    );

    let tags = parse_tags(&request).expect("tags");

    assert_eq!(
        tags,
        vec![
            RdsTag {
                key: "env".to_string(),
                value: "dev".to_string(),
            },
            RdsTag {
                key: "team".to_string(),
                value: "core".to_string(),
            }
        ]
    );
}

#[test]
fn parse_tag_keys_reads_member_shape() {
    let request = request(
        "RemoveTagsFromResource",
        &[("TagKeys.member.1", "env"), ("TagKeys.member.2", "team")],
    );

    let tag_keys = parse_tag_keys(&request).expect("tag keys");

    assert_eq!(tag_keys, vec!["env".to_string(), "team".to_string()]);
}

#[test]
fn merge_tags_updates_existing_values() {
    let mut tags = vec![RdsTag {
        key: "env".to_string(),
        value: "dev".to_string(),
    }];

    merge_tags(
        &mut tags,
        &[
            RdsTag {
                key: "env".to_string(),
                value: "prod".to_string(),
            },
            RdsTag {
                key: "team".to_string(),
                value: "core".to_string(),
            },
        ],
    );

    assert_eq!(tags.len(), 2);
    assert_eq!(tags[0].value, "prod");
    assert_eq!(tags[1].key, "team");
}

#[tokio::test]
async fn describe_engine_versions_returns_xml_body() {
    let service = RdsService::new(Arc::new(RwLock::new(
        fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
    )));
    let request = request("DescribeDBEngineVersions", &[("Engine", "postgres")]);

    let response = service.handle(request).await.expect("response");
    let body = String::from_utf8(response.body.expect_bytes().to_vec()).expect("utf8");

    assert!(body.contains("<DescribeDBEngineVersionsResponse"));
    assert!(body.contains("<Engine>postgres</Engine>"));
    assert!(body.contains("<DBParameterGroupFamily>postgres16</DBParameterGroupFamily>"));
}

fn request(action: &str, params: &[(&str, &str)]) -> AwsRequest {
    let mut query_params = HashMap::from([("Action".to_string(), action.to_string())]);
    for (key, value) in params {
        query_params.insert((*key).to_string(), (*value).to_string());
    }

    AwsRequest {
        service: "rds".to_string(),
        action: action.to_string(),
        region: "us-east-1".to_string(),
        account_id: "123456789012".to_string(),
        request_id: "test-request-id".to_string(),
        headers: HeaderMap::new(),
        query_params,
        body: Bytes::new(),
        body_stream: parking_lot::Mutex::new(None),
        path_segments: vec![],
        raw_path: "/".to_string(),
        raw_query: String::new(),
        method: Method::POST,
        is_query_protocol: true,
        access_key_id: None,
        principal: None,
    }
}

// ── Helpers for handler tests ────────────────────────────────────

fn make_service() -> RdsService {
    RdsService::new(Arc::new(RwLock::new(
        fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
    )))
}

#[derive(Default)]
struct CapturedEvent {
    source: String,
    detail_type: String,
    detail: String,
}

#[derive(Default)]
struct RecordingEb {
    events: std::sync::Mutex<Vec<CapturedEvent>>,
}

impl fakecloud_core::delivery::EventBridgeDelivery for RecordingEb {
    fn put_event(&self, source: &str, detail_type: &str, detail: &str, _bus: &str) {
        self.events.lock().unwrap().push(CapturedEvent {
            source: source.to_string(),
            detail_type: detail_type.to_string(),
            detail: detail.to_string(),
        });
    }
}

fn make_service_with_recorder() -> (RdsService, Arc<RecordingEb>) {
    let recorder = Arc::new(RecordingEb::default());
    let bus = Arc::new(DeliveryBus::new().with_eventbridge(recorder.clone()));
    let svc = RdsService::new(Arc::new(RwLock::new(
        fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
    )))
    .with_delivery_bus(bus);
    (svc, recorder)
}

#[test]
fn emit_event_emits_aws_rds_event_via_bus() {
    let (svc, rec) = make_service_with_recorder();
    svc.emit_event(
        RdsSourceType::DbInstance,
        "my-db",
        "arn:aws:rds:us-east-1:123456789012:db:my-db",
        "RDS-EVENT-0005",
        &["creation"],
        "DB instance created",
    );
    let events = rec.events.lock().unwrap();
    assert_eq!(events.len(), 1);
    let e = &events[0];
    assert_eq!(e.source, "aws.rds");
    assert_eq!(e.detail_type, "RDS DB Instance Event");
    let detail: serde_json::Value = serde_json::from_str(&e.detail).unwrap();
    assert_eq!(detail["EventID"], "RDS-EVENT-0005");
    assert_eq!(detail["SourceType"], "DB_INSTANCE");
    assert_eq!(detail["SourceIdentifier"], "my-db");
    assert_eq!(detail["Message"], "DB instance created");
    assert_eq!(detail["EventCategories"][0], "creation");
}

#[test]
fn emit_event_no_op_without_bus() {
    let svc = make_service();
    svc.emit_event(
        RdsSourceType::DbSnapshot,
        "snap",
        "arn:aws:rds:us-east-1:123456789012:snapshot:snap",
        "RDS-EVENT-0042",
        &["creation"],
        "Manual snapshot created",
    );
}

#[test]
fn rds_source_type_detail_type_mapping() {
    assert_eq!(
        RdsSourceType::DbInstance.detail_type(),
        "RDS DB Instance Event"
    );
    assert_eq!(
        RdsSourceType::DbSnapshot.detail_type(),
        "RDS DB Snapshot Event"
    );
    assert_eq!(
        RdsSourceType::DbParameterGroup.detail_type(),
        "RDS DB Parameter Group Event"
    );
}

fn body_of(resp: fakecloud_core::service::AwsResponse) -> String {
    String::from_utf8(resp.body.expect_bytes().to_vec()).expect("utf8")
}

fn seed_instance(svc: &RdsService, identifier: &str) -> String {
    let arn = format!("arn:aws:rds:us-east-1:123456789012:db:{identifier}");
    let mut accounts = svc.state.write();
    let state = accounts.default_mut();
    state.instances.insert(
        identifier.to_string(),
        DbInstance {
            db_instance_identifier: identifier.to_string(),
            db_instance_arn: arn.clone(),
            db_instance_class: "db.t3.micro".to_string(),
            engine: "postgres".to_string(),
            engine_version: "16.3".to_string(),
            db_instance_status: "available".to_string(),
            master_username: "admin".to_string(),
            db_name: Some("appdb".to_string()),
            endpoint_address: "127.0.0.1".to_string(),
            port: 15432,
            allocated_storage: 20,
            publicly_accessible: true,
            deletion_protection: false,
            created_at: Utc::now(),
            dbi_resource_id: format!("db-{}", Uuid::new_v4().simple()),
            master_user_password: "secret".to_string(),
            container_id: "container".to_string(),
            host_port: 15432,
            tags: Vec::new(),
            read_replica_source_db_instance_identifier: None,
            read_replica_db_instance_identifiers: Vec::new(),
            vpc_security_group_ids: vec!["sg-12345678".to_string()],
            db_parameter_group_name: Some("default.postgres16".to_string()),
            backup_retention_period: 1,
            preferred_backup_window: "03:00-04:00".to_string(),
            preferred_maintenance_window: None,
            latest_restorable_time: None,
            option_group_name: None,
            multi_az: false,
            pending_modified_values: None,
            availability_zone: None,
            storage_type: None,
            storage_encrypted: false,
            kms_key_id: None,
            iam_database_authentication_enabled: false,
            iops: None,
            monitoring_interval: None,
            monitoring_role_arn: None,
            performance_insights_enabled: false,
            performance_insights_kms_key_id: None,
            performance_insights_retention_period: None,
            enabled_cloudwatch_logs_exports: Vec::new(),
            ca_certificate_identifier: None,
            network_type: None,
            character_set_name: None,
            auto_minor_version_upgrade: None,
            copy_tags_to_snapshot: None,
            master_user_secret_arn: None,
            master_user_secret_kms_key_id: None,
            license_model: None,
            max_allocated_storage: None,
            multi_tenant: None,
            storage_throughput: None,
            tde_credential_arn: None,
            delete_automated_backups: None,
            db_security_groups: Vec::new(),
            domain: None,
            domain_fqdn: None,
            domain_ou: None,
            domain_iam_role_name: None,
            domain_auth_secret_arn: None,
            domain_dns_ips: Vec::new(),
            db_cluster_identifier: None,
        },
    );
    arn
}

fn assert_code<T>(result: Result<T, AwsServiceError>, expected_code: &str) -> AwsServiceError {
    match result {
        Ok(_) => panic!("expected error {expected_code}, got Ok"),
        Err(e) => {
            assert_eq!(e.code(), expected_code, "wrong error code");
            e
        }
    }
}

// ── Tag operations ───────────────────────────────────────────────

#[test]
fn add_tags_requires_resource_name() {
    let svc = make_service();
    let req = request("AddTagsToResource", &[]);
    assert_code(svc.add_tags_to_resource(&req), "MissingParameter");
}

#[test]
fn add_tags_with_no_tag_keys_is_a_noop() {
    // AWS RDS' Smithy model declares no `MissingParameter` analogue on
    // AddTagsToResource, so we accept an empty Tags list as a no-op
    // rather than emit an undeclared error code the strict conformance
    // probe would reject.
    let svc = make_service();
    let arn = seed_instance(&svc, "db1");
    let req = request("AddTagsToResource", &[("ResourceName", arn.as_str())]);
    svc.add_tags_to_resource(&req).expect("noop ok");
}

#[test]
fn add_tags_appends_then_list_tags_returns_them() {
    let svc = make_service();
    let arn = seed_instance(&svc, "db1");
    let add_req = request(
        "AddTagsToResource",
        &[
            ("ResourceName", arn.as_str()),
            ("Tags.Tag.1.Key", "env"),
            ("Tags.Tag.1.Value", "dev"),
        ],
    );
    svc.add_tags_to_resource(&add_req).unwrap();

    let list_req = request("ListTagsForResource", &[("ResourceName", arn.as_str())]);
    let body = body_of(svc.list_tags_for_resource(&list_req).unwrap());
    assert!(body.contains("<Key>env</Key>"));
    assert!(body.contains("<Value>dev</Value>"));
}

#[test]
fn list_tags_ignores_unsupported_filters_param() {
    // Smithy doesn't declare an "unsupported filter" error on
    // ListTagsForResource. Real AWS silently ignores unknown filters
    // and so do we, returning the tag list as if no filter was set.
    let svc = make_service();
    let arn = seed_instance(&svc, "db1");
    let req = request(
        "ListTagsForResource",
        &[
            ("ResourceName", arn.as_str()),
            ("Filters.Filter.1.Name", "x"),
        ],
    );
    svc.list_tags_for_resource(&req).expect("filters ignored");
}

#[test]
fn list_tags_missing_db_instance_returns_typed_not_found() {
    let svc = make_service();
    let req = request(
        "ListTagsForResource",
        &[("ResourceName", "arn:aws:rds:us-east-1:123456789012:db:nope")],
    );
    assert_code(svc.list_tags_for_resource(&req), "DBInstanceNotFound");
}

#[test]
fn list_tags_unknown_arn_resource_type_errors() {
    let svc = make_service();
    let req = request(
        "ListTagsForResource",
        &[(
            "ResourceName",
            "arn:aws:rds:us-east-1:123456789012:bogus:nope",
        )],
    );
    assert_code(svc.list_tags_for_resource(&req), "DBInstanceNotFound");
}

#[test]
fn list_tags_malformed_arn_errors() {
    let svc = make_service();
    let req = request(
        "ListTagsForResource",
        &[("ResourceName", "not-even-an-arn")],
    );
    assert_code(svc.list_tags_for_resource(&req), "DBInstanceNotFound");
}

#[test]
fn add_tags_to_snapshot_arn_persists() {
    let svc = make_service();
    seed_snapshot(&svc, "snap-1", "db1");
    let arn = {
        let __a = svc.state.read();
        __a.default_ref()
            .snapshots
            .get("snap-1")
            .unwrap()
            .db_snapshot_arn
            .clone()
    };
    let req = request(
        "AddTagsToResource",
        &[
            ("ResourceName", arn.as_str()),
            ("Tags.Tag.1.Key", "team"),
            ("Tags.Tag.1.Value", "platform"),
        ],
    );
    svc.add_tags_to_resource(&req).unwrap();
    let __a = svc.state.read();
    let snap = __a.default_ref().snapshots.get("snap-1").unwrap();
    assert_eq!(snap.tags.len(), 1);
    assert_eq!(snap.tags[0].key, "team");
    assert_eq!(snap.tags[0].value, "platform");
}

#[test]
fn add_tags_to_parameter_group_arn_persists_and_lists() {
    let svc = make_service();
    create_param_group(&svc, "pg1");
    let arn = {
        let __a = svc.state.read();
        __a.default_ref()
            .parameter_groups
            .get("pg1")
            .unwrap()
            .db_parameter_group_arn
            .clone()
    };
    let req = request(
        "AddTagsToResource",
        &[
            ("ResourceName", arn.as_str()),
            ("Tags.Tag.1.Key", "env"),
            ("Tags.Tag.1.Value", "prod"),
        ],
    );
    svc.add_tags_to_resource(&req).unwrap();

    let req = request("ListTagsForResource", &[("ResourceName", arn.as_str())]);
    let resp = svc.list_tags_for_resource(&req).unwrap();
    let body = String::from_utf8(resp.body.expect_bytes().to_vec()).unwrap();
    assert!(body.contains("<Key>env</Key>"));
    assert!(body.contains("<Value>prod</Value>"));
}

#[test]
fn add_tags_to_subnet_group_arn_persists() {
    let svc = make_service();
    let arn = {
        let mut __a = svc.state.write();
        let state = __a.default_mut();
        let arn = state.db_subnet_group_arn("sg1");
        state.subnet_groups.insert(
            "sg1".to_string(),
            crate::state::DbSubnetGroup {
                db_subnet_group_name: "sg1".to_string(),
                db_subnet_group_arn: arn.clone(),
                db_subnet_group_description: "desc".to_string(),
                vpc_id: "vpc-1".to_string(),
                subnet_ids: Vec::new(),
                subnet_availability_zones: Vec::new(),
                tags: Vec::new(),
            },
        );
        arn
    };
    let req = request(
        "AddTagsToResource",
        &[
            ("ResourceName", arn.as_str()),
            ("Tags.Tag.1.Key", "owner"),
            ("Tags.Tag.1.Value", "team-a"),
        ],
    );
    svc.add_tags_to_resource(&req).unwrap();
    let __a = svc.state.read();
    let g = __a.default_ref().subnet_groups.get("sg1").unwrap();
    assert_eq!(g.tags.len(), 1);
    assert_eq!(g.tags[0].key, "owner");
}

#[test]
fn remove_tags_from_parameter_group_only_listed_keys() {
    let svc = make_service();
    create_param_group(&svc, "pg1");
    let arn = {
        let __a = svc.state.read();
        __a.default_ref()
            .parameter_groups
            .get("pg1")
            .unwrap()
            .db_parameter_group_arn
            .clone()
    };
    let add = request(
        "AddTagsToResource",
        &[
            ("ResourceName", arn.as_str()),
            ("Tags.Tag.1.Key", "k1"),
            ("Tags.Tag.1.Value", "v1"),
            ("Tags.Tag.2.Key", "k2"),
            ("Tags.Tag.2.Value", "v2"),
        ],
    );
    svc.add_tags_to_resource(&add).unwrap();
    let remove = request(
        "RemoveTagsFromResource",
        &[("ResourceName", arn.as_str()), ("TagKeys.member.1", "k1")],
    );
    svc.remove_tags_from_resource(&remove).unwrap();
    let __a = svc.state.read();
    let pg = __a.default_ref().parameter_groups.get("pg1").unwrap();
    assert_eq!(pg.tags.len(), 1);
    assert_eq!(pg.tags[0].key, "k2");
}

#[test]
fn add_tags_to_extras_resource_arn_stores_on_json() {
    // Cluster ARNs are extras-stored; tags land in a `Tags` array on
    // the JSON entry so they survive serde round-trips.
    let svc = make_service();
    let cluster_arn = {
        let mut __a = svc.state.write();
        let state = __a.default_mut();
        let arn = format!(
            "arn:aws:rds:us-east-1:{}:cluster:my-cluster",
            state.account_id
        );
        state
            .extras
            .entry("clusters".to_string())
            .or_default()
            .insert(
                "my-cluster".to_string(),
                serde_json::json!({"DBClusterIdentifier": "my-cluster"}),
            );
        arn
    };
    let req = request(
        "AddTagsToResource",
        &[
            ("ResourceName", cluster_arn.as_str()),
            ("Tags.Tag.1.Key", "team"),
            ("Tags.Tag.1.Value", "data"),
        ],
    );
    svc.add_tags_to_resource(&req).unwrap();
    let __a = svc.state.read();
    let entry = __a
        .default_ref()
        .extras
        .get("clusters")
        .unwrap()
        .get("my-cluster")
        .unwrap();
    let tags = entry.get("Tags").and_then(|t| t.as_array()).unwrap();
    assert_eq!(tags.len(), 1);
    assert_eq!(tags[0].get("Key").and_then(|k| k.as_str()), Some("team"));
}

/// Seed an extras-backed RDS resource with a minimal JSON entry so the
/// tagging dispatcher can locate it. The kind/bucket pairs mirror the
/// ones used by the create-time handlers in `extras.rs`; keeping this
/// helper local to the test module avoids leaking test-only surface
/// into the prod crate API.
fn seed_extras_entry(svc: &RdsService, bucket: &str, name: &str) {
    let mut accounts = svc.state.write();
    let state = accounts.default_mut();
    state
        .extras
        .entry(bucket.to_string())
        .or_default()
        .insert(name.to_string(), serde_json::json!({"Name": name}));
}

#[test]
fn tags_dispatch_covers_every_supported_resource_type() {
    // One tag round-trip (add -> list -> remove) per ARN segment, so the
    // dispatcher and tag_resource_not_found mapping stay in lockstep
    // with the resource buckets the rest of the crate writes to.
    let svc = make_service();
    let region = "us-east-1";
    let acct = "123456789012";

    // State-backed: db / snapshot / pg / subgrp.
    let _db_arn = seed_instance(&svc, "db1");
    seed_snapshot(&svc, "snap-1", "db1");
    create_param_group(&svc, "pg1");
    create_subnet_group(&svc, "sub1");

    // Extras-backed: cluster / cluster-snapshot / cluster-pg / og /
    // secgrp / es / db-proxy.
    seed_extras_entry(&svc, "clusters", "cluster-1");
    seed_extras_entry(&svc, "cluster_snapshots", "csnap-1");
    seed_extras_entry(&svc, "cluster_param_groups", "cpg-1");
    seed_extras_entry(&svc, "option_groups", "og-1");
    seed_extras_entry(&svc, "security_groups", "secgrp-1");
    seed_extras_entry(&svc, "event_subscriptions", "es-1");
    seed_extras_entry(&svc, "proxies", "proxy-1");

    let cases: &[(&str, &str)] = &[
        ("db", "db1"),
        ("snapshot", "snap-1"),
        ("pg", "pg1"),
        ("subgrp", "sub1"),
        ("cluster", "cluster-1"),
        ("cluster-snapshot", "csnap-1"),
        ("cluster-pg", "cpg-1"),
        ("og", "og-1"),
        ("secgrp", "secgrp-1"),
        ("es", "es-1"),
        ("db-proxy", "proxy-1"),
    ];

    for (kind, name) in cases {
        let arn = format!("arn:aws:rds:{region}:{acct}:{kind}:{name}");

        let add = request(
            "AddTagsToResource",
            &[
                ("ResourceName", arn.as_str()),
                ("Tags.Tag.1.Key", "env"),
                ("Tags.Tag.1.Value", "prod"),
            ],
        );
        svc.add_tags_to_resource(&add)
            .unwrap_or_else(|e| panic!("AddTags failed for kind={kind}: {e:?}"));

        let list = request("ListTagsForResource", &[("ResourceName", arn.as_str())]);
        let body = body_of(
            svc.list_tags_for_resource(&list)
                .unwrap_or_else(|e| panic!("ListTags failed for kind={kind}: {e:?}")),
        );
        assert!(
            body.contains("<Key>env</Key>") && body.contains("<Value>prod</Value>"),
            "ListTags for kind={kind} should echo the tag, body was: {body}"
        );

        let rm = request(
            "RemoveTagsFromResource",
            &[("ResourceName", arn.as_str()), ("TagKeys.member.1", "env")],
        );
        svc.remove_tags_from_resource(&rm)
            .unwrap_or_else(|e| panic!("RemoveTags failed for kind={kind}: {e:?}"));

        let body = body_of(svc.list_tags_for_resource(&list).unwrap());
        assert!(
            !body.contains("<Key>env</Key>"),
            "RemoveTags for kind={kind} should strip the tag, body was: {body}"
        );
    }
}

#[test]
fn tags_dispatch_typed_not_found_per_resource_type() {
    // Each known resource-type must surface its own NotFound code
    // rather than the generic `DBInstanceNotFound` fallback we use for
    // malformed ARNs.
    let svc = make_service();
    let region = "us-east-1";
    let acct = "123456789012";

    let cases: &[(&str, &str)] = &[
        ("db", "DBInstanceNotFound"),
        ("snapshot", "DBSnapshotNotFound"),
        ("cluster", "DBClusterNotFoundFault"),
        ("cluster-snapshot", "DBClusterSnapshotNotFoundFault"),
        ("pg", "DBParameterGroupNotFound"),
        ("cluster-pg", "DBParameterGroupNotFound"),
        ("og", "OptionGroupNotFoundFault"),
        ("subgrp", "DBSubnetGroupNotFoundFault"),
        ("secgrp", "DBSecurityGroupNotFound"),
        ("db-proxy", "DBProxyNotFoundFault"),
        ("es", "SubscriptionNotFound"),
    ];

    for (kind, expected_code) in cases {
        let arn = format!("arn:aws:rds:{region}:{acct}:{kind}:ghost");
        let req = request("ListTagsForResource", &[("ResourceName", arn.as_str())]);
        assert_code(svc.list_tags_for_resource(&req), expected_code);
    }
}

#[test]
fn remove_tags_strips_only_listed_keys() {
    let svc = make_service();
    let arn = seed_instance(&svc, "db1");
    {
        let mut __a = svc.state.write();
        let state = __a.default_mut();
        let inst = state.instances.get_mut("db1").unwrap();
        inst.tags = vec![
            RdsTag {
                key: "env".to_string(),
                value: "dev".to_string(),
            },
            RdsTag {
                key: "team".to_string(),
                value: "core".to_string(),
            },
        ];
    }
    let req = request(
        "RemoveTagsFromResource",
        &[("ResourceName", arn.as_str()), ("TagKeys.member.1", "env")],
    );
    svc.remove_tags_from_resource(&req).unwrap();

    let __a = svc.state.read();
    let state = __a.default_ref();
    let tags = &state.instances.get("db1").unwrap().tags;
    assert_eq!(tags.len(), 1);
    assert_eq!(tags[0].key, "team");
}

#[test]
fn remove_tags_with_no_keys_is_a_noop() {
    // RemoveTagsFromResource declares no `MissingParameter`-equivalent
    // wire shape in Smithy; treat empty TagKeys as a no-op rather than
    // emit an undeclared error code.
    let svc = make_service();
    let arn = seed_instance(&svc, "db1");
    let req = request("RemoveTagsFromResource", &[("ResourceName", arn.as_str())]);
    svc.remove_tags_from_resource(&req).expect("noop ok");
}

// ── DB Subnet Groups ─────────────────────────────────────────────

fn create_subnet_group(svc: &RdsService, name: &str) {
    let req = request(
        "CreateDBSubnetGroup",
        &[
            ("DBSubnetGroupName", name),
            ("DBSubnetGroupDescription", "test"),
            ("SubnetIds.SubnetIdentifier.1", "subnet-aaa"),
            ("SubnetIds.SubnetIdentifier.2", "subnet-bbb"),
        ],
    );
    svc.create_db_subnet_group(&req).unwrap();
}

#[test]
fn create_db_subnet_group_requires_two_subnets() {
    let svc = make_service();
    let req = request(
        "CreateDBSubnetGroup",
        &[
            ("DBSubnetGroupName", "sg1"),
            ("DBSubnetGroupDescription", "t"),
            ("SubnetIds.SubnetIdentifier.1", "subnet-aaa"),
        ],
    );
    assert_code(
        svc.create_db_subnet_group(&req),
        "DBSubnetGroupDoesNotCoverEnoughAZs",
    );
}

#[test]
fn create_db_subnet_group_rejects_empty_subnets() {
    // Folded into the `subnet_ids.len() < 2` check that emits the
    // Smithy-declared `DBSubnetGroupDoesNotCoverEnoughAZs` shape.
    let svc = make_service();
    let req = request(
        "CreateDBSubnetGroup",
        &[
            ("DBSubnetGroupName", "sg1"),
            ("DBSubnetGroupDescription", "t"),
        ],
    );
    assert_code(
        svc.create_db_subnet_group(&req),
        "DBSubnetGroupDoesNotCoverEnoughAZs",
    );
}

#[test]
fn create_db_subnet_group_rejects_duplicates() {
    let svc = make_service();
    create_subnet_group(&svc, "sg1");
    let req = request(
        "CreateDBSubnetGroup",
        &[
            ("DBSubnetGroupName", "sg1"),
            ("DBSubnetGroupDescription", "t"),
            ("SubnetIds.SubnetIdentifier.1", "subnet-x"),
            ("SubnetIds.SubnetIdentifier.2", "subnet-y"),
        ],
    );
    assert_code(
        svc.create_db_subnet_group(&req),
        "DBSubnetGroupAlreadyExists",
    );
}

#[test]
fn describe_db_subnet_groups_by_name_or_list() {
    let svc = make_service();
    create_subnet_group(&svc, "sg-alpha");
    create_subnet_group(&svc, "sg-beta");

    let by_name = request(
        "DescribeDBSubnetGroups",
        &[("DBSubnetGroupName", "sg-alpha")],
    );
    let body = body_of(svc.describe_db_subnet_groups(&by_name).unwrap());
    assert!(body.contains("sg-alpha"));
    assert!(!body.contains("sg-beta"));

    let list_all = request("DescribeDBSubnetGroups", &[]);
    let body = body_of(svc.describe_db_subnet_groups(&list_all).unwrap());
    assert!(body.contains("sg-alpha"));
    assert!(body.contains("sg-beta"));
}

#[test]
fn describe_db_subnet_groups_unknown_name_errors() {
    let svc = make_service();
    let req = request("DescribeDBSubnetGroups", &[("DBSubnetGroupName", "ghost")]);
    assert_code(
        svc.describe_db_subnet_groups(&req),
        "DBSubnetGroupNotFoundFault",
    );
}

#[test]
fn delete_db_subnet_group_unknown_errors() {
    let svc = make_service();
    let req = request("DeleteDBSubnetGroup", &[("DBSubnetGroupName", "ghost")]);
    assert_code(
        svc.delete_db_subnet_group(&req),
        "DBSubnetGroupNotFoundFault",
    );
}

#[test]
fn delete_db_subnet_group_removes_entry() {
    let svc = make_service();
    create_subnet_group(&svc, "sg1");
    let req = request("DeleteDBSubnetGroup", &[("DBSubnetGroupName", "sg1")]);
    svc.delete_db_subnet_group(&req).unwrap();
    assert!(svc.state.read().default_ref().subnet_groups.is_empty());
}

#[test]
fn modify_db_subnet_group_updates_subnet_ids() {
    let svc = make_service();
    create_subnet_group(&svc, "sg1");
    let req = request(
        "ModifyDBSubnetGroup",
        &[
            ("DBSubnetGroupName", "sg1"),
            ("SubnetIds.SubnetIdentifier.1", "subnet-new1"),
            ("SubnetIds.SubnetIdentifier.2", "subnet-new2"),
        ],
    );
    svc.modify_db_subnet_group(&req).unwrap();

    let __a = svc.state.read();
    let state = __a.default_ref();
    let sg = state.subnet_groups.get("sg1").unwrap();
    assert_eq!(sg.subnet_ids, vec!["subnet-new1", "subnet-new2"]);
}

// ── DB Parameter Groups ──────────────────────────────────────────

fn create_param_group(svc: &RdsService, name: &str) {
    let req = request(
        "CreateDBParameterGroup",
        &[
            ("DBParameterGroupName", name),
            ("DBParameterGroupFamily", "postgres16"),
            ("Description", "test"),
        ],
    );
    svc.create_db_parameter_group(&req).unwrap();
}

#[test]
fn create_db_parameter_group_accepts_unknown_family() {
    // Smithy declares no `InvalidParameterValue` shape on
    // CreateDBParameterGroup, so we accept any family verbatim
    // rather than emit an undeclared wire code.
    let svc = make_service();
    let req = request(
        "CreateDBParameterGroup",
        &[
            ("DBParameterGroupName", "pg1"),
            ("DBParameterGroupFamily", "oracle19"),
            ("Description", "t"),
        ],
    );
    svc.create_db_parameter_group(&req)
        .expect("unknown family accepted");
}

#[test]
fn create_db_parameter_group_rejects_duplicates() {
    let svc = make_service();
    create_param_group(&svc, "pg1");
    let req = request(
        "CreateDBParameterGroup",
        &[
            ("DBParameterGroupName", "pg1"),
            ("DBParameterGroupFamily", "postgres16"),
            ("Description", "t"),
        ],
    );
    assert_code(
        svc.create_db_parameter_group(&req),
        "DBParameterGroupAlreadyExists",
    );
}

#[test]
fn describe_db_parameter_groups_by_name_or_list() {
    let svc = make_service();
    create_param_group(&svc, "pg-alpha");
    create_param_group(&svc, "pg-beta");
    let by_name = request(
        "DescribeDBParameterGroups",
        &[("DBParameterGroupName", "pg-alpha")],
    );
    let body = body_of(svc.describe_db_parameter_groups(&by_name).unwrap());
    assert!(body.contains("pg-alpha"));
    assert!(!body.contains("pg-beta"));
    let list = request("DescribeDBParameterGroups", &[]);
    let body = body_of(svc.describe_db_parameter_groups(&list).unwrap());
    assert!(body.contains("pg-alpha"));
    assert!(body.contains("pg-beta"));
}

#[test]
fn describe_db_parameter_groups_unknown_name_errors() {
    let svc = make_service();
    let req = request(
        "DescribeDBParameterGroups",
        &[("DBParameterGroupName", "ghost")],
    );
    assert_code(
        svc.describe_db_parameter_groups(&req),
        "DBParameterGroupNotFound",
    );
}

#[test]
fn delete_db_parameter_group_rejects_default_groups() {
    let svc = make_service();
    let req = request(
        "DeleteDBParameterGroup",
        &[("DBParameterGroupName", "default.postgres16")],
    );
    assert_code(
        svc.delete_db_parameter_group(&req),
        "InvalidDBParameterGroupState",
    );
}

#[test]
fn delete_db_parameter_group_unknown_errors() {
    let svc = make_service();
    let req = request(
        "DeleteDBParameterGroup",
        &[("DBParameterGroupName", "ghost")],
    );
    assert_code(
        svc.delete_db_parameter_group(&req),
        "DBParameterGroupNotFound",
    );
}

#[test]
fn delete_db_parameter_group_removes_entry() {
    let svc = make_service();
    create_param_group(&svc, "pg1");
    let req = request("DeleteDBParameterGroup", &[("DBParameterGroupName", "pg1")]);
    svc.delete_db_parameter_group(&req).unwrap();
    assert!(!svc
        .state
        .read()
        .default_ref()
        .parameter_groups
        .contains_key("pg1"));
}

#[test]
fn modify_db_parameter_group_updates_description() {
    let svc = make_service();
    create_param_group(&svc, "pg1");
    let req = request(
        "ModifyDBParameterGroup",
        &[
            ("DBParameterGroupName", "pg1"),
            ("Description", "shiny new"),
        ],
    );
    svc.modify_db_parameter_group(&req).unwrap();
    let __a = svc.state.read();
    let state = __a.default_ref();
    assert_eq!(
        state.parameter_groups.get("pg1").unwrap().description,
        "shiny new"
    );
}

#[test]
fn modify_db_parameter_group_unknown_errors() {
    let svc = make_service();
    let req = request(
        "ModifyDBParameterGroup",
        &[("DBParameterGroupName", "ghost"), ("Description", "x")],
    );
    assert_code(
        svc.modify_db_parameter_group(&req),
        "DBParameterGroupNotFound",
    );
}

#[test]
fn modify_db_parameter_group_persists_parameters() {
    let svc = make_service();
    create_param_group(&svc, "pg1");
    let req = request(
        "ModifyDBParameterGroup",
        &[
            ("DBParameterGroupName", "pg1"),
            ("Parameters.member.1.ParameterName", "max_connections"),
            ("Parameters.member.1.ParameterValue", "200"),
            ("Parameters.member.1.ApplyMethod", "immediate"),
            ("Parameters.member.2.ParameterName", "shared_buffers"),
            ("Parameters.member.2.ParameterValue", "256MB"),
            ("Parameters.member.2.ApplyMethod", "pending-reboot"),
        ],
    );
    svc.modify_db_parameter_group(&req).unwrap();
    let __a = svc.state.read();
    let state = __a.default_ref();
    let pg = state.parameter_groups.get("pg1").unwrap();
    assert_eq!(
        pg.parameters.get("max_connections").map(String::as_str),
        Some("200")
    );
    assert_eq!(
        pg.parameters.get("shared_buffers").map(String::as_str),
        Some("256MB")
    );
}

#[test]
fn describe_db_parameters_returns_user_set_values() {
    let svc = make_service();
    create_param_group(&svc, "pg1");
    let req = request(
        "ModifyDBParameterGroup",
        &[
            ("DBParameterGroupName", "pg1"),
            ("Parameters.member.1.ParameterName", "max_connections"),
            ("Parameters.member.1.ParameterValue", "200"),
        ],
    );
    svc.modify_db_parameter_group(&req).unwrap();

    let req = request("DescribeDBParameters", &[("DBParameterGroupName", "pg1")]);
    let resp = svc.describe_db_parameters_real(&req).unwrap();
    let body = String::from_utf8(resp.body.expect_bytes().to_vec()).unwrap();
    assert!(body.contains("<ParameterName>max_connections</ParameterName>"));
    assert!(body.contains("<ParameterValue>200</ParameterValue>"));
    assert!(body.contains("<Source>user</Source>"));
}

#[test]
fn describe_db_parameters_with_engine_default_source_omits_user_params() {
    let svc = make_service();
    create_param_group(&svc, "pg1");
    // Modify a parameter that is NOT seeded as an engine default so the
    // `engine-default` source filter has a clean way to demonstrate it
    // skips user-only parameters.
    let req = request(
        "ModifyDBParameterGroup",
        &[
            ("DBParameterGroupName", "pg1"),
            ("Parameters.member.1.ParameterName", "user_only_knob"),
            ("Parameters.member.1.ParameterValue", "42"),
        ],
    );
    svc.modify_db_parameter_group(&req).unwrap();

    let req = request(
        "DescribeDBParameters",
        &[
            ("DBParameterGroupName", "pg1"),
            ("Source", "engine-default"),
        ],
    );
    let resp = svc.describe_db_parameters_real(&req).unwrap();
    let body = String::from_utf8(resp.body.expect_bytes().to_vec()).unwrap();
    // User-only parameter is hidden when filtering on engine defaults.
    assert!(!body.contains("user_only_knob"));
    // Engine defaults still surface (postgres16 seeds `max_connections`).
    assert!(body.contains("max_connections"));
    assert!(body.contains("<Source>engine-default</Source>"));
    assert!(!body.contains("<Source>user</Source>"));
}

#[test]
fn describe_db_parameters_with_no_source_returns_user_and_engine_defaults() {
    let svc = make_service();
    create_param_group(&svc, "pg1");
    let req = request(
        "ModifyDBParameterGroup",
        &[
            ("DBParameterGroupName", "pg1"),
            ("Parameters.member.1.ParameterName", "max_connections"),
            ("Parameters.member.1.ParameterValue", "200"),
        ],
    );
    svc.modify_db_parameter_group(&req).unwrap();

    let req = request("DescribeDBParameters", &[("DBParameterGroupName", "pg1")]);
    let resp = svc.describe_db_parameters_real(&req).unwrap();
    let body = String::from_utf8(resp.body.expect_bytes().to_vec()).unwrap();
    // User override of `max_connections` shadows the engine default so
    // the parameter appears exactly once with `Source=user`.
    assert_eq!(
        body.matches("<ParameterName>max_connections</ParameterName>")
            .count(),
        1
    );
    assert!(body.contains("<ParameterValue>200</ParameterValue>"));
    // Other engine defaults (e.g. work_mem) still come through.
    assert!(body.contains("<ParameterName>work_mem</ParameterName>"));
    assert!(body.contains("<Source>engine-default</Source>"));
}

#[test]
fn describe_db_parameters_unknown_group_returns_not_found() {
    let svc = make_service();
    let req = request("DescribeDBParameters", &[("DBParameterGroupName", "ghost")]);
    assert_code(
        svc.describe_db_parameters_real(&req),
        "DBParameterGroupNotFound",
    );
}

// ── DescribeDBInstances ──────────────────────────────────────────

#[test]
fn describe_db_instances_by_id_returns_only_one() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    seed_instance(&svc, "db2");
    let req = request("DescribeDBInstances", &[("DBInstanceIdentifier", "db1")]);
    let body = body_of(svc.describe_db_instances(&req).unwrap());
    assert!(body.contains("<DBInstanceIdentifier>db1</DBInstanceIdentifier>"));
    assert!(!body.contains("<DBInstanceIdentifier>db2</DBInstanceIdentifier>"));
}

#[test]
fn describe_db_instances_unknown_id_errors() {
    let svc = make_service();
    let req = request("DescribeDBInstances", &[("DBInstanceIdentifier", "ghost")]);
    assert_code(svc.describe_db_instances(&req), "DBInstanceNotFound");
}

#[test]
fn describe_db_instances_lists_all_when_unbounded() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    seed_instance(&svc, "db2");
    seed_instance(&svc, "db3");
    let req = request("DescribeDBInstances", &[]);
    let body = body_of(svc.describe_db_instances(&req).unwrap());
    for id in ["db1", "db2", "db3"] {
        assert!(body.contains(&format!(
            "<DBInstanceIdentifier>{id}</DBInstanceIdentifier>"
        )));
    }
}

// ── ModifyDBInstance ─────────────────────────────────────────────

#[test]
fn modify_db_instance_with_no_changes_is_a_noop() {
    // Smithy declares no `InvalidParameterCombination` shape on
    // ModifyDBInstance. A modify call that touches nothing returns
    // the unchanged DB instance description rather than an error.
    let svc = make_service();
    seed_instance(&svc, "db1");
    let req = request("ModifyDBInstance", &[("DBInstanceIdentifier", "db1")]);
    svc.modify_db_instance(&req).expect("noop modify ok");
}

#[test]
fn modify_db_instance_unknown_errors() {
    let svc = make_service();
    let req = request(
        "ModifyDBInstance",
        &[
            ("DBInstanceIdentifier", "ghost"),
            ("DBInstanceClass", "db.t3.small"),
        ],
    );
    assert_code(svc.modify_db_instance(&req), "DBInstanceNotFound");
}

#[test]
fn modify_db_instance_apply_immediately_updates_class() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    let req = request(
        "ModifyDBInstance",
        &[
            ("DBInstanceIdentifier", "db1"),
            ("DBInstanceClass", "db.t3.small"),
            ("ApplyImmediately", "true"),
        ],
    );
    svc.modify_db_instance(&req).unwrap();
    let __a = svc.state.read();
    let state = __a.default_ref();
    assert_eq!(
        state.instances.get("db1").unwrap().db_instance_class,
        "db.t3.small"
    );
}

#[test]
fn modify_db_instance_pending_when_not_apply_immediately() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    let req = request(
        "ModifyDBInstance",
        &[
            ("DBInstanceIdentifier", "db1"),
            ("DBInstanceClass", "db.t3.small"),
            ("ApplyImmediately", "false"),
        ],
    );
    svc.modify_db_instance(&req).unwrap();
    let __a = svc.state.read();
    let state = __a.default_ref();
    let inst = state.instances.get("db1").unwrap();
    assert_eq!(inst.db_instance_class, "db.t3.micro");
    assert_eq!(
        inst.pending_modified_values
            .as_ref()
            .unwrap()
            .db_instance_class
            .as_deref(),
        Some("db.t3.small"),
    );
}

#[test]
fn modify_db_instance_apply_immediately_updates_engine_and_storage() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    let req = request(
        "ModifyDBInstance",
        &[
            ("DBInstanceIdentifier", "db1"),
            ("EngineVersion", "16.4"),
            ("AllocatedStorage", "100"),
            ("Iops", "3000"),
            ("StorageType", "io2"),
            ("PreferredMaintenanceWindow", "Mon:00:00-Mon:01:00"),
            ("MultiAZ", "true"),
            ("ApplyImmediately", "true"),
        ],
    );
    svc.modify_db_instance(&req).unwrap();
    let __a = svc.state.read();
    let state = __a.default_ref();
    let inst = state.instances.get("db1").unwrap();
    assert_eq!(inst.engine_version, "16.4");
    assert_eq!(inst.allocated_storage, 100);
    assert_eq!(inst.iops, Some(3000));
    assert_eq!(inst.storage_type.as_deref(), Some("io2"));
    assert_eq!(
        inst.preferred_maintenance_window.as_deref(),
        Some("Mon:00:00-Mon:01:00")
    );
    assert!(inst.multi_az);
    assert!(inst.pending_modified_values.is_none());
}

#[test]
fn modify_db_instance_pending_stages_extended_fields() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    let req = request(
        "ModifyDBInstance",
        &[
            ("DBInstanceIdentifier", "db1"),
            ("EngineVersion", "16.4"),
            ("AllocatedStorage", "100"),
            ("PreferredBackupWindow", "04:00-05:00"),
            ("DBParameterGroupName", "custom-pg"),
            ("MultiAZ", "true"),
            ("ApplyImmediately", "false"),
        ],
    );
    svc.modify_db_instance(&req).unwrap();
    let __a = svc.state.read();
    let state = __a.default_ref();
    let inst = state.instances.get("db1").unwrap();
    let pending = inst.pending_modified_values.as_ref().unwrap();
    assert_eq!(pending.engine_version.as_deref(), Some("16.4"));
    assert_eq!(pending.allocated_storage, Some(100));
    assert_eq!(
        pending.preferred_backup_window.as_deref(),
        Some("04:00-05:00")
    );
    assert_eq!(
        pending.db_parameter_group_name.as_deref(),
        Some("custom-pg")
    );
    assert_eq!(pending.multi_az, Some(true));
    // Live values unchanged.
    assert_eq!(inst.engine_version, "16.3");
    assert_eq!(inst.allocated_storage, 20);
}

#[test]
fn modify_db_instance_immediate_only_fields_apply_with_apply_immediately_false() {
    // CACertificateIdentifier, MasterUserSecretKmsKeyId, and the
    // CloudwatchLogsExportConfiguration are AWS-immediate fields:
    // ApplyImmediately=false must not stage them.
    let svc = make_service();
    seed_instance(&svc, "db1");
    let req = request(
        "ModifyDBInstance",
        &[
            ("DBInstanceIdentifier", "db1"),
            ("CACertificateIdentifier", "rds-ca-2024"),
            ("MasterUserSecretKmsKeyId", "alias/aws/rds"),
            (
                "CloudwatchLogsExportConfiguration.EnableLogTypes.member.1",
                "postgresql",
            ),
            ("ApplyImmediately", "false"),
        ],
    );
    svc.modify_db_instance(&req).unwrap();
    let __a = svc.state.read();
    let state = __a.default_ref();
    let inst = state.instances.get("db1").unwrap();
    assert_eq!(
        inst.ca_certificate_identifier.as_deref(),
        Some("rds-ca-2024")
    );
    assert_eq!(
        inst.master_user_secret_kms_key_id.as_deref(),
        Some("alias/aws/rds")
    );
    assert!(inst
        .enabled_cloudwatch_logs_exports
        .iter()
        .any(|t| t == "postgresql"));
    // No pending values for these — they were applied directly.
    assert!(inst.pending_modified_values.is_none());
}

#[test]
fn modify_db_instance_cloudwatch_disable_log_types_removes_existing() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    {
        let mut __a = svc.state.write();
        let state = __a.default_mut();
        let inst = state.instances.get_mut("db1").unwrap();
        inst.enabled_cloudwatch_logs_exports =
            vec!["postgresql".to_string(), "upgrade".to_string()];
    }
    let req = request(
        "ModifyDBInstance",
        &[
            ("DBInstanceIdentifier", "db1"),
            (
                "CloudwatchLogsExportConfiguration.DisableLogTypes.member.1",
                "upgrade",
            ),
        ],
    );
    svc.modify_db_instance(&req).unwrap();
    let __a = svc.state.read();
    let state = __a.default_ref();
    let inst = state.instances.get("db1").unwrap();
    assert_eq!(inst.enabled_cloudwatch_logs_exports, vec!["postgresql"]);
}

// ── Snapshots (sync ops only) ────────────────────────────────────

fn seed_snapshot(svc: &RdsService, snapshot_id: &str, instance_id: &str) {
    let mut __a = svc.state.write();
    let state = __a.default_mut();
    let arn = state.db_snapshot_arn(snapshot_id);
    state.snapshots.insert(
        snapshot_id.to_string(),
        crate::state::DbSnapshot {
            db_snapshot_identifier: snapshot_id.to_string(),
            db_snapshot_arn: arn,
            db_instance_identifier: instance_id.to_string(),
            snapshot_create_time: Utc::now(),
            engine: "postgres".to_string(),
            engine_version: "16.3".to_string(),
            allocated_storage: 20,
            status: "available".to_string(),
            port: 5432,
            master_username: "admin".to_string(),
            db_name: Some("appdb".to_string()),
            dbi_resource_id: format!("db-{}", Uuid::new_v4().simple()),
            snapshot_type: "manual".to_string(),
            master_user_password: "secret".to_string(),
            tags: Vec::new(),
            dump_data: Vec::new(),
            availability_zone: None,
            vpc_id: None,
            instance_create_time: None,
            license_model: None,
            iops: None,
            option_group_name: None,
            percent_progress: None,
            storage_type: None,
            encrypted: false,
            kms_key_id: None,
            iam_database_authentication_enabled: false,
            timezone: None,
            storage_throughput: None,
        },
    );
}

#[test]
fn delete_db_snapshot_removes_entry() {
    let svc = make_service();
    seed_snapshot(&svc, "snap1", "db1");
    let req = request("DeleteDBSnapshot", &[("DBSnapshotIdentifier", "snap1")]);
    svc.delete_db_snapshot(&req).unwrap();
    assert!(svc.state.read().default_ref().snapshots.is_empty());
}

#[test]
fn delete_db_snapshot_unknown_errors() {
    let svc = make_service();
    let req = request("DeleteDBSnapshot", &[("DBSnapshotIdentifier", "ghost")]);
    assert_code(svc.delete_db_snapshot(&req), "DBSnapshotNotFound");
}

#[test]
fn describe_db_snapshots_accepts_both_filters() {
    // Both snapshot id + instance id is tolerated: the snapshot id
    // takes precedence below. Smithy doesn't declare an
    // `InvalidParameterCombination` error shape on this op.
    let svc = make_service();
    let req = request(
        "DescribeDBSnapshots",
        &[("DBSnapshotIdentifier", "s"), ("DBInstanceIdentifier", "i")],
    );
    assert_code(svc.describe_db_snapshots(&req), "DBSnapshotNotFound");
}

#[test]
fn describe_db_snapshots_by_id_or_instance() {
    let svc = make_service();
    seed_snapshot(&svc, "snap1", "db1");
    seed_snapshot(&svc, "snap2", "db2");

    let by_id = request("DescribeDBSnapshots", &[("DBSnapshotIdentifier", "snap1")]);
    let body = body_of(svc.describe_db_snapshots(&by_id).unwrap());
    assert!(body.contains("snap1"));
    assert!(!body.contains("snap2"));

    let by_instance = request("DescribeDBSnapshots", &[("DBInstanceIdentifier", "db2")]);
    let body = body_of(svc.describe_db_snapshots(&by_instance).unwrap());
    assert!(body.contains("snap2"));
    assert!(!body.contains("snap1"));

    let list_all = request("DescribeDBSnapshots", &[]);
    let body = body_of(svc.describe_db_snapshots(&list_all).unwrap());
    assert!(body.contains("snap1"));
    assert!(body.contains("snap2"));
}

#[test]
fn describe_db_snapshots_unknown_id_errors() {
    let svc = make_service();
    let req = request("DescribeDBSnapshots", &[("DBSnapshotIdentifier", "ghost")]);
    assert_code(svc.describe_db_snapshots(&req), "DBSnapshotNotFound");
}

// ── Error branch tests ──

#[test]
fn describe_db_instances_not_found() {
    let svc = make_service();
    let req = request("DescribeDBInstances", &[("DBInstanceIdentifier", "ghost")]);
    assert_code(svc.describe_db_instances(&req), "DBInstanceNotFound");
}

#[tokio::test]
async fn delete_db_instance_not_found() {
    let svc = make_service();
    let req = request(
        "DeleteDBInstance",
        &[
            ("DBInstanceIdentifier", "ghost"),
            ("SkipFinalSnapshot", "true"),
        ],
    );
    assert_code(svc.delete_db_instance(&req).await, "DBInstanceNotFound");
}

#[test]
fn modify_db_instance_not_found() {
    let svc = make_service();
    let req = request(
        "ModifyDBInstance",
        &[
            ("DBInstanceIdentifier", "ghost"),
            ("AllocatedStorage", "20"),
        ],
    );
    // AllocatedStorage is a valid mutable field — validation passes
    // and the existence check fires next.
    assert_code(svc.modify_db_instance(&req), "DBInstanceNotFound");
}

#[test]
fn modify_db_instance_no_fields_against_missing_instance_returns_not_found() {
    // After dropping the synthetic `InvalidParameterCombination` check
    // (no Smithy analogue), the resource-not-found path is what
    // surfaces for a probe-style empty Modify against a non-existent
    // instance.
    let svc = make_service();
    let req = request("ModifyDBInstance", &[("DBInstanceIdentifier", "anyone")]);
    assert_code(svc.modify_db_instance(&req), "DBInstanceNotFound");
}

#[tokio::test]
async fn reboot_db_instance_not_found() {
    let svc = make_service();
    let req = request("RebootDBInstance", &[("DBInstanceIdentifier", "ghost")]);
    assert_code(svc.reboot_db_instance(&req).await, "DBInstanceNotFound");
}

#[tokio::test]
async fn create_db_snapshot_instance_not_found() {
    let svc = make_service();
    let req = request(
        "CreateDBSnapshot",
        &[
            ("DBInstanceIdentifier", "ghost"),
            ("DBSnapshotIdentifier", "snap1"),
        ],
    );
    // Instance lookup happens before the runtime probe so a missing
    // source surfaces the declared `DBInstanceNotFoundFault` shape.
    assert_code(svc.create_db_snapshot(&req).await, "DBInstanceNotFound");
}

#[tokio::test]
async fn restore_db_instance_snapshot_not_found() {
    let svc = make_service();
    let req = request(
        "RestoreDBInstanceFromDBSnapshot",
        &[
            ("DBInstanceIdentifier", "restored"),
            ("DBSnapshotIdentifier", "ghost-snap"),
        ],
    );
    assert_code(
        svc.restore_db_instance_from_db_snapshot(&req).await,
        "DBSnapshotNotFound",
    );
}

#[tokio::test]
async fn create_db_instance_read_replica_source_not_found() {
    let svc = make_service();
    let req = request(
        "CreateDBInstanceReadReplica",
        &[
            ("DBInstanceIdentifier", "replica"),
            ("SourceDBInstanceIdentifier", "ghost"),
        ],
    );
    assert_code(
        svc.create_db_instance_read_replica(&req).await,
        "DBInstanceNotFound",
    );
}

#[test]
fn describe_db_engine_versions_basic() {
    let svc = make_service();
    let req = request("DescribeDBEngineVersions", &[]);
    let resp = svc.describe_db_engine_versions(&req).unwrap();
    let body = body_of(resp);
    assert!(body.contains("<DBEngineVersions>"));
}

#[test]
fn describe_orderable_db_instance_options_basic() {
    let svc = make_service();
    let req = request("DescribeOrderableDBInstanceOptions", &[("Engine", "mysql")]);
    let resp = svc.describe_orderable_db_instance_options(&req).unwrap();
    let body = body_of(resp);
    assert!(body.contains("<OrderableDBInstanceOptions>"));
}

#[test]
fn describe_db_parameter_group_not_found() {
    let svc = make_service();
    let req = request(
        "DescribeDBParameterGroups",
        &[("DBParameterGroupName", "ghost")],
    );
    assert_code(
        svc.describe_db_parameter_groups(&req),
        "DBParameterGroupNotFound",
    );
}

#[test]
fn delete_db_parameter_group_not_found() {
    let svc = make_service();
    let req = request(
        "DeleteDBParameterGroup",
        &[("DBParameterGroupName", "ghost")],
    );
    assert_code(
        svc.delete_db_parameter_group(&req),
        "DBParameterGroupNotFound",
    );
}

#[test]
fn describe_db_subnet_group_not_found() {
    let svc = make_service();
    let req = request("DescribeDBSubnetGroups", &[("DBSubnetGroupName", "ghost")]);
    assert_code(
        svc.describe_db_subnet_groups(&req),
        "DBSubnetGroupNotFoundFault",
    );
}

#[test]
fn delete_db_subnet_group_not_found() {
    let svc = make_service();
    let req = request("DeleteDBSubnetGroup", &[("DBSubnetGroupName", "ghost")]);
    assert_code(
        svc.delete_db_subnet_group(&req),
        "DBSubnetGroupNotFoundFault",
    );
}

#[test]
fn add_tags_resource_not_found() {
    let svc = make_service();
    let req = request(
        "AddTagsToResource",
        &[
            ("ResourceName", "arn:aws:rds:us-east-1:123:db:ghost"),
            ("Tags.member.1.Key", "k"),
            ("Tags.member.1.Value", "v"),
        ],
    );
    // `Tags.member.N` is the generic awsQuery wire form that the
    // conformance probe (and modern SDKs) emit — accepted alongside the
    // canonical `Tags.Tag.N.Key`/`Tags.Tag.N.Value` form. With non-empty
    // tags the call advances to the resource-lookup step and returns
    // the declared `DBInstanceNotFound`.
    assert_code(svc.add_tags_to_resource(&req), "DBInstanceNotFound");
}

#[test]
fn list_tags_resource_not_found() {
    let svc = make_service();
    let req = request(
        "ListTagsForResource",
        &[("ResourceName", "arn:aws:rds:us-east-1:123:db:ghost")],
    );
    assert_code(svc.list_tags_for_resource(&req), "DBInstanceNotFound");
}

// ── snapshot operations ──

#[tokio::test]
async fn create_db_snapshot_missing_id_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBSnapshot",
        &[("DBInstanceIdentifier", "nonexistent")],
    );
    assert_code(svc.create_db_snapshot(&req).await, "MissingParameter");
}

#[tokio::test]
async fn create_db_snapshot_unknown_instance_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBSnapshot",
        &[
            ("DBSnapshotIdentifier", "snap1"),
            ("DBInstanceIdentifier", "ghost"),
        ],
    );
    assert!(svc.create_db_snapshot(&req).await.is_err());
}

// ── delete_db_instance ──

#[tokio::test]
async fn delete_db_instance_missing_id_errors() {
    let svc = make_service();
    let req = request("DeleteDBInstance", &[]);
    assert_code(svc.delete_db_instance(&req).await, "MissingParameter");
}

// ── reboot_db_instance ──

#[tokio::test]
async fn reboot_db_instance_missing_id_errors() {
    let svc = make_service();
    let req = request("RebootDBInstance", &[]);
    assert_code(svc.reboot_db_instance(&req).await, "MissingParameter");
}

// ── create_db_instance validation ──

#[tokio::test]
async fn create_db_instance_missing_id_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBInstance",
        &[
            ("Engine", "postgres"),
            ("DBInstanceClass", "db.t3.micro"),
            ("AllocatedStorage", "20"),
            ("MasterUsername", "admin"),
            ("MasterUserPassword", "secretpass"),
        ],
    );
    assert!(svc.create_db_instance(&req).await.is_err());
}

#[tokio::test]
async fn create_db_instance_unsupported_engine_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBInstance",
        &[
            ("DBInstanceIdentifier", "db1"),
            ("Engine", "mongodb"),
            ("DBInstanceClass", "db.t3.micro"),
            ("AllocatedStorage", "20"),
            ("MasterUsername", "admin"),
            ("MasterUserPassword", "secretpass"),
        ],
    );
    assert!(svc.create_db_instance(&req).await.is_err());
}

#[tokio::test]
async fn create_db_instance_persists_request_tags() {
    // Real RDS applies create-time `Tags` immediately. The stub runtime
    // lets the handler reach the synchronously-stored "creating"
    // placeholder without a container daemon; the background start fails
    // harmlessly and never touches `tags`.
    let svc = make_service().with_runtime(Arc::new(crate::runtime::RdsRuntime::new_stub()));
    let req = request(
        "CreateDBInstance",
        &[
            ("DBInstanceIdentifier", "tagged-db"),
            ("Engine", "postgres"),
            ("DBInstanceClass", "db.t3.micro"),
            ("AllocatedStorage", "20"),
            ("MasterUsername", "admin"),
            ("MasterUserPassword", "secretpass"),
            ("Tags.Tag.1.Key", "env"),
            ("Tags.Tag.1.Value", "prod"),
            ("Tags.Tag.2.Key", "owner"),
            ("Tags.Tag.2.Value", "platform"),
        ],
    );
    svc.create_db_instance(&req).await.expect("create ok");

    let accounts = svc.state.read();
    let state = accounts.default_ref();
    let inst = state
        .instances
        .get("tagged-db")
        .expect("placeholder stored");
    assert_eq!(
        inst.tags,
        vec![
            RdsTag {
                key: "env".to_string(),
                value: "prod".to_string()
            },
            RdsTag {
                key: "owner".to_string(),
                value: "platform".to_string()
            },
        ]
    );
}

// ── restore_db_instance_from_db_snapshot ──

#[tokio::test]
async fn restore_db_instance_missing_ids_errors() {
    let svc = make_service();
    let req = request("RestoreDBInstanceFromDBSnapshot", &[]);
    assert!(svc
        .restore_db_instance_from_db_snapshot(&req)
        .await
        .is_err());
}

#[tokio::test]
async fn restore_db_instance_unknown_snapshot_errors() {
    let svc = make_service();
    let req = request(
        "RestoreDBInstanceFromDBSnapshot",
        &[
            ("DBInstanceIdentifier", "restored"),
            ("DBSnapshotIdentifier", "missing"),
        ],
    );
    assert!(svc
        .restore_db_instance_from_db_snapshot(&req)
        .await
        .is_err());
}

#[tokio::test]
async fn restore_db_instance_from_db_snapshot_persists_tags() {
    // Real round-trip: the handler parses `Tags.Tag.N.{Key,Value}` via
    // `parse_tags` then forwards them to `build_restored_instance`,
    // which writes them onto the new `DbInstance.tags`. The runtime
    // call between those two steps doesn't touch tags, so we can stub
    // out the running container struct and assert end-state.
    let req = request(
        "RestoreDBInstanceFromDBSnapshot",
        &[
            ("DBInstanceIdentifier", "restored"),
            ("DBSnapshotIdentifier", "snap"),
            ("Tags.Tag.1.Key", "env"),
            ("Tags.Tag.1.Value", "prod"),
            ("Tags.Tag.2.Key", "owner"),
            ("Tags.Tag.2.Value", "platform"),
        ],
    );
    let tags = parse_tags(&req).expect("tags parse");

    let snapshot = crate::state::DbSnapshot {
        db_snapshot_identifier: "snap".to_string(),
        db_snapshot_arn: "arn:aws:rds:us-east-1:123456789012:snapshot:snap".to_string(),
        db_instance_identifier: "src".to_string(),
        snapshot_create_time: Utc::now(),
        engine: "postgres".to_string(),
        engine_version: "16.3".to_string(),
        allocated_storage: 20,
        status: "available".to_string(),
        port: 5432,
        master_username: "admin".to_string(),
        db_name: Some("appdb".to_string()),
        dbi_resource_id: "db-rid".to_string(),
        snapshot_type: "manual".to_string(),
        master_user_password: "secret".to_string(),
        tags: Vec::new(),
        dump_data: Vec::new(),
        availability_zone: None,
        vpc_id: None,
        instance_create_time: None,
        license_model: None,
        iops: None,
        option_group_name: None,
        percent_progress: None,
        storage_type: None,
        encrypted: false,
        kms_key_id: None,
        iam_database_authentication_enabled: false,
        timezone: None,
        storage_throughput: None,
    };
    let running = crate::runtime::RunningDbContainer {
        container_id: "c-restored".to_string(),
        host_port: 15432,
        endpoint_address: "127.0.0.1".to_string(),
        endpoint_port: 15432,
    };
    let instance = build_restored_instance(
        "restored",
        "arn:aws:rds:us-east-1:123456789012:db:restored".to_string(),
        "db-restored".to_string(),
        Utc::now(),
        Vec::new(),
        &snapshot,
        &running,
        tags,
    );
    assert_eq!(
        instance.tags,
        vec![
            RdsTag {
                key: "env".to_string(),
                value: "prod".to_string()
            },
            RdsTag {
                key: "owner".to_string(),
                value: "platform".to_string()
            },
        ]
    );
}

#[tokio::test]
async fn restore_db_instance_to_point_in_time_missing_ids_errors() {
    let svc = make_service();
    let req = request("RestoreDBInstanceToPointInTime", &[]);
    assert!(svc
        .restore_db_instance_to_point_in_time(&req)
        .await
        .is_err());
}

#[tokio::test]
async fn restore_db_instance_to_point_in_time_missing_target_errors() {
    let svc = make_service();
    let req = request(
        "RestoreDBInstanceToPointInTime",
        &[("SourceDBInstanceIdentifier", "src")],
    );
    assert!(svc
        .restore_db_instance_to_point_in_time(&req)
        .await
        .is_err());
}

#[tokio::test]
async fn restore_db_instance_to_point_in_time_unknown_source_errors() {
    let svc = make_service();
    let req = request(
        "RestoreDBInstanceToPointInTime",
        &[
            ("SourceDBInstanceIdentifier", "ghost"),
            ("TargetDBInstanceIdentifier", "restored"),
        ],
    );
    let err = svc
        .restore_db_instance_to_point_in_time(&req)
        .await
        .err()
        .expect("unknown source should error");
    assert_eq!(err.code(), "DBInstanceNotFound");
}

#[tokio::test]
async fn restore_db_instance_from_s3_missing_ids_errors() {
    let svc = make_service();
    let req = request("RestoreDBInstanceFromS3", &[]);
    assert!(svc.restore_db_instance_from_s3(&req).await.is_err());
}

#[tokio::test]
async fn restore_db_instance_from_s3_without_bus_errors() {
    let svc = make_service();
    let req = request(
        "RestoreDBInstanceFromS3",
        &[
            ("DBInstanceIdentifier", "restored"),
            ("S3BucketName", "backups"),
            ("S3Prefix", "dump.sql"),
            ("MasterUsername", "admin"),
            ("MasterUserPassword", "password"),
            ("Engine", "postgres"),
        ],
    );
    let err = svc
        .restore_db_instance_from_s3(&req)
        .await
        .err()
        .expect("missing bus should error");
    assert_eq!(err.code(), "InvalidS3BucketFault");
}

// ── create_db_instance_read_replica ──

#[tokio::test]
async fn create_read_replica_missing_source_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBInstanceReadReplica",
        &[("DBInstanceIdentifier", "replica1")],
    );
    assert!(svc.create_db_instance_read_replica(&req).await.is_err());
}

#[tokio::test]
async fn create_read_replica_unknown_source_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBInstanceReadReplica",
        &[
            ("DBInstanceIdentifier", "replica1"),
            ("SourceDBInstanceIdentifier", "ghost"),
        ],
    );
    assert!(svc.create_db_instance_read_replica(&req).await.is_err());
}

// ── describe_db_snapshots with filters ──

#[test]
fn describe_db_snapshots_by_snapshot_id_only() {
    let svc = make_service();
    seed_snapshot(&svc, "s1", "inst1");
    let req = request("DescribeDBSnapshots", &[("DBSnapshotIdentifier", "s1")]);
    let resp = svc.describe_db_snapshots(&req).unwrap();
    let b = body_of(resp);
    assert!(b.contains("<DBSnapshotIdentifier>s1</DBSnapshotIdentifier>"));
}

#[test]
fn describe_db_snapshots_by_instance_id_returns_matching() {
    let svc = make_service();
    seed_snapshot(&svc, "s1", "inst1");
    seed_snapshot(&svc, "s2", "inst2");
    let req = request("DescribeDBSnapshots", &[("DBInstanceIdentifier", "inst1")]);
    let resp = svc.describe_db_snapshots(&req).unwrap();
    let b = body_of(resp);
    assert!(b.contains("s1"));
    assert!(!b.contains("<DBSnapshotIdentifier>s2</DBSnapshotIdentifier>"));
}

// ── modify_db_parameter_group ──

#[test]
fn modify_db_parameter_group_missing_name() {
    let svc = make_service();
    let req = request("ModifyDBParameterGroup", &[]);
    assert!(svc.modify_db_parameter_group(&req).is_err());
}

// ── modify_db_subnet_group ──

#[test]
fn modify_db_subnet_group_unknown_errors() {
    let svc = make_service();
    let req = request(
        "ModifyDBSubnetGroup",
        &[
            ("DBSubnetGroupName", "ghost"),
            ("SubnetIds.SubnetIdentifier.1", "subnet-a"),
            ("SubnetIds.SubnetIdentifier.2", "subnet-b"),
        ],
    );
    assert!(svc.modify_db_subnet_group(&req).is_err());
}

// ── describe_db_instances ──

#[test]
fn describe_db_instances_empty_returns_xml() {
    let svc = make_service();
    let req = request("DescribeDBInstances", &[]);
    let resp = svc.describe_db_instances(&req).unwrap();
    let b = body_of(resp);
    assert!(b.contains("DescribeDBInstancesResult"));
}

#[test]
fn describe_db_snapshots_empty_returns_empty_list() {
    let svc = make_service();
    let req = request("DescribeDBSnapshots", &[]);
    let resp = svc.describe_db_snapshots(&req).unwrap();
    let b = body_of(resp);
    assert!(b.contains("DescribeDBSnapshotsResult"));
}

#[test]
fn add_tags_unknown_resource_errors() {
    let svc = make_service();
    let req = request(
        "AddTagsToResource",
        &[
            ("ResourceName", "arn:aws:rds:us-east-1:123:db:ghost"),
            ("Tags.member.1.Key", "k"),
            ("Tags.member.1.Value", "v"),
        ],
    );
    assert!(svc.add_tags_to_resource(&req).is_err());
}

#[test]
fn remove_tags_unknown_resource_errors() {
    let svc = make_service();
    let req = request(
        "RemoveTagsFromResource",
        &[
            ("ResourceName", "arn:aws:rds:us-east-1:123:db:ghost"),
            ("TagKeys.member.1", "k"),
        ],
    );
    assert!(svc.remove_tags_from_resource(&req).is_err());
}

#[test]
fn create_db_parameter_group_missing_name_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBParameterGroup",
        &[
            ("DBParameterGroupFamily", "postgres16"),
            ("Description", "d"),
        ],
    );
    assert!(svc.create_db_parameter_group(&req).is_err());
}

#[test]
fn create_db_subnet_group_missing_desc_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBSubnetGroup",
        &[
            ("DBSubnetGroupName", "sg1"),
            ("SubnetIds.SubnetIdentifier.1", "subnet-a"),
            ("SubnetIds.SubnetIdentifier.2", "subnet-b"),
        ],
    );
    assert!(svc.create_db_subnet_group(&req).is_err());
}

#[tokio::test]
async fn create_db_instance_missing_class_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBInstance",
        &[
            ("DBInstanceIdentifier", "miss-class"),
            ("Engine", "postgres"),
            ("AllocatedStorage", "20"),
            ("MasterUsername", "admin"),
            ("MasterUserPassword", "secretpass"),
        ],
    );
    assert!(svc.create_db_instance(&req).await.is_err());
}

#[tokio::test]
async fn create_db_instance_missing_master_username_errors() {
    let svc = make_service();
    let req = request(
        "CreateDBInstance",
        &[
            ("DBInstanceIdentifier", "miss-mu"),
            ("Engine", "postgres"),
            ("DBInstanceClass", "db.t3.micro"),
            ("AllocatedStorage", "20"),
            ("MasterUserPassword", "secretpass"),
        ],
    );
    assert!(svc.create_db_instance(&req).await.is_err());
}

#[test]
fn modify_db_instance_missing_id_errors() {
    let svc = make_service();
    let req = request("ModifyDBInstance", &[]);
    assert!(svc.modify_db_instance(&req).is_err());
}

#[test]
fn modify_db_parameter_group_unknown_pg_errors() {
    let svc = make_service();
    let req = request(
        "ModifyDBParameterGroup",
        &[
            ("DBParameterGroupName", "ghost"),
            ("Parameters.member.1.ParameterName", "p"),
            ("Parameters.member.1.ParameterValue", "v"),
            ("Parameters.member.1.ApplyMethod", "immediate"),
        ],
    );
    assert!(svc.modify_db_parameter_group(&req).is_err());
}

#[test]
fn describe_db_parameter_groups_unknown_errors() {
    let svc = make_service();
    let req = request(
        "DescribeDBParameterGroups",
        &[("DBParameterGroupName", "ghost")],
    );
    assert!(svc.describe_db_parameter_groups(&req).is_err());
}

#[test]
fn describe_db_subnet_groups_unknown_errors() {
    let svc = make_service();
    let req = request("DescribeDBSubnetGroups", &[("DBSubnetGroupName", "ghost")]);
    assert!(svc.describe_db_subnet_groups(&req).is_err());
}

/// Issue #914: the bg container-start task flips status from `creating`
/// to `available`. Without persisting after the flip, a restart loaded a
/// `creating` placeholder which the load path then dropped, making the
/// DB instance disappear. `save_snapshot_static` is the free fn the bg
/// task calls — exercise it directly to lock the contract: the latest
/// state lands on disk for every caller, not just service handlers.
#[tokio::test]
async fn save_snapshot_static_persists_status_flip_from_bg_task() {
    fn make_instance(id: &str, status: &str) -> DbInstance {
        let now = Utc::now();
        DbInstance {
            db_instance_identifier: id.to_string(),
            db_instance_arn: format!("arn:aws:rds:us-east-1:123456789012:db:{id}"),
            db_instance_class: "db.t3.micro".to_string(),
            engine: "postgres".to_string(),
            engine_version: "16.3".to_string(),
            db_instance_status: status.to_string(),
            master_username: "admin".to_string(),
            db_name: Some("appdb".to_string()),
            endpoint_address: String::new(),
            port: 0,
            allocated_storage: 20,
            publicly_accessible: true,
            deletion_protection: false,
            created_at: now,
            dbi_resource_id: format!("db-{id}"),
            master_user_password: "secret123".to_string(),
            container_id: String::new(),
            host_port: 0,
            tags: Vec::new(),
            read_replica_source_db_instance_identifier: None,
            read_replica_db_instance_identifiers: Vec::new(),
            vpc_security_group_ids: Vec::new(),
            db_parameter_group_name: None,
            backup_retention_period: 1,
            preferred_backup_window: "03:00-04:00".to_string(),
            preferred_maintenance_window: None,
            latest_restorable_time: Some(now),
            option_group_name: None,
            multi_az: false,
            pending_modified_values: None,
            availability_zone: None,
            storage_type: None,
            storage_encrypted: false,
            kms_key_id: None,
            iam_database_authentication_enabled: false,
            iops: None,
            monitoring_interval: None,
            monitoring_role_arn: None,
            performance_insights_enabled: false,
            performance_insights_kms_key_id: None,
            performance_insights_retention_period: None,
            enabled_cloudwatch_logs_exports: Vec::new(),
            ca_certificate_identifier: None,
            network_type: None,
            character_set_name: None,
            auto_minor_version_upgrade: None,
            copy_tags_to_snapshot: None,
            master_user_secret_arn: None,
            master_user_secret_kms_key_id: None,
            license_model: None,
            max_allocated_storage: None,
            multi_tenant: None,
            storage_throughput: None,
            tde_credential_arn: None,
            delete_automated_backups: None,
            db_security_groups: Vec::new(),
            domain: None,
            domain_fqdn: None,
            domain_ou: None,
            domain_iam_role_name: None,
            domain_auth_secret_arn: None,
            domain_dns_ips: Vec::new(),
            db_cluster_identifier: None,
        }
    }

    let tmp = tempfile::tempdir().unwrap();
    let path = tmp.path().join("rds.snapshot.json");
    let store: Arc<dyn SnapshotStore> = Arc::new(DiskSnapshotStore::new(path.clone()));
    let lock = Arc::new(AsyncMutex::new(()));

    let state: SharedRdsState = Arc::new(RwLock::new(
        fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
    ));
    {
        let mut accounts = state.write();
        let s = accounts.get_or_create("123456789012");
        s.instances
            .insert("db-1".to_string(), make_instance("db-1", "creating"));
    }

    // First save: simulates the synchronous CreateDBInstance handler save.
    save_snapshot_static(state.clone(), Some(store.clone()), lock.clone()).await;
    let bytes = std::fs::read(&path).expect("snapshot file should exist");
    let snap: RdsSnapshot = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(snap.schema_version, RDS_SNAPSHOT_SCHEMA_VERSION);
    let acc = snap.accounts.expect("multi-account");
    let s = acc.get("123456789012").expect("account state");
    assert_eq!(s.instances["db-1"].db_instance_status, "creating");

    // Bg task flips the status and saves again — the regression path.
    {
        let mut accounts = state.write();
        let s = accounts.get_or_create("123456789012");
        let inst = s.instances.get_mut("db-1").expect("placeholder still here");
        inst.db_instance_status = "available".to_string();
        inst.host_port = 15432;
        inst.port = 15432;
        inst.endpoint_address = "127.0.0.1".to_string();
        inst.container_id = "container-id".to_string();
    }
    save_snapshot_static(state.clone(), Some(store.clone()), lock.clone()).await;

    let bytes = std::fs::read(&path).unwrap();
    let snap: RdsSnapshot = serde_json::from_slice(&bytes).unwrap();
    let acc = snap.accounts.expect("multi-account");
    let s = acc.get("123456789012").expect("account state");
    assert_eq!(
        s.instances["db-1"].db_instance_status, "available",
        "post-bg-task save must overwrite the `creating` placeholder",
    );
    assert_eq!(s.instances["db-1"].host_port, 15432);
}

/// Memory mode: no store wired, save is a no-op. Guards against
/// accidentally requiring a store for the bg-task path.
#[tokio::test]
async fn save_snapshot_static_is_noop_without_store() {
    let lock = Arc::new(AsyncMutex::new(()));
    let state: SharedRdsState = Arc::new(RwLock::new(
        fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
    ));
    save_snapshot_static(state, None, lock).await;
}

// ── DescribeDBLogFiles / DownloadDBLogFilePortion (M10) ─────────

#[tokio::test]
async fn describe_db_log_files_returns_synthetic_files_when_runtime_absent() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    let req = request("DescribeDBLogFiles", &[("DBInstanceIdentifier", "db1")]);
    let body = body_of(svc.describe_db_log_files(&req).await.unwrap());
    assert!(
        body.contains("<LogFileName>error/postgres.log</LogFileName>"),
        "expected error/postgres.log entry in {body}"
    );
    assert!(body.contains("<LastWritten>"));
    assert!(body.contains("<Size>"));
}

#[tokio::test]
async fn describe_db_log_files_unknown_instance_returns_not_found() {
    let svc = make_service();
    let req = request("DescribeDBLogFiles", &[("DBInstanceIdentifier", "ghost")]);
    assert_code(svc.describe_db_log_files(&req).await, "DBInstanceNotFound");
}

#[tokio::test]
async fn describe_db_log_files_filename_contains_filter_applied() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    let req = request(
        "DescribeDBLogFiles",
        &[
            ("DBInstanceIdentifier", "db1"),
            ("FilenameContains", "trace"),
        ],
    );
    let body = body_of(svc.describe_db_log_files(&req).await.unwrap());
    assert!(
        body.contains("<LogFileName>trace/postgres-trace.log</LogFileName>"),
        "trace file should pass filter: {body}"
    );
    assert!(
        !body.contains("<LogFileName>error/postgres.log</LogFileName>"),
        "error file should be filtered out: {body}"
    );
}

#[tokio::test]
async fn download_db_log_file_portion_unknown_instance_errors() {
    let svc = make_service();
    let req = request(
        "DownloadDBLogFilePortion",
        &[
            ("DBInstanceIdentifier", "ghost"),
            ("LogFileName", "error/postgres.log"),
        ],
    );
    assert_code(
        svc.download_db_log_file_portion(&req).await,
        "DBInstanceNotFound",
    );
}

#[tokio::test]
async fn download_db_log_file_portion_returns_empty_when_runtime_absent() {
    let svc = make_service();
    seed_instance(&svc, "db1");
    let req = request(
        "DownloadDBLogFilePortion",
        &[
            ("DBInstanceIdentifier", "db1"),
            ("LogFileName", "error/postgres.log"),
        ],
    );
    let body = body_of(svc.download_db_log_file_portion(&req).await.unwrap());
    assert!(
        body.contains("<LogFileData></LogFileData>"),
        "expected empty LogFileData in {body}"
    );
    assert!(body.contains("<AdditionalDataPending>false</AdditionalDataPending>"));
    assert!(body.contains("<Marker>0</Marker>"));
}

/// No snapshot store (memory mode) -> no persist hook for the CFN provisioner.
#[test]
fn snapshot_hook_is_none_without_store() {
    let svc = make_service();
    assert!(svc.snapshot_hook().is_none());
}

/// With a store, the hook is present and invoking it runs the whole-state
/// persist path the CloudFormation provisioner uses after mutating RDS state
/// directly.
#[tokio::test]
async fn snapshot_hook_fires_with_store() {
    let store: Arc<dyn SnapshotStore> = Arc::new(fakecloud_persistence::MemorySnapshotStore::new());
    let svc = make_service().with_snapshot_store(store);
    let hook = svc
        .snapshot_hook()
        .expect("hook present when a store is set");
    hook().await;
}