//! Tests for `CapabilityExecutor` and `ResponseCache`

use super::*;
use crate::capability::CapabilityExecutionContext;
use crate::capability::response_cache::ResponseCache;
use crate::identity_grants::GrantSubject;
use axum::{
    Json, Router,
    body::Body,
    http::header,
    response::Response as AxumResponse,
    routing::{get, post},
};
use base64::{Engine as _, engine::general_purpose::STANDARD};

#[test]
fn test_build_url() {
    let executor = CapabilityExecutor::new();
    let config = RestConfig {
        base_url: "https://api.example.com".to_string(),
        path: "/users/{id}/posts/{post_id}".to_string(),
        ..Default::default()
    };

    let params = serde_json::json!({
        "id": "123",
        "post_id": 456
    });

    let url = executor.build_url(&config, &params).unwrap();
    assert_eq!(url, "https://api.example.com/users/123/posts/456");
}

#[test]
fn test_substitute_string() {
    let executor = CapabilityExecutor::new();
    let template = "Hello {name}, your score is {score}";
    let params = serde_json::json!({
        "name": "World",
        "score": 100
    });

    let result = executor.substitute_string(template, &params).unwrap();
    assert_eq!(result, "Hello World, your score is 100");
}

#[test]
fn test_extract_path() {
    let executor = CapabilityExecutor::new();
    let value = serde_json::json!({
        "data": {
            "users": [
                {"name": "Alice"},
                {"name": "Bob"}
            ]
        }
    });

    let result = executor.extract_path(&value, "data.users").unwrap();
    assert!(result.is_array());
    assert_eq!(result.as_array().unwrap().len(), 2);

    let result = executor.extract_path(&value, "data.users.0.name").unwrap();
    assert_eq!(result, "Alice");
}

#[test]
fn test_cache() {
    let cache = ResponseCache::new();
    let value = serde_json::json!({"test": true});

    cache.set("key1", &value, 60);
    assert_eq!(cache.get("key1"), Some(value));

    assert_eq!(cache.get("nonexistent"), None);
}

#[test]
fn test_fetch_from_file_simple() {
    let executor = CapabilityExecutor::new();
    let dir = std::env::temp_dir().join("mcp-gateway-test-cred");
    std::fs::create_dir_all(&dir).unwrap();
    let file = dir.join("tokens.json");
    std::fs::write(
        &file,
        r#"{"access_token": "test-token-123", "refresh_token": "refresh-456"}"#,
    )
    .unwrap();

    let spec = format!("{}:access_token", file.display());
    let result = executor.fetch_from_file(&spec).unwrap();
    assert_eq!(result, "test-token-123");

    let spec = format!("{}:refresh_token", file.display());
    let result = executor.fetch_from_file(&spec).unwrap();
    assert_eq!(result, "refresh-456");

    // Cleanup
    std::fs::remove_dir_all(&dir).ok();
}

#[test]
fn test_fetch_from_file_nested() {
    let executor = CapabilityExecutor::new();
    let dir = std::env::temp_dir().join("mcp-gateway-test-cred-nested");
    std::fs::create_dir_all(&dir).unwrap();
    let file = dir.join("config.json");
    std::fs::write(
        &file,
        r#"{"auth": {"google": {"token": "nested-token"}, "count": 42}}"#,
    )
    .unwrap();

    let spec = format!("{}:auth.google.token", file.display());
    let result = executor.fetch_from_file(&spec).unwrap();
    assert_eq!(result, "nested-token");

    // Numeric values work too
    let spec = format!("{}:auth.count", file.display());
    let result = executor.fetch_from_file(&spec).unwrap();
    assert_eq!(result, "42");

    // Cleanup
    std::fs::remove_dir_all(&dir).ok();
}

#[test]
fn test_fetch_from_file_missing_field() {
    let executor = CapabilityExecutor::new();
    let dir = std::env::temp_dir().join("mcp-gateway-test-cred-missing");
    std::fs::create_dir_all(&dir).unwrap();
    let file = dir.join("tokens.json");
    std::fs::write(&file, r#"{"access_token": "value"}"#).unwrap();

    let spec = format!("{}:nonexistent", file.display());
    let result = executor.fetch_from_file(&spec);
    assert!(result.is_err());
    assert!(result.unwrap_err().to_string().contains("not found"));

    // Cleanup
    std::fs::remove_dir_all(&dir).ok();
}

#[test]
fn test_fetch_from_file_missing_file() {
    let executor = CapabilityExecutor::new();
    let result = executor.fetch_from_file("/nonexistent/path/tokens.json:field");
    assert!(result.is_err());
    assert!(result.unwrap_err().to_string().contains("Failed to read"));
}

#[test]
fn test_fetch_from_file_invalid_format() {
    let executor = CapabilityExecutor::new();
    // No colon separator for field
    let result = executor.fetch_from_file("/path/to/file.json");
    assert!(result.is_err());
    assert!(
        result
            .unwrap_err()
            .to_string()
            .contains("Invalid file credential format")
    );
}

#[test]
fn test_substitute_params_skips_unresolved_placeholders() {
    let executor = CapabilityExecutor::new();
    let mut template = std::collections::HashMap::new();
    template.insert("resolved".to_string(), "{name}".to_string());
    template.insert("unresolved".to_string(), "{missing_param}".to_string());
    template.insert("static_val".to_string(), "hello".to_string());

    let params = serde_json::json!({"name": "world"});
    let result = executor.substitute_params(&template, &params).unwrap();

    // resolved and static_val should be present, unresolved should be filtered
    let keys: Vec<&str> = result.iter().map(|(k, _)| k.as_str()).collect();
    assert!(keys.contains(&"resolved"));
    assert!(keys.contains(&"static_val"));
    assert!(!keys.contains(&"unresolved"));
}

fn personal_auth_capability() -> CapabilityDefinition {
    crate::capability::parse_capability(
        r"
name: personal_calendar
description: Personal calendar read
auth:
  required: true
  type: bearer
  key: env:MIK6553_SHOULD_NOT_BE_READ
metadata:
  exposure: personal
  identity_owner:
    authority: cloudflare_access
    subject: owner-1
    label: owner@example.com
providers:
  primary:
    service: rest
    config:
      base_url: https://example.com
      path: /calendar
      method: GET
",
    )
    .unwrap()
}

#[tokio::test]
async fn personal_capability_missing_identity_denies_before_auth_lookup() {
    let executor = CapabilityExecutor::new();
    let cap = personal_auth_capability();

    let err = executor
        .execute_with_context(
            &cap,
            serde_json::json!({}),
            CapabilityExecutionContext::default(),
        )
        .await
        .unwrap_err()
        .to_string();

    assert!(err.contains("caller identity is required"), "{err}");
    assert!(!err.contains("MIK6553_SHOULD_NOT_BE_READ"), "{err}");
}

#[tokio::test]
async fn personal_capability_owner_mismatch_denies_before_auth_lookup() {
    let executor = CapabilityExecutor::new();
    let cap = personal_auth_capability();
    let context = CapabilityExecutionContext::with_caller_identity(GrantSubject::new(
        "cloudflare_access",
        "other-user",
        Some("other@example.com".to_string()),
    ));

    let err = executor
        .execute_with_context(&cap, serde_json::json!({}), context)
        .await
        .unwrap_err()
        .to_string();

    assert!(err.contains("does not match owner"), "{err}");
    assert!(!err.contains("MIK6553_SHOULD_NOT_BE_READ"), "{err}");
}

#[tokio::test]
async fn personal_capability_matching_owner_reaches_auth_lookup() {
    let executor = CapabilityExecutor::new();
    let cap = personal_auth_capability();
    let context = CapabilityExecutionContext::with_caller_identity(GrantSubject::new(
        "cloudflare_access",
        "owner-1",
        Some("owner@example.com".to_string()),
    ));

    let err = executor
        .execute_with_context(&cap, serde_json::json!({}), context)
        .await
        .unwrap_err()
        .to_string();

    assert!(err.contains("MIK6553_SHOULD_NOT_BE_READ"), "{err}");
    assert!(!err.contains("access denied"), "{err}");
}

#[tokio::test]
async fn shared_required_auth_preserves_legacy_auth_lookup() {
    let executor = CapabilityExecutor::new();
    let cap = crate::capability::parse_capability(
        r"
name: shared_weather
description: Shared weather lookup
auth:
  required: true
  type: bearer
  key: env:MIK6553_SHARED_AUTH_LOOKUP
providers:
  primary:
    service: rest
    config:
      base_url: https://example.com
      path: /weather
      method: GET
",
    )
    .unwrap();

    let err = executor
        .execute_with_context(
            &cap,
            serde_json::json!({}),
            CapabilityExecutionContext::default(),
        )
        .await
        .unwrap_err()
        .to_string();

    assert!(err.contains("MIK6553_SHARED_AUTH_LOOKUP"), "{err}");
    assert!(!err.contains("access denied"), "{err}");
}

// ── static_params integration tests ──────────────────────────────────────────

#[test]
fn static_params_flow_into_url_substitution() {
    // GIVEN: RestConfig with static_params containing a path placeholder value
    // WHEN: build_url is called with empty caller params
    // THEN: static_params are substituted into the URL path
    let executor = CapabilityExecutor::new();
    let config = RestConfig {
        base_url: "https://api.example.com".to_string(),
        path: "/v1/forecast/{format}".to_string(),
        static_params: [("format".to_string(), serde_json::json!("json"))]
            .into_iter()
            .collect(),
        ..Default::default()
    };

    let caller_params = serde_json::json!({});
    let effective = config.merge_with_static_params(&caller_params);
    let url = executor.build_url(&config, &effective).unwrap();
    assert_eq!(url, "https://api.example.com/v1/forecast/json");
}

#[test]
fn static_params_flow_into_query_params() {
    // GIVEN: RestConfig with static_params and a query params template referencing them
    // WHEN: substitute_params is called with merged effective params
    // THEN: static values appear in the resolved query parameters
    let executor = CapabilityExecutor::new();
    let config = RestConfig {
        base_url: "https://api.example.com".to_string(),
        path: "/forecast".to_string(),
        params: [
            ("current".to_string(), "{current}".to_string()),
            ("timezone".to_string(), "{timezone}".to_string()),
        ]
        .into_iter()
        .collect(),
        static_params: [
            (
                "current".to_string(),
                serde_json::json!("temperature_2m,weather_code"),
            ),
            ("timezone".to_string(), serde_json::json!("auto")),
        ]
        .into_iter()
        .collect(),
        ..Default::default()
    };

    let caller_params = serde_json::json!({});
    let effective = config.merge_with_static_params(&caller_params);
    let resolved = executor
        .substitute_params(&config.params, &effective)
        .unwrap();

    let map: std::collections::HashMap<_, _> = resolved.into_iter().collect();
    assert_eq!(map["current"], "temperature_2m,weather_code");
    assert_eq!(map["timezone"], "auto");
}

#[test]
fn caller_params_override_static_params_on_collision() {
    // GIVEN: static_params defines "timezone" = "auto"
    // WHEN: caller supplies "timezone" = "America/New_York"
    // THEN: caller value wins
    let config = RestConfig {
        static_params: [("timezone".to_string(), serde_json::json!("auto"))]
            .into_iter()
            .collect(),
        ..Default::default()
    };

    let caller_params = serde_json::json!({ "timezone": "America/New_York" });
    let effective = config.merge_with_static_params(&caller_params);
    assert_eq!(effective["timezone"], "America/New_York");
}

#[test]
fn static_params_preserved_when_caller_provides_no_collision() {
    // GIVEN: static_params has "format" key, caller provides "lat" key
    // WHEN: merging
    // THEN: both keys are present in the effective params
    let config = RestConfig {
        static_params: [("format".to_string(), serde_json::json!("json"))]
            .into_iter()
            .collect(),
        ..Default::default()
    };

    let caller_params = serde_json::json!({ "lat": 60.17 });
    let effective = config.merge_with_static_params(&caller_params);
    assert_eq!(effective["format"], "json");
    assert!((effective["lat"].as_f64().unwrap() - 60.17).abs() < f64::EPSILON);
}

#[test]
fn empty_static_params_returns_borrowed_caller_params() {
    // GIVEN: RestConfig with no static_params
    // WHEN: merging with caller params
    // THEN: returned Cow is Borrowed (zero allocation)
    let config = RestConfig::default();
    let caller_params = serde_json::json!({ "q": "rust" });
    let effective = config.merge_with_static_params(&caller_params);
    assert!(matches!(effective, std::borrow::Cow::Borrowed(_)));
    assert_eq!(effective["q"], "rust");
}

#[test]
fn static_params_support_numeric_and_boolean_values() {
    // GIVEN: static_params with integer, float, and boolean values
    // WHEN: merging with empty caller params
    // THEN: all typed values are preserved
    let config = RestConfig {
        static_params: [
            ("count".to_string(), serde_json::json!(10)),
            ("ratio".to_string(), serde_json::json!(0.5)),
            ("enabled".to_string(), serde_json::json!(true)),
        ]
        .into_iter()
        .collect(),
        ..Default::default()
    };

    let empty = serde_json::json!({});
    let effective = config.merge_with_static_params(&empty);
    assert_eq!(effective["count"], serde_json::json!(10));
    assert!((effective["ratio"].as_f64().unwrap() - 0.5).abs() < f64::EPSILON);
    assert_eq!(effective["enabled"], serde_json::json!(true));
}

#[test]
fn static_params_deserialization_from_yaml() {
    // GIVEN: YAML config with static_params containing mixed types
    // WHEN: deserializing RestConfig
    // THEN: all static_params are correctly parsed
    let yaml = r"
base_url: https://api.open-meteo.com
path: /v1/forecast
static_params:
  current: 'temperature_2m,weather_code'
  timezone: auto
  forecast_days: 7
";
    let config: RestConfig = serde_yaml::from_str(yaml).unwrap();
    assert_eq!(
        config.static_params["current"],
        "temperature_2m,weather_code"
    );
    assert_eq!(config.static_params["timezone"], "auto");
    assert_eq!(config.static_params["forecast_days"], serde_json::json!(7));
}

#[test]
fn static_params_serialization_round_trips() {
    // GIVEN: RestConfig with static_params
    // WHEN: serialized to JSON and back
    // THEN: static_params are preserved exactly
    let mut static_params = std::collections::HashMap::new();
    static_params.insert("key".to_string(), serde_json::json!("value"));
    static_params.insert("num".to_string(), serde_json::json!(42));

    let config = RestConfig {
        base_url: "https://example.com".to_string(),
        static_params,
        ..Default::default()
    };

    let json = serde_json::to_string(&config).unwrap();
    let restored: RestConfig = serde_json::from_str(&json).unwrap();
    assert_eq!(restored.static_params["key"], "value");
    assert_eq!(restored.static_params["num"], serde_json::json!(42));
}

#[test]
fn static_params_with_array_value_preserved() {
    // GIVEN: static_params containing a JSON array value
    // WHEN: merging
    // THEN: array is preserved in the effective params
    let config = RestConfig {
        static_params: [(
            "fields".to_string(),
            serde_json::json!(["id", "name", "email"]),
        )]
        .into_iter()
        .collect(),
        ..Default::default()
    };

    let empty = serde_json::json!({});
    let effective = config.merge_with_static_params(&empty);
    assert_eq!(
        effective["fields"],
        serde_json::json!(["id", "name", "email"])
    );
}

#[test]
fn build_url_with_static_params_substitution_in_endpoint() {
    // GIVEN: endpoint-style URL with static params embedded as placeholders
    // WHEN: merged with empty caller params and URL built
    // THEN: static values are substituted into the endpoint URL
    let executor = CapabilityExecutor::new();
    let config = RestConfig {
        endpoint: "https://api.example.com/v1/data?format={fmt}&version={ver}".to_string(),
        static_params: [
            ("fmt".to_string(), serde_json::json!("json")),
            ("ver".to_string(), serde_json::json!("2")),
        ]
        .into_iter()
        .collect(),
        ..Default::default()
    };

    let caller_params = serde_json::json!({});
    let effective = config.merge_with_static_params(&caller_params);
    let url = executor.build_url(&config, &effective).unwrap();
    assert!(
        url.contains("format=json"),
        "URL should contain format=json"
    );
    assert!(url.contains("version=2"), "URL should contain version=2");
}

// ── body_content_type tests ──────────────────────────────────────────────────

#[test]
fn body_content_type_text_plain_uses_raw_string_body() {
    // Verify that when body_content_type = "text/plain" and the body template
    // is a JSON string, attach_request_body builds a raw-string request (not
    // JSON-encoded).  We can't easily inspect the built request in a unit test
    // without a live HTTP server, so we at least verify that the RestConfig
    // deserialises correctly from YAML and that substitute_string works.
    let executor = CapabilityExecutor::new();
    let config = RestConfig {
        body: Some(serde_json::Value::String(
            "SELECT * FROM bus_msg WHERE topic = '{topic}' LIMIT {max_msg}".to_string(),
        )),
        body_content_type: "text/plain".to_string(),
        ..Default::default()
    };

    let params = serde_json::json!({"topic": "bus.demo.test", "max_msg": 50});

    // substitute_string is the path taken for plain-text bodies.
    let sql = executor
        .substitute_string(config.body.as_ref().unwrap().as_str().unwrap(), &params)
        .unwrap();

    assert!(
        sql.contains("bus.demo.test"),
        "SQL should contain topic: {sql}"
    );
    assert!(sql.contains("50"), "SQL should contain max_msg: {sql}");
    assert!(
        !sql.contains('{'),
        "All placeholders should be resolved: {sql}"
    );
}

#[tokio::test]
async fn handle_response_binary_returns_base64_payload() {
    async fn binary_handler() -> AxumResponse {
        AxumResponse::builder()
            .status(200)
            .header(header::CONTENT_TYPE, "video/mp4")
            .body(Body::from(vec![0_u8, 1, 2, 3, 4, 5]))
            .unwrap()
    }

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, Router::new().route("/video", get(binary_handler)))
            .await
            .unwrap();
    });

    let executor = CapabilityExecutor::new();
    let response = executor
        .client
        .get(format!("http://{addr}/video"))
        .send()
        .await
        .unwrap();
    let config = RestConfig {
        response_format: "binary".to_string(),
        ..Default::default()
    };

    let body = executor.handle_response(response, &config).await.unwrap();
    assert_eq!(body["mime_type"], "video/mp4");
    assert_eq!(body["size"], 6);
    assert_eq!(body["data"], STANDARD.encode([0_u8, 1, 2, 3, 4, 5]));
}

#[tokio::test]
async fn handle_response_graphql_error_with_null_projection_returns_error() {
    async fn graphql_error_handler() -> Json<serde_json::Value> {
        Json(serde_json::json!({
            "data": null,
            "errors": [
                { "message": "Field \"createAsUser\" is not defined by type \"IssueCreateInput\"." },
                { "message": "Field \"displayIconUrl\" is not defined by type \"IssueCreateInput\"." }
            ]
        }))
    }

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(
            listener,
            Router::new().route("/graphql", post(graphql_error_handler)),
        )
        .await
        .unwrap();
    });

    let executor = CapabilityExecutor::new();
    let response = executor
        .client
        .post(format!("http://{addr}/graphql"))
        .json(&serde_json::json!({ "query": "mutation Broken { issueCreate { success } }" }))
        .send()
        .await
        .unwrap();
    let config = RestConfig {
        response_path: Some("data.issueCreate".to_string()),
        ..Default::default()
    };

    let err = executor
        .handle_response(response, &config)
        .await
        .unwrap_err();
    let message = err.to_string();
    assert!(message.contains("GraphQL error"), "{message}");
    assert!(message.contains("createAsUser"), "{message}");
    assert!(message.contains("displayIconUrl"), "{message}");
}

#[tokio::test]
async fn handle_response_graphql_success_with_response_path_is_unchanged() {
    async fn graphql_success_handler() -> Json<serde_json::Value> {
        Json(serde_json::json!({
            "data": {
                "issueCreate": {
                    "success": true,
                    "issue": {
                        "id": "lin-123",
                        "identifier": "MIK-3181"
                    }
                }
            }
        }))
    }

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(
            listener,
            Router::new().route("/graphql", post(graphql_success_handler)),
        )
        .await
        .unwrap();
    });

    let executor = CapabilityExecutor::new();
    let response = executor
        .client
        .post(format!("http://{addr}/graphql"))
        .json(&serde_json::json!({ "query": "mutation Ok { issueCreate { success } }" }))
        .send()
        .await
        .unwrap();
    let config = RestConfig {
        response_path: Some("data.issueCreate".to_string()),
        ..Default::default()
    };

    let body = executor.handle_response(response, &config).await.unwrap();
    assert_eq!(body["success"], true);
    assert_eq!(body["issue"]["identifier"], "MIK-3181");
}

#[test]
fn body_content_type_empty_defaults_to_json() {
    // Default behaviour: body_content_type is empty → use JSON body.
    // RestConfig::default() should produce empty body_content_type.
    let config = RestConfig::default();
    assert!(
        config.body_content_type.is_empty(),
        "Default body_content_type must be empty (→ JSON)"
    );
}

#[test]
fn body_content_type_deserialises_from_yaml() {
    let yaml = r#"
base_url: "http://127.0.0.1:8000"
path: "/sql"
method: "POST"
body_content_type: "text/plain"
body: "SELECT * FROM bus_msg LIMIT 10"
"#;
    let config: RestConfig = serde_yaml::from_str(yaml).unwrap();
    assert_eq!(config.body_content_type, "text/plain");
    assert_eq!(
        config.body.unwrap().as_str().unwrap(),
        "SELECT * FROM bus_msg LIMIT 10"
    );
}

#[tokio::test]
async fn send_with_retry_recovers_from_transient_timeouts() {
    use std::io::{Read, Write};
    use std::net::TcpListener;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    // Local server: the first two connections hang past the client's per-attempt
    // timeout (a transient timeout -> retry); the third responds 200 at once.
    // Verifies MIK-5081: transient outbound failures are retried with backoff
    // instead of surfacing as an immediate BACKEND_ERROR.
    let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
    let addr = listener.local_addr().unwrap();
    let counter = Arc::new(AtomicUsize::new(0));
    let counter_srv = Arc::clone(&counter);
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let Ok(mut stream) = stream else { continue };
            let n = counter_srv.fetch_add(1, Ordering::SeqCst);
            std::thread::spawn(move || {
                let mut buf = [0u8; 1024];
                let _ = stream.read(&mut buf);
                if n < 2 {
                    std::thread::sleep(std::time::Duration::from_millis(400));
                } else {
                    let _ = stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}");
                    let _ = stream.flush();
                }
            });
        }
    });

    let client = reqwest::Client::new();
    let url = format!("http://{addr}/");
    let req = client
        .get(&url)
        .timeout(std::time::Duration::from_millis(120));

    // Idempotent (retry_timeouts = true): timeouts are retried.
    let health = crate::failsafe::HealthTracker::new("test");
    let resp = send_with_retry(req, "test", true, &health).await;
    assert!(
        resp.is_ok(),
        "retry should recover from transient timeouts, got {resp:?}"
    );
    assert_eq!(resp.unwrap().status(), 200);
    assert!(
        health.is_healthy(),
        "a recovered call records transport success"
    );
    assert_eq!(
        counter.load(Ordering::SeqCst),
        3,
        "should have taken exactly 3 attempts (2 transient + 1 success)"
    );
}

#[tokio::test]
async fn send_with_retry_records_transport_failures() {
    // A refused port (listener bound then dropped) yields connection errors,
    // which are always retried and recorded as transport failures. After
    // enough consecutive failures the health tracker flips unhealthy (MIK-5080).
    let addr = {
        let l = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        l.local_addr().unwrap()
        // listener dropped here -> port now refuses connections
    };

    let client = reqwest::Client::new();
    let url = format!("http://{addr}/");
    let health = crate::failsafe::HealthTracker::new("test");

    assert!(health.is_healthy(), "fresh tracker is healthy");
    for _ in 0..3 {
        let req = client
            .get(&url)
            .timeout(std::time::Duration::from_millis(200));
        let resp = send_with_retry(req, "test", false, &health).await;
        assert!(resp.is_err(), "connect to a refused port must fail");
    }
    assert!(
        !health.is_healthy(),
        "consecutive transport failures should flip the tracker unhealthy"
    );
}

#[tokio::test]
async fn send_with_retry_does_not_retry_timeouts_when_not_idempotent() {
    // A server that always hangs past the client timeout. With retry_timeouts
    // = false (non-idempotent), the timeout is NOT retried: exactly one
    // connection attempt is made. Protects against duplicate side effects.
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
    let addr = listener.local_addr().unwrap();
    let counter = Arc::new(AtomicUsize::new(0));
    let counter_srv = Arc::clone(&counter);
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let Ok(stream) = stream else { continue };
            counter_srv.fetch_add(1, Ordering::SeqCst);
            // Hold the connection open and never respond.
            std::thread::spawn(move || {
                std::thread::sleep(std::time::Duration::from_millis(500));
                drop(stream);
            });
        }
    });

    let client = reqwest::Client::new();
    let url = format!("http://{addr}/");
    let health = crate::failsafe::HealthTracker::new("test");
    let req = client
        .post(&url)
        .timeout(std::time::Duration::from_millis(120));

    let resp = send_with_retry(req, "test", false, &health).await;
    assert!(resp.is_err(), "the hung request should time out");
    assert_eq!(
        counter.load(Ordering::SeqCst),
        1,
        "a non-idempotent timeout must NOT be retried (single attempt)"
    );
}
