camel-integration-test 0.41.0

//! Partner verification for the `validate` action's `partner`
//! target (ADR-0069 §5): the filtered recorded-request count, the
//! deadline poll with its early-settle and ceiling rules, and the
//! mismatch-detail renderers. Split out of the runner core so the
//! message-grammar validation and the partner count assertion stay
//! separately navigable; the runner dispatches the `partner` target
//! here.

#[cfg(feature = "http")]
use std::collections::BTreeMap;
use std::time::Duration;

use crate::adapters::PartnerRouter;
#[cfg(feature = "http")]
use crate::adapters::http::HttpWireRequest;
#[cfg(feature = "http")]
use crate::adapters::redact_wire_path;
use crate::document::PartnerExpectation;
#[cfg(feature = "http")]
use crate::document::{CountBound, PathFilter};

use super::ScenarioFailure;

/// The poll interval of a partner validate with a deadline
/// (feature `http`): a fresh recorded-request snapshot every 100 ms
/// until the filtered count settles or the deadline passes. The sleep
/// between snapshots means the poll never busy-waits.
#[cfg(feature = "http")]
const PARTNER_POLL_INTERVAL: Duration = Duration::from_millis(100);

/// Counts the recorded requests that pass all filters (feature
/// `http`): `method` compares ASCII-case-insensitively (wire records
/// are uppercased; the expectation may declare any casing), the path
/// filter matches the recorded path-and-query — `Exact` byte-for-byte,
/// `Contains` by substring, `Matches` by regex — and the `query`
/// subset requires every declared pair to appear among the request's
/// percent-decoded query pairs, in any position order. A `None`
/// filter passes everything, and all filters combine conjunctively.
///
/// This layer is the only place wire leniency lives: arrival-lane
/// keys stay strict raw wire bytes, never canonicalized.
#[cfg(feature = "http")]
pub(crate) fn matching_requests(
    requests: &[HttpWireRequest],
    method: Option<&str>,
    path_filter: Option<&PathFilter>,
    query: Option<&BTreeMap<String, String>>,
) -> usize {
    // The regex of a `Matches` filter compiles once per call, not once
    // per recorded request. An invalid pattern (the parser rejects it
    // first) matches nothing, failing closed.
    let matches_regex = match path_filter {
        Some(PathFilter::Matches(pattern)) => regex::Regex::new(pattern).ok(),
        _ => None,
    };
    requests
        .iter()
        .filter(|request| {
            let path_matches = match path_filter {
                None => true,
                Some(PathFilter::Exact(p)) => p.as_str() == request.path.as_str(),
                Some(PathFilter::Contains(s)) => request.path.contains(s.as_str()),
                Some(PathFilter::Matches(_)) => matches_regex
                    .as_ref()
                    .is_some_and(|re| re.is_match(&request.path)),
            };
            let query_subset = query.is_none_or(|declared| {
                let pairs = query_pairs(&request.path);
                declared
                    .iter()
                    .all(|(key, value)| pairs.iter().any(|(k, v)| k == key && v == value))
            });
            method.is_none_or(|m| m.eq_ignore_ascii_case(&request.method))
                && path_matches
                && query_subset
        })
        .count()
}

/// The percent-decoded query pairs of a recorded path-and-query
/// (feature `http`): everything after the first `?`, parsed with
/// `form_urlencoded`, which percent-decodes `%XX` and `+`. A path
/// without a query yields no pairs.
#[cfg(feature = "http")]
fn query_pairs(path_and_query: &str) -> Vec<(String, String)> {
    match path_and_query.split_once('?') {
        Some((_, query)) => form_urlencoded::parse(query.as_bytes())
            .map(|(key, value)| (key.into_owned(), value.into_owned()))
            .collect(),
        None => Vec::new(),
    }
}

/// Whether one snapshot's filtered count satisfies the bound: the
/// decision predicate of the no-deadline read and of the final expiry
/// snapshot.
#[cfg(feature = "http")]
fn bound_holds(bound: &CountBound, actual: usize) -> bool {
    let actual = actual as u64;
    match bound {
        CountBound::Exact(n) => actual == *n,
        CountBound::AtLeast(n) => actual >= *n,
        CountBound::AtMost(n) => actual <= *n,
        CountBound::Range(min, max) => actual >= *min && actual <= *max,
    }
}

/// Whether the poll may settle early on this snapshot. `Exact`
/// settles at equality and `AtLeast` at its floor — both sound
/// because arrivals only add, so a reached count stays reached.
/// `AtMost` and a `Range` never settle early: a passing snapshot
/// cannot prove the count stays within bounds while the window is
/// open.
#[cfg(feature = "http")]
fn settles_early(bound: &CountBound, actual: usize) -> bool {
    match bound {
        CountBound::Exact(_) | CountBound::AtLeast(_) => bound_holds(bound, actual),
        CountBound::AtMost(_) | CountBound::Range(..) => false,
    }
}

/// Whether this snapshot has already broken an upper bound beyond
/// recovery (`AtMost` above its ceiling, a `Range` above its
/// maximum): arrivals only add, so the claim fails on the first
/// observation instead of waiting the window out.
#[cfg(feature = "http")]
fn above_ceiling(bound: &CountBound, actual: usize) -> bool {
    let actual = actual as u64;
    match bound {
        CountBound::Exact(_) | CountBound::AtLeast(_) => false,
        CountBound::AtMost(n) => actual > *n,
        CountBound::Range(_, max) => actual > *max,
    }
}

/// Asserts the partner expectation against the router's
/// recorded-request snapshots for the declared endpoint key
/// (ADR-0069 §5: what crossed the wire is the normative proof).
///
/// Every snapshot filters by the expectation's `method`, `path`, and
/// `query` subset ([`matching_requests`]) and decides the filtered
/// count per [`CountBound`] (arrivals only add, so the count is
/// monotone non-decreasing). Without a deadline one immediate
/// snapshot decides for every bound. With a deadline the poll runs at
/// [`PARTNER_POLL_INTERVAL`]: `Exact` settles at equality and
/// `AtLeast` at its floor, while `AtMost` and a `Range` are absence
/// claims over the window — they wait the full deadline, fail
/// immediately on any snapshot above the ceiling, and decide on the
/// final snapshot. On expiry one final snapshot decides with its own
/// count the reported actual. Every snapshot clones out of the
/// recorder's lock before any await, so no lock spans an await point
/// and the poll sleeps between snapshots. Each iteration snapshots
/// the recorder once and derives both the filtered count and a
/// mismatch's evidence paths from that single snapshot, so the
/// evidence can never list an arrival the count missed.
#[cfg(feature = "http")]
pub(super) async fn partner_validate_action(
    index: usize,
    uri: &str,
    expected: &PartnerExpectation,
    deadline: Option<Duration>,
    router: &PartnerRouter,
) -> Result<(), ScenarioFailure> {
    // One recorder read per iteration: the filtered count and the
    // evidence paths both derive from the same snapshot, so a request
    // arriving between two reads can never make the evidence list
    // inconsistent with the reported actual.
    let snapshot = || {
        let requests = router.recorded_requests(uri);
        let actual = matching_requests(
            &requests,
            expected.method.as_deref(),
            expected.path.as_ref(),
            expected.query.as_ref(),
        );
        (requests, actual)
    };
    let mismatch = |actual: usize, requests: &[HttpWireRequest]| {
        // The recorded paths of the same snapshot, in arrival order:
        // the wire evidence the mismatch detail lists (redacted).
        let recorded: Vec<String> = requests
            .iter()
            .map(|request| request.path.clone())
            .collect();
        ScenarioFailure::ValidationMismatch {
            action: index,
            detail: partner_mismatch_detail(
                uri,
                expected,
                actual,
                &recorded,
                &router.secret_query_keys(),
            ),
        }
    };
    match deadline {
        // No deadline: one immediate snapshot decides for every
        // bound.
        None => {
            let (requests, actual) = snapshot();
            if bound_holds(&expected.bound, actual) {
                Ok(())
            } else {
                Err(mismatch(actual, &requests))
            }
        }
        // Poll: early success for `Exact`/`AtLeast`, immediate
        // failure above an `AtMost`/`Range` ceiling, and the final
        // snapshot decides at expiry.
        Some(deadline) => {
            let until = tokio::time::Instant::now() + deadline;
            loop {
                let (requests, actual) = snapshot();
                if above_ceiling(&expected.bound, actual) {
                    return Err(mismatch(actual, &requests));
                }
                if settles_early(&expected.bound, actual) {
                    return Ok(());
                }
                let now = tokio::time::Instant::now();
                if now >= until {
                    return if bound_holds(&expected.bound, actual) {
                        Ok(())
                    } else {
                        Err(mismatch(actual, &requests))
                    };
                }
                tokio::time::sleep((until - now).min(PARTNER_POLL_INTERVAL)).await;
            }
        }
    }
}

/// Partner verification needs the http adapter's recording (feature
/// `http`); without the feature the arm fails with the verdict-class
/// mismatch instead of passing silently.
#[cfg(not(feature = "http"))]
pub(super) async fn partner_validate_action(
    index: usize,
    uri: &str,
    expected: &PartnerExpectation,
    deadline: Option<Duration>,
    router: &PartnerRouter,
) -> Result<(), ScenarioFailure> {
    let _ = (uri, expected, deadline, router);
    Err(ScenarioFailure::ValidationMismatch {
        action: index,
        detail: "partner validation requires the `http` feature".to_string(),
    })
}

/// The mismatch detail of a failed partner assertion: the partner
/// URI, the applied filters rendered by kind, the bound in its own
/// grammar with the expected-versus-actual counts, and the recorded
/// request paths — `recorded` carries the RAW wire paths; every
/// secret-marked query value is masked through the shared redactor
/// before it reaches the detail (ADR-0051).
#[cfg(feature = "http")]
pub(crate) fn partner_mismatch_detail(
    uri: &str,
    expected: &PartnerExpectation,
    actual: usize,
    recorded: &[String],
    secret_keys: &[String],
) -> String {
    // The partner URI header may itself carry query bytes (a
    // query-bearing declaration): render it redacted like every
    // recorded path below (ADR-0051).
    let mut detail = format!("partner {}", redact_wire_path(uri, secret_keys));
    let filters = render_filters(expected, secret_keys);
    if !filters.is_empty() {
        detail.push_str(&format!(" ({filters})"));
    }
    detail.push_str(&format!(
        ", {}, actual {actual}",
        render_bound(&expected.bound)
    ));
    // The recorded request paths, deduplicated in arrival order: what
    // actually crossed the wire, every secret-marked query value
    // masked before it reaches the detail (ADR-0051).
    let mut unique: Vec<&str> = Vec::new();
    for path in recorded {
        if !unique.contains(&path.as_str()) {
            unique.push(path);
        }
    }
    if !unique.is_empty() {
        let redacted: Vec<String> = unique
            .iter()
            .map(|path| redact_wire_path(path, secret_keys))
            .collect();
        detail.push_str(&format!(", recorded: [{}]", redacted.join(", ")));
    }
    detail
}

/// Renders a count bound in its own grammar for mismatch details:
/// `Exact(3)` renders `expected 3` — the historical phrasing the
/// exact-count tests pin byte-for-byte — `AtLeast(3)` renders
/// `expected at least 3`, `AtMost(2)` renders `expected at most 2`,
/// and `Range(2, 4)` renders `expected between 2 and 4`.
#[cfg(feature = "http")]
pub(crate) fn render_bound(bound: &CountBound) -> String {
    match bound {
        CountBound::Exact(n) => format!("expected {n}"),
        CountBound::AtLeast(n) => format!("expected at least {n}"),
        CountBound::AtMost(n) => format!("expected at most {n}"),
        CountBound::Range(min, max) => format!("expected between {min} and {max}"),
    }
}

/// Renders the applied filters of a partner expectation for mismatch
/// details (the ADR-0051 redaction law, extended to filter payloads):
/// the method renders `method GET`; an `Exact` path filter renders
/// via [`redact_wire_path`] (a declared filter may carry query bytes,
/// and redaction is idempotent) while `Contains`/`Matches` render by
/// KIND only — the pattern payload never prints; each declared
/// `query` pair renders `k=v` except keys in `secret_keys`, which
/// render `k=<redacted>`. An expectation with no filters renders the
/// empty string.
#[cfg(feature = "http")]
pub(crate) fn render_filters(expected: &PartnerExpectation, secret_keys: &[String]) -> String {
    let mut clauses: Vec<String> = Vec::new();
    if let Some(method) = expected.method.as_deref() {
        clauses.push(format!("method {method}"));
    }
    match expected.path.as_ref() {
        Some(PathFilter::Exact(path)) => {
            clauses.push(format!("path {}", redact_wire_path(path, secret_keys)));
        }
        Some(PathFilter::Contains(_)) => {
            clauses.push("pathContains <pattern elided>".to_string());
        }
        Some(PathFilter::Matches(_)) => {
            clauses.push("pathMatches <pattern elided>".to_string());
        }
        None => {}
    }
    if let Some(query) = expected.query.as_ref() {
        for (key, value) in query {
            if secret_keys.iter().any(|secret| secret == key) {
                clauses.push(format!("{key}=<redacted>"));
            } else {
                clauses.push(format!("{key}={value}"));
            }
        }
    }
    clauses.join(", ")
}