//! A request this crate's client cannot read is not dropped in silence.
//!
//! `draft-ietf-moq-transport-22` §9.20: "An endpoint that receives an unknown Message
//! Parameter MUST close the session with PROTOCOL_VIOLATION". Found by a run against another
//! implementation's relay, whose SUBSCRIBE to this client's publisher carried a parameter
//! this codec did not read: the client dropped the request stream, and the relay could only
//! report a request that ended with no response.

use joule_moq::draft22::{ControlMessage, Parameters, SessionError, TrackRequest};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::transport::mux::{memory_pair, MuxConfig};
use joule_moq_relay::transport::{SharedConnection, TransportError};
use std::sync::Arc;
use std::time::Duration;
use tokio::time::timeout;

#[tokio::test]
async fn an_unknown_message_parameter_in_a_request_closes_the_session() {
    let (a, b, _) = memory_pair(MuxConfig::default(), false);
    let (a, b): (SharedConnection, SharedConnection) = (Arc::new(a), Arc::new(b));
    let (ca, cb) = tokio::join!(Client::connect(a.clone(), ClientOptions::default()), Client::connect(b.clone(), ClientOptions::default()));
    let (_ca, _cb) = (ca.expect("one end's SETUP"), cb.expect("the other's"));
    // A SUBSCRIBE with no parameter, and then one of a type the draft does not define put
    // where its empty list was: the count 1, the type 0x3e (even, so a varint follows), 0.
    let m = ControlMessage::Subscribe(TrackRequest { request_id: 1, namespace: TrackNamespace::from_strs(["a"]), name: TrackName::from_str("t"), params: Parameters::new() });
    let mut bytes = m.encode().expect("a SUBSCRIBE");
    assert_eq!(bytes.pop(), Some(0), "the empty parameter list is the message's last byte");
    bytes.extend_from_slice(&[1, 0x3e, 0]);
    let len = u16::from_be_bytes([bytes[1], bytes[2]]) + 2;
    bytes[1..3].copy_from_slice(&len.to_be_bytes());
    let (mut tx, _rx) = b.open_bi().await.expect("a request stream");
    tx.write_all(&bytes).await.expect("written");
    let why = timeout(Duration::from_secs(5), b.closed()).await.expect("the session is closed, not left open");
    assert!(matches!(why, TransportError::Closed { code, by_peer: true, .. } if code == SessionError::PROTOCOL_VIOLATION.0), "{why:?}");
}