camel-integration-test 0.40.0

//! Scenario action runner tests (ADR-0069 sections 5 and 7).
//!
//! Unit-test module of the lib target, declared in `src/lib.rs` under
//! `#[cfg(test)]`. Every test wraps a [`FakeAdapter`] in a
//! single-entry [`PartnerRouter`] and drives [`run_scenario`] under a
//! tokio runtime; deadlines are real monotonic time and stay at
//! test-scale magnitudes.

use std::collections::BTreeMap;
use std::time::Duration;

use camel_api::Value;
use futures::future::BoxFuture;

use crate::adapters::{
    FakeAdapter, IncomingMessage, OutgoingMessage, PartnerAdapter, PartnerRouter, ReceiveError,
    TransportError,
};
use crate::document::{
    EndpointRef, Expectation, Provisioning, RouteSource, ScenarioAction, ScenarioDocument,
    ScenarioTarget, ValidateExpectation,
};
use crate::runner::{
    DocumentOutcome, ScenarioFailure, ScenarioVars, ScenarioVerdict, fill_bind_vars,
    interpolate_value, resolve_placeholders, run_scenario, run_scenario_document,
};

#[cfg(feature = "http")]
use crate::adapters::http::{HttpPartner, HttpWireRequest};
#[cfg(feature = "http")]
use crate::document::PartnerExpectation;
#[cfg(feature = "http")]
use crate::runner::matching_requests;

/// A bare endpoint reference with no provisioning and no bind variable.
fn endpoint(uri: &str) -> EndpointRef {
    EndpointRef {
        endpoint: uri.to_string(),
        provisioning: None,
        bind_var: None,
    }
}

/// A minimal document with the given actions and file-based routes.
fn doc_with(actions: Vec<ScenarioAction>) -> ScenarioDocument {
    ScenarioDocument {
        route_source: RouteSource::RouteFiles(vec!["routes.yaml".into()]),
        scenario: actions,
        partners: None,
        env: None,
        env_passthrough: None,
        profile: None,
    }
}

/// A single-entry router over one fake adapter, keyed by endpoint URI.
fn router_for(uri: &str, fake: FakeAdapter) -> PartnerRouter {
    PartnerRouter::new(BTreeMap::from([(
        uri.to_string(),
        Box::new(fake) as Box<dyn PartnerAdapter>,
    )]))
}

/// An incoming message with a string body and no headers.
fn text_message(body: &str) -> IncomingMessage {
    IncomingMessage {
        body: Value::String(body.to_string()),
        headers: BTreeMap::new(),
        status: None,
        method: None,
        path: None,
    }
}

#[tokio::test]
async fn send_then_receive_within_deadline() {
    let fake = FakeAdapter::scripted(vec![text_message("hello")]);
    let router = router_for("partner://fake", fake);
    let doc = doc_with(vec![
        ScenarioAction::Send {
            to: endpoint("partner://fake"),
            body: Some(Value::String("hello".to_string())),
            headers: None,
            method: "POST".to_string(),
        },
        ScenarioAction::Receive {
            from: endpoint("partner://fake"),
            deadline: Duration::from_secs(1),
            extract: None,
        },
        ScenarioAction::Validate {
            target: ScenarioTarget::LastReceived(endpoint("partner://fake")),
            expectation: ValidateExpectation::Message(Expectation::Equals(Value::String(
                "hello".to_string(),
            ))),
            deadline: None,
        },
    ]);
    let mut vars = ScenarioVars::new();
    let verdict = run_scenario(&doc, &router, &mut vars).await;
    assert_eq!(verdict, Ok(ScenarioVerdict::Pass));
}

#[tokio::test]
async fn receive_timeout_is_verdict_failure() {
    let fake = FakeAdapter::scripted(Vec::new());
    let router = router_for("partner://fake", fake);
    let doc = doc_with(vec![ScenarioAction::Receive {
        from: endpoint("partner://fake"),
        deadline: Duration::from_millis(50),
        extract: None,
    }]);
    let mut vars = ScenarioVars::new();
    let failure = run_scenario(&doc, &router, &mut vars)
        .await
        .expect_err("empty queue must time out");
    assert!(
        matches!(failure, ScenarioFailure::ReceiveTimeout { .. }),
        "expected ReceiveTimeout, got {failure:?}"
    );
    assert!(
        failure.to_string().starts_with("receive-timeout"),
        "error must name the receive-timeout class: {failure}"
    );
}

#[tokio::test]
async fn variable_extraction_flows_forward() {
    fn scripted_with_id(id: &str) -> FakeAdapter {
        FakeAdapter::scripted(vec![IncomingMessage {
            body: Value::String("payload".to_string()),
            headers: BTreeMap::from([("X-Id".to_string(), Value::String(id.to_string()))]),
            status: None,
            method: None,
            path: None,
        }])
    }
    fn extraction_doc() -> ScenarioDocument {
        doc_with(vec![
            ScenarioAction::Receive {
                from: endpoint("partner://fake"),
                deadline: Duration::from_secs(1),
                extract: Some(BTreeMap::from([(
                    "id".to_string(),
                    "headers.X-Id".to_string(),
                )])),
            },
            ScenarioAction::Validate {
                target: ScenarioTarget::Variable("id".to_string()),
                expectation: ValidateExpectation::Message(Expectation::Equals(Value::String(
                    "abc-123".to_string(),
                ))),
                deadline: None,
            },
        ])
    }

    // Matching header: extraction sets the variable, validation passes.
    let router = router_for("partner://fake", scripted_with_id("abc-123"));
    let mut vars = ScenarioVars::new();
    let verdict = run_scenario(&extraction_doc(), &router, &mut vars).await;
    assert_eq!(verdict, Ok(ScenarioVerdict::Pass));
    assert_eq!(
        vars.get("id"),
        Some(&Value::String("abc-123".to_string())),
        "extraction must persist the variable for later actions"
    );

    // Mismatched header: validation fails with the action index named.
    let router = router_for("partner://fake", scripted_with_id("nope"));
    let mut vars = ScenarioVars::new();
    let failure = run_scenario(&extraction_doc(), &router, &mut vars)
        .await
        .expect_err("mismatched header must fail validation");
    assert!(
        matches!(
            failure,
            ScenarioFailure::ValidationMismatch { action: 1, .. }
        ),
        "expected ValidationMismatch on action 1, got {failure:?}"
    );
}

#[tokio::test]
async fn transport_error_is_apparatus_failure() {
    let fake = FakeAdapter::failing_send("connection refused");
    let router = router_for("partner://fake", fake);
    let doc = doc_with(vec![ScenarioAction::Send {
        to: endpoint("partner://fake"),
        body: None,
        headers: None,
        method: "GET".to_string(),
    }]);
    let mut vars = ScenarioVars::new();
    let failure = run_scenario(&doc, &router, &mut vars)
        .await
        .expect_err("failing send must fail the scenario");
    assert!(
        matches!(failure, ScenarioFailure::ActionTransport { action: 0, .. }),
        "expected ActionTransport on action 0, got {failure:?}"
    );
    assert!(
        failure.to_string().starts_with("action-transport-failure"),
        "error must name the action-transport-failure class: {failure}"
    );
}

/// A receive that fails at the transport mid-scenario is apparatus
/// class (`action-transport-failure`), not a verdict-class timeout.
#[tokio::test]
async fn receive_transport_error_is_apparatus_failure() {
    let fake = FakeAdapter::failing_receive("connection reset");
    let router = router_for("partner://fake", fake);
    let doc = doc_with(vec![ScenarioAction::Receive {
        from: endpoint("partner://fake"),
        deadline: Duration::from_secs(1),
        extract: None,
    }]);
    let mut vars = ScenarioVars::new();
    let failure = run_scenario(&doc, &router, &mut vars)
        .await
        .expect_err("failing receive must fail the scenario");
    assert!(
        matches!(failure, ScenarioFailure::ActionTransport { action: 0, .. }),
        "expected ActionTransport on action 0, got {failure:?}"
    );
    assert!(
        failure.to_string().starts_with("action-transport-failure"),
        "error must name the action-transport-failure class: {failure}"
    );
}

// -------------------------------------------------------------------------
// Adapter-level contract checks (dispatch, recording, message shapes)
// -------------------------------------------------------------------------

/// The router dispatches by endpoint equality and records sends on the
/// owning fake.
#[tokio::test]
async fn router_dispatches_and_fake_records_sends() {
    let fake = FakeAdapter::scripted(Vec::new());
    let handle = fake.recorder();
    let router = router_for("partner://fake", fake);
    let sent = OutgoingMessage {
        body: Value::String("recorded".to_string()),
        headers: BTreeMap::from([("X-Trace".to_string(), Value::String("t1".to_string()))]),
        method: "POST".to_string(),
    };
    router
        .send("partner://fake", "partner://fake", sent)
        .await
        .expect("send must succeed");
    let recorded = handle.sent_messages();
    assert_eq!(recorded.len(), 1);
    assert_eq!(recorded[0].endpoint, "partner://fake");
    assert_eq!(
        recorded[0].message.body,
        Value::String("recorded".to_string())
    );

    // Unknown endpoint: the send fails at the transport, and the
    // receive fails at the transport too — no partner exists that
    // could ever deliver, so the failure is apparatus class, not a
    // verdict-class timeout, and the call never hangs.
    let err = router
        .send(
            "partner://other",
            "partner://other",
            OutgoingMessage {
                body: Value::Null,
                headers: BTreeMap::new(),
                method: "GET".to_string(),
            },
        )
        .await
        .expect_err("unbound endpoint must fail");
    assert!(matches!(err, TransportError::Unbound { .. }));
    let failure = router
        .receive(
            "partner://other",
            "partner://other",
            Duration::from_secs(30),
        )
        .await
        .expect_err("unbound endpoint must never deliver");
    assert!(
        matches!(
            failure,
            ReceiveError::Transport(TransportError::Unbound { .. })
        ),
        "expected Transport(Unbound), got {failure:?}"
    );
}

// -------------------------------------------------------------------------
// Selector grammar (status / method / path heads, case-insensitive
// header lookup — ADR-0069 section 5 partner-side validation)
// -------------------------------------------------------------------------

/// `extract` selectors reach the transport status, the request method,
/// and the request path a partner adapter reports.
#[tokio::test]
async fn selector_extracts_status_method_and_path() {
    let fake = FakeAdapter::scripted(vec![IncomingMessage {
        body: Value::String("payload".to_string()),
        headers: BTreeMap::new(),
        status: Some(201),
        method: Some("POST".to_string()),
        path: Some("/orders".to_string()),
    }]);
    let router = router_for("partner://fake", fake);
    let doc = doc_with(vec![
        ScenarioAction::Receive {
            from: endpoint("partner://fake"),
            deadline: Duration::from_secs(1),
            extract: Some(BTreeMap::from([
                ("status".to_string(), "status".to_string()),
                ("method".to_string(), "method".to_string()),
                ("path".to_string(), "path".to_string()),
            ])),
        },
        ScenarioAction::Validate {
            target: ScenarioTarget::Variable("status".to_string()),
            expectation: ValidateExpectation::Message(Expectation::Equals(Value::Number(
                201.into(),
            ))),
            deadline: None,
        },
        ScenarioAction::Validate {
            target: ScenarioTarget::Variable("method".to_string()),
            expectation: ValidateExpectation::Message(Expectation::Equals(Value::String(
                "POST".to_string(),
            ))),
            deadline: None,
        },
        ScenarioAction::Validate {
            target: ScenarioTarget::Variable("path".to_string()),
            expectation: ValidateExpectation::Message(Expectation::Equals(Value::String(
                "/orders".to_string(),
            ))),
            deadline: None,
        },
    ]);
    let mut vars = ScenarioVars::new();
    let verdict = run_scenario(&doc, &router, &mut vars).await;
    assert_eq!(verdict, Ok(ScenarioVerdict::Pass));
}

/// Header lookup is ASCII-case-insensitive: the same selector behaves
/// identically whether the adapter preserved author casing (`X-Trace`)
/// or the wire normalized it to lowercase (`x-trace`), and vice versa.
#[tokio::test]
async fn selector_header_lookup_is_case_insensitive() {
    fn scripted_header(header_key: &str) -> FakeAdapter {
        FakeAdapter::scripted(vec![IncomingMessage {
            body: Value::Null,
            headers: BTreeMap::from([(header_key.to_string(), Value::String("t-42".to_string()))]),
            status: None,
            method: None,
            path: None,
        }])
    }
    fn doc(selector: &str) -> ScenarioDocument {
        doc_with(vec![
            ScenarioAction::Receive {
                from: endpoint("partner://fake"),
                deadline: Duration::from_secs(1),
                extract: Some(BTreeMap::from([(
                    "trace".to_string(),
                    selector.to_string(),
                )])),
            },
            ScenarioAction::Validate {
                target: ScenarioTarget::Variable("trace".to_string()),
                expectation: ValidateExpectation::Message(Expectation::Equals(Value::String(
                    "t-42".to_string(),
                ))),
                deadline: None,
            },
        ])
    }

    // Author-cased header, author-cased selector (the FakeAdapter shape).
    let router = router_for("partner://fake", scripted_header("X-Trace"));
    let mut vars = ScenarioVars::new();
    let verdict = run_scenario(&doc("headers.X-Trace"), &router, &mut vars).await;
    assert_eq!(verdict, Ok(ScenarioVerdict::Pass));

    // Wire-lowercased header, author-cased selector (the hyper shape).
    let router = router_for("partner://fake", scripted_header("x-trace"));
    let mut vars = ScenarioVars::new();
    let verdict = run_scenario(&doc("headers.X-Trace"), &router, &mut vars).await;
    assert_eq!(verdict, Ok(ScenarioVerdict::Pass));

    // Author-cased header, lowercase selector.
    let router = router_for("partner://fake", scripted_header("X-Trace"));
    let mut vars = ScenarioVars::new();
    let verdict = run_scenario(&doc("headers.x-trace"), &router, &mut vars).await;
    assert_eq!(verdict, Ok(ScenarioVerdict::Pass));
}

// -------------------------------------------------------------------------
// Document-level execution (run_scenario_document)
// -------------------------------------------------------------------------

/// The document run records one outcome per executed action, passes
/// the verdict when every action passed, and leaves the
/// post-shutdown slot empty for the caller.
#[tokio::test]
async fn document_run_all_pass_records_verdict() {
    let fake = FakeAdapter::scripted(vec![text_message("one"), text_message("two")]);
    let router = router_for("partner://fake", fake);
    let doc = doc_with(vec![
        ScenarioAction::Receive {
            from: endpoint("partner://fake"),
            deadline: Duration::from_secs(1),
            extract: None,
        },
        ScenarioAction::Validate {
            target: ScenarioTarget::Variable("unset".to_string()),
            expectation: ValidateExpectation::Message(Expectation::Exists),
            deadline: None,
        },
    ]);
    // Seed the variable so the `Exists` validation passes.
    let mut vars = ScenarioVars::new();
    vars.set("unset", Value::String("set".to_string()));

    let outcome = run_scenario_document(&doc, &router, &mut vars).await;
    assert_eq!(
        outcome,
        DocumentOutcome {
            per_action: vec![Ok(ScenarioVerdict::Pass), Ok(ScenarioVerdict::Pass)],
            verdict: Some(ScenarioVerdict::Pass),
            final_failure: None,
        }
    );
}

/// The document run stops at the first failure: later actions never
/// execute, each executed action carries its own outcome, and no
/// verdict is recorded.
#[tokio::test]
async fn document_run_stops_at_first_failure() {
    let fake = FakeAdapter::scripted(vec![text_message("one")]);
    let router = router_for("partner://fake", fake);
    let doc = doc_with(vec![
        ScenarioAction::Receive {
            from: endpoint("partner://fake"),
            deadline: Duration::from_secs(1),
            extract: None,
        },
        ScenarioAction::Receive {
            from: endpoint("partner://fake"),
            deadline: Duration::from_millis(50),
            extract: None,
        },
        ScenarioAction::Receive {
            from: endpoint("partner://fake"),
            deadline: Duration::from_secs(1),
            extract: None,
        },
    ]);
    let mut vars = ScenarioVars::new();
    let outcome = run_scenario_document(&doc, &router, &mut vars).await;
    assert_eq!(outcome.per_action.len(), 2, "only two actions ran");
    assert_eq!(outcome.per_action[0], Ok(ScenarioVerdict::Pass));
    assert!(matches!(
        outcome.per_action[1],
        Err(ScenarioFailure::ReceiveTimeout { .. })
    ));
    assert_eq!(outcome.verdict, None, "no verdict after a failure");
    assert_eq!(outcome.final_failure, None);
    assert!(
        vars.last_received("partner://fake").is_some(),
        "executed actions' side effects must persist"
    );
}

/// A validation mismatch on a variable names the variable, so a
/// corrupted-header regression is diagnosable from the failure text.
#[tokio::test]
async fn variable_mismatch_names_the_variable() {
    let fake = FakeAdapter::scripted(vec![IncomingMessage {
        body: Value::Null,
        headers: BTreeMap::from([(
            "X-Order-Type".to_string(),
            Value::String("priority".to_string()),
        )]),
        status: None,
        method: None,
        path: None,
    }]);
    let router = router_for("partner://fake", fake);
    let doc = doc_with(vec![
        ScenarioAction::Receive {
            from: endpoint("partner://fake"),
            deadline: Duration::from_secs(1),
            extract: Some(BTreeMap::from([(
                "orderType".to_string(),
                "headers.X-Order-Type".to_string(),
            )])),
        },
        ScenarioAction::Validate {
            target: ScenarioTarget::Variable("orderType".to_string()),
            expectation: ValidateExpectation::Message(Expectation::Equals(Value::String(
                "express".to_string(),
            ))),
            deadline: None,
        },
    ]);
    let mut vars = ScenarioVars::new();
    let outcome = run_scenario_document(&doc, &router, &mut vars).await;
    assert_eq!(outcome.verdict, None);
    match &outcome.per_action[1] {
        Err(ScenarioFailure::ValidationMismatch { action: 1, detail }) => {
            assert!(
                detail.contains("orderType"),
                "mismatch must name the variable: {detail}"
            );
            assert!(
                detail.contains("express") && detail.contains("priority"),
                "mismatch must show expected and actual: {detail}"
            );
        }
        other => panic!("expected ValidationMismatch on action 1, got {other:?}"),
    }
}

/// Compile-time shape check: the trait is object-safe and the trait
/// object is Send + Sync, as the runner and the router map require.
#[test]
fn partner_adapter_trait_object_is_send_sync() {
    fn assert_send_sync<T: Send + Sync + ?Sized>() {}
    assert_send_sync::<dyn PartnerAdapter>();
    assert_send_sync::<Box<dyn PartnerAdapter>>();
}

// -------------------------------------------------------------------------
// Placeholder resolution (${name} in scenario strings, ADR-0069 §5)
// -------------------------------------------------------------------------

/// A known variable substitutes its string value into the placeholder.
#[test]
fn resolve_substitutes_known_var() {
    let mut vars = ScenarioVars::new();
    vars.set("PARTNER", Value::String("127.0.0.1:9".to_string()));
    assert_eq!(
        resolve_placeholders("http://${PARTNER}/orders", &vars),
        Ok("http://127.0.0.1:9/orders".to_string())
    );
}

/// `$${` escapes to a literal `${`; the rest is scanned literally and
/// no lookup happens.
#[test]
fn resolve_escape_yields_literal() {
    let vars = ScenarioVars::new();
    assert_eq!(
        resolve_placeholders("$${not_a_var}", &vars),
        Ok("${not_a_var}".to_string())
    );
}

/// An unset variable fails with the variable's name named.
#[test]
fn resolve_unset_var_names_it() {
    let vars = ScenarioVars::new();
    assert_eq!(
        resolve_placeholders("${missing}", &vars),
        Err(ScenarioFailure::VarUnresolved {
            name: "missing".to_string()
        })
    );
}

/// A non-string variable substitutes its JSON representation.
#[test]
fn resolve_non_string_stringifies() {
    let mut vars = ScenarioVars::new();
    vars.set("N", Value::Number(42.into()));
    assert_eq!(resolve_placeholders("${N}", &vars), Ok("42".to_string()));
}

/// A name that does not match `[A-Za-z0-9_]+` stays literal.
#[test]
fn resolve_invalid_name_stays_literal() {
    let vars = ScenarioVars::new();
    assert_eq!(
        resolve_placeholders("${a-b}", &vars),
        Ok("${a-b}".to_string())
    );
}

/// An env-style placeholder (a colon after the name) stays literal:
/// `${env:}` never resolves in scenarios.
#[test]
fn resolve_env_placeholder_stays_literal() {
    let vars = ScenarioVars::new();
    assert_eq!(
        resolve_placeholders("${env:FOO}", &vars),
        Ok("${env:FOO}".to_string())
    );
}

/// Interpolation rebuilds maps and arrays recursively, substituting
/// string leaves and leaving other leaves untouched.
#[test]
fn interpolate_walks_nested_leaves() {
    let mut vars = ScenarioVars::new();
    vars.set("x", Value::String("1".to_string()));
    vars.set("y", Value::String("2".to_string()));
    let body = Value::Object(
        [
            (
                "a".to_string(),
                Value::Array(vec![
                    Value::String("${x}".to_string()),
                    Value::Number(1.into()),
                ]),
            ),
            (
                "b".to_string(),
                Value::Object(
                    [("c".to_string(), Value::String("${y}".to_string()))]
                        .into_iter()
                        .collect(),
                ),
            ),
        ]
        .into_iter()
        .collect(),
    );
    let expected = Value::Object(
        [
            (
                "a".to_string(),
                Value::Array(vec![
                    Value::String("1".to_string()),
                    Value::Number(1.into()),
                ]),
            ),
            (
                "b".to_string(),
                Value::Object(
                    [("c".to_string(), Value::String("2".to_string()))]
                        .into_iter()
                        .collect(),
                ),
            ),
        ]
        .into_iter()
        .collect(),
    );
    assert_eq!(interpolate_value(&body, &vars), Ok(expected));
}

/// An unset variable inside a nested body propagates the failure.
#[test]
fn interpolate_unset_in_body_propagates() {
    let vars = ScenarioVars::new();
    let body = Value::Object(
        [(
            "a".to_string(),
            Value::Object(
                [("b".to_string(), Value::String("${missing}".to_string()))]
                    .into_iter()
                    .collect(),
            ),
        )]
        .into_iter()
        .collect(),
    );
    assert_eq!(
        interpolate_value(&body, &vars),
        Err(ScenarioFailure::VarUnresolved {
            name: "missing".to_string()
        })
    );
}

// -------------------------------------------------------------------------
// Interpolation and bind vars in the action path (ADR-0069 §5, §9)
// -------------------------------------------------------------------------

/// An adapter standing in for one that owns a listener: it reports a
/// fixed bound authority and nothing else.
struct StaticAuthority(&'static str);

impl PartnerAdapter for StaticAuthority {
    fn receive<'a>(
        &'a self,
        _lane_key: &'a str,
        source_uri: &'a str,
        _deadline: Duration,
    ) -> BoxFuture<'a, Result<IncomingMessage, ReceiveError>> {
        Box::pin(async move {
            Err(ReceiveError::Transport(TransportError::Other {
                message: format!("{source_uri} has no receive role in this test"),
            }))
        })
    }

    fn bound_authority(&self) -> Option<String> {
        Some(self.0.to_string())
    }
}

/// A send to a dynamic `http://${PARTNER}/...` reference dials the
/// partner's bound authority: the interpolated URI resolves to the
/// registered partner by authority, and the partner listener records
/// the request path.
#[tokio::test]
#[cfg(feature = "http")]
async fn send_interpolates_endpoint() {
    let partner = HttpPartner::start_permissive(200)
        .await
        .expect("partner must bind 127.0.0.1:0");
    let recorder = partner.recorder();
    let authority = partner.bound_addr().to_string();
    let router = PartnerRouter::new(BTreeMap::from([(
        "http://127.0.0.1:0/orders".to_string(),
        Box::new(partner) as Box<dyn PartnerAdapter>,
    )]));
    let doc = doc_with(vec![
        ScenarioAction::Send {
            to: endpoint("http://${PARTNER}/orders"),
            body: None,
            headers: None,
            method: "POST".to_string(),
        },
        // The client lane dials in a spawned task (task 1.5 makes the
        // send await the connect); the receive takes the parked
        // roundtrip and synchronizes the server-side recording, the
        // same pattern the http partner tests use.
        ScenarioAction::Receive {
            from: endpoint("http://${PARTNER}/orders"),
            deadline: Duration::from_secs(5),
            extract: None,
        },
    ]);
    let mut vars = ScenarioVars::new();
    vars.set("PARTNER", Value::String(authority));
    run_scenario(&doc, &router, &mut vars)
        .await
        .expect("the interpolated send must reach the partner");
    let recorded = recorder.recorded_requests();
    assert_eq!(recorded.len(), 1, "exactly one request must reach the wire");
    assert_eq!(recorded[0].path, "/orders");
}

/// A send interpolates its body's string leaves and its header
/// values: the recorded wire request carries the substituted bytes.
#[tokio::test]
#[cfg(feature = "http")]
async fn send_interpolates_body_and_headers() {
    let partner = HttpPartner::start_permissive(200)
        .await
        .expect("partner must bind 127.0.0.1:0");
    let recorder = partner.recorder();
    let uri = format!("http://{}/orders", partner.bound_addr());
    let router = PartnerRouter::new(BTreeMap::from([(
        uri.clone(),
        Box::new(partner) as Box<dyn PartnerAdapter>,
    )]));
    let doc = doc_with(vec![
        ScenarioAction::Send {
            to: endpoint(&uri),
            body: Some(Value::Object(
                [("sku".to_string(), Value::String("${SKU}".to_string()))]
                    .into_iter()
                    .collect(),
            )),
            headers: Some(BTreeMap::from([(
                "X-Trace".to_string(),
                Value::String("${SKU}".to_string()),
            )])),
            method: "POST".to_string(),
        },
        // Synchronizes the spawned client-lane exchange and the
        // server-side recording.
        ScenarioAction::Receive {
            from: endpoint(&uri),
            deadline: Duration::from_secs(5),
            extract: None,
        },
    ]);
    let mut vars = ScenarioVars::new();
    vars.set("SKU", Value::String("x1".to_string()));
    run_scenario(&doc, &router, &mut vars)
        .await
        .expect("the send must reach the partner");
    let recorded = recorder.recorded_requests();
    assert_eq!(recorded.len(), 1);
    assert!(
        String::from_utf8_lossy(&recorded[0].body).contains("x1"),
        "recorded body must carry the substituted SKU: {:?}",
        recorded[0].body,
    );
    assert_eq!(
        recorded[0].headers.get("x-trace").map(String::as_str),
        Some("x1"),
        "recorded header must carry the substituted value"
    );
}

/// `fill_bind_vars` writes the partner's bound authority —
/// `host:port`, no scheme — into the scenario variable named by the
/// wired reference's `bindVar`.
#[test]
fn fill_bind_vars_sets_authority_without_scheme() {
    let uri = "http://127.0.0.1:0/orders";
    let router = PartnerRouter::new(BTreeMap::from([(
        uri.to_string(),
        Box::new(StaticAuthority("127.0.0.1:45678")) as Box<dyn PartnerAdapter>,
    )]));
    let wired = vec![EndpointRef {
        endpoint: uri.to_string(),
        provisioning: Some(Provisioning::Harness),
        bind_var: Some("PARTNER".to_string()),
    }];
    let mut vars = ScenarioVars::new();
    fill_bind_vars(&wired, &router, &mut vars);
    assert_eq!(
        vars.get("PARTNER"),
        Some(&Value::String("127.0.0.1:45678".to_string())),
        "the bind variable must carry the bare host:port authority"
    );
}

// -------------------------------------------------------------------------
// Partner validation (recorded-request counts, ADR-0069 §5)
// -------------------------------------------------------------------------

/// The declared harness endpoint every partner validate here reads.
#[cfg(feature = "http")]
const ORDERS: &str = "http://127.0.0.1:0/orders";

/// A single-entry router with `partner` registered under the declared
/// `:0` orders endpoint.
#[cfg(feature = "http")]
fn orders_router(partner: HttpPartner) -> PartnerRouter {
    PartnerRouter::new(BTreeMap::from([(
        ORDERS.to_string(),
        Box::new(partner) as Box<dyn PartnerAdapter>,
    )]))
}

/// A POST send to the declared `:0` orders endpoint.
#[cfg(feature = "http")]
fn orders_send() -> ScenarioAction {
    ScenarioAction::Send {
        to: endpoint(ORDERS),
        body: None,
        headers: None,
        method: "POST".to_string(),
    }
}

/// A partner validate on the declared orders endpoint: the count
/// expectation with optional method/path filters and an optional poll
/// deadline.
#[cfg(feature = "http")]
fn partner_validate(
    count: u64,
    method: Option<&str>,
    path: Option<&str>,
    deadline: Option<Duration>,
) -> ScenarioAction {
    ScenarioAction::Validate {
        target: ScenarioTarget::Partner(endpoint(ORDERS)),
        expectation: ValidateExpectation::Partner(PartnerExpectation {
            count,
            method: method.map(str::to_string),
            path: path.map(str::to_string),
        }),
        deadline,
    }
}

/// One raw HTTP/1.1 exchange straight to the partner's bound address —
/// the foreign-client arrival path no router lane owns.
/// `connection: close` makes it one write and one drained read, and
/// the partner records the request before it answers, so a completed
/// call means a recorded arrival.
#[cfg(feature = "http")]
async fn raw_request(authority: &str, method: &str, path: &str) {
    use tokio::io::AsyncReadExt;
    use tokio::io::AsyncWriteExt;
    let mut stream = tokio::net::TcpStream::connect(authority)
        .await
        .expect("the partner's bound address must accept");
    let request = format!(
        "{method} {path} HTTP/1.1\r\nhost: {authority}\r\nconnection: close\r\ncontent-length: 0\r\n\r\n"
    );
    stream
        .write_all(request.as_bytes())
        .await
        .expect("the raw request must leave");
    let mut sink = Vec::new();
    stream
        .read_to_end(&mut sink)
        .await
        .expect("the partner must close after its response");
}

/// The first failure of an outcome that must have failed.
#[cfg(feature = "http")]
fn first_failure(outcome: &DocumentOutcome) -> &ScenarioFailure {
    outcome
        .per_action
        .iter()
        .find_map(|result| result.as_ref().err())
        .expect("the document must have failed")
}

/// Filter semantics of `matching_requests`: the method filter folds
/// ASCII case, the path filter is the exact path-and-query, and `None`
/// filters pass everything.
#[test]
#[cfg(feature = "http")]
fn matching_requests_filters_method_case_insensitive_and_exact_path() {
    let wire = |method: &str, path: &str| HttpWireRequest {
        method: method.to_string(),
        path: path.to_string(),
        headers: BTreeMap::new(),
        body: Vec::new(),
    };
    let requests = vec![
        wire("POST", "/orders"),
        wire("GET", "/orders"),
        wire("GET", "/orders?page=2"),
        wire("GET", "/health"),
        wire("delete", "/orders"),
    ];
    // `None` filters pass every request.
    assert_eq!(matching_requests(&requests, None, None), 5);
    // The method filter folds ASCII case in both directions.
    assert_eq!(matching_requests(&requests, Some("get"), None), 3);
    assert_eq!(matching_requests(&requests, Some("DELETE"), None), 1);
    // The path filter is the exact path-and-query: no prefix and no
    // query-blind matching.
    assert_eq!(matching_requests(&requests, None, Some("/orders")), 3);
    assert_eq!(
        matching_requests(&requests, None, Some("/orders?page=2")),
        1
    );
    // Both filters combine conjunctively.
    assert_eq!(
        matching_requests(&requests, Some("get"), Some("/orders")),
        1
    );
}

/// The immediate snapshot: exact equality passes without a deadline,
/// and a mismatch names the partner URI and both counts. The receive
/// synchronizes the send's spawned client-lane exchange, so the
/// validates read a settled recorder.
#[tokio::test]
#[cfg(feature = "http")]
async fn immediate_count_passes_and_mismatch_names_counts() {
    let partner = HttpPartner::start_permissive(200)
        .await
        .expect("partner must bind 127.0.0.1:0");
    let router = orders_router(partner);
    let doc = doc_with(vec![
        orders_send(),
        ScenarioAction::Receive {
            from: endpoint(ORDERS),
            deadline: Duration::from_secs(5),
            extract: None,
        },
        partner_validate(1, None, None, None),
        partner_validate(2, None, None, None),
    ]);
    let mut vars = ScenarioVars::new();
    let outcome = run_scenario_document(&doc, &router, &mut vars).await;

    assert!(
        matches!(outcome.per_action[2], Ok(ScenarioVerdict::Pass)),
        "the exact immediate count must pass: {outcome:?}"
    );
    assert_eq!(outcome.verdict, None, "the count: 2 validate must fail");
    let ScenarioFailure::ValidationMismatch { detail, .. } = first_failure(&outcome) else {
        panic!(
            "expected ValidationMismatch, got {:?}",
            first_failure(&outcome)
        );
    };
    assert!(
        detail.contains("partner http://127.0.0.1:0/orders"),
        "the mismatch must name the partner URI: {detail}"
    );
    assert!(
        detail.contains("expected 2, actual 1"),
        "the mismatch must name both counts: {detail}"
    );
}

/// A filtered count mismatch names the applied filter clauses: the one
/// recorded POST to `/orders` matches both filters, so an expectation
/// of 2 fails with `method post, path /orders` spelled out in the
/// detail.
#[tokio::test]
#[cfg(feature = "http")]
async fn filtered_mismatch_names_method_and_path_clauses() {
    let partner = HttpPartner::start_permissive(200)
        .await
        .expect("partner must bind 127.0.0.1:0");
    let router = orders_router(partner);
    let doc = doc_with(vec![
        orders_send(),
        ScenarioAction::Receive {
            from: endpoint(ORDERS),
            deadline: Duration::from_secs(5),
            extract: None,
        },
        partner_validate(2, Some("post"), Some("/orders"), None),
    ]);
    let mut vars = ScenarioVars::new();
    let outcome = run_scenario_document(&doc, &router, &mut vars).await;

    assert_eq!(outcome.verdict, None, "the filtered count must fail");
    let ScenarioFailure::ValidationMismatch { detail, .. } = first_failure(&outcome) else {
        panic!(
            "expected ValidationMismatch, got {:?}",
            first_failure(&outcome)
        );
    };
    assert!(
        detail.contains("method post"),
        "the mismatch must name the method filter: {detail}"
    );
    assert!(
        detail.contains("path /orders"),
        "the mismatch must name the path filter: {detail}"
    );
    assert!(
        detail.contains("expected 2, actual 1"),
        "the mismatch must name both counts: {detail}"
    );
}

/// The polled snapshot settles: one arrival lands before the run, two
/// more at 300 ms while the validate polls, and the count reaches its
/// expectation long before the 5 s deadline — the pass comes from a
/// poll seeing the settle, not from waiting the deadline out.
#[tokio::test]
#[cfg(feature = "http")]
async fn poll_passes_once_count_settles() {
    let partner = HttpPartner::start_permissive(200)
        .await
        .expect("partner must bind 127.0.0.1:0");
    let authority = partner.bound_addr().to_string();
    // One arrival before the run: every early snapshot reads 1, below
    // the expectation, so the validate must keep polling.
    raw_request(&authority, "POST", "/orders").await;
    let router = orders_router(partner);
    let settling = tokio::spawn({
        let authority = authority.clone();
        async move {
            tokio::time::sleep(Duration::from_millis(300)).await;
            raw_request(&authority, "POST", "/orders").await;
            raw_request(&authority, "POST", "/orders").await;
        }
    });
    let doc = doc_with(vec![partner_validate(
        3,
        None,
        None,
        Some(Duration::from_secs(5)),
    )]);
    let mut vars = ScenarioVars::new();
    let outcome = run_scenario_document(&doc, &router, &mut vars).await;
    settling.await.expect("the settling task must finish");

    assert_eq!(
        outcome.verdict,
        Some(ScenarioVerdict::Pass),
        "the polled count must settle to 3: {outcome:?}"
    );
}

/// A count above the expectation never passes: arrivals only add, so
/// every polled snapshot and the final one read 4 against an
/// expectation of 3.
#[tokio::test]
#[cfg(feature = "http")]
async fn overshoot_never_passes() {
    let partner = HttpPartner::start_permissive(200)
        .await
        .expect("partner must bind 127.0.0.1:0");
    let authority = partner.bound_addr().to_string();
    for _ in 0..4 {
        raw_request(&authority, "POST", "/orders").await;
    }
    let router = orders_router(partner);
    let doc = doc_with(vec![partner_validate(
        3,
        None,
        None,
        Some(Duration::from_secs(1)),
    )]);
    let mut vars = ScenarioVars::new();
    let outcome = run_scenario_document(&doc, &router, &mut vars).await;

    assert_eq!(
        outcome.verdict, None,
        "a count above the expectation must never pass: {outcome:?}"
    );
    let ScenarioFailure::ValidationMismatch { detail, .. } = first_failure(&outcome) else {
        panic!(
            "expected ValidationMismatch, got {:?}",
            first_failure(&outcome)
        );
    };
    assert!(
        detail.contains("expected 3, actual 4"),
        "the mismatch must name the final counts: {detail}"
    );
}

/// Deadline expiry reports the final snapshot's count as the actual:
/// one arrival, an expectation of 3, and after the 1 s deadline the
/// failure names actual 1.
#[tokio::test]
#[cfg(feature = "http")]
async fn deadline_expiry_reports_final_actual() {
    let partner = HttpPartner::start_permissive(200)
        .await
        .expect("partner must bind 127.0.0.1:0");
    let authority = partner.bound_addr().to_string();
    raw_request(&authority, "POST", "/orders").await;
    let router = orders_router(partner);
    let doc = doc_with(vec![partner_validate(
        3,
        None,
        None,
        Some(Duration::from_secs(1)),
    )]);
    let mut vars = ScenarioVars::new();
    let outcome = run_scenario_document(&doc, &router, &mut vars).await;

    assert_eq!(outcome.verdict, None, "the count must never reach 3");
    let ScenarioFailure::ValidationMismatch { detail, .. } = first_failure(&outcome) else {
        panic!(
            "expected ValidationMismatch, got {:?}",
            first_failure(&outcome)
        );
    };
    assert!(
        detail.contains("actual 1"),
        "the mismatch must report the final snapshot's count: {detail}"
    );
}