fakecloud-stepfunctions 0.52.0

//! `aws-sdk:<service>:<action>` Tasks over each service's real wire protocol.
//!
//! A definition names an action and its parameters; the Smithy model of the
//! service says how those become a request. The table in
//! `data/sdk_models.json` (generated by `scripts/gen-sfn-sdk-models.py`) carries
//! each service's protocol, HTTP bindings and shapes, and this module drives it
//! in both directions, so no action needs code of its own:
//!
//! - Parameter names match model members whatever their case, and map keys are
//!   never touched, because the model says which values are maps.
//! - Results come back with PascalCase member names and ISO-8601 dates, as they
//!   do on AWS. Optimized integrations use the same conversion with epoch
//!   millisecond dates.

use std::collections::HashMap;
use std::io::Read;
use std::sync::{Arc, Mutex, OnceLock};

use base64::Engine;
use bytes::Bytes;
use chrono::{DateTime, SecondsFormat, Utc};
use fakecloud_core::auth::{Principal, PrincipalType};
use fakecloud_core::registry::ServiceRegistry;
use fakecloud_core::service::{AwsRequest, ResponseBody};
use http::{HeaderMap, HeaderName, HeaderValue, Method};
use percent_encoding::{utf8_percent_encode, AsciiSet, NON_ALPHANUMERIC};
use quick_xml::events::Event;
use quick_xml::Reader;
use serde_json::{json, Map, Value};
use tokio::io::AsyncReadExt;

/// RFC 3986 unreserved characters.
const UNRESERVED: &AsciiSet = &NON_ALPHANUMERIC
    .remove(b'-')
    .remove(b'_')
    .remove(b'.')
    .remove(b'~');
/// A greedy path label keeps its slashes.
const GREEDY_LABEL: &AsciiSet = &UNRESERVED.remove(b'/');

static TABLE: &str = include_str!("../data/sdk_models.json");

/// How a result presents dates.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub(crate) enum DateStyle {
    /// `aws-sdk:*`: ISO-8601, with milliseconds only when they are non-zero.
    Iso,
    /// Optimized integrations: integer epoch milliseconds.
    EpochMillis,
}

/// A service's protocol table. Each is decompressed and parsed on first use.
#[derive(Debug)]
pub(crate) struct SdkService {
    doc: Value,
}

/// Where an SDK call failed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum FailureOrigin {
    /// The service answered with an error (or could not be reached).
    Service,
    /// The SDK client could not use what the service answered: no status, no
    /// request id, and the code and message exactly as recorded.
    Client,
}

/// The failure of an SDK call.
#[derive(Debug)]
pub(crate) struct SdkFailure {
    pub(crate) code: String,
    pub(crate) message: String,
    pub(crate) status: u16,
    pub(crate) origin: FailureOrigin,
}

/// Why a service's answer could not be read into a result.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum ReplyError {
    /// A blob payload member whose bytes aren't text, as the SDK client refuses it.
    NotText { member: String },
    /// A body that doesn't parse as the protocol says.
    Unreadable(String),
}

impl From<String> for ReplyError {
    fn from(message: String) -> Self {
        ReplyError::Unreadable(message)
    }
}

impl From<ReplyError> for SdkFailure {
    fn from(error: ReplyError) -> Self {
        match error {
            ReplyError::NotText { member } => SdkFailure {
                code: "InvalidContent".to_string(),
                message: format!("Failed to convert '{member}' to string"),
                status: 0,
                origin: FailureOrigin::Client,
            },
            ReplyError::Unreadable(message) => SdkFailure {
                code: "InternalError".to_string(),
                message,
                status: 500,
                origin: FailureOrigin::Service,
            },
        }
    }
}

/// A raw service response.
pub(crate) struct RawReply {
    pub(crate) headers: HeaderMap,
    pub(crate) content_type: String,
    pub(crate) status: u16,
    pub(crate) body: Bytes,
}

pub(crate) struct Caller<'a> {
    pub(crate) registry: &'a Arc<ServiceRegistry>,
    pub(crate) account_id: &'a str,
    pub(crate) region: &'a str,
    /// Who the call is made as: the execution's role, for a Task.
    pub(crate) principal: Option<Principal>,
}

/// The assumed-role session a state machine's Tasks run as.
pub(crate) fn execution_principal(role_arn: &str, execution_name: &str) -> Option<Principal> {
    let account = fakecloud_aws::arn::account_of(role_arn)?;
    let role = role_arn.rsplit('/').next()?;
    let session = if execution_name.is_empty() {
        "execution"
    } else {
        execution_name
    };
    Some(Principal {
        arn: format!("arn:aws:sts::{account}:assumed-role/{role}/{session}"),
        user_id: format!("AROAFAKECLOUDSTATES:{session}"),
        account_id: account.to_string(),
        principal_type: PrincipalType::AssumedRole,
        source_identity: None,
        tags: None,
    })
}

/// The service a Task names: an SDK id (`sfn`, `cloudwatchlogs`), or the
/// `states` of a nested-execution ARN.
pub(crate) fn service(id: &str) -> Option<Arc<SdkService>> {
    static RAW: OnceLock<HashMap<String, String>> = OnceLock::new();
    static PARSED: OnceLock<Mutex<HashMap<String, Arc<SdkService>>>> = OnceLock::new();
    let key = match id {
        "states" => "sfn",
        other => other,
    };
    let parsed = PARSED.get_or_init(Default::default);
    if let Some(found) = parsed.lock().unwrap_or_else(|e| e.into_inner()).get(key) {
        return Some(found.clone());
    }
    let raw = RAW.get_or_init(|| serde_json::from_str(TABLE).expect("generated table is JSON"));
    let compressed = base64::engine::general_purpose::STANDARD
        .decode(raw.get(key)?)
        .expect("generated table is base64");
    let mut text = Vec::new();
    flate2::read::GzDecoder::new(compressed.as_slice())
        .read_to_end(&mut text)
        .expect("generated table is gzip");
    let doc = serde_json::from_slice(&text).expect("generated table entry is JSON");
    let service = Arc::new(SdkService { doc });
    parsed
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .insert(key.to_string(), service.clone());
    Some(service)
}

fn upper_first(key: &str) -> String {
    let mut chars = key.chars();
    match chars.next() {
        Some(first) => first.to_ascii_uppercase().to_string() + chars.as_str(),
        None => String::new(),
    }
}

/// The member of a structure a key names: exactly, or differing only in case.
fn find_member<'m>(members: &'m Map<String, Value>, key: &str) -> Option<(&'m String, &'m Value)> {
    members.get_key_value(key).or_else(|| {
        members.iter().find(|(name, member)| {
            name.eq_ignore_ascii_case(key)
                || member["j"]
                    .as_str()
                    .is_some_and(|wire| wire.eq_ignore_ascii_case(key))
        })
    })
}

fn scalar_text(value: &Value) -> String {
    crate::io_processing::value_text(value)
}

fn to_epoch_millis(value: &Value) -> Option<i64> {
    match value {
        Value::Number(n) => n.as_f64().map(|secs| (secs * 1000.0).round() as i64),
        Value::String(text) => DateTime::parse_from_rfc3339(text)
            .or_else(|_| DateTime::parse_from_rfc2822(text))
            .ok()
            .map(|t| t.timestamp_millis())
            .or_else(|| {
                text.parse::<f64>()
                    .ok()
                    .map(|s| (s * 1000.0).round() as i64)
            }),
        _ => None,
    }
}

fn present_date(value: &Value, style: DateStyle) -> Value {
    let Some(millis) = to_epoch_millis(value) else {
        return value.clone();
    };
    match style {
        DateStyle::EpochMillis => json!(millis),
        DateStyle::Iso => {
            let at = DateTime::<Utc>::from_timestamp_millis(millis).unwrap_or_default();
            let form = if millis % 1000 == 0 {
                SecondsFormat::Secs
            } else {
                SecondsFormat::Millis
            };
            json!(at.to_rfc3339_opts(form, true))
        }
    }
}

/// A timestamp parameter in the form its member's model asks for.
fn timestamp_wire(value: &Value, format: Option<&str>, default: &str) -> Value {
    let Some(millis) = to_epoch_millis(value) else {
        return value.clone();
    };
    let Some(at) = DateTime::<Utc>::from_timestamp_millis(millis) else {
        return value.clone();
    };
    match format.unwrap_or(default) {
        "epoch-seconds" => json!(millis as f64 / 1000.0),
        "http-date" => json!(at.format("%a, %d %b %Y %H:%M:%S GMT").to_string()),
        _ => json!(at.to_rfc3339_opts(SecondsFormat::Millis, true)),
    }
}

/// Any JSON a definition gives for a text member is sent as its JSON text.
fn text_param(value: &Value) -> String {
    scalar_text(value)
}

/// A response XML element.
#[derive(Debug, Default)]
struct Node {
    name: String,
    text: String,
    attrs: Vec<(String, String)>,
    children: Vec<Node>,
}

impl Node {
    fn parse(xml: &[u8]) -> Result<Node, String> {
        let text = std::str::from_utf8(xml).map_err(|e| e.to_string())?;
        let mut reader = Reader::from_str(text);
        let mut stack = vec![Node::default()];
        loop {
            let event = reader.read_event().map_err(|e| e.to_string())?;
            match event {
                Event::Start(start) => stack.push(Node::open(&start)),
                Event::Empty(start) => {
                    let node = Node::open(&start);
                    stack
                        .last_mut()
                        .ok_or("unbalanced XML")?
                        .children
                        .push(node);
                }
                Event::Text(raw) => {
                    let unescaped = raw.unescape().map_err(|e| e.to_string())?;
                    stack
                        .last_mut()
                        .ok_or("unbalanced XML")?
                        .text
                        .push_str(&unescaped);
                }
                Event::CData(raw) => {
                    let data = String::from_utf8_lossy(raw.as_ref()).into_owned();
                    stack
                        .last_mut()
                        .ok_or("unbalanced XML")?
                        .text
                        .push_str(&data);
                }
                Event::End(_) => {
                    let node = stack.pop().ok_or("unbalanced XML")?;
                    stack
                        .last_mut()
                        .ok_or("unbalanced XML")?
                        .children
                        .push(node);
                }
                Event::Eof => break,
                _ => {}
            }
        }
        let mut document = stack.pop().ok_or("unbalanced XML")?;
        document
            .children
            .pop()
            .ok_or_else(|| "empty XML".to_string())
    }

    fn open(start: &quick_xml::events::BytesStart<'_>) -> Node {
        let local = start.local_name();
        Node {
            name: String::from_utf8_lossy(local.as_ref()).into_owned(),
            attrs: start
                .attributes()
                .flatten()
                .map(|a| {
                    (
                        String::from_utf8_lossy(a.key.local_name().as_ref()).into_owned(),
                        a.unescape_value()
                            .map(|v| v.into_owned())
                            .unwrap_or_default(),
                    )
                })
                .collect(),
            ..Node::default()
        }
    }

    fn children_named<'n>(&'n self, name: &'n str) -> impl Iterator<Item = &'n Node> + 'n {
        self.children
            .iter()
            .filter(move |child| child.name.eq_ignore_ascii_case(name))
    }
}

fn xml_escape(text: &str) -> String {
    quick_xml::escape::escape(text).into_owned()
}

impl SdkService {
    /// The name to report in errors: `S3.NoSuchKeyException`, `Sfn.*`.
    pub(crate) fn error_prefix(&self) -> String {
        let name = self.doc["n"].as_str().unwrap_or_default().replace(' ', "");
        if name == "DynamoDB" {
            return "DynamoDb".to_string();
        }
        if name.chars().any(|c| c.is_ascii_lowercase()) || name.len() <= 2 {
            return upper_first(&name);
        }
        // An all-capitals id reads as one word: `SFN` -> `Sfn`, `EC2` -> `Ec2`.
        upper_first(&name.to_ascii_lowercase())
    }

    pub(crate) fn protocol(&self) -> &str {
        self.doc["p"].as_str().unwrap_or_default()
    }

    /// The name the service is registered under here.
    pub(crate) fn registered_in<'r>(
        &self,
        registry: &'r ServiceRegistry,
    ) -> Option<(&'r Arc<dyn fakecloud_core::service::AwsService>, String)> {
        for key in ["r", "sg"] {
            let name = self.doc[key].as_str().unwrap_or_default();
            if let Some(found) = registry.get(name) {
                return Some((found, name.to_string()));
            }
        }
        None
    }

    /// The operation an action names, exactly or differing only in case.
    pub(crate) fn operation(&self, action: &str) -> Option<String> {
        let operations = self.doc["o"].as_object()?;
        if operations.contains_key(action) {
            return Some(action.to_string());
        }
        operations
            .keys()
            .find(|name| name.eq_ignore_ascii_case(action))
            .cloned()
    }

    fn shape(&self, target: &str) -> Option<&Value> {
        self.doc["h"].get(target)
    }

    fn op_field(&self, operation: &str, index: usize) -> &str {
        self.doc["o"][operation][index].as_str().unwrap_or_default()
    }

    fn input_shape(&self, operation: &str) -> &str {
        self.op_field(operation, 2)
    }

    fn output_shape(&self, operation: &str) -> &str {
        self.op_field(operation, 3)
    }

    /// Definition parameters as the operation's JSON request body.
    pub(crate) fn request_json(&self, operation: &str, parameters: &Value) -> Value {
        self.json_in(self.input_shape(operation), parameters)
    }

    fn json_in(&self, target: &str, value: &Value) -> Value {
        match target {
            "~s" => match value {
                Value::Null => Value::Null,
                other => Value::String(text_param(other)),
            },
            "~t" => to_epoch_millis(value)
                .map(|millis| json!(millis as f64 / 1000.0))
                .unwrap_or_else(|| value.clone()),
            _ => match (self.shape(target), value) {
                (Some(shape), Value::Object(object)) if shape["k"] == "s" => {
                    let empty = Map::new();
                    let members = shape["m"].as_object().unwrap_or(&empty);
                    Value::Object(
                        object
                            .iter()
                            .map(|(key, member_value)| match find_member(members, key) {
                                Some((name, member)) => (
                                    member["j"].as_str().unwrap_or(name).to_string(),
                                    if member["t"] == "~t" {
                                        timestamp_wire(
                                            member_value,
                                            member["tf"].as_str(),
                                            "epoch-seconds",
                                        )
                                    } else {
                                        self.json_in(
                                            member["t"].as_str().unwrap_or("~d"),
                                            member_value,
                                        )
                                    },
                                ),
                                None => (key.clone(), member_value.clone()),
                            })
                            .collect(),
                    )
                }
                (Some(shape), Value::Object(object)) if shape["k"] == "m" => {
                    let inner = shape["t"].as_str().unwrap_or("~d");
                    Value::Object(
                        object
                            .iter()
                            .map(|(key, v)| (key.clone(), self.json_in(inner, v)))
                            .collect(),
                    )
                }
                (Some(shape), Value::Array(items)) if shape["k"] == "l" => {
                    let inner = shape["t"].as_str().unwrap_or("~d");
                    Value::Array(items.iter().map(|v| self.json_in(inner, v)).collect())
                }
                _ => value.clone(),
            },
        }
    }

    /// An operation's JSON response as a Task result.
    pub(crate) fn result_json(&self, operation: &str, response: &Value, style: DateStyle) -> Value {
        self.json_out(self.output_shape(operation), response, style)
    }

    fn json_out(&self, target: &str, value: &Value, style: DateStyle) -> Value {
        if target == "~t" {
            return present_date(value, style);
        }
        match (self.shape(target), value) {
            (Some(shape), Value::Object(object)) if shape["k"] == "s" => {
                let empty = Map::new();
                let members = shape["m"].as_object().unwrap_or(&empty);
                Value::Object(
                    object
                        .iter()
                        // An empty pagination token is no token.
                        .filter(|(key, v)| {
                            !v.is_null() && !(key.starts_with("Next") && v.as_str() == Some(""))
                        })
                        .map(|(key, v)| match find_member(members, key) {
                            Some((name, member)) => (
                                upper_first(name),
                                self.json_out(member["t"].as_str().unwrap_or("~d"), v, style),
                            ),
                            None => (upper_first(key), v.clone()),
                        })
                        .collect(),
                )
            }
            (Some(shape), Value::Object(object)) if shape["k"] == "m" => {
                let inner = shape["t"].as_str().unwrap_or("~d");
                Value::Object(
                    object
                        .iter()
                        .map(|(key, v)| (key.clone(), self.json_out(inner, v, style)))
                        .collect(),
                )
            }
            (Some(shape), Value::Array(items)) if shape["k"] == "l" => {
                let inner = shape["t"].as_str().unwrap_or("~d");
                Value::Array(
                    items
                        .iter()
                        .map(|v| self.json_out(inner, v, style))
                        .collect(),
                )
            }
            _ => value.clone(),
        }
    }

    /// Flatten parameters into `awsQuery` / `ec2Query` request parameters.
    fn query_in(&self, prefix: &str, target: &str, value: &Value, out: &mut Vec<(String, String)>) {
        let ec2 = self.protocol() == "ec2";
        match (self.shape(target), value) {
            (Some(shape), Value::Object(object)) if shape["k"] == "s" => {
                let empty = Map::new();
                let members = shape["m"].as_object().unwrap_or(&empty);
                for (key, member_value) in object {
                    let Some((name, member)) = find_member(members, key) else {
                        continue;
                    };
                    let wire = if ec2 {
                        member["e"]
                            .as_str()
                            .map(String::from)
                            .or_else(|| member["x"].as_str().map(upper_first))
                            .unwrap_or_else(|| upper_first(name))
                    } else {
                        member["x"].as_str().unwrap_or(name).to_string()
                    };
                    let path = if prefix.is_empty() {
                        wire
                    } else {
                        format!("{prefix}.{wire}")
                    };
                    let inner = member["t"].as_str().unwrap_or("~s");
                    // An absent value is no parameter, a timestamp's included.
                    if member_value.is_null() {
                        continue;
                    }
                    if inner == "~t" {
                        let wire = timestamp_wire(member_value, member["tf"].as_str(), "date-time");
                        out.push((path, scalar_text(&wire)));
                        continue;
                    }
                    if member["f"] == 1 && !ec2 {
                        // A flattened list has no `member` step of its own.
                        if let (Some(list), Value::Array(items)) = (self.shape(inner), member_value)
                        {
                            let element = list["t"].as_str().unwrap_or("~s");
                            for (i, item) in items.iter().enumerate() {
                                self.query_in(&format!("{path}.{}", i + 1), element, item, out);
                            }
                            continue;
                        }
                    }
                    self.query_in(&path, inner, member_value, out);
                }
            }
            (Some(shape), Value::Array(items)) if shape["k"] == "l" => {
                let element = shape["t"].as_str().unwrap_or("~s");
                let step = shape["x"].as_str().unwrap_or("member");
                for (i, item) in items.iter().enumerate() {
                    let path = if ec2 {
                        format!("{prefix}.{}", i + 1)
                    } else {
                        format!("{prefix}.{step}.{}", i + 1)
                    };
                    self.query_in(&path, element, item, out);
                }
            }
            (Some(shape), Value::Object(object)) if shape["k"] == "m" => {
                let inner = shape["t"].as_str().unwrap_or("~s");
                let key_name = shape["kx"].as_str().unwrap_or("key");
                let value_name = shape["vx"].as_str().unwrap_or("value");
                for (i, (key, v)) in object.iter().enumerate() {
                    let entry = format!("{prefix}.entry.{}", i + 1);
                    out.push((format!("{entry}.{key_name}"), key.clone()));
                    self.query_in(&format!("{entry}.{value_name}"), inner, v, out);
                }
            }
            (_, Value::Null) => {}
            (_, other) => out.push((prefix.to_string(), scalar_text(other))),
        }
    }

    fn xml_in(&self, name: &str, target: &str, value: &Value, namespace: bool) -> String {
        let attr = match (namespace, self.doc["xn"].as_str()) {
            (true, Some(uri)) => format!(" xmlns=\"{uri}\""),
            _ => String::new(),
        };
        match (self.shape(target), value) {
            (Some(shape), Value::Object(object)) if shape["k"] == "s" => {
                let empty = Map::new();
                let members = shape["m"].as_object().unwrap_or(&empty);
                let mut inner = String::new();
                let mut attr = attr;
                for (key, member_value) in object {
                    let Some((member_name, member)) = find_member(members, key) else {
                        continue;
                    };
                    if member["l"] == 1 || !member["q"].is_null() || !member["h"].is_null() {
                        continue;
                    }
                    let wire = member["x"].as_str().unwrap_or(member_name);
                    if member["a"] == 1 {
                        // S3 grantees are the one modeled use of a prefixed attribute.
                        if wire.starts_with("xsi:") {
                            attr.push_str(
                                " xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\"",
                            );
                        }
                        attr.push_str(&format!(
                            " {wire}=\"{}\"",
                            xml_escape(&scalar_text(member_value))
                        ));
                        continue;
                    }
                    let member_target = member["t"].as_str().unwrap_or("~s");
                    if member_target == "~t" {
                        if member_value.is_null() {
                            continue;
                        }
                        let text = timestamp_wire(member_value, member["tf"].as_str(), "date-time");
                        inner.push_str(&self.xml_in(wire, "~s", &text, false));
                        continue;
                    }
                    if member["f"] == 1 {
                        if let (Some(list), Value::Array(items)) =
                            (self.shape(member_target), member_value)
                        {
                            let element = list["t"].as_str().unwrap_or("~s");
                            for item in items {
                                inner.push_str(&self.xml_in(wire, element, item, false));
                            }
                            continue;
                        }
                    }
                    inner.push_str(&self.xml_in(wire, member_target, member_value, false));
                }
                format!("<{name}{attr}>{inner}</{name}>")
            }
            (Some(shape), Value::Array(items)) if shape["k"] == "l" => {
                let element = shape["t"].as_str().unwrap_or("~s");
                let step = shape["x"].as_str().unwrap_or("member");
                let inner: String = items
                    .iter()
                    .map(|item| self.xml_in(step, element, item, false))
                    .collect();
                format!("<{name}{attr}>{inner}</{name}>")
            }
            (Some(shape), Value::Object(object)) if shape["k"] == "m" => {
                let value_target = shape["t"].as_str().unwrap_or("~s");
                let key_name = shape["kx"].as_str().unwrap_or("key");
                let value_name = shape["vx"].as_str().unwrap_or("value");
                let inner: String = object
                    .iter()
                    .map(|(key, v)| {
                        format!(
                            "<entry><{key_name}>{}</{key_name}>{}</entry>",
                            xml_escape(key),
                            self.xml_in(value_name, value_target, v, false)
                        )
                    })
                    .collect();
                format!("<{name}{attr}>{inner}</{name}>")
            }
            (_, other) => format!("<{name}{attr}>{}</{name}>", xml_escape(&scalar_text(other))),
        }
    }

    fn xml_scalar(&self, target: &str, text: &str, style: DateStyle) -> Value {
        match target {
            "~n" => text
                .parse::<i64>()
                .map(Value::from)
                .or_else(|_| text.parse::<f64>().map(|f| json!(f)))
                .unwrap_or_else(|_| json!(text)),
            "~B" => json!(text == "true"),
            "~t" => present_date(&json!(text), style),
            _ => json!(text),
        }
    }

    fn xml_out(&self, target: &str, node: &Node, style: DateStyle) -> Value {
        let Some(shape) = self.shape(target) else {
            return self.xml_scalar(target, &node.text, style);
        };
        match shape["k"].as_str() {
            Some("s") => {
                let empty = Map::new();
                let members = shape["m"].as_object().unwrap_or(&empty);
                let mut out = Map::new();
                for (name, member) in members {
                    if member["h"].is_string()
                        || member["hp"].is_string()
                        || member["rc"] == 1
                        || member["p"] == 1
                    {
                        continue;
                    }
                    let member_target = member["t"].as_str().unwrap_or("~s");
                    let wire = member["x"].as_str().unwrap_or(name);
                    if member["a"] == 1 {
                        // Attributes are read by local name: `xsi:type` is `type`.
                        let local = wire.rsplit(':').next().unwrap_or(wire);
                        if let Some((_, value)) = node
                            .attrs
                            .iter()
                            .find(|(key, _)| key.eq_ignore_ascii_case(local))
                        {
                            out.insert(upper_first(name), json!(value));
                        }
                        continue;
                    }
                    let found: Vec<&Node> = node.children_named(wire).collect();
                    if member["f"] == 1 {
                        if found.is_empty() {
                            continue;
                        }
                        let value = match self.shape(member_target) {
                            Some(list) if list["k"] == "l" => {
                                let element = list["t"].as_str().unwrap_or("~s");
                                Value::Array(
                                    found
                                        .iter()
                                        .map(|n| self.xml_out(element, n, style))
                                        .collect(),
                                )
                            }
                            Some(map) if map["k"] == "m" => {
                                self.xml_map_entries(map, &found, style)
                            }
                            _ => self.xml_out(member_target, found[0], style),
                        };
                        out.insert(upper_first(name), value);
                        continue;
                    }
                    if let Some(child) = found.first() {
                        out.insert(upper_first(name), self.xml_out(member_target, child, style));
                    }
                }
                Value::Object(out)
            }
            Some("l") => {
                let element = shape["t"].as_str().unwrap_or("~s");
                Value::Array(
                    node.children
                        .iter()
                        .map(|child| self.xml_out(element, child, style))
                        .collect(),
                )
            }
            Some("m") => {
                let entries: Vec<&Node> = node.children.iter().collect();
                self.xml_map_entries(shape, &entries, style)
            }
            _ => self.xml_scalar(target, &node.text, style),
        }
    }

    fn xml_map_entries(&self, map: &Value, entries: &[&Node], style: DateStyle) -> Value {
        let value_target = map["t"].as_str().unwrap_or("~s");
        let key_name = map["kx"].as_str().unwrap_or("key");
        let value_name = map["vx"].as_str().unwrap_or("value");
        let mut out = Map::new();
        for entry in entries {
            let key = entry.children_named(key_name).next();
            let value = entry.children_named(value_name).next();
            if let (Some(key), Some(value)) = (key, value) {
                out.insert(key.text.clone(), self.xml_out(value_target, value, style));
            }
        }
        Value::Object(out)
    }

    /// Run one operation over the service's wire protocol.
    pub(crate) async fn invoke(
        &self,
        caller: &Caller<'_>,
        operation: &str,
        parameters: &Value,
        style: DateStyle,
    ) -> Result<Value, SdkFailure> {
        let raw = self.send(caller, operation, parameters).await?;
        self.parse_reply(operation, raw, style)
            .map_err(SdkFailure::from)
    }

    /// Run one operation and return its raw response.
    pub(crate) async fn send(
        &self,
        caller: &Caller<'_>,
        operation: &str,
        parameters: &Value,
    ) -> Result<RawReply, SdkFailure> {
        let request = self.build_request(caller, operation, parameters)?;
        let (service, _) = self
            .registered_in(caller.registry)
            .ok_or_else(|| SdkFailure {
                code: "ServiceUnavailable".to_string(),
                message: format!("{} is not available", self.doc["n"]),
                status: 503,
                origin: FailureOrigin::Service,
            })?;
        let response = match service.handle(request).await {
            Ok(response) => response,
            Err(err) => {
                let failure = SdkFailure {
                    code: err.code().to_string(),
                    message: err.message(),
                    status: err.status().as_u16(),
                    origin: FailureOrigin::Service,
                };
                return Err(self
                    .missing_bucket_failure(caller, parameters, &failure)
                    .await
                    .unwrap_or(failure));
            }
        };
        let status = response.status.as_u16();
        let body = read_body(response.body)
            .await
            .map_err(|message| SdkFailure {
                code: "InternalError".to_string(),
                message,
                status: 500,
                origin: FailureOrigin::Service,
            })?;
        if status >= 400 {
            return Err(SdkFailure {
                code: status.to_string(),
                message: String::from_utf8_lossy(&body).into_owned(),
                status,
                origin: FailureOrigin::Service,
            });
        }
        Ok(RawReply {
            headers: response.headers,
            content_type: response.content_type,
            status,
            body,
        })
    }

    /// S3 answers `NoSuchKey` for an object call on a missing bucket, because
    /// that is the only error the model declares for those operations. The SDK
    /// behind a Task sees `NoSuchBucket`, so a missing key is checked against the
    /// bucket here.
    async fn missing_bucket_failure(
        &self,
        caller: &Caller<'_>,
        parameters: &Value,
        failure: &SdkFailure,
    ) -> Option<SdkFailure> {
        if self.doc["r"] != "s3" || failure.code != "NoSuchKey" {
            return None;
        }
        let bucket = parameters
            .get("Bucket")
            .or_else(|| parameters.get("bucket"))?;
        let probe = self
            .build_request(caller, "HeadBucket", &json!({ "Bucket": bucket }))
            .ok()?;
        let (service, _) = self.registered_in(caller.registry)?;
        match service.handle(probe).await {
            Err(err) if err.status().as_u16() == 404 => Some(SdkFailure {
                code: "NoSuchBucket".to_string(),
                message: "The specified bucket does not exist".to_string(),
                status: 404,
                origin: FailureOrigin::Service,
            }),
            _ => None,
        }
    }

    fn build_request(
        &self,
        caller: &Caller<'_>,
        operation: &str,
        parameters: &Value,
    ) -> Result<AwsRequest, SdkFailure> {
        let (_, registered) = self
            .registered_in(caller.registry)
            .ok_or_else(|| SdkFailure {
                code: "ServiceUnavailable".to_string(),
                message: format!("{} is not available", self.doc["n"]),
                status: 503,
                origin: FailureOrigin::Service,
            })?;
        let mut request = AwsRequest {
            service: registered,
            action: operation.to_string(),
            region: caller.region.to_string(),
            account_id: caller.account_id.to_string(),
            request_id: uuid::Uuid::new_v4().to_string(),
            headers: HeaderMap::new(),
            query_params: HashMap::new(),
            body: Bytes::new(),
            body_stream: parking_lot::Mutex::new(None),
            path_segments: Vec::new(),
            raw_path: "/".to_string(),
            raw_query: String::new(),
            method: Method::POST,
            is_query_protocol: false,
            access_key_id: None,
            principal: caller.principal.clone(),
        };
        match self.protocol() {
            "json" => {
                let body = self.request_json(operation, parameters);
                request.body = Bytes::from(body.to_string());
            }
            "query" | "ec2" => {
                let mut flat = vec![
                    ("Action".to_string(), operation.to_string()),
                    (
                        "Version".to_string(),
                        self.doc["v"].as_str().unwrap_or_default().to_string(),
                    ),
                ];
                self.query_in("", self.input_shape(operation), parameters, &mut flat);
                request.is_query_protocol = true;
                set_query(&mut request, flat);
            }
            _ => self.build_rest_request(&mut request, operation, parameters)?,
        }
        Ok(request)
    }

    fn build_rest_request(
        &self,
        request: &mut AwsRequest,
        operation: &str,
        parameters: &Value,
    ) -> Result<(), SdkFailure> {
        let invalid = |message: String| SdkFailure {
            code: "ValidationException".to_string(),
            message,
            status: 400,
            origin: FailureOrigin::Service,
        };
        let method = self.op_field(operation, 0);
        let (template, fixed_query) = {
            let uri = self.op_field(operation, 1);
            uri.split_once('?').unwrap_or((uri, ""))
        };
        let input_target = self.input_shape(operation);
        let empty = Map::new();
        let members = self
            .shape(input_target)
            .and_then(|shape| shape["m"].as_object())
            .unwrap_or(&empty);
        let params = parameters.as_object().unwrap_or(&empty);

        let mut path = template.to_string();
        let mut query: Vec<(String, String)> = fixed_query
            .split('&')
            .filter(|part| !part.is_empty())
            .map(|part| {
                let (name, value) = part.split_once('=').unwrap_or((part, ""));
                (name.to_string(), value.to_string())
            })
            .collect();
        let mut headers = HeaderMap::new();
        let mut body_fields = Map::new();
        let mut payload: Option<(&String, &Value, &Value)> = None;

        for (key, value) in params {
            let Some((name, member)) = find_member(members, key) else {
                body_fields.insert(key.clone(), value.clone());
                continue;
            };
            if value.is_null() {
                continue;
            }
            let target = member["t"].as_str().unwrap_or("~s");
            if member["l"] == 1 {
                let text = scalar_text(value);
                let greedy = format!("{{{name}+}}");
                if path.contains(&greedy) {
                    let encoded = utf8_percent_encode(&text, GREEDY_LABEL).to_string();
                    path = path.replace(&greedy, &encoded);
                } else {
                    let encoded = utf8_percent_encode(&text, UNRESERVED).to_string();
                    path = path.replace(&format!("{{{name}}}"), &encoded);
                }
            } else if let Some(wire) = member["q"].as_str() {
                match value {
                    other if target == "~t" => {
                        let text = timestamp_wire(other, member["tf"].as_str(), "date-time");
                        query.push((wire.to_string(), scalar_text(&text)));
                    }
                    Value::Array(items) => {
                        for item in items {
                            query.push((wire.to_string(), scalar_text(item)));
                        }
                    }
                    other => query.push((wire.to_string(), scalar_text(other))),
                }
            } else if member["qp"] == 1 {
                for (k, v) in value.as_object().unwrap_or(&empty) {
                    query.push((k.clone(), scalar_text(v)));
                }
            } else if let Some(wire) = member["h"].as_str() {
                let text = match (target, value) {
                    ("~t", other) => {
                        scalar_text(&timestamp_wire(other, member["tf"].as_str(), "http-date"))
                    }
                    (_, Value::Array(items)) => {
                        items.iter().map(scalar_text).collect::<Vec<_>>().join(",")
                    }
                    (_, other) => scalar_text(other),
                };
                insert_header(&mut headers, wire, &text)
                    .map_err(|m| invalid(format!("{name}: {m}")))?;
            } else if let Some(prefix) = member["hp"].as_str() {
                for (k, v) in value.as_object().unwrap_or(&empty) {
                    insert_header(&mut headers, &format!("{prefix}{k}"), &scalar_text(v))
                        .map_err(|m| invalid(format!("{name}: {m}")))?;
                }
            } else if member["p"] == 1 {
                payload = Some((name, member, value));
            } else {
                body_fields.insert(name.clone(), value.clone());
            }
        }

        if let Some(unfilled) = path
            .split('{')
            .nth(1)
            .and_then(|rest| rest.split('}').next())
        {
            return Err(invalid(format!(
                "The field '{}' is required",
                unfilled.trim_end_matches('+')
            )));
        }

        let xml = self.protocol() == "restXml";
        request.body = match payload {
            Some((name, member, value)) => {
                let target = member["t"].as_str().unwrap_or("~b");
                match self.shape(target) {
                    None => {
                        // As recorded, an S3 object uploaded by a Task with no type
                        // of its own is stored as text; any other blob payload is
                        // sent as bytes.
                        if !headers.contains_key(http::header::CONTENT_TYPE) {
                            let content_type = if is_text_upload(&self.doc, operation) {
                                TEXT_UPLOAD_CONTENT_TYPE
                            } else {
                                "application/octet-stream"
                            };
                            headers.insert(
                                http::header::CONTENT_TYPE,
                                HeaderValue::from_static(content_type),
                            );
                        }
                        Bytes::from(scalar_text(value))
                    }
                    Some(shape) if xml => {
                        let root = member["x"]
                            .as_str()
                            .or_else(|| shape["x"].as_str())
                            .unwrap_or(name);
                        Bytes::from(self.xml_in(root, target, value, true))
                    }
                    Some(_) => Bytes::from(self.json_in(target, value).to_string()),
                }
            }
            None if xml && !body_fields.is_empty() => {
                let root = self
                    .shape(input_target)
                    .and_then(|shape| shape["x"].as_str())
                    .unwrap_or(input_target);
                let fields = Value::Object(body_fields);
                Bytes::from(self.xml_in(root, input_target, &fields, true))
            }
            None if !xml
                && (!body_fields.is_empty() || matches!(method, "POST" | "PUT" | "PATCH")) =>
            {
                let fields = Value::Object(body_fields);
                Bytes::from(self.json_in(input_target, &fields).to_string())
            }
            None => Bytes::new(),
        };
        if !request.body.is_empty() && !headers.contains_key(http::header::CONTENT_TYPE) {
            let content_type = if xml {
                "application/xml"
            } else {
                "application/json"
            };
            headers.insert(
                http::header::CONTENT_TYPE,
                HeaderValue::from_static(content_type),
            );
        }

        request.method = Method::from_bytes(method.as_bytes()).unwrap_or(Method::POST);
        request.path_segments = fakecloud_core::path::split_path_segments(&path);
        request.raw_path = path;
        request.headers = headers;
        set_query(request, query);

        // The SDK behind a Task checksums uploads by default.
        if self.doc["r"] == "s3" && operation == "PutObject" {
            let crc = crc32fast::hash(&request.body);
            let encoded = base64::engine::general_purpose::STANDARD.encode(crc.to_be_bytes());
            insert_header(&mut request.headers, "x-amz-checksum-crc32", &encoded)
                .map_err(invalid)?;
        }

        // S3 takes object bodies only as a stream, and drains it for the rest.
        if self.doc["r"] == "s3"
            && request.method == Method::PUT
            && request.path_segments.len() >= 2
        {
            let body = std::mem::take(&mut request.body);
            request.body_stream = parking_lot::Mutex::new(Some(axum::body::Body::from(body)));
        }
        Ok(())
    }

    fn parse_reply(
        &self,
        operation: &str,
        reply: RawReply,
        style: DateStyle,
    ) -> Result<Value, ReplyError> {
        let target = self.output_shape(operation);
        match self.protocol() {
            "json" => {
                if reply.body.is_empty() {
                    return Ok(json!({}));
                }
                let parsed: Value = serde_json::from_slice(&reply.body)
                    .map_err(|e| format!("unreadable response: {e}"))?;
                Ok(self.json_out(target, &parsed, style))
            }
            "query" | "ec2" => {
                if reply.body.is_empty() || target.is_empty() {
                    return Ok(json!({}));
                }
                let root = Node::parse(&reply.body)?;
                let wrapper = format!("{operation}Result");
                let node = root.children_named(&wrapper).next().unwrap_or(&root);
                Ok(self.xml_out(target, node, style))
            }
            _ => self.parse_rest_reply(target, &reply, style),
        }
    }

    fn parse_rest_reply(
        &self,
        target: &str,
        reply: &RawReply,
        style: DateStyle,
    ) -> Result<Value, ReplyError> {
        let xml = self.protocol() == "restXml";
        let empty = Map::new();
        let members = self
            .shape(target)
            .and_then(|shape| shape["m"].as_object())
            .unwrap_or(&empty);
        let mut out = Map::new();
        let mut payload_member = None;

        for (name, member) in members {
            let member_target = member["t"].as_str().unwrap_or("~s");
            if member["rc"] == 1 {
                out.insert(upper_first(name), json!(reply.status));
            } else if member["p"] == 1 {
                payload_member = Some((name, member_target));
            } else if let Some(prefix) = member["hp"].as_str() {
                let mut map = Map::new();
                for (header, value) in &reply.headers {
                    if let (Some(key), Ok(text)) =
                        (header.as_str().strip_prefix(prefix), value.to_str())
                    {
                        map.insert(key.to_string(), json!(text));
                    }
                }
                out.insert(upper_first(name), Value::Object(map));
            } else if let Some(wire) = member["h"].as_str() {
                let found = reply
                    .headers
                    .get(wire.to_ascii_lowercase().as_str())
                    .and_then(|v| v.to_str().ok())
                    .map(String::from)
                    .or_else(|| match wire.to_ascii_lowercase().as_str() {
                        "content-type" if !reply.content_type.is_empty() => {
                            Some(reply.content_type.clone())
                        }
                        "content-length" => Some(reply.body.len().to_string()),
                        _ => None,
                    });
                if let Some(text) = found {
                    out.insert(
                        upper_first(name),
                        self.xml_scalar(member_target, &text, style),
                    );
                }
            }
        }

        match payload_member {
            Some((name, payload_target)) => {
                if self.shape(payload_target).is_none() {
                    // A blob payload (an S3 object's data) reads as its text, as
                    // recorded; bytes that aren't text fail the task.
                    let member = upper_first(name);
                    let Ok(body) = std::str::from_utf8(&reply.body) else {
                        return Err(ReplyError::NotText { member });
                    };
                    out.insert(member, json!(body));
                } else if !reply.body.is_empty() {
                    let value = if xml {
                        self.xml_out(payload_target, &Node::parse(&reply.body)?, style)
                    } else {
                        let parsed: Value = serde_json::from_slice(&reply.body)
                            .map_err(|e| format!("unreadable response: {e}"))?;
                        self.json_out(payload_target, &parsed, style)
                    };
                    out.insert(upper_first(name), value);
                }
            }
            None if reply.body.is_empty() => {}
            None if xml => {
                if let Value::Object(fields) =
                    self.xml_out(target, &Node::parse(&reply.body)?, style)
                {
                    out.extend(fields);
                }
            }
            None => {
                let parsed: Value = serde_json::from_slice(&reply.body)
                    .map_err(|e| format!("unreadable response: {e}"))?;
                if let Value::Object(fields) = self.json_out(target, &parsed, style) {
                    out.extend(fields);
                }
            }
        }
        Ok(Value::Object(out))
    }
}

/// What an S3 `PutObject` Task stores when it names no `ContentType`, as recorded.
const TEXT_UPLOAD_CONTENT_TYPE: &str = "text/plain; charset=UTF-8";

/// Whether an operation's blob payload is uploaded as text by default: only S3's
/// `PutObject`.
fn is_text_upload(doc: &Value, operation: &str) -> bool {
    doc["r"] == "s3" && operation == "PutObject"
}

fn set_query(request: &mut AwsRequest, query: Vec<(String, String)>) {
    request.raw_query = query
        .iter()
        .map(|(name, value)| {
            format!(
                "{}={}",
                utf8_percent_encode(name, UNRESERVED),
                utf8_percent_encode(value, UNRESERVED)
            )
        })
        .collect::<Vec<_>>()
        .join("&");
    request.query_params = query.into_iter().collect();
}

fn insert_header(headers: &mut HeaderMap, name: &str, value: &str) -> Result<(), String> {
    let name = HeaderName::try_from(name).map_err(|_| format!("invalid header {name}"))?;
    let value = HeaderValue::from_str(value).map_err(|_| "invalid header value".to_string())?;
    headers.append(name, value);
    Ok(())
}

/// A service answers with bytes or, for a disk-backed object, an open file.
pub(crate) async fn read_body(body: ResponseBody) -> Result<Bytes, String> {
    match body {
        ResponseBody::Bytes(bytes) => Ok(bytes),
        ResponseBody::File { mut file, size } => {
            let mut buf = Vec::with_capacity(size as usize);
            file.read_to_end(&mut buf)
                .await
                .map_err(|e| format!("failed to read the response body: {e}"))?;
            Ok(Bytes::from(buf))
        }
    }
}

/// An `aws-sdk` failure as a state error: the SDK exception name and its message.
pub(crate) fn task_error(service: &SdkService, failure: &SdkFailure) -> (String, String) {
    let prefix = service.error_prefix();
    if failure.origin == FailureOrigin::Client {
        return (
            format!("{prefix}.{}", failure.code),
            failure.message.clone(),
        );
    }
    let code = failure.code.rsplit('.').next().unwrap_or(&failure.code);
    let code = if code.ends_with("Exception") {
        code.to_string()
    } else {
        format!("{code}Exception")
    };
    let request_id = uuid::Uuid::new_v4().simple().to_string().to_uppercase();
    let extended = if prefix == "S3" {
        format!(", Extended Request ID: {}", &request_id[..16])
    } else {
        String::new()
    };
    (
        format!("{prefix}.{code}"),
        format!(
            "{} (Service: {prefix}, Status Code: {}, Request ID: {}{extended}) (SDK Attempt Count: 1)",
            failure.message, failure.status, &request_id[..16]
        ),
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    fn svc(id: &str) -> Arc<SdkService> {
        service(id).unwrap_or_else(|| panic!("no table for {id}"))
    }

    #[test]
    fn execution_principal_session_is_the_execution_name() {
        let role = "arn:aws:iam::123456789012:role/sfn-role";
        let p = execution_principal(role, "run-1").unwrap();
        assert_eq!(
            p.arn,
            "arn:aws:sts::123456789012:assumed-role/sfn-role/run-1"
        );
        assert_eq!(p.user_id, "AROAFAKECLOUDSTATES:run-1");
        assert_eq!(p.account_id, "123456789012");
        let unnamed = execution_principal(role, "").unwrap();
        assert_eq!(unnamed.user_id, "AROAFAKECLOUDSTATES:execution");
    }

    #[test]
    fn services_load_by_sdk_id_with_their_protocol() {
        assert_eq!(svc("sfn").protocol(), "json");
        assert_eq!(svc("states").protocol(), "json");
        assert_eq!(svc("cloudwatchlogs").protocol(), "json");
        assert_eq!(svc("lambda").protocol(), "restJson");
        assert_eq!(svc("s3").protocol(), "restXml");
        assert_eq!(svc("iam").protocol(), "query");
        assert_eq!(svc("ec2").protocol(), "ec2");
        assert!(service("nosuchservice").is_none());
    }

    #[test]
    fn a_null_timestamp_is_no_query_parameter() {
        let cloudwatch = svc("cloudwatch");
        let mut out = Vec::new();
        cloudwatch.query_in(
            "",
            cloudwatch.input_shape("GetMetricStatistics"),
            &json!({"Namespace": "n", "StartTime": null, "EndTime": "2026-11-09T12:58:06Z"}),
            &mut out,
        );
        let names: Vec<&str> = out.iter().map(|(name, _)| name.as_str()).collect();
        assert!(names.contains(&"Namespace"), "{names:?}");
        assert!(names.contains(&"EndTime"), "{names:?}");
        assert!(!names.contains(&"StartTime"), "{names:?}");
    }

    #[test]
    fn only_s3_put_object_defaults_to_a_text_upload() {
        assert!(is_text_upload(&svc("s3").doc, "PutObject"));
        assert!(!is_text_upload(&svc("s3").doc, "UploadPart"));
        assert!(!is_text_upload(&svc("lambda").doc, "PutObject"));
        assert!(!is_text_upload(&svc("lambda").doc, "Invoke"));
    }

    #[test]
    fn error_prefixes_read_like_the_sdk() {
        assert_eq!(svc("s3").error_prefix(), "S3");
        assert_eq!(svc("sfn").error_prefix(), "Sfn");
        assert_eq!(svc("dynamodb").error_prefix(), "DynamoDb");
        assert_eq!(svc("ec2").error_prefix(), "Ec2");
        assert_eq!(svc("cloudwatchlogs").error_prefix(), "CloudWatchLogs");
    }

    #[test]
    fn operations_match_whatever_the_action_case() {
        assert_eq!(
            svc("s3").operation("ListObjectsV2").unwrap(),
            "ListObjectsV2"
        );
        assert_eq!(
            svc("s3").operation("listobjectsv2").unwrap(),
            "ListObjectsV2"
        );
        assert!(svc("s3").operation("Nope").is_none());
    }

    #[test]
    fn nested_parameters_take_the_models_casing() {
        let request = svc("ecs").request_json(
            "RunTask",
            &json!({
                "Cluster": "c",
                "NetworkConfiguration": {
                    "AwsvpcConfiguration": {"Subnets": ["s"], "AssignPublicIp": "ENABLED"}
                },
                "Overrides": {
                    "ContainerOverrides": [{
                        "Name": "app",
                        "Environment": [{"Name": "KEY", "Value": "VALUE"}]
                    }]
                }
            }),
        );
        assert_eq!(request["cluster"], "c");
        assert_eq!(
            request["networkConfiguration"]["awsvpcConfiguration"]["assignPublicIp"],
            "ENABLED"
        );
        let container = &request["overrides"]["containerOverrides"][0];
        assert_eq!(container["name"], "app");
        assert_eq!(container["environment"][0]["value"], "VALUE");
    }

    #[test]
    fn free_form_map_keys_and_pascal_case_shapes_keep_their_names() {
        let task = svc("ecs").request_json(
            "RegisterTaskDefinition",
            &json!({"ContainerDefinitions": [{
                "DockerLabels": {"Team": "a"},
                "LogConfiguration": {"LogDriver": "awslogs", "Options": {"Awslogs-Group": "g"}}
            }]}),
        );
        let container = &task["containerDefinitions"][0];
        assert_eq!(container["dockerLabels"]["Team"], "a");
        assert_eq!(
            container["logConfiguration"]["options"]["Awslogs-Group"],
            "g"
        );

        // ECR's request members are camelCase but its Tag shape is PascalCase.
        let repo = svc("ecr").request_json(
            "CreateRepository",
            &json!({"RepositoryName": "r", "Tags": [{"Key": "k", "Value": "v"}]}),
        );
        assert_eq!(repo["repositoryName"], "r");
        assert_eq!(repo["tags"][0], json!({"Key": "k", "Value": "v"}));
    }

    #[test]
    fn structured_values_for_text_members_become_their_json_text() {
        let request = svc("sfn").request_json(
            "StartExecution",
            &json!({"StateMachineArn": "arn", "Input": {"KeepCase": 1}, "Name": "n"}),
        );
        assert_eq!(request["stateMachineArn"], "arn");
        assert_eq!(request["input"], "{\"KeepCase\":1}");
        let output = svc("sfn").request_json(
            "SendTaskSuccess",
            &json!({"TaskToken": "t", "Output": {"Done": true}}),
        );
        assert_eq!(output["output"], "{\"Done\":true}");
    }

    #[test]
    fn results_are_pascal_case_with_presented_dates() {
        let sfn = svc("sfn");
        // As recorded from `aws-sdk:sfn:describeStateMachine`.
        let response = json!({"name": "n", "creationDate": 1794229071.256, "description": null});
        assert_eq!(
            sfn.result_json("DescribeStateMachine", &response, DateStyle::Iso),
            json!({"Name": "n", "CreationDate": "2026-11-09T12:57:51.256Z"})
        );
        let dated = json!({"startDate": 1794229071.944, "executionArn": "e"});
        assert_eq!(
            sfn.result_json("StartExecution", &dated, DateStyle::EpochMillis),
            json!({"StartDate": 1794229071944i64, "ExecutionArn": "e"})
        );
    }

    #[test]
    fn whole_second_dates_omit_milliseconds() {
        assert_eq!(
            present_date(&json!("2026-11-09T12:58:06.000Z"), DateStyle::Iso),
            "2026-11-09T12:58:06Z"
        );
        assert_eq!(
            present_date(&json!(1794229086), DateStyle::EpochMillis),
            1794229086000i64
        );
    }

    #[test]
    fn query_parameters_flatten_by_protocol() {
        let mut iam = Vec::new();
        svc("iam").query_in(
            "",
            svc("iam").input_shape("CreateUser"),
            &json!({"UserName": "u", "Tags": [{"Key": "k", "Value": "v"}]}),
            &mut iam,
        );
        assert!(iam.contains(&("UserName".to_string(), "u".to_string())));
        assert!(iam.contains(&("Tags.member.1.Key".to_string(), "k".to_string())));

        let mut ec2 = Vec::new();
        svc("ec2").query_in(
            "",
            svc("ec2").input_shape("DescribeVpcs"),
            &json!({"VpcIds": ["vpc-1"], "Filters": [{"Name": "n", "Values": ["x"]}]}),
            &mut ec2,
        );
        assert!(
            ec2.contains(&("VpcId.1".to_string(), "vpc-1".to_string())),
            "{ec2:?}"
        );
        assert!(
            ec2.contains(&("Filter.1.Name".to_string(), "n".to_string())),
            "{ec2:?}"
        );
        assert!(
            ec2.contains(&("Filter.1.Value.1".to_string(), "x".to_string())),
            "{ec2:?}"
        );
    }

    #[test]
    fn query_results_unwrap_lists_and_result_elements() {
        let iam = svc("iam");
        let body = br#"<ListUsersResponse xmlns="https://iam.amazonaws.com/doc/2010-05-08/">
          <ListUsersResult><IsTruncated>false</IsTruncated><Users><member>
            <UserName>u</UserName><CreateDate>2026-11-09T12:58:06Z</CreateDate>
          </member></Users></ListUsersResult></ListUsersResponse>"#;
        let reply = RawReply {
            headers: HeaderMap::new(),
            content_type: String::new(),
            status: 200,
            body: Bytes::from_static(body),
        };
        let out = iam.parse_reply("ListUsers", reply, DateStyle::Iso).unwrap();
        assert_eq!(out["IsTruncated"], false);
        assert_eq!(out["Users"][0]["UserName"], "u");
        assert_eq!(out["Users"][0]["CreateDate"], "2026-11-09T12:58:06Z");
    }

    #[test]
    fn rest_xml_results_read_flattened_lists_headers_and_spaces() {
        let s3 = svc("s3");
        let body = br#"<ListBucketResult><Name>b</Name><KeyCount>2</KeyCount><IsTruncated>false</IsTruncated>
          <Contents><Key> spaced </Key><Size>7</Size><ETag>&quot;e&quot;</ETag><LastModified>2026-11-09T12:58:06.000Z</LastModified></Contents>
          <Contents><Key>b</Key><Size>8</Size></Contents>
          <CommonPrefixes><Prefix>p/</Prefix></CommonPrefixes></ListBucketResult>"#;
        let reply = RawReply {
            headers: HeaderMap::new(),
            content_type: "application/xml".to_string(),
            status: 200,
            body: Bytes::from_static(body),
        };
        let out = s3
            .parse_reply("ListObjectsV2", reply, DateStyle::Iso)
            .unwrap();
        assert_eq!(out["KeyCount"], 2);
        assert_eq!(out["IsTruncated"], false);
        assert_eq!(out["Contents"][0]["Key"], " spaced ");
        assert_eq!(out["Contents"][0]["ETag"], "\"e\"");
        assert_eq!(out["Contents"][0]["LastModified"], "2026-11-09T12:58:06Z");
        assert_eq!(out["Contents"][1]["Size"], 8);
        assert_eq!(out["CommonPrefixes"][0]["Prefix"], "p/");
    }

    #[test]
    fn merged_models_keep_shapes_that_share_a_short_name_apart() {
        let timestream = svc("timestreamwrite");
        let shapes = timestream.doc["h"].as_object().unwrap();
        assert!(
            shapes.keys().any(|name| name.contains('#')),
            "no qualified shape"
        );
        // Each operation resolves to a shape of its own namespace.
        let write = timestream.input_shape("WriteRecords");
        assert!(shapes.contains_key(write), "{write}");
        for name in shapes.keys() {
            let reduced = name.rsplit('#').next().unwrap();
            if name != reduced {
                assert!(!shapes.contains_key(reduced), "{reduced} is ambiguous");
            }
        }
    }

    #[test]
    fn timestamps_take_their_modeled_format() {
        let at = json!("2026-11-09T12:58:06Z");
        assert_eq!(
            timestamp_wire(&at, None, "epoch-seconds"),
            json!(1794229086.0)
        );
        assert_eq!(
            timestamp_wire(&at, Some("date-time"), "epoch-seconds"),
            json!("2026-11-09T12:58:06.000Z")
        );
        assert_eq!(
            timestamp_wire(&json!(1794229086), None, "http-date"),
            json!("Mon, 09 Nov 2026 12:58:06 GMT")
        );
    }

    #[test]
    fn xml_attributes_round_trip() {
        let s3 = svc("s3");
        let body = s3.xml_in(
            "Grantee",
            "Grantee",
            &json!({"Type": "CanonicalUser", "ID": "abc"}),
            false,
        );
        assert!(body.contains(" xsi:type=\"CanonicalUser\""), "{body}");
        assert!(body.contains("xmlns:xsi="), "{body}");
        assert!(body.contains("<ID>abc</ID>"), "{body}");
        assert!(!body.contains("<xsi:type>"), "{body}");

        let node = Node::parse(body.as_bytes()).unwrap();
        let back = s3.xml_out("Grantee", &node, DateStyle::Iso);
        assert_eq!(back["Type"], "CanonicalUser", "{back}");
    }

    #[test]
    fn non_text_blob_payloads_fail_as_the_sdk_client_does() {
        let s3 = svc("s3");
        let reply = RawReply {
            headers: HeaderMap::new(),
            content_type: String::new(),
            status: 200,
            body: Bytes::from_static(&[0xff, 0x00, 0xfe]),
        };
        let error = s3
            .parse_reply("GetObject", reply, DateStyle::Iso)
            .unwrap_err();
        assert_eq!(
            error,
            ReplyError::NotText {
                member: "Body".to_string()
            }
        );
        let failure = SdkFailure::from(error);
        assert_eq!(failure.origin, FailureOrigin::Client);
        assert_eq!(
            task_error(&s3, &failure),
            (
                "S3.InvalidContent".to_string(),
                "Failed to convert 'Body' to string".to_string()
            )
        );
    }

    #[test]
    fn xml_bodies_follow_the_model() {
        let s3 = svc("s3");
        let body = s3.xml_in(
            "Tagging",
            "Tagging",
            &json!({"TagSet": [{"Key": "k", "Value": "<v>"}]}),
            true,
        );
        assert_eq!(
            body,
            "<Tagging xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><TagSet><Tag><Key>k</Key>\
             <Value>&lt;v&gt;</Value></Tag></TagSet></Tagging>"
        );
    }
}