ocppsim 0.2.0

use futures_util::{SinkExt, StreamExt};
use http::HeaderValue;
use http::header::SEC_WEBSOCKET_PROTOCOL;
use serde_json::{Value, json};
use tokio::net::TcpListener;
use tokio_tungstenite::tungstenite::handshake::server::{
  ErrorResponse, Request, Response,
};
use tokio_tungstenite::tungstenite::protocol::Message;
use tokio_tungstenite::{accept_async, accept_hdr_async, connect_async};

use crate::ocpp::{OcppFrame, build_call, build_call_result, parse_frame};

use super::*;

// The tungstenite handshake callback trait requires this large error type
// and result wrapping.
#[allow(clippy::result_large_err, clippy::unnecessary_wraps)]
fn accept_v1_6_subprotocol(
  _request: &Request,
  mut response: Response,
) -> Result<Response, ErrorResponse> {
  response
    .headers_mut()
    .insert(SEC_WEBSOCKET_PROTOCOL, HeaderValue::from_static("ocpp1.6"));
  Ok(response)
}

async fn capture_inbound_call_response(
  protocol: OcppVersion,
  action: &str,
  payload: Value,
) -> (OcppFrame, Simulator) {
  let (frame, simulator, _events) =
    capture_inbound_call_response_with_events(protocol, false, action, payload)
      .await;
  (frame, simulator)
}

async fn capture_inbound_call_response_with_strict(
  protocol: OcppVersion,
  strict: bool,
  action: &str,
  payload: Value,
) -> (OcppFrame, Simulator) {
  let (frame, simulator, _events) = capture_inbound_call_response_with_events(
    protocol, strict, action, payload,
  )
  .await;
  (frame, simulator)
}

async fn capture_inbound_call_response_with_events(
  protocol: OcppVersion,
  strict: bool,
  action: &str,
  payload: Value,
) -> (OcppFrame, Simulator, Vec<UiEvent>) {
  let listener = TcpListener::bind("127.0.0.1:0")
    .await
    .expect("bind listener");
  let address = listener.local_addr().expect("local address");
  let server = tokio::spawn(async move {
    let (stream, _) = listener.accept().await.expect("accept client");
    let websocket = accept_async(stream).await.expect("accept websocket");
    let (_server_write, mut server_read) = websocket.split();
    let message = server_read
      .next()
      .await
      .expect("response frame")
      .expect("response frame ok");
    parse_frame(message.to_text().expect("text frame")).expect("parse response")
  });

  let (client_stream, _) = connect_async(format!("ws://{address}"))
    .await
    .expect("connect client");
  let (mut write, _read) = client_stream.split();
  let (mut simulator, mut ui_rx) =
    simulator_for_tests_with_protocol_and_ui(protocol);
  simulator.config.strict = strict;
  simulator
    .handle_incoming_call(&mut write, "test-message", action, payload)
    .await
    .expect("handle inbound call");
  drop(write);

  let frame = server.await.expect("server task");
  let mut events = Vec::new();
  while let Ok(event) = ui_rx.try_recv() {
    events.push(event);
  }
  (frame, simulator, events)
}

async fn capture_ws_text_response_with_events(
  protocol: OcppVersion,
  text: String,
) -> (OcppFrame, Simulator, Vec<UiEvent>) {
  let listener = TcpListener::bind("127.0.0.1:0")
    .await
    .expect("bind listener");
  let address = listener.local_addr().expect("local address");
  let server = tokio::spawn(async move {
    let (stream, _) = listener.accept().await.expect("accept client");
    let websocket = accept_async(stream).await.expect("accept websocket");
    let (_server_write, mut server_read) = websocket.split();
    let message = server_read
      .next()
      .await
      .expect("response frame")
      .expect("response frame ok");
    parse_frame(message.to_text().expect("text frame")).expect("parse response")
  });

  let (client_stream, _) = connect_async(format!("ws://{address}"))
    .await
    .expect("connect client");
  let (mut write, _read) = client_stream.split();
  let (mut simulator, mut ui_rx) =
    simulator_for_tests_with_protocol_and_ui(protocol);
  simulator.config.trace_frames = true;
  simulator
    .handle_ws_text(text, &mut write)
    .await
    .expect("handle text");
  drop(write);

  let frame = server.await.expect("server task");
  let mut events = Vec::new();
  while let Ok(event) = ui_rx.try_recv() {
    events.push(event);
  }
  (frame, simulator, events)
}

fn assert_trace_redacted(events: &[UiEvent], secret: &str) {
  let log_messages = events
    .iter()
    .filter_map(|event| {
      if let UiEvent::Log { message, .. } = event {
        Some(message.as_str())
      } else {
        None
      }
    })
    .collect::<Vec<_>>();
  assert!(
    log_messages.iter().all(|message| !message.contains(secret)),
    "secret appeared in logs: {log_messages:?}"
  );
  assert!(
    log_messages
      .iter()
      .any(|message| message.contains("<redacted>")),
    "redacted marker missing from logs: {log_messages:?}"
  );
}

#[test]
fn validates_negotiated_subprotocol() {
  assert_eq!(
    validate_negotiated_subprotocol("ocpp1.6", Some("ocpp1.6"))
      .expect("subprotocol should match"),
    "ocpp1.6"
  );
  assert!(validate_negotiated_subprotocol("ocpp1.6", None).is_err());
  assert!(
    validate_negotiated_subprotocol("ocpp1.6", Some("ocpp2.0.1")).is_err()
  );
}

#[tokio::test]
async fn trace_frames_redacts_change_configuration_authorization_key() {
  let password = "0123456789abcdef0123456789abcdef";
  let frame = build_call(
    "secret-message",
    "ChangeConfiguration",
    &json!({
      "key": "AuthorizationKey",
      "value": password
    }),
  );

  let (_response, simulator, events) =
    capture_ws_text_response_with_events(OcppVersion::V1_6, frame).await;

  assert_eq!(
    simulator.security.basic_auth_password.as_deref(),
    Some(password)
  );
  assert_trace_redacted(&events, password);
}

#[tokio::test]
async fn trace_frames_redacts_set_variables_basic_auth_password() {
  let password = "0123456789abcdef0123456789abcdef";
  let frame = build_call(
    "secret-message",
    "SetVariables",
    &json!({
      "setVariableData": [
        set_variable_data("BasicAuthPassword", password)
      ]
    }),
  );

  let (_response, simulator, events) =
    capture_ws_text_response_with_events(OcppVersion::V2_0_1, frame).await;

  assert_eq!(
    simulator.security.basic_auth_password.as_deref(),
    Some(password)
  );
  assert_trace_redacted(&events, password);
}

#[tokio::test]
async fn boot_response_starts_heartbeat_from_interval() {
  let mut simulator = simulator_for_tests();
  simulator
    .apply_call_result_context(
      &PendingContext::Boot,
      &json!({
        "status": "Accepted",
        "currentTime": now_timestamp(),
        "interval": 17
      }),
    )
    .expect("boot response should apply");

  assert_eq!(
    simulator.heartbeat.as_ref().map(|item| item.seconds),
    Some(17)
  );
  assert_eq!(
    simulator
      .configuration
      .get(&ConfigurationKey::HeartbeatInterval)
      .map(|entry| entry.value.as_str()),
    Some("17")
  );
  simulator.stop_heartbeat();
}

#[tokio::test]
async fn mock_csms_boot_lifecycle_updates_heartbeat() {
  let listener = TcpListener::bind("127.0.0.1:0")
    .await
    .expect("bind listener");
  let address = listener.local_addr().expect("local address");
  let server = tokio::spawn(async move {
    let (stream, _) = listener.accept().await.expect("accept client");
    let mut websocket = accept_hdr_async(stream, accept_v1_6_subprotocol)
      .await
      .expect("accept websocket");
    let message = websocket
      .next()
      .await
      .expect("boot frame")
      .expect("boot frame ok");
    let frame =
      parse_frame(message.to_text().expect("text frame")).expect("parse frame");
    let OcppFrame::Call {
      message_id, action, ..
    } = frame
    else {
      panic!("expected CALL frame");
    };
    assert_eq!(action, "BootNotification");
    let response = build_call_result(
      &message_id,
      &json!({
        "status": "Accepted",
        "currentTime": now_timestamp(),
        "interval": 9
      }),
    );
    websocket
      .send(Message::Text(response.into()))
      .await
      .expect("send boot response");
  });

  let mut simulator = simulator_for_tests();
  simulator.config.ws_url = format!("ws://{address}");
  let mut connection = simulator.connect().await.expect("connect");
  simulator
    .try_send_next(&mut connection.write)
    .await
    .expect("send boot");
  let message = connection
    .read
    .next()
    .await
    .expect("boot response")
    .expect("boot response ok");
  simulator
    .handle_ws_message(message, &mut connection.write)
    .await
    .expect("handle response");

  assert_eq!(
    simulator.heartbeat.as_ref().map(|item| item.seconds),
    Some(9)
  );
  simulator.stop_heartbeat();
  server.await.expect("server task");
}

#[tokio::test]
async fn malformed_remote_start_returns_call_error() {
  let listener = TcpListener::bind("127.0.0.1:0")
    .await
    .expect("bind listener");
  let address = listener.local_addr().expect("local address");
  let server = tokio::spawn(async move {
    let (stream, _) = listener.accept().await.expect("accept client");
    let websocket = accept_async(stream).await.expect("accept websocket");
    let (_server_write, mut server_read) = websocket.split();
    let message = server_read
      .next()
      .await
      .expect("response frame")
      .expect("response frame ok");
    parse_frame(message.to_text().expect("text frame")).expect("parse response")
  });

  let (client_stream, _) = connect_async(format!("ws://{address}"))
    .await
    .expect("connect client");
  let (mut write, _read) = client_stream.split();
  let mut simulator = simulator_for_tests();
  simulator
    .handle_incoming_call_v1_6(
      &mut write,
      "bad-remote-start",
      "RemoteStartTransaction",
      json!({}),
    )
    .await
    .expect("handle malformed remote start");
  drop(write);

  let frame = server.await.expect("server task");
  let OcppFrame::CallError { code, .. } = frame else {
    panic!("expected CALLERROR frame");
  };
  assert_eq!(code, "FormationViolation");
  assert!(
    simulator
      .connectors
      .values()
      .all(|connector| connector.transaction.is_none())
  );
}

#[tokio::test]
async fn remote_start_v1_6_authorizes_before_start_when_configured() {
  let (frame, simulator) = capture_inbound_call_response(
    OcppVersion::V1_6,
    "RemoteStartTransaction",
    json!({ "idTag": "TOKEN" }),
  )
  .await;

  let OcppFrame::CallResult { payload, .. } = frame else {
    panic!("expected CALLRESULT frame");
  };
  assert_eq!(payload["status"], json!(ResponseStatus::Accepted.as_str()));
  assert!(
    simulator
      .connectors
      .values()
      .all(|connector| connector.transaction.is_none())
  );
  let authorize = simulator.queue.front().expect("queued authorize");
  assert_eq!(authorize.action, "Authorize");
  assert!(matches!(
    &authorize.context,
    PendingContext::RemoteStartAuthorizeV1_6 {
      connector: 1,
      id_token,
    } if id_token == "TOKEN"
  ));
}

#[tokio::test]
async fn malformed_request_start_v2_x_returns_call_error() {
  let malformed_payloads = [
    json!({ "idToken": { "idToken": "TOKEN" } }),
    json!({ "remoteStartId": 11, "idToken": {} }),
    json!({
      "remoteStartId": 11,
      "idToken": { "idToken": "TOKEN" },
      "evseId": "bad"
    }),
  ];

  for protocol in v2_x_protocols() {
    for payload in malformed_payloads.clone() {
      let (frame, simulator) = capture_inbound_call_response(
        protocol,
        "RequestStartTransaction",
        payload,
      )
      .await;
      let OcppFrame::CallError { code, .. } = frame else {
        panic!("expected CALLERROR frame");
      };
      assert_eq!(code, "FormationViolation");
      assert!(
        simulator
          .connectors
          .values()
          .all(|connector| connector.transaction.is_none())
      );
    }
  }
}

#[tokio::test]
async fn malformed_request_stop_v2_x_returns_call_error() {
  for protocol in v2_x_protocols() {
    let (frame, simulator) = capture_inbound_call_response(
      protocol,
      "RequestStopTransaction",
      json!({}),
    )
    .await;

    let OcppFrame::CallError { code, .. } = frame else {
      panic!("expected CALLERROR frame");
    };
    assert_eq!(code, "FormationViolation");
    assert!(
      simulator
        .connectors
        .values()
        .all(|connector| connector.transaction.is_none())
    );
  }
}

#[tokio::test]
async fn malformed_supported_requests_return_formation_violation() {
  let mut cases = vec![
    (
      OcppVersion::V1_6,
      "ReserveNow",
      json!({ "reservationId": 1 }),
    ),
    (OcppVersion::V1_6, "UnlockConnector", json!({})),
    (
      OcppVersion::V1_6,
      "SetChargingProfile",
      json!({ "csChargingProfiles": {} }),
    ),
    (
      OcppVersion::V1_6,
      "GetCompositeSchedule",
      json!({ "connectorId": 1 }),
    ),
    (OcppVersion::V1_6, "TriggerMessage", json!({})),
  ];

  for protocol in v2_x_protocols() {
    cases.extend([
      (protocol, "UnlockConnector", json!({ "evseId": 1 })),
      (
        protocol,
        "SetChargingProfile",
        json!({ "chargingProfile": {} }),
      ),
      (protocol, "GetCompositeSchedule", json!({ "evseId": 1 })),
      (protocol, "TriggerMessage", json!({})),
    ]);
  }

  for (protocol, action, payload) in cases {
    let (frame, simulator) =
      capture_inbound_call_response(protocol, action, payload).await;
    let OcppFrame::CallError { code, .. } = frame else {
      panic!("expected CALLERROR frame for {action}");
    };
    assert_eq!(code, "FormationViolation");
    assert!(
      simulator
        .connectors
        .values()
        .all(|connector| connector.transaction.is_none())
    );
  }
}

#[tokio::test]
async fn v2_x_unlock_rejects_unsupported_connector_id() {
  let cases = [
    (OcppVersion::V2_0_1, 0),
    (OcppVersion::V2_0_1, 2),
    (OcppVersion::V2_1, 2),
  ];

  for (protocol, connector_id) in cases {
    let (frame, simulator) = capture_inbound_call_response(
      protocol,
      "UnlockConnector",
      json!({
        "evseId": 1,
        "connectorId": connector_id,
      }),
    )
    .await;

    let OcppFrame::CallResult { payload, .. } = frame else {
      panic!("expected CALLRESULT frame");
    };
    assert_eq!(
      payload["status"],
      json!(ResponseStatus::UnknownConnector.as_str())
    );
    assert!(simulator.queue.is_empty());
  }
}

#[tokio::test]
async fn unsupported_actions_return_protocol_call_errors() {
  let cases = [
    (OcppVersion::V2_0_1, "GetBaseReport", "NotSupported"),
    (OcppVersion::V2_1, "BatterySwap", "NotSupported"),
    (OcppVersion::V1_6, "TotallyUnknown", "NotImplemented"),
    (OcppVersion::V2_0_1, "TotallyUnknown", "NotImplemented"),
  ];

  for (protocol, action, expected_code) in cases {
    let (frame, _) =
      capture_inbound_call_response(protocol, action, json!({})).await;
    let OcppFrame::CallError { code, .. } = frame else {
      panic!("expected CALLERROR frame for {action}");
    };
    assert_eq!(code, expected_code);
  }
}

#[tokio::test]
async fn v1_6_update_firmware_returns_not_supported_for_whitepaper() {
  let (frame, simulator) = capture_inbound_call_response(
    OcppVersion::V1_6,
    "UpdateFirmware",
    json!({
      "location": "https://csms.example/firmware.bin",
      "retrieveDate": now_timestamp()
    }),
  )
  .await;

  let OcppFrame::CallError {
    code, description, ..
  } = frame
  else {
    panic!("expected CALLERROR frame");
  };
  assert_eq!(code, "NotSupported");
  assert!(description.contains("SignedUpdateFirmware"));
  assert!(simulator.queue.is_empty());
}

#[tokio::test]
async fn strict_mode_rejects_schema_invalid_v1_6_requests() {
  let (frame, simulator) = capture_inbound_call_response_with_strict(
    OcppVersion::V1_6,
    true,
    "DataTransfer",
    json!({
      "vendorId": "ocppsim",
      "data": { "not": "a string" }
    }),
  )
  .await;

  let OcppFrame::CallError { code, .. } = frame else {
    panic!("expected CALLERROR frame");
  };
  assert_eq!(code, "FormationViolation");
  assert!(simulator.queue.is_empty());
}

#[tokio::test]
async fn strict_mode_rejects_oversized_install_certificate_v1_6() {
  let (frame, simulator) = capture_inbound_call_response_with_strict(
    OcppVersion::V1_6,
    true,
    "InstallCertificate",
    json!({
      "certificateType": "CentralSystemRootCertificate",
      "certificate": "A".repeat(5_501)
    }),
  )
  .await;

  let OcppFrame::CallError { code, .. } = frame else {
    panic!("expected CALLERROR frame");
  };
  assert_eq!(code, "FormationViolation");
  assert!(simulator.security.certificates.is_empty());
}

#[tokio::test]
async fn strict_mode_rejects_schema_invalid_v2_x_requests() {
  for protocol in v2_x_protocols() {
    let (frame, simulator) = capture_inbound_call_response_with_strict(
      protocol,
      true,
      "Reset",
      json!({}),
    )
    .await;

    let OcppFrame::CallError { code, .. } = frame else {
      panic!("expected CALLERROR frame");
    };
    assert_eq!(code, "FormationViolation");
    assert!(simulator.queue.is_empty());
  }
}

#[tokio::test]
async fn strict_mode_warns_when_request_schema_is_missing() {
  let (frame, simulator, events) = capture_inbound_call_response_with_events(
    OcppVersion::V1_6,
    true,
    "TotallyUnknown",
    json!({}),
  )
  .await;

  let OcppFrame::CallError { code, .. } = frame else {
    panic!("expected CALLERROR frame");
  };
  assert_eq!(code, "NotImplemented");
  assert!(simulator.queue.is_empty());
  assert!(events.iter().any(|event| {
    matches!(
      event,
      UiEvent::Log {
        level: UiLogLevel::Warn,
        message,
      } if message.contains("Strict schema coverage is missing")
        && message.contains("TotallyUnknown")
    )
  }));
}

#[tokio::test]
async fn non_strict_mode_keeps_pragmatic_v2_x_request_handling() {
  for protocol in v2_x_protocols() {
    let (frame, _) =
      capture_inbound_call_response(protocol, "Reset", json!({})).await;

    let OcppFrame::CallResult { payload, .. } = frame else {
      panic!("expected CALLRESULT frame");
    };
    assert_eq!(payload["status"], json!(ResponseStatus::Accepted.as_str()));
  }
}

#[tokio::test]
async fn trigger_message_v1_6_enqueues_requested_simulator_calls() {
  let (v1_6_frame, v1_6_simulator) = capture_inbound_call_response(
    OcppVersion::V1_6,
    "TriggerMessage",
    json!({ "requestedMessage": "Heartbeat" }),
  )
  .await;
  let OcppFrame::CallResult { payload, .. } = v1_6_frame else {
    panic!("expected CALLRESULT frame");
  };
  assert_eq!(payload["status"], json!(ResponseStatus::Accepted.as_str()));
  assert!(
    v1_6_simulator
      .queue
      .iter()
      .any(|call| call.action == "Heartbeat")
  );
}

#[tokio::test]
async fn trigger_message_v1_6_standard_and_extended_are_separate() {
  let (frame, simulator) = capture_inbound_call_response(
    OcppVersion::V1_6,
    "TriggerMessage",
    json!({ "requestedMessage": "DiagnosticsStatusNotification" }),
  )
  .await;
  let OcppFrame::CallResult { payload, .. } = frame else {
    panic!("expected CALLRESULT frame");
  };
  assert_eq!(payload["status"], json!(ResponseStatus::Accepted.as_str()));
  assert!(
    simulator
      .queue
      .iter()
      .any(|call| { call.action == "DiagnosticsStatusNotification" })
  );

  let (frame, simulator) = capture_inbound_call_response(
    OcppVersion::V1_6,
    "TriggerMessage",
    json!({ "requestedMessage": "LogStatusNotification" }),
  )
  .await;
  let OcppFrame::CallResult { payload, .. } = frame else {
    panic!("expected CALLRESULT frame");
  };
  assert_eq!(
    payload["status"],
    json!(ResponseStatus::NotImplemented.as_str())
  );
  assert!(simulator.queue.is_empty());

  let (frame, simulator) = capture_inbound_call_response(
    OcppVersion::V1_6,
    "ExtendedTriggerMessage",
    json!({ "requestedMessage": "LogStatusNotification" }),
  )
  .await;
  let OcppFrame::CallResult { payload, .. } = frame else {
    panic!("expected CALLRESULT frame");
  };
  assert_eq!(payload["status"], json!(ResponseStatus::Accepted.as_str()));
  assert!(
    simulator
      .queue
      .iter()
      .any(|call| { call.action == "LogStatusNotification" })
  );
}

#[tokio::test]
async fn trigger_message_v2_x_enqueues_requested_simulator_calls() {
  for protocol in v2_x_protocols() {
    let (frame, simulator) = capture_inbound_call_response(
      protocol,
      "TriggerMessage",
      json!({
        "requestedMessage": "StatusNotification",
        "evse": { "id": 2 }
      }),
    )
    .await;
    let OcppFrame::CallResult { payload, .. } = frame else {
      panic!("expected CALLRESULT frame");
    };
    assert_eq!(payload["status"], json!(ResponseStatus::Accepted.as_str()));
    let status_payload = queued_payload(&simulator, "StatusNotification");
    assert_eq!(status_payload["evseId"], json!(2));
  }
}

#[tokio::test]
async fn trigger_message_v2_x_uses_version_specific_values() {
  let (frame, simulator) = capture_inbound_call_response(
    OcppVersion::V2_0_1,
    "TriggerMessage",
    json!({ "requestedMessage": "SignV2G20Certificate" }),
  )
  .await;
  let OcppFrame::CallResult { payload, .. } = frame else {
    panic!("expected CALLRESULT frame");
  };
  assert_eq!(
    payload["status"],
    json!(ResponseStatus::NotImplemented.as_str())
  );
  assert!(simulator.queue.is_empty());

  let (frame, simulator) = capture_inbound_call_response(
    OcppVersion::V2_1,
    "TriggerMessage",
    json!({ "requestedMessage": "SignV2G20Certificate" }),
  )
  .await;
  let OcppFrame::CallResult { payload, .. } = frame else {
    panic!("expected CALLRESULT frame");
  };
  assert_eq!(payload["status"], json!(ResponseStatus::Accepted.as_str()));
  let sign_payload = queued_payload(&simulator, "SignCertificate");
  assert_eq!(sign_payload["certificateType"], "V2G20Certificate");
}

#[test]
fn trigger_message_v2_x_can_trigger_active_transaction_event() {
  for protocol in v2_x_protocols() {
    let mut simulator = simulator_for_tests_with_protocol(protocol);
    simulator
      .start_transaction(1, "TOKEN".to_string(), false, None, true)
      .expect("start transaction");
    simulator.queue.clear();

    let status = simulator
      .trigger_message_v2_x(
        crate::ocpp::TriggerMessage_V2_X::TransactionEvent,
        Some(1),
      )
      .expect("trigger transaction event");
    assert_eq!(status, ResponseStatus::Accepted);
    let payload = queued_payload(&simulator, "TransactionEvent");
    assert_eq!(payload["triggerReason"], "Trigger");
    assert_eq!(payload["eventType"], "Updated");
  }
}

#[tokio::test]
async fn mock_csms_remote_start_meter_and_stop_v2_x_lifecycle() {
  for protocol in v2_x_protocols() {
    let listener = TcpListener::bind("127.0.0.1:0")
      .await
      .expect("bind listener");
    let address = listener.local_addr().expect("local address");
    let server = tokio::spawn(async move {
      let (stream, _) = listener.accept().await.expect("accept client");
      let websocket = accept_async(stream).await.expect("accept websocket");
      let (_server_write, mut server_read) = websocket.split();
      let mut frames = Vec::new();
      for _ in 0..2 {
        let message = server_read
          .next()
          .await
          .expect("response frame")
          .expect("response frame ok");
        frames.push(
          parse_frame(message.to_text().expect("text frame"))
            .expect("parse response"),
        );
      }
      frames
    });

    let (client_stream, _) = connect_async(format!("ws://{address}"))
      .await
      .expect("connect client");
    let (mut write, _read) = client_stream.split();
    let mut simulator = simulator_for_tests_with_protocol(protocol);
    simulator
      .handle_incoming_call_v2_x(
        &mut write,
        "remote-start",
        "RequestStartTransaction",
        json!({
          "remoteStartId": 99,
          "idToken": { "idToken": "TOKEN" },
          "evseId": 1
        }),
      )
      .await
      .expect("handle request start");
    simulator
      .send_meter(1, true)
      .expect("meter should enqueue update");
    let transaction_id = simulator
      .active_transaction_uid(1)
      .expect("active transaction uid");
    simulator
      .handle_incoming_call_v2_x(
        &mut write,
        "remote-stop",
        "RequestStopTransaction",
        json!({ "transactionId": transaction_id }),
      )
      .await
      .expect("handle request stop");
    drop(write);

    let frames = server.await.expect("server task");
    assert_eq!(frames.len(), 2);
    assert!(frames.iter().all(|frame| {
      matches!(
        frame,
        OcppFrame::CallResult {
          payload,
          ..
        } if payload.get("status").and_then(Value::as_str)
          == Some(ResponseStatus::Accepted.as_str())
      )
    }));
    assert!(
      simulator
        .queue
        .iter()
        .filter(|call| call.action == "TransactionEvent")
        .count()
        >= 3
    );
    assert_eq!(
      simulator
        .connectors
        .get(&1)
        .map(|item| item.status.display()),
      Some("Finishing")
    );
  }
}

#[tokio::test]
async fn malformed_ws_text_returns_protocol_error() {
  let listener = TcpListener::bind("127.0.0.1:0")
    .await
    .expect("bind listener");
  let address = listener.local_addr().expect("local address");
  let server = tokio::spawn(async move {
    let (stream, _) = listener.accept().await.expect("accept client");
    let websocket = accept_async(stream).await.expect("accept websocket");
    let (_server_write, mut server_read) = websocket.split();
    let message = server_read
      .next()
      .await
      .expect("response frame")
      .expect("response frame ok");
    parse_frame(message.to_text().expect("text frame")).expect("parse response")
  });

  let (client_stream, _) = connect_async(format!("ws://{address}"))
    .await
    .expect("connect client");
  let (mut write, _read) = client_stream.split();
  let mut simulator = simulator_for_tests();
  simulator
    .handle_ws_text("not-json".to_string(), &mut write)
    .await
    .expect("handle malformed text");
  drop(write);

  let frame = server.await.expect("server task");
  let OcppFrame::CallError { code, .. } = frame else {
    panic!("expected CALLERROR frame");
  };
  assert_eq!(code, "ProtocolError");
}