//! Property and unit tests for `mcptest_core::runner::*`.
//!
//! ## What is covered today
//!
//! The runner module today exposes two helpers, `time_step` (async) and
//! `time_step_sync` (sync), that wrap a step's work with wall-clock
//! timing and produce a populated `TestStep`. The full runner engine
//! has not landed yet.
//!
//! Real tests in this file exercise the helpers across their happy path,
//! cache-hit short circuit, label round-trip, RFC 3339 timestamp shape,
//! and sleep precision. Property tests via `proptest` flush out edge
//! cases on the label round-trip, the cache-hit bool round-trip, and the
//! `started_at <= ended_at` invariant.
//!
//! The runner engine has since landed, so the `engine_state_machine` module
//! exercises it over the public API: bounded parallelism (`--parallel N`),
//! per-test timeout, retry-to-pass, and bail/`max_failures`. Per-test timeouts
//! are independent, so one hanging test does not block its siblings.
//!
//! ## What is deferred
//!
//! Graceful shutdown on SIGINT (in-flight tests finish or cancel cleanly, an
//! "interrupted" exit code) stays under `#[ignore]`: the engine has no signal
//! path, and portable SIGINT injection belongs in a CLI integration test, not a
//! library unit test. Tracked in WOR-1434. There is no suite-wide timeout in the
//! engine; the per-test-timeout-does-not-block-siblings test covers the closest
//! real contract.

use std::time::Duration;

use mcptest_core::runner::{time_step, time_step_sync};
use proptest::prelude::*;

/// Happy path: a cold sync step records a duration close to the sleep
/// and carries a populated label, RFC 3339 timestamps, and `cache_hit:
/// false`.
#[test]
fn cold_sync_step_records_observed_duration() {
    let step = time_step_sync("probe", false, || {
        std::thread::sleep(Duration::from_millis(50));
    });
    assert_eq!(step.name, "probe");
    assert!(!step.cache_hit);
    assert!(
        step.duration_ms >= 45,
        "expected a ~50ms step to record at least 45ms, got {}",
        step.duration_ms
    );
    assert!(step.started_at.is_some());
    assert!(step.ended_at.is_some());
}

/// Cache-hit sync step short-circuits to the canonical `1ms` regardless
/// of how long the inner closure actually ran.
#[test]
fn cache_hit_sync_step_emits_one_millisecond() {
    let step = time_step_sync("warm-call", true, || {
        std::thread::sleep(Duration::from_millis(20));
    });
    assert_eq!(step.duration_ms, 1);
    assert!(step.cache_hit);
    assert_eq!(step.started_at, step.ended_at);
}

/// Async helper matches the sync helper on cold-path timing.
#[tokio::test]
async fn cold_async_step_records_observed_duration() {
    let step = time_step("async-cold", false, || async {
        tokio::time::sleep(Duration::from_millis(50)).await;
    })
    .await;
    assert!(
        step.duration_ms >= 45,
        "expected a ~50ms step to record at least 45ms, got {}",
        step.duration_ms
    );
    assert!(!step.cache_hit);
}

/// Async helper short-circuits cache hits to `1ms`.
#[tokio::test]
async fn cache_hit_async_step_emits_one_millisecond() {
    let step = time_step("async-hit", true, || async {
        tokio::time::sleep(Duration::from_millis(20)).await;
    })
    .await;
    assert_eq!(step.duration_ms, 1);
    assert!(step.cache_hit);
}

/// Even at zero work, the timestamps are well-formed RFC 3339 strings
/// and `started_at <= ended_at` (lexical compare is fine because RFC
/// 3339 sorts lexically when the offset and precision are fixed).
#[test]
fn rfc3339_timestamps_are_lexically_ordered() {
    let step = time_step_sync("zero-work", false, || {});
    let started = step.started_at.expect("started_at present");
    let ended = step.ended_at.expect("ended_at present");
    assert!(
        started.as_str() <= ended.as_str(),
        "started_at {started} must precede or equal ended_at {ended}"
    );
    // Shape check: YYYY-MM-DDTHH:MM:SSZ is 20 chars and ends with Z.
    assert_eq!(started.len(), 20);
    assert!(started.ends_with('Z'));
}

/// Inner closures may panic-free side effect; the helper does not
/// swallow their work and the timing still reflects the closure body.
#[test]
fn inner_closure_side_effects_observable() {
    let mut seen = false;
    let step = time_step_sync("side-effect", false, || {
        seen = true;
    });
    assert!(seen, "closure must run");
    assert_eq!(step.name, "side-effect");
}

// proptest section ---------------------------------------------------

proptest! {
    /// Property: any non-empty ASCII label round-trips through the
    /// helper into `TestStep.name` unchanged. We bound the label to 64
    /// chars so the corpus stays fast; the helper has no length cap.
    #[test]
    fn label_round_trips_through_test_step(
        label in "[ -~]{1,64}"
    ) {
        let step = time_step_sync(label.clone(), false, || {});
        prop_assert_eq!(step.name, label);
    }

    /// Property: the `cache_hit` argument always equals the field on
    /// the produced `TestStep`. The helper never silently flips it.
    #[test]
    fn cache_hit_bool_round_trips(hit in proptest::bool::ANY) {
        let step = time_step_sync("x", hit, || {});
        prop_assert_eq!(step.cache_hit, hit);
        if hit {
            prop_assert_eq!(step.duration_ms, 1);
        }
    }

    /// Property: `started_at <= ended_at` for every cold step. RFC 3339
    /// strings produced by the helper share a fixed precision and
    /// offset (`Z`), so lexical compare is monotonic with time.
    #[test]
    fn started_at_never_exceeds_ended_at(
        label in "[ -~]{0,16}",
        hit in proptest::bool::ANY,
    ) {
        let step = time_step_sync(label, hit, || {});
        let s = step.started_at.unwrap();
        let e = step.ended_at.unwrap();
        prop_assert!(s.as_str() <= e.as_str());
    }
}

// Runner-engine state-machine tests ---------------------------------
//
// These drive the runner engine (`Runner::run_all`) over its public API with
// in-memory mock executors, from the integration-crate boundary (the engine's
// own `executor` field is private, so peak-concurrency is read via a shared
// atomic the test holds). SIGINT graceful shutdown stays deferred to WOR-1434:
// the engine has no signal path and portable SIGINT injection belongs in a CLI
// integration test, not a library unit test.
mod engine_state_machine {
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;
    use std::time::Duration;

    use async_trait::async_trait;
    use mcptest_core::report::TestStatus;
    use mcptest_core::runner::{
        Parallelism, Runner, RunnerOptions, TestExecutor, TestOutcome, TestSpec,
    };

    fn specs(names: &[&str]) -> Vec<TestSpec> {
        names.iter().map(|n| TestSpec::new(*n)).collect()
    }

    /// Tracks peak concurrency via shared atomics the test reads.
    struct ConcurrencyProbe {
        in_flight: Arc<AtomicUsize>,
        max_in_flight: Arc<AtomicUsize>,
        delay: Duration,
    }

    #[async_trait]
    impl TestExecutor for ConcurrencyProbe {
        async fn execute_one(&self, _test: &TestSpec) -> TestOutcome {
            let now = self.in_flight.fetch_add(1, Ordering::AcqRel) + 1;
            self.max_in_flight.fetch_max(now, Ordering::AcqRel);
            tokio::time::sleep(self.delay).await;
            self.in_flight.fetch_sub(1, Ordering::AcqRel);
            TestOutcome::pass()
        }
    }

    /// Bounded parallelism: with `--parallel N` no more than N tests run at
    /// once, even for a larger suite.
    #[tokio::test]
    async fn parallel_n_keeps_at_most_n_in_flight() {
        let max_in_flight = Arc::new(AtomicUsize::new(0));
        let probe = ConcurrencyProbe {
            in_flight: Arc::new(AtomicUsize::new(0)),
            max_in_flight: Arc::clone(&max_in_flight),
            delay: Duration::from_millis(30),
        };
        let options = RunnerOptions {
            parallel: Parallelism::Fixed(3),
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, probe);
        runner
            .run_all(&specs(&["a", "b", "c", "d", "e", "f", "g"]))
            .await
            .expect("run");
        let peak = max_in_flight.load(Ordering::Acquire);
        assert!(
            (1..=3).contains(&peak),
            "peak concurrency {peak} must respect the bound of 3"
        );
    }

    /// A test that exceeds its per-test timeout is cancelled and recorded as a
    /// failure with the timeout reason.
    #[tokio::test]
    async fn per_test_timeout_cancels_long_test() {
        struct Hanging;
        #[async_trait]
        impl TestExecutor for Hanging {
            async fn execute_one(&self, _test: &TestSpec) -> TestOutcome {
                tokio::time::sleep(Duration::from_secs(5)).await;
                TestOutcome::pass()
            }
        }
        let options = RunnerOptions {
            default_timeout: Duration::from_millis(50),
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, Hanging);
        let report = runner.run_all(&specs(&["slow"])).await.expect("run");
        assert_eq!(report.summary.failed, 1);
        let failure = report.tests[0].failure.as_ref().expect("failure recorded");
        assert!(
            failure.message.contains("timed out"),
            "got `{}`",
            failure.message
        );
    }

    /// A per-test timeout cancels only the slow test; its siblings still run
    /// and pass. (The engine has no suite-wide timeout; that is out of scope.)
    #[tokio::test]
    async fn per_test_timeout_does_not_block_sibling_tests() {
        struct SlowOnly;
        #[async_trait]
        impl TestExecutor for SlowOnly {
            async fn execute_one(&self, test: &TestSpec) -> TestOutcome {
                if test.name == "slow" {
                    tokio::time::sleep(Duration::from_secs(5)).await;
                }
                TestOutcome::pass()
            }
        }
        let options = RunnerOptions {
            parallel: Parallelism::Fixed(1),
            default_timeout: Duration::from_millis(50),
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, SlowOnly);
        let report = runner
            .run_all(&specs(&["slow", "fast_one", "fast_two"]))
            .await
            .expect("run");
        assert_eq!(report.summary.failed, 1, "only the slow test times out");
        assert_eq!(report.summary.passed, 2, "the fast siblings still run");
    }

    /// Retry replays a flaky test until it passes; the recorded outcome is the
    /// final passing attempt, and the runner stops retrying once it passes.
    #[tokio::test]
    async fn retry_records_final_attempt_outcome() {
        struct Flaky {
            calls: Arc<AtomicUsize>,
            fail_until: usize,
        }
        #[async_trait]
        impl TestExecutor for Flaky {
            async fn execute_one(&self, _test: &TestSpec) -> TestOutcome {
                let n = self.calls.fetch_add(1, Ordering::AcqRel) + 1;
                if n <= self.fail_until {
                    TestOutcome::fail("flaky attempt")
                } else {
                    TestOutcome::pass()
                }
            }
        }
        let calls = Arc::new(AtomicUsize::new(0));
        let options = RunnerOptions {
            retry: 3,
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(
            options,
            Flaky {
                calls: Arc::clone(&calls),
                fail_until: 2,
            },
        );
        let report = runner.run_all(&specs(&["flaky"])).await.expect("run");
        assert_eq!(report.summary.passed, 1, "the final attempt passes");
        assert_eq!(report.summary.failed, 0);
        assert_eq!(
            calls.load(Ordering::Acquire),
            3,
            "fails twice then passes; no further retries"
        );
    }

    /// Bail (`max_failures: 1`): the first failure halts the run; the rest are
    /// recorded as skipped, not failed, so every test still has a record.
    #[tokio::test]
    async fn bail_on_first_failure_skips_subsequent_tests() {
        struct AlwaysFail;
        #[async_trait]
        impl TestExecutor for AlwaysFail {
            async fn execute_one(&self, _test: &TestSpec) -> TestOutcome {
                TestOutcome::fail("boom")
            }
        }
        let options = RunnerOptions {
            parallel: Parallelism::Fixed(1),
            max_failures: Some(1),
            ..RunnerOptions::default()
        };
        let mut runner = Runner::new(options, AlwaysFail);
        let report = runner
            .run_all(&specs(&["a", "b", "c", "d"]))
            .await
            .expect("run");
        assert_eq!(report.summary.failed, 1);
        assert_eq!(report.summary.skipped, 3);
        assert_eq!(report.summary.passed, 0);
        // Every test still carries a record (the report is well-formed).
        assert_eq!(report.tests.len(), 4);
        assert_eq!(report.tests[1].status, TestStatus::Skip);
    }

    /// Graceful shutdown on SIGINT stays deferred (WOR-1434): the engine has no
    /// signal path, and portable SIGINT injection belongs in a CLI test.
    #[tokio::test]
    #[ignore = "WOR-1434: SIGINT graceful-shutdown surface lives in the CLI; not a portable library unit test"]
    async fn sigint_shuts_runner_down_cleanly() {
        // Lands with WOR-1434 (CLI signal handling + interrupted exit code).
    }
}

// OAuth refresh + HTTP transport configuration tests ------

mod oauth_refresh_550 {
    //! Live wiremock-backed tests for the reqwest refresher.
    //!
    //! These pin the refresher's HTTP contract (token exchange, 5xx retry,
    //! non-retryable 4xx) directly. Wiring a 401-triggered refresh into the
    //! request path is a transport/CLI concern handled outside this engine
    //! crate; these tests keep the refresher contract stable regardless.

    use mcptest_core::auth::{AuthError, Token};
    use mcptest_core::auth::{Refresher, ReqwestRefresher, ReqwestRefresherConfig};
    use reqwest::blocking::Client;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;
    use std::time::Duration;
    use wiremock::matchers::{body_string_contains, method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    fn cfg(token_endpoint: &str) -> ReqwestRefresherConfig {
        ReqwestRefresherConfig {
            token_endpoint: token_endpoint.to_string(),
            client_id: "cid".into(),
            client_secret: Some("csec".into()),
            timeout: Duration::from_secs(2),
            max_retries: 2,
            backoff_steps_ms: vec![1, 1, 1],
        }
    }

    /// Run the blocking refresh on a real OS thread so the
    /// `reqwest::blocking::Client`'s internal tokio runtime can be
    /// dropped without panicking. Using `tokio::task::spawn_blocking`
    /// keeps us inside the test runtime and trips
    /// `Cannot drop a runtime in a context where blocking is not
    /// allowed` when the inner client is torn down.
    fn refresh_blocking(token_url: String, refresh_token: &str) -> Result<Token, AuthError> {
        let refresh_token = refresh_token.to_string();
        std::thread::spawn(move || {
            let client = Client::builder()
                .timeout(Duration::from_secs(2))
                .build()
                .expect("build client");
            let r = ReqwestRefresher::with_client(cfg(&token_url), client);
            r.refresh(&refresh_token)
        })
        .join()
        .expect("thread join")
    }

    #[tokio::test]
    async fn refresh_exchanges_token_against_a_live_token_endpoint() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .and(body_string_contains("grant_type=refresh_token"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "AT-new",
                "refresh_token": "RT-new",
                "expires_in": 3600,
                "token_type": "Bearer",
            })))
            .mount(&server)
            .await;
        let url = format!("{}/token", server.uri());
        let token = refresh_blocking(url, "RT-old").expect("refresh ok");
        assert_eq!(token.access_token, "AT-new");
        assert_eq!(token.refresh_token.as_deref(), Some("RT-new"));
    }

    #[tokio::test]
    async fn refresh_retries_transient_5xx_then_succeeds() {
        let server = MockServer::start().await;
        // The first 2 responses 503; the 3rd attempt succeeds. The
        // refresher is configured for max_retries=2 (so up to 3
        // attempts total).
        Mock::given(method("POST"))
            .and(path("/token"))
            .respond_with(ResponseTemplate::new(503))
            .up_to_n_times(2)
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .and(path("/token"))
            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
                "access_token": "AT-late",
                "expires_in": 600,
                "token_type": "Bearer",
            })))
            .mount(&server)
            .await;

        let url = format!("{}/token", server.uri());
        let token = refresh_blocking(url, "RT");
        assert!(
            token.is_ok(),
            "should have recovered after 2 retries: {token:?}"
        );
        assert_eq!(token.unwrap().access_token, "AT-late");
    }

    #[tokio::test]
    async fn refresh_surfaces_non_retryable_4xx_immediately() {
        let server = MockServer::start().await;
        let attempts = Arc::new(AtomicUsize::new(0));
        let attempts_clone = attempts.clone();
        Mock::given(method("POST"))
            .and(path("/token"))
            .respond_with(move |_req: &wiremock::Request| {
                attempts_clone.fetch_add(1, Ordering::SeqCst);
                ResponseTemplate::new(400).set_body_string("invalid_grant")
            })
            .mount(&server)
            .await;

        let url = format!("{}/token", server.uri());
        let result = refresh_blocking(url, "RT");
        match result {
            Err(AuthError::RefreshFailed {
                retryable: false,
                reason,
            }) => {
                assert!(reason.contains("400"), "unexpected reason: {reason}");
            }
            other => panic!("expected non-retryable refresh failure, got {other:?}"),
        }
        // 4xx must NOT retry.
        assert_eq!(
            attempts.load(Ordering::SeqCst),
            1,
            "non-retryable status should fire exactly once"
        );
    }

    #[tokio::test]
    async fn header_validator_rejects_authorization_at_load_time() {
        // The runner runs HeaderValidator::validate as soon as YAML
        // loads. This sanity check
        // re-asserts the contract so changes to the validator surface
        // here as a runner-flavored test failure too.
        use mcptest_core::transport::{HeaderValidator, HeaderValue};
        use std::collections::HashMap;
        let mut h: HashMap<String, HeaderValue> = HashMap::new();
        h.insert("Authorization".into(), HeaderValue::literal("Bearer x"));
        let err = HeaderValidator::validate(&h).expect_err("must reject");
        assert!(err.to_string().contains("auth:"), "got: {err}");
    }
}
