#![cfg(feature = "dynamic")]
#![allow(clippy::unwrap_used)] // integration-test setup helpers are not #[test] fns

use std::time::Duration;

use nifi_rust_client::dynamic::DynamicClient;
use nifi_rust_client::wait::{
    ControllerServiceTargetState, ProcessGroupTargetState, ProcessorTargetState, WaitConfig,
    controller_service_state_dynamic, controller_service_verify_config_dynamic,
    empty_all_connections_dynamic, flow_analysis_rule_verify_config_dynamic, flowfile_drop_dynamic,
    flowfile_listing_dynamic, parameter_context_update_dynamic,
    parameter_context_validation_dynamic, parameter_provider_apply_parameters_dynamic,
    parameter_provider_verify_config_dynamic, process_group_controller_services_state_dynamic,
    process_group_state_dynamic, processor_state_dynamic, processor_verify_config_dynamic,
    provenance_lineage_dynamic, provenance_query_dynamic, reporting_task_verify_config_dynamic,
    versioned_flow_revert_dynamic, versioned_flow_update_dynamic,
};
use nifi_rust_client::{NifiClientBuilder, NifiError};
use serde_json::json;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};

fn fast_config(timeout_ms: u64) -> WaitConfig {
    WaitConfig {
        timeout: Duration::from_millis(timeout_ms),
        poll_interval: Duration::from_millis(10),
        initial_delay: Duration::ZERO,
        cleanup: true,
    }
}

async fn mount_about(server: &MockServer) {
    Mock::given(method("GET"))
        .and(path("/nifi-api/flow/about"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "about": {
                "title": "NiFi",
                "version": "2.9.0",
                "uri": "https://localhost:8443/nifi-api",
                "contentViewerUrl": "../nifi-content-viewer/",
                "timezone": "UTC",
                "buildTag": "nifi-2.9.0",
                "buildTimestamp": "2024-01-01T00:00:00Z"
            }
        })))
        .mount(server)
        .await;
}

async fn dynamic_client(server: &MockServer) -> DynamicClient {
    let client = NifiClientBuilder::new(&server.uri())
        .unwrap()
        .build_dynamic()
        .unwrap();
    client.detect_version().await.unwrap();
    client.inner().set_token("jwt".to_string()).await;
    client
}

#[tokio::test]
async fn processor_state_dynamic_reaches_target() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path("/nifi-api/processors/abc"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "component": { "id": "abc", "state": "RUNNING" }
        })))
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let entity = processor_state_dynamic(
        &client,
        "abc",
        ProcessorTargetState::Running,
        fast_config(1000),
    )
    .await
    .unwrap();
    assert!(entity.component.is_some());
}

#[tokio::test]
async fn controller_service_state_dynamic_reaches_target() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    // First ENABLING, then ENABLED — helper must poll through the transient state.
    Mock::given(method("GET"))
        .and(path("/nifi-api/controller-services/cs-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "component": { "id": "cs-1", "state": "ENABLING" }
        })))
        .up_to_n_times(1)
        .mount(&mock_server)
        .await;
    Mock::given(method("GET"))
        .and(path("/nifi-api/controller-services/cs-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "component": { "id": "cs-1", "state": "ENABLED" }
        })))
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let entity = controller_service_state_dynamic(
        &client,
        "cs-1",
        ControllerServiceTargetState::Enabled,
        fast_config(1000),
    )
    .await
    .unwrap();
    assert!(entity.component.is_some());
}

#[tokio::test]
async fn parameter_context_update_dynamic_succeeds_and_cleans_up() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/parameter-contexts/ctx-1/update-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(json!({ "request": { "complete": false } })),
        )
        .up_to_n_times(1)
        .mount(&mock_server)
        .await;
    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/parameter-contexts/ctx-1/update-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(json!({ "request": { "complete": true } })),
        )
        .mount(&mock_server)
        .await;

    // cleanup: true — DELETE fires exactly once after poll resolves.
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/parameter-contexts/ctx-1/update-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(json!({ "request": { "complete": true } })),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let entity = parameter_context_update_dynamic(&client, "ctx-1", "req-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(entity.request.and_then(|r| r.complete), Some(true));
}

#[tokio::test]
async fn parameter_context_update_dynamic_reports_failure() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/parameter-contexts/ctx-1/update-requests/req-1",
        ))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "request": { "complete": true, "failureReason": "validation failed" }
        })))
        .mount(&mock_server)
        .await;

    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/parameter-contexts/ctx-1/update-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(json!({ "request": { "complete": true } })),
        )
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let err = parameter_context_update_dynamic(&client, "ctx-1", "req-1", fast_config(1000))
        .await
        .unwrap_err();
    match err {
        NifiError::Api { status, message } => {
            assert_eq!(status, 500);
            assert!(message.contains("validation failed"));
        }
        other => panic!("expected Api, got {other:?}"),
    }
}

#[tokio::test]
async fn provenance_query_dynamic_succeeds_and_cleans_up() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path("/nifi-api/provenance/q-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "provenance": { "id": "q-1", "finished": false, "percentCompleted": 50 }
        })))
        .up_to_n_times(1)
        .mount(&mock_server)
        .await;
    Mock::given(method("GET"))
        .and(path("/nifi-api/provenance/q-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "provenance": { "id": "q-1", "finished": true, "percentCompleted": 100 }
        })))
        .mount(&mock_server)
        .await;

    Mock::given(method("DELETE"))
        .and(path("/nifi-api/provenance/q-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "provenance": { "id": "q-1", "finished": true }
        })))
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto = provenance_query_dynamic(&client, "q-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(dto.finished, Some(true));
}

fn drop_request_entity(finished: bool, failure: Option<&str>) -> serde_json::Value {
    let mut req = json!({
        "id": "drop-1",
        "finished": finished,
        "percentCompleted": if finished { 100 } else { 50 },
    });
    if let Some(reason) = failure {
        req["failureReason"] = json!(reason);
    }
    json!({ "dropRequest": req })
}

#[tokio::test]
async fn flowfile_drop_dynamic_succeeds_and_cleans_up() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path("/nifi-api/flowfile-queues/q-1/drop-requests/drop-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(drop_request_entity(true, None)))
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path("/nifi-api/flowfile-queues/q-1/drop-requests/drop-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(drop_request_entity(true, None)))
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto = flowfile_drop_dynamic(&client, "q-1", "drop-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(dto.finished, Some(true));
}

fn listing_request_entity(finished: bool, failure: Option<&str>) -> serde_json::Value {
    let mut req = json!({
        "id": "list-1",
        "finished": finished,
        "percentCompleted": if finished { 100 } else { 50 },
    });
    if let Some(reason) = failure {
        req["failureReason"] = json!(reason);
    }
    json!({ "listingRequest": req })
}

#[tokio::test]
async fn flowfile_listing_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/flowfile-queues/q-1/listing-requests/list-1",
        ))
        .respond_with(ResponseTemplate::new(200).set_body_json(listing_request_entity(true, None)))
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/flowfile-queues/q-1/listing-requests/list-1",
        ))
        .respond_with(ResponseTemplate::new(200).set_body_json(listing_request_entity(true, None)))
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto = flowfile_listing_dynamic(&client, "q-1", "list-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(dto.finished, Some(true));
}

#[tokio::test]
async fn empty_all_connections_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/process-groups/pg-1/empty-all-connections-requests/drop-1",
        ))
        .respond_with(ResponseTemplate::new(200).set_body_json(drop_request_entity(true, None)))
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/process-groups/pg-1/empty-all-connections-requests/drop-1",
        ))
        .respond_with(ResponseTemplate::new(200).set_body_json(drop_request_entity(true, None)))
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto = empty_all_connections_dynamic(&client, "pg-1", "drop-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(dto.finished, Some(true));
}

#[tokio::test]
async fn provenance_lineage_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path("/nifi-api/provenance/lineage/lin-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "lineage": { "id": "lin-1", "finished": true, "percentCompleted": 100 }
        })))
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path("/nifi-api/provenance/lineage/lin-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "lineage": { "id": "lin-1", "finished": true }
        })))
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto = provenance_lineage_dynamic(&client, "lin-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(dto.finished, Some(true));
}

fn verify_config_request_entity(complete: bool, failure: Option<&str>) -> serde_json::Value {
    let mut req = json!({
        "requestId": "req-1",
        "complete": complete,
        "percentCompleted": if complete { 100 } else { 50 },
    });
    if let Some(reason) = failure {
        req["failureReason"] = json!(reason);
    }
    json!({ "request": req })
}

#[tokio::test]
async fn processor_verify_config_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/processors/p-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/processors/p-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto = processor_verify_config_dynamic(&client, "p-1", "req-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(dto.complete, Some(true));
}

#[tokio::test]
async fn controller_service_verify_config_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/controller-services/cs-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/controller-services/cs-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto = controller_service_verify_config_dynamic(&client, "cs-1", "req-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(dto.complete, Some(true));
}

#[tokio::test]
async fn reporting_task_verify_config_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/reporting-tasks/rt-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/reporting-tasks/rt-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto = reporting_task_verify_config_dynamic(&client, "rt-1", "req-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(dto.complete, Some(true));
}

#[tokio::test]
async fn parameter_provider_verify_config_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/parameter-providers/pp-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/parameter-providers/pp-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto = parameter_provider_verify_config_dynamic(&client, "pp-1", "req-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(dto.complete, Some(true));
}

#[tokio::test]
async fn flow_analysis_rule_verify_config_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/controller/flow-analysis-rules/far-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/controller/flow-analysis-rules/far-1/config/verification-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(verify_config_request_entity(true, None)),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto =
        flow_analysis_rule_verify_config_dynamic(&client, "far-1", "req-1", fast_config(1000))
            .await
            .unwrap();
    assert_eq!(dto.complete, Some(true));
}

fn apply_parameters_request_entity(complete: bool, failure: Option<&str>) -> serde_json::Value {
    let mut req = json!({
        "requestId": "req-1",
        "complete": complete,
        "percentCompleted": if complete { 100 } else { 50 },
    });
    if let Some(reason) = failure {
        req["failureReason"] = json!(reason);
    }
    json!({ "request": req })
}

#[tokio::test]
async fn parameter_provider_apply_parameters_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/parameter-providers/pp-1/apply-parameters-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(apply_parameters_request_entity(true, None)),
        )
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/parameter-providers/pp-1/apply-parameters-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(apply_parameters_request_entity(true, None)),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let dto =
        parameter_provider_apply_parameters_dynamic(&client, "pp-1", "req-1", fast_config(1000))
            .await
            .unwrap();
    assert_eq!(dto.complete, Some(true));
}

// ── versioned_flow_update_dynamic ──────────────────────────────────────────

fn versioned_flow_update_entity(complete: bool, failure: Option<&str>) -> serde_json::Value {
    let mut req = json!({
        "requestId": "req-1",
        "complete": complete,
        "percentCompleted": if complete { 100 } else { 50 },
    });
    if let Some(reason) = failure {
        req["failureReason"] = json!(reason);
    }
    json!({ "request": req })
}

#[tokio::test]
async fn versioned_flow_update_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path("/nifi-api/versions/update-requests/req-1"))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(versioned_flow_update_entity(true, None)),
        )
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path("/nifi-api/versions/update-requests/req-1"))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(versioned_flow_update_entity(true, None)),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let entity = versioned_flow_update_dynamic(&client, "req-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(entity.request.and_then(|r| r.complete), Some(true));
}

#[tokio::test]
async fn versioned_flow_revert_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path("/nifi-api/versions/revert-requests/req-1"))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(versioned_flow_update_entity(true, None)),
        )
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path("/nifi-api/versions/revert-requests/req-1"))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(versioned_flow_update_entity(true, None)),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let entity = versioned_flow_revert_dynamic(&client, "req-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(entity.request.and_then(|r| r.complete), Some(true));
}

fn validation_request_entity(complete: bool, failure: Option<&str>) -> serde_json::Value {
    let mut req = json!({
        "requestId": "req-1",
        "complete": complete,
        "percentCompleted": if complete { 100 } else { 50 },
    });
    if let Some(reason) = failure {
        req["failureReason"] = json!(reason);
    }
    json!({ "request": req })
}

#[tokio::test]
async fn parameter_context_validation_dynamic_succeeds() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/parameter-contexts/ctx-1/validation-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(validation_request_entity(true, None)),
        )
        .mount(&mock_server)
        .await;
    Mock::given(method("DELETE"))
        .and(path(
            "/nifi-api/parameter-contexts/ctx-1/validation-requests/req-1",
        ))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(validation_request_entity(true, None)),
        )
        .expect(1)
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let entity = parameter_context_validation_dynamic(&client, "ctx-1", "req-1", fast_config(1000))
        .await
        .unwrap();
    assert_eq!(entity.request.and_then(|r| r.complete), Some(true));
}

// ── process_group_state_dynamic ──────────────────────────────────────────────

#[tokio::test]
async fn process_group_state_dynamic_reaches_running() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    Mock::given(method("GET"))
        .and(path("/nifi-api/process-groups/pg-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "id": "pg-1", "runningCount": 3, "stoppedCount": 1
        })))
        .up_to_n_times(1)
        .mount(&mock_server)
        .await;
    Mock::given(method("GET"))
        .and(path("/nifi-api/process-groups/pg-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "id": "pg-1", "runningCount": 4, "stoppedCount": 0
        })))
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let entity = process_group_state_dynamic(
        &client,
        "pg-1",
        ProcessGroupTargetState::Running,
        fast_config(1000),
    )
    .await
    .unwrap();
    assert_eq!(entity.stopped_count, Some(0));
}

// ── process_group_controller_services_state_dynamic ──────────────────────────

#[tokio::test]
async fn pg_controller_services_state_dynamic_reaches_disabled() {
    let mock_server = MockServer::start().await;
    mount_about(&mock_server).await;

    // First poll: one service still DISABLING. Second: all DISABLED.
    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/flow/process-groups/pg-1/controller-services",
        ))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "controllerServices": [
                { "component": { "state": "DISABLED", "validationStatus": "VALID" } },
                { "component": { "state": "DISABLING", "validationStatus": "VALID" } }
            ]
        })))
        .up_to_n_times(1)
        .mount(&mock_server)
        .await;
    Mock::given(method("GET"))
        .and(path(
            "/nifi-api/flow/process-groups/pg-1/controller-services",
        ))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "controllerServices": [
                { "component": { "state": "DISABLED", "validationStatus": "VALID" } },
                { "component": { "state": "DISABLED", "validationStatus": "VALID" } }
            ]
        })))
        .mount(&mock_server)
        .await;

    let client = dynamic_client(&mock_server).await;

    let entity = process_group_controller_services_state_dynamic(
        &client,
        "pg-1",
        ControllerServiceTargetState::Disabled,
        fast_config(1000),
    )
    .await
    .unwrap();
    assert!(entity.controller_services.is_some());
}
