videosdk-server-sdk 0.1.0

//! End-to-end tests for the Phase 5 resources, against a mock HTTP server.

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use serde_json::{json, Value};
use wiremock::matchers::{body_json, method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};

use crate::resources::agents::{
    DeployAgentParams, DispatchAgentParams, GeneralDispatchAgentParams, ListDeploymentsParams,
};
use crate::resources::batch_calls::{
    BatchCallCancelMode, ExportBatchCallRecordsParams, ListBatchCallRecordsParams,
    UploadBatchCallParams,
};
use crate::resources::connectors::{ConnectorProvider, CreateConnectorParams};
use crate::resources::ingress::{
    CreateSocketIngressParams, CreateWhepParams, CreateWhipParams, WhepSource,
};
use crate::resources::random_uuid;
use crate::resources::resource_pool::{AcquireResourceParams, ComposerType};
use crate::resources::transcodings::{MeetingRecordingMergeParams, MeetingRecordingRef};
use crate::resources::transcription::StartRealtimeTranscriptionParams;
use crate::test_support::{body, client, header, only_request, requests};

async fn mount(server: &MockServer, verb: &str, route: &str, response: Value) {
    Mock::given(method(verb))
        .and(path(route))
        .respond_with(ResponseTemplate::new(200).set_body_json(response))
        .mount(server)
        .await;
}

/// Writes a temporary file. The caller removes it.
fn temp_file(name: &str, contents: &[u8]) -> std::path::PathBuf {
    let path = std::env::temp_dir().join(format!("videosdk-rs-{}-{name}", random_uuid()));
    std::fs::write(&path, contents).expect("write temp file");
    path
}

/* ----------------------------------- agents ----------------------------------- */

#[tokio::test]
async fn dispatch_folds_the_meeting_id_and_room_id_aliases() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/agent/dispatch"))
        // Only `room_id` was supplied, yet both go on the wire.
        .and(body_json(json!({
            "agentId": "a-1",
            "meetingId": "r-1",
            "roomId": "r-1",
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"data": {"jobId": "j-1"}})))
        .expect(1)
        .mount(&server)
        .await;

    let result = client(&server)
        .agents()
        .dispatch(DispatchAgentParams {
            agent_id: Some("a-1".into()),
            room_id: Some("r-1".into()),
            ..Default::default()
        })
        .await
        .unwrap();
    assert_eq!(result.job_id.as_deref(), Some("j-1"));
}

#[tokio::test]
async fn dispatch_requires_a_room() {
    let server = MockServer::start().await;
    let err = client(&server)
        .agents()
        .dispatch(DispatchAgentParams::default())
        .await
        .unwrap_err();

    assert!(err.is_validation());
    assert!(err.to_string().contains("meeting_id"));
    assert!(requests(&server).await.is_empty());
}

#[tokio::test]
async fn general_dispatch_drops_version_tag_and_wait() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/agent/general/dispatch"))
        .and(body_json(json!({
            "versionId": "v-1", "meetingId": "r-1", "roomId": "r-1",
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"data": {}})))
        .expect(1)
        .mount(&server)
        .await;

    client(&server)
        .agents()
        .general_dispatch(GeneralDispatchAgentParams {
            meeting_id: Some("r-1".into()),
            version_id: Some("v-1".into()),
            ..Default::default()
        })
        .await
        .unwrap();
}

#[tokio::test]
async fn init_config_unwraps_data() {
    let server = MockServer::start().await;
    mount(
        &server,
        "POST",
        "/v2/agent/init-config",
        json!({"data": {"registryUrl": "wss://registry"}}),
    )
    .await;

    let config = client(&server).agents().init_config().await.unwrap();
    assert_eq!(config.registry_url, "wss://registry");
}

#[tokio::test]
async fn deploying_without_env_skips_the_secret_and_omits_agent_id() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/ai/v1/cloud/deployments"))
        // `agentId` must be absent, not "": the server's `agentId ?? generate()`
        // would not catch an empty string, and its schema then rejects it.
        .and(body_json(json!({"name": "my agent"})))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(json!({"id": "d-1", "agentId": "a-1"})),
        )
        .expect(1)
        .mount(&server)
        .await;
    Mock::given(method("POST"))
        .and(path("/ai/v1/cloud/versions"))
        .and(body_json(json!({
            "agentId": "a-1",
            "deploymentId": "d-1",
            "image": {
                "imageUrl": "registry.videosdk.live/acme/agent:1.4",
                "imageCR": "registry.videosdk.live",
                "imageType": "public",
            },
            "profile": "cpu-small",
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"id": "v-1"})))
        .expect(1)
        .mount(&server)
        .await;

    let deployed = client(&server)
        .agents()
        .deployment()
        .create(DeployAgentParams {
            name: "my agent".into(),
            image: Some("registry.videosdk.live/acme/agent:1.4".into()),
            ..Default::default()
        })
        .await
        .unwrap();

    assert_eq!(deployed.deployment.id, "d-1");
    assert_eq!(deployed.version.id, "v-1");
    assert!(deployed.secret_id.is_none());
    assert_eq!(requests(&server).await.len(), 2, "no secret call was made");
}

#[tokio::test]
async fn deploying_with_env_creates_a_secret_first_and_attaches_it() {
    let server = MockServer::start().await;
    mount(
        &server,
        "POST",
        "/ai/v1/cloud/secrets",
        json!({"data": {"id": "s-1"}}),
    )
    .await;
    mount(
        &server,
        "POST",
        "/ai/v1/cloud/deployments",
        json!({"id": "d-1", "agentId": "a-1"}),
    )
    .await;
    mount(
        &server,
        "POST",
        "/ai/v1/cloud/versions",
        json!({"id": "v-1"}),
    )
    .await;

    let deployed = client(&server)
        .agents()
        .deployment()
        .create(DeployAgentParams {
            name: "My Agent!".into(),
            image: Some("acme/agent".into()),
            env: HashMap::from([("KEY".to_string(), "value".to_string())]),
            env_secret_name: Some("fixed-name".into()),
            ..Default::default()
        })
        .await
        .unwrap();

    assert_eq!(deployed.secret_id.as_deref(), Some("s-1"));

    let sent = requests(&server).await;
    assert_eq!(sent.len(), 3, "secret, deployment, version");
    assert_eq!(
        body(&sent[0]),
        json!({"name": "fixed-name", "keys": {"KEY": "value"}, "type": "NORMAL"})
    );
    assert_eq!(
        body(&sent[1]),
        json!({"name": "My Agent!", "envSecret": "s-1"})
    );
    // A short reference falls back to Docker Hub.
    assert_eq!(body(&sent[2])["image"]["imageCR"], json!("docker.io"));
}

#[tokio::test]
async fn a_generated_secret_name_is_slugified_from_the_deployment_name() {
    let server = MockServer::start().await;
    mount(
        &server,
        "POST",
        "/ai/v1/cloud/secrets",
        json!({"data": {"id": "s-1"}}),
    )
    .await;
    mount(
        &server,
        "POST",
        "/ai/v1/cloud/deployments",
        json!({"id": "d-1"}),
    )
    .await;
    mount(
        &server,
        "POST",
        "/ai/v1/cloud/versions",
        json!({"id": "v-1"}),
    )
    .await;

    client(&server)
        .agents()
        .deployment()
        .create(DeployAgentParams {
            name: "My Agent!".into(),
            image: Some("agent".into()),
            env: HashMap::from([("K".to_string(), "v".to_string())]),
            ..Default::default()
        })
        .await
        .unwrap();

    let name = body(&requests(&server).await[0])["name"]
        .as_str()
        .unwrap()
        .to_string();
    assert!(name.starts_with("my-agent-env-"), "{name}");
    assert_eq!(name.len(), "my-agent-env-".len() + 8);
}

#[tokio::test]
async fn getting_a_deployment_filters_the_list_and_reports_not_found() {
    let server = MockServer::start().await;
    Mock::given(method("GET"))
        .and(query_param("deploymentId", "d-1"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "deployments": [{"id": "other"}, {"id": "d-1", "name": "match"}],
        })))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"deployments": []})))
        .mount(&server)
        .await;

    let client = client(&server);
    let deployment = client.agents().deployment().get("d-1").await.unwrap();
    assert_eq!(deployment.name.as_deref(), Some("match"));

    let err = client
        .agents()
        .deployment()
        .get("absent")
        .await
        .unwrap_err();
    assert!(err.is_not_found());
}

#[tokio::test]
async fn deployment_logs_read_the_logs_data_key() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/ai/v1/cloud/deployments/d-1/console-logs",
        json!({"logs": [{"log": "hello"}, {"log": "world"}]}),
    )
    .await;

    let page = client(&server)
        .agents()
        .deployment()
        .logs("d-1", Default::default())
        .await
        .unwrap();
    assert_eq!(page.data.len(), 2);
    assert_eq!(page.data[0].log, "hello");
}

#[tokio::test]
async fn deleting_a_deployment_posts_force() {
    let server = MockServer::start().await;
    mount(
        &server,
        "POST",
        "/ai/v1/cloud/deployments/d-1/delete",
        json!({"message": "deleted"}),
    )
    .await;

    client(&server)
        .agents()
        .deployment()
        .delete("d-1", true)
        .await
        .unwrap();
    assert_eq!(body(&only_request(&server).await), json!({"force": true}));
}

#[tokio::test]
async fn listing_deployments_reads_the_deployments_data_key() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/ai/v1/cloud/deployments",
        json!({"deployments": [{"id": "d-1"}]}),
    )
    .await;

    let page = client(&server)
        .agents()
        .deployment()
        .list(ListDeploymentsParams {
            status: Some("running".into()),
            ..Default::default()
        })
        .await
        .unwrap();
    assert_eq!(page.data[0].id, "d-1");
}

/* --------------------------------- batch calls -------------------------------- */

/// Mocks the presign, PUT, parse and poll sequence. Each status answers once, so
/// the poll walks them in order.
async fn mount_upload(server: &MockServer, statuses: &[&str]) {
    Mock::given(method("POST"))
        .and(path("/v2/batch-calls/upload"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "presignedUrl": format!("{}/storage/put?X-Amz-Signature=secret", server.uri()),
            "batchId": "b-1",
        })))
        .expect(1)
        .mount(server)
        .await;
    Mock::given(method("PUT"))
        .and(path("/storage/put"))
        .respond_with(ResponseTemplate::new(200))
        .expect(1)
        .mount(server)
        .await;
    Mock::given(method("POST"))
        .and(path("/v2/batch-calls/parse"))
        .respond_with(ResponseTemplate::new(204))
        .expect(1)
        .mount(server)
        .await;

    for status in statuses {
        Mock::given(method("GET"))
            .and(path("/v2/batch-calls/b-1"))
            .respond_with(
                ResponseTemplate::new(200)
                    .set_body_json(json!({"batchId": "b-1", "status": status})),
            )
            .up_to_n_times(1)
            .mount(server)
            .await;
    }
}

#[tokio::test]
async fn uploading_presigns_puts_parses_and_polls_until_parsed() {
    let server = MockServer::start().await;
    mount_upload(&server, &["parsing", "parsed"]).await;
    let file = temp_file("recipients.csv", b"phone\n+15550001\n");

    let parsing_called = Arc::new(AtomicBool::new(false));
    let flag = parsing_called.clone();

    let batch = client(&server)
        .batch_call()
        .upload(UploadBatchCallParams {
            file_path: file.to_string_lossy().into_owned(),
            poll_interval: Some(Duration::from_millis(5)),
            on_parsing: Some(Box::new(move || flag.store(true, Ordering::SeqCst))),
            ..Default::default()
        })
        .await
        .unwrap();

    assert_eq!(batch.batch_id, "b-1");
    assert_eq!(batch.status.unwrap().as_str(), "parsed");
    assert!(parsing_called.load(Ordering::SeqCst));

    // The PUT carries the file bytes and no Authorization header — sending one
    // would invalidate the presigned signature.
    let put = requests(&server)
        .await
        .into_iter()
        .find(|request| request.method.as_str() == "PUT")
        .expect("a PUT was issued");
    assert!(header(&put, "authorization").is_none());
    assert_eq!(header(&put, "content-type").as_deref(), Some("text/csv"));
    assert_eq!(put.body, b"phone\n+15550001\n");

    std::fs::remove_file(file).ok();
}

#[tokio::test]
async fn uploading_can_skip_waiting_for_the_parse() {
    let server = MockServer::start().await;
    mount_upload(&server, &["parsing"]).await;
    let file = temp_file("recipients.csv", b"phone\n");

    let batch = client(&server)
        .batch_call()
        .upload(UploadBatchCallParams {
            file_path: file.to_string_lossy().into_owned(),
            wait_for_parse: Some(false),
            ..Default::default()
        })
        .await
        .unwrap();

    assert_eq!(batch.status.unwrap().as_str(), "parsing");
    std::fs::remove_file(file).ok();
}

#[tokio::test]
async fn a_failed_parse_carries_a_machine_readable_code() {
    let server = MockServer::start().await;
    mount_upload(&server, &["parse_failed"]).await;
    let file = temp_file("recipients.csv", b"bad\n");

    let err = client(&server)
        .batch_call()
        .upload(UploadBatchCallParams {
            file_path: file.to_string_lossy().into_owned(),
            poll_interval: Some(Duration::from_millis(5)),
            ..Default::default()
        })
        .await
        .unwrap_err();

    assert_eq!(err.code(), Some("batchcall_parse_failed"));
    assert_eq!(err.status(), None, "a result, not a response");
    std::fs::remove_file(file).ok();
}

#[tokio::test]
async fn a_parse_that_never_finishes_times_out() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/batch-calls/upload"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "presignedUrl": format!("{}/storage/put", server.uri()),
            "batchId": "b-1",
        })))
        .mount(&server)
        .await;
    Mock::given(method("PUT"))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;
    Mock::given(method("POST"))
        .and(path("/v2/batch-calls/parse"))
        .respond_with(ResponseTemplate::new(204))
        .mount(&server)
        .await;
    Mock::given(method("GET"))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"batchId": "b-1", "status": "parsing"})),
        )
        .mount(&server)
        .await;

    let file = temp_file("recipients.csv", b"phone\n");
    let err = client(&server)
        .batch_call()
        .upload(UploadBatchCallParams {
            file_path: file.to_string_lossy().into_owned(),
            poll_interval: Some(Duration::from_millis(1)),
            poll_timeout: Some(Duration::from_millis(20)),
            ..Default::default()
        })
        .await
        .unwrap_err();

    assert!(err.is_timeout());
    assert_eq!(err.code(), Some("batchcall_parse_timeout"));
    std::fs::remove_file(file).ok();
}

#[tokio::test]
async fn uploading_rejects_bad_extensions_and_missing_files_before_any_request() {
    let server = MockServer::start().await;
    let client = client(&server);

    let file = temp_file("recipients.txt", b"nope");
    let err = client
        .batch_call()
        .upload(UploadBatchCallParams {
            file_path: file.to_string_lossy().into_owned(),
            ..Default::default()
        })
        .await
        .unwrap_err();
    assert!(err.is_validation());
    assert!(err.to_string().contains(".csv"));
    std::fs::remove_file(file).ok();

    let err = client
        .batch_call()
        .upload(UploadBatchCallParams {
            file_path: "/nonexistent/file.csv".into(),
            ..Default::default()
        })
        .await
        .unwrap_err();
    assert!(err.is_validation());
    assert!(err.to_string().contains("could not read"));

    assert!(requests(&server).await.is_empty());
}

#[tokio::test]
async fn batch_stats_expose_the_per_status_breakdown() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/v2/batch-calls/b-1/stats",
        json!({"batchId": "b-1", "status": "running", "total": 8, "completed": 5}),
    )
    .await;

    let stats = client(&server).batch_call().stats("b-1").await.unwrap();
    assert_eq!(stats.total, Some(8));
    assert_eq!(stats.counts["completed"], 5);
}

#[tokio::test]
async fn cancelling_a_batch_posts_its_mode() {
    let server = MockServer::start().await;
    mount(
        &server,
        "POST",
        "/v2/batch-calls/b-1/cancel",
        json!({"batchId": "b-1"}),
    )
    .await;

    client(&server)
        .batch_call()
        .cancel("b-1", BatchCallCancelMode::GRACEFUL)
        .await
        .unwrap();
    assert_eq!(
        body(&only_request(&server).await),
        json!({"mode": "graceful"})
    );
}

#[tokio::test]
async fn records_read_the_records_key_and_rename_search_to_query() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/v2/batch-calls/b-1/records",
        json!({"records": [{"recordId": "r-1"}]}),
    )
    .await;

    let page = client(&server)
        .batch_call()
        .records()
        .list(
            "b-1",
            ListBatchCallRecordsParams {
                search: Some("+1555".into()),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    assert_eq!(page.data[0].record_id, "r-1");

    let request = only_request(&server).await;
    let query: HashMap<_, _> = request.url.query_pairs().collect();
    assert_eq!(query["query"], "+1555", "search is sent as `query` here");
    assert!(!query.contains_key("search"));
}

#[tokio::test]
async fn exporting_records_returns_raw_csv() {
    let server = MockServer::start().await;
    Mock::given(method("GET"))
        .and(path("/v2/batch-calls/b-1/records/export"))
        .respond_with(ResponseTemplate::new(200).set_body_string("recordId,phone\nr-1,+1\n"))
        .mount(&server)
        .await;

    let csv = client(&server)
        .batch_call()
        .records()
        .export("b-1", ExportBatchCallRecordsParams::default())
        .await
        .unwrap();
    assert_eq!(csv, "recordId,phone\nr-1,+1\n");
}

/* --------------------------------- connectors --------------------------------- */

#[tokio::test]
async fn creating_a_connector_renames_room_id_to_default_room_id() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/connectors"))
        .and(body_json(
            json!({"provider": "twilio", "defaultRoomId": "r-1"}),
        ))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"data": {"id": "c-1"}})))
        .expect(1)
        .mount(&server)
        .await;

    let connector = client(&server)
        .connectors()
        .create(CreateConnectorParams {
            provider: Some(ConnectorProvider::TWILIO),
            room_id: Some("r-1".into()),
            ..Default::default()
        })
        .await
        .unwrap();
    assert_eq!(connector.id, "c-1");
}

#[tokio::test]
async fn listing_connectors_wraps_a_flat_array_in_a_complete_page() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/v2/connectors",
        json!({"data": [{"id": "c-1"}, {"id": "c-2"}]}),
    )
    .await;

    let page = client(&server).connectors().list().await.unwrap();
    assert_eq!(page.data.len(), 2);
    assert_eq!(page.total(), 2);
    assert!(!page.has_next_page());
}

/* -------------------------------- resource pool ------------------------------- */

#[tokio::test]
async fn acquiring_units_unwraps_the_data_array() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/resource/acquire"))
        .and(body_json(
            json!({"type": "hls", "webhookUrl": "https://x", "units": 2}),
        ))
        .respond_with(
            ResponseTemplate::new(200)
                .set_body_json(json!({"data": [{"id": "u-1"}, {"id": "u-2"}]})),
        )
        .expect(1)
        .mount(&server)
        .await;

    let units = client(&server)
        .resource_pool()
        .acquire(AcquireResourceParams {
            kind: Some(ComposerType::HLS),
            webhook_url: "https://x".into(),
            units: Some(2),
            ..Default::default()
        })
        .await
        .unwrap();
    assert_eq!(units.len(), 2);
    assert_eq!(units[1].id, "u-2");
}

#[tokio::test]
async fn releasing_units_posts_their_ids() {
    let server = MockServer::start().await;
    mount(
        &server,
        "POST",
        "/v2/resource/release",
        json!({"data": [{"id": "u-1", "success": true}]}),
    )
    .await;

    let results = client(&server)
        .resource_pool()
        .release(&["u-1".to_string()])
        .await
        .unwrap();
    assert!(results[0].success);
    assert_eq!(body(&only_request(&server).await), json!({"ids": ["u-1"]}));
}

/* -------------------------------- transcodings -------------------------------- */

#[tokio::test]
async fn a_meeting_recording_merge_serializes_mixed_refs() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/transcodings/meeting-recording-merge"))
        .and(body_json(json!({
            "recordingIds": ["rec-1", {"id": "rec-2", "presignedUrl": "https://s3/get"}],
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"id": "t-1"})))
        .expect(1)
        .mount(&server)
        .await;

    let transcoding = client(&server)
        .transcodings()
        .meeting_recording_merge(MeetingRecordingMergeParams {
            recording_ids: vec![
                "rec-1".into(),
                MeetingRecordingRef {
                    id: "rec-2".into(),
                    presigned_url: Some("https://s3/get".into()),
                },
            ],
            ..Default::default()
        })
        .await
        .unwrap();
    assert_eq!(transcoding.id, "t-1");
}

#[tokio::test]
async fn cancelling_a_transcoding_hits_its_route() {
    let server = MockServer::start().await;
    mount(
        &server,
        "POST",
        "/v2/transcodings/t-1/cancel",
        json!({"id": "t-1", "status": "cancelled"}),
    )
    .await;

    let transcoding = client(&server).transcodings().cancel("t-1").await.unwrap();
    assert_eq!(transcoding.status.unwrap().as_str(), "cancelled");
}

/* -------------------------------- transcription ------------------------------- */

#[tokio::test]
async fn starting_a_transcription_populates_the_room_id() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/ai/v1/realtime-transcriptions/start"))
        .and(body_json(json!({"roomId": "r-1", "language": "en-US"})))
        // The start response carries no roomId; the SDK fills it in.
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"id": "t-1"})))
        .expect(1)
        .mount(&server)
        .await;

    let transcription = client(&server)
        .transcription()
        .realtime()
        .start(
            "r-1",
            StartRealtimeTranscriptionParams {
                language: Some("en-US".into()),
                ..Default::default()
            },
        )
        .await
        .unwrap();
    assert_eq!(transcription.room_id.as_deref(), Some("r-1"));
}

#[tokio::test]
async fn listing_transcriptions_lifts_the_nested_room_id() {
    let server = MockServer::start().await;
    mount(
        &server,
        "GET",
        "/ai/v1/realtime-transcriptions",
        json!({"transcriptions": [
            {"id": "t-1", "extension": {"extensionConfig": {"roomId": "r-9"}}},
        ]}),
    )
    .await;

    let page = client(&server)
        .transcription()
        .realtime()
        .list(Default::default())
        .await
        .unwrap();
    assert_eq!(page.data[0].room_id.as_deref(), Some("r-9"));
}

#[tokio::test]
async fn stopping_a_transcription_requires_a_room_id() {
    let server = MockServer::start().await;
    let err = client(&server)
        .transcription()
        .realtime()
        .stop("")
        .await
        .unwrap_err();
    assert!(err.is_validation());
    assert!(requests(&server).await.is_empty());
}

/* ----------------------------------- ingress ---------------------------------- */

#[tokio::test]
async fn whip_create_is_synchronous_and_makes_no_request() {
    let server = MockServer::start().await;
    let client = client(&server);

    let ingress = client
        .whip()
        .create(
            "r-1",
            CreateWhipParams {
                name: Some("Publisher".into()),
                use_existing_peer: Some(true),
                ..Default::default()
            },
        )
        .unwrap();

    assert!(ingress.url.starts_with(&server.uri()));
    assert!(ingress.url.contains("roomId=r-1"));
    assert!(ingress.url.contains("useExistingPeer=true"));
    assert!(ingress.participant_id.starts_with("whip-"));
    // The stream key is an alias of the token.
    assert_eq!(ingress.token, ingress.stream_key);
    assert_eq!(ingress.token.split('.').count(), 3, "a compact JWT");

    assert!(requests(&server).await.is_empty(), "no network call");
}

#[tokio::test]
async fn whip_create_omits_absent_query_parameters() {
    let server = MockServer::start().await;
    let ingress = client(&server)
        .whip()
        .create("r-1", CreateWhipParams::default())
        .unwrap();
    assert!(!ingress.url.contains("displayName"));
    assert!(!ingress.url.contains("useExistingPeer"));
}

#[tokio::test]
async fn whep_create_carries_the_remote_peer() {
    let server = MockServer::start().await;
    let playback = client(&server)
        .whep()
        .create(
            "r-1",
            CreateWhepParams {
                participant_id: Some("sub-1".into()),
                source: Some(WhepSource {
                    participant_id: "pub-1".into(),
                }),
                ..Default::default()
            },
        )
        .unwrap();

    assert!(playback.url.contains("remotePeerId=pub-1"));
    assert_eq!(playback.participant_id, "sub-1");
    assert_eq!(playback.remote_peer_id.as_deref(), Some("pub-1"));
}

#[tokio::test]
async fn whip_delete_resolves_the_active_session() {
    let server = MockServer::start().await;
    Mock::given(method("GET"))
        .and(path("/v2/sessions"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"data": [{"id": "s-9"}]})))
        .mount(&server)
        .await;
    Mock::given(method("DELETE"))
        .and(path("/v2/whip/sessions/s-9"))
        .respond_with(ResponseTemplate::new(204))
        .expect(1)
        .mount(&server)
        .await;

    let client = client(&server);
    let ingress = client
        .whip()
        .create("r-1", CreateWhipParams::default())
        .unwrap();
    client.whip().delete(&ingress).await.unwrap();
}

#[tokio::test]
async fn whip_delete_reports_not_found_without_a_live_session() {
    let server = MockServer::start().await;
    mount(&server, "GET", "/v2/sessions", json!({"data": []})).await;

    let client = client(&server);
    let ingress = client
        .whip()
        .create("r-1", CreateWhipParams::default())
        .unwrap();
    let err = client.whip().delete(&ingress).await.unwrap_err();
    assert!(err.is_not_found());
}

#[tokio::test]
async fn socket_ingress_parses_the_ref_out_of_the_ws_url() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/ingest/sessions"))
        .and(body_json(json!({
            "roomId": "r-1",
            "participant": {"id": "p-1", "name": "Streamer"},
        })))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({"data": {
            "wsUrl": "wss://ingest.videosdk.live/socket?ref=abc123",
            "roomId": "r-1",
            "expiresIn": 90,
        }})))
        .expect(1)
        .mount(&server)
        .await;

    let ingress = client(&server)
        .socket_ingress()
        .create(
            "r-1",
            CreateSocketIngressParams {
                participant_id: Some("p-1".into()),
                name: Some("Streamer".into()),
                ..Default::default()
            },
        )
        .await
        .unwrap();

    assert_eq!(ingress.ws_ref.as_deref(), Some("abc123"));
    assert_eq!(ingress.expires_in, 90);
}

#[tokio::test]
async fn socket_ingress_omits_the_participant_when_nothing_is_set() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(body_json(json!({"roomId": "r-1"})))
        .respond_with(
            ResponseTemplate::new(200).set_body_json(json!({"data": {"wsUrl": "wss://x/s"}})),
        )
        .expect(1)
        .mount(&server)
        .await;

    let ingress = client(&server)
        .socket_ingress()
        .create("r-1", CreateSocketIngressParams::default())
        .await
        .unwrap();
    assert_eq!(ingress.ws_ref, None);
}

/* --------------------------------- alert rules -------------------------------- */

#[tokio::test]
async fn creating_an_alert_rule_unwraps_the_alert_rule_envelope() {
    use crate::resources::alert_rules::{
        AlertConditionOperator, AlertRuleCondition, AlertRuleConfig, AlertRuleQuery, AlertSeverity,
        CreateAlertRuleParams,
    };

    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/v2/alertRules"))
        .and(body_json(json!({
            "name": "High bitrate",
            "severity": "critical",
            "ruleConfig": {
                "query": {"metric": "stat_consumer_video_bitrate"},
                "condition": {"operator": ">", "threshold": 2500.0},
            },
            "notificationChannelIds": ["ch-1"],
        })))
        // Response is wrapped in `alertRule`, matching Go's doWrappedJSON.
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "alertRule": {"alertRuleId": "ar-1", "status": "active"},
        })))
        .expect(1)
        .mount(&server)
        .await;

    let mut params = CreateAlertRuleParams::new(
        "High bitrate",
        AlertSeverity::CRITICAL,
        AlertRuleConfig::new(
            AlertRuleQuery::metric("stat_consumer_video_bitrate"),
            AlertRuleCondition::new(AlertConditionOperator::GREATER_THAN, 2500.0),
        ),
    );
    params.notification_channel_ids = vec!["ch-1".into()];

    let rule = client(&server).alert_rules().create(params).await.unwrap();
    assert_eq!(rule.alert_rule_id, "ar-1");
    assert_eq!(rule.status.unwrap().as_str(), "active");
}