open-standards/joule-code/joule-code-rs/crates/joule-moq-web/src/tests.rs

//! The surface's pure-Rust half, on the host: what a page sends decodes as the codec reads
//! it, and what a publisher signs a subscriber verifies, and refuses when it is changed.

use super::*;
use joule_moq::carriage::{from_subgroup_object, read_object_properties};
use joule_moq::draft22::decode_uni_stream;
use joule_moq::draft22::kvp::KvValue;

fn parse(b: &[u8]) -> ControlMessage {
    ControlMessage::decode_exact(b).expect("the bytes are one control message")
}

#[test]
fn setup_carries_the_token_and_the_receipt_extension() {
    let b = setup_bytes(Some(b"grant-bytes"), "joule-moq-web/test", true, 4).unwrap();
    let ControlMessage::Setup(s) = parse(&b) else { panic!("not SETUP") };
    assert_eq!(s.options.implementation(), Some(&b"joule-moq-web/test"[..]));
    assert_eq!(s.options.max_request_updates(), 4);
    assert_eq!(s.options.authorization_tokens(), vec![Token::UseValue { token_type: 0, value: b"grant-bytes".to_vec() }]);
    let ext = s.options.0.iter().find(|p| p.ty == SETUP_OPTION_JWP_RECEIPTS).expect("the receipt extension");
    assert_eq!(ext.value, KvValue::Varint(JWP_MOQ_EXTENSION_VERSION as u64));
    // A client's SETUP over WebTransport: no PATH, no AUTHORITY, a token by value.
    s.check_from(joule_moq::draft22::Perspective::Client, true).unwrap();
}

#[test]
fn requests_encode_as_the_codec_reads_them() {
    let b = control_bytes(
        r#"{"message":"SUBSCRIBE","request_id":2,"namespace":["room","alice","video"],"name":"0",
            "params":{"subscriber_priority":10,"group_order":1,"forward":true,"location_filter":{"type":"next"},
                      "fill":{"location_filter":{"type":"relative","start_group":3}},"authorization_token":"6869"}}"#,
    )
    .unwrap();
    let ControlMessage::Subscribe(r) = parse(&b) else { panic!("not SUBSCRIBE") };
    assert_eq!(r.request_id, 2);
    assert_eq!(r.namespace, TrackNamespace::from_strs(["room", "alice", "video"]));
    assert_eq!(r.name, TrackName::from_str("0"));
    let want = Parameters::new()
        .with(Parameter::SubscriberPriority(10))
        .with(Parameter::GroupOrder(GroupOrder::Ascending))
        .with(Parameter::Forward(true))
        .with(Parameter::LocationFilter(LocationFilter::NextObject))
        .with(Parameter::FillParameters(Parameters::new().with(Parameter::LocationFilter(LocationFilter::RelativeStart { start_group: 3 }))))
        .with(Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: b"hi".to_vec() }));
    // serde_json's map is ordered by key, so the parameters come in key order.
    let mut got = r.params.0.clone();
    let mut want = want.0;
    got.sort_by_key(Parameter::type_code);
    want.sort_by_key(Parameter::type_code);
    assert_eq!(got, want);

    let b = control_bytes(r#"{"message":"PUBLISH","request_id":0,"namespace":["r","p","audio"],"name":"0","track_alias":7,"track_properties":{"publisher_priority":32}}"#).unwrap();
    let ControlMessage::Publish(p) = parse(&b) else { panic!("not PUBLISH") };
    assert_eq!((p.request_id, p.track_alias), (0, 7));
    assert_eq!(p.track_properties.find(prop::DEFAULT_PUBLISHER_PRIORITY).unwrap().and_then(|x| x.as_varint()), Some(32));

    let b = control_bytes(r#"{"message":"FETCH","request_id":4,"namespace":["r","p","video"],"name":"0","params":{"location_filter":{"type":"range","start":[0,0],"end_group_delta":9,"end_object":0},"group_order":1}}"#).unwrap();
    let ControlMessage::Fetch(f) = parse(&b) else { panic!("not FETCH") };
    assert!(f.params.0.contains(&Parameter::LocationFilter(LocationFilter::AbsoluteRange { start: Location::new(0, 0), end_group_delta: 9, end_object: 0 })));

    for desc in [
        r#"{"message":"PUBLISH_NAMESPACE","request_id":6,"prefix":["r","p"]}"#,
        r#"{"message":"SUBSCRIBE_NAMESPACE","request_id":8,"prefix":["r"]}"#,
        r#"{"message":"REQUEST_OK"}"#,
        r#"{"message":"SUBSCRIBE_OK","track_alias":3}"#,
        r#"{"message":"REQUEST_UPDATE","request_id":2,"params":{"subscriber_priority":1}}"#,
        r#"{"message":"PUBLISH_DONE","status":0,"stream_count":4,"reason":"done"}"#,
    ] {
        let b = control_bytes(desc).unwrap();
        let m = parse(&b);
        let v: Value = serde_json::from_str(desc).unwrap();
        assert_eq!(json::control(&m)["message"], v["message"], "{desc}");
    }
    assert!(control_bytes(r#"{"message":"SETUP"}"#).is_err(), "SETUP has its own function");
    assert!(control_bytes(r#"{"message":"SUBSCRIBE","request_id":0,"namespace":["r"],"name":"0","params":{"subscriber_priority":300}}"#).is_err());
}

#[test]
fn the_message_stream_waits_for_whole_messages() {
    let a = control_bytes(r#"{"message":"REQUEST_OK"}"#).unwrap();
    let b = control_bytes(r#"{"message":"SUBSCRIBE_OK","track_alias":9}"#).unwrap();
    let all: Vec<u8> = a.iter().chain(&b).copied().collect();
    let mut s = MessageStream::default();
    let mut got = Vec::new();
    for byte in &all {
        s.buf.push(*byte);
        while let Some(m) = s.next_message().unwrap() {
            got.push(m);
        }
    }
    assert_eq!(got.len(), 2);
    assert!(matches!(got[1], ControlMessage::SubscribeOk(SubscribeOk { track_alias: 9, .. })));
    assert!(s.buf.is_empty());
}

fn signer(seed: u8) -> Signer {
    Signer::new([seed; 32], 8, 20, 21, 64, "page".into(), Some(500))
}

#[test]
fn a_signed_group_is_one_receipts_object_and_one_media_object() {
    let mut s = signer(1);
    let g = s.sign(3, b"WAI1-envelope", "2026-10-18T00:00:00Z", "2026-10-18T00:00:00Z").unwrap();
    // The receipts stream: group 3 of the receipts track, one bundle.
    let (rh, bundle) = ReceiptBundle::decode_stream(&g.receipts_stream).unwrap();
    assert_eq!((rh.track_alias, rh.group_id, rh.publisher_priority), (21, 3, Some(64)));
    assert_eq!(bundle.group.group_id, GroupId(3));
    assert_eq!(bundle.group.track_alias, TrackAlias(8), "the receipts sign the wai_receipt_alias");
    assert_eq!(bundle.group.parent_receipt_hash, None, "the first group starts the chain");
    // The media stream: one object, its WAI_RECEIPT naming the group receipt.
    let ev = decode_uni_stream(&g.media_stream, GroupOrder::Ascending).unwrap();
    let [UniEvent::Header(StreamHeader::Subgroup(sh)), UniEvent::SubgroupObject(o)] = ev.as_slice() else { panic!("{ev:?}") };
    assert_eq!((sh.track_alias, sh.group_id), (20, 3));
    assert_eq!(o.payload, b"WAI1-envelope");
    let (h, _, extras) = from_subgroup_object(sh, o).unwrap();
    assert!(h.wai_envelope);
    assert_eq!(extras.delivery_timeout_ms, Some(500));
    let w = extras.wai_receipt.expect("the WAI_RECEIPT property");
    assert!(w.agrees_with(&bundle.objects[0], &bundle.group));
    assert_eq!(w.joules_micro, 0, "the unmetered marker");
    let mut hh = h;
    read_object_properties(&o.properties, &mut hh).unwrap();
    // The next group's parent is this one's link.
    let g2 = s.sign(4, b"next", "t", "t").unwrap();
    let (_, b2) = ReceiptBundle::decode_stream(&g2.receipts_stream).unwrap();
    assert_eq!(b2.group.parent_receipt_hash, Some(g.link));
}

fn bundle_payload(stream: &[u8]) -> Vec<u8> {
    let ev = decode_uni_stream(stream, GroupOrder::Ascending).unwrap();
    let UniEvent::SubgroupObject(o) = &ev[1] else { panic!() };
    o.payload.clone()
}

fn media_payload(stream: &[u8]) -> Vec<u8> {
    let ev = decode_uni_stream(stream, GroupOrder::Ascending).unwrap();
    let UniEvent::SubgroupObject(o) = &ev[1] else { panic!() };
    o.payload.clone()
}

#[test]
fn a_subscriber_verifies_every_object_in_any_order_and_refuses_a_change() {
    let mut s = signer(2);
    let groups: Vec<Signed> = (0..6).map(|g| s.sign(g, format!("frame {g}").as_bytes(), "t0", "t1").unwrap()).collect();
    let mut v = Verifier::new(s.public_key(), 8).unwrap();
    // Receipts out of order: 2 anchors the joined chain, 1 and then 0 fill it back as
    // history, the rest link on; 5 before 4 is held until 4 arrives.
    for g in [2u64, 1, 0, 3, 5, 4] {
        let r = v.receipts(g, &bundle_payload(&groups[g as usize].receipts_stream), 0);
        assert_eq!(r["ok"], true, "group {g}: {r}");
    }
    let st = v.stats();
    assert_eq!(st["admissions"]["anchor"], 1, "{st}");
    assert_eq!(st["chain_held"], 0, "nothing is left held: {st}");
    for (g, sg) in groups.iter().enumerate() {
        let payload = media_payload(&sg.media_stream);
        let r = v.object(g as u64, 0, &payload);
        assert_eq!(r["ok"], true);
        // What a frame decoded from it names (WAI browser-sink §7): the object's content hash,
        // and the link hash of the group receipt it verified under.
        assert_eq!(r["content_hash"], hex::encode(ObjectReceipt::hash_payload(&payload)), "group {g}");
        let bundle = ReceiptBundle::from_payload(&bundle_payload(&sg.receipts_stream)).unwrap();
        assert_eq!(r["group_receipt"], hex::encode(bundle.group.link_hash()), "group {g}");
        assert_eq!(v.inclusion(g as u64, 0)["ok"], true);
    }
    // A changed payload, a receipt for another group, another publisher's receipts.
    assert_eq!(v.object(1, 0, b"frame 9")["reason"], "payload_mismatch");
    assert_eq!(v.receipts(7, &bundle_payload(&groups[3].receipts_stream), 0)["reason"], "moqt_mismatch");
    let mut other = signer(3);
    let og = other.sign(6, b"x", "t", "t").unwrap();
    assert_eq!(v.receipts(6, &bundle_payload(&og.receipts_stream), 0)["reason"], "unverified");
    assert_eq!(v.object(9, 0, b"x")["reason"], "no_receipt");
    // A receipt whose wai_receipt_alias is not the catalog's.
    let mut wrong = Signer::new([2; 32], 9, 20, 21, 64, "page".into(), None);
    let wg = wrong.sign(8, b"x", "t", "t").unwrap();
    assert_eq!(v.receipts(8, &bundle_payload(&wg.receipts_stream), 0)["reason"], "moqt_mismatch");
}

#[test]
fn a_subgroup_writer_writes_a_stream_the_decoder_reads_incrementally() {
    let mut w = SubgroupWriter::new(5, 0, 200, false);
    let mut b = w.header();
    for i in 0..4u64 {
        let mut out = Vec::new();
        w.codec.encode_object(&SubgroupObject::normal(i, format!("event {i}").into_bytes()), &mut out).unwrap();
        b.extend(out);
    }
    let mut d = UniStreamDecoder::new(GroupOrder::Ascending, Limits::default());
    let mut objects = 0;
    for chunk in b.chunks(3) {
        d.push(chunk);
        while let Some(e) = d.next().unwrap() {
            if let UniEvent::SubgroupObject(o) = e {
                assert_eq!(o.payload, format!("event {objects}").into_bytes());
                objects += 1;
            }
        }
    }
    d.finish().unwrap();
    assert_eq!(objects, 4);
    assert!(matches!(header_of(&b), Some(StreamHeader::Subgroup(h)) if h.track_alias == 5 && h.publisher_priority == Some(200)));
}

#[test]
fn a_grant_rides_setup_with_its_proof() {
    let key = SigningKey::from_bytes(&[0x51; 32]);
    let mut record = b"WRG1\x01".to_vec();
    record.extend_from_slice(&[3; 32]);
    record.extend_from_slice(&key.verifying_key().to_bytes());
    record.extend_from_slice(&[0; 150]);
    let challenge = [0x2c; 32];
    let target = "//127.0.0.1:4433/moq";
    let b = setup_with_grant_bytes(&record, [0x51; 32], challenge, target, "joule-moq-web/test", true, 0).unwrap();
    let ControlMessage::Setup(s) = parse(&b) else { panic!("not SETUP") };
    assert_eq!(s.options.authorization_tokens(), vec![Token::UseValue { token_type: 0, value: record.clone() }]);
    let proofs = grant_proof::proofs_of(&s.options);
    assert_eq!(proofs.len(), 1);
    let (digest, principal) = grant_proof::grant_digest_and_principal(&record).unwrap();
    let p = grant_proof::check_proof(&digest, &principal, Some(proofs[0]), &challenge, |_| false, &format!("SETUP {target}")).expect("the relay's check passes");
    assert_eq!(p.to_bytes().len(), 4 + 1 + 32 * 3 + 2 + "SETUP ".len() + target.len() + 64);
    assert!(setup_with_grant_bytes(&record, [0x52; 32], challenge, target, "t", true, 0).is_err(), "only the principal's seed signs");
    // Over HTTP (§10.5.3), the proof a room service's co-performance request carries.
    let http = crate::http_proof_bytes(&record, [0x51; 32], challenge, "POST", "//room.example/room-api/co/schedule").unwrap();
    grant_proof::check_proof(&digest, &principal, Some(&http), &challenge, |_| false, "POST //room.example/room-api/co/schedule").expect("the service's check passes");
    assert_eq!(grant_proof::check_proof(&digest, &principal, Some(&http), &challenge, |_| false, "POST //room.example/co/schedule"), Err(grant_proof::ProofRefusal::WrongContext));
    assert!(crate::http_proof_bytes(&record, [0x52; 32], challenge, "POST", "//x/y").is_err(), "only the principal's seed signs");
}

/// The size of one receipts-track object for a group of one object, as the browser's room
/// signs them (publisher and origin a participant's name, RFC 3339 times with milliseconds):
/// the binary form, and the same receipts in this crate's JSON serialisation.
#[test]
fn the_receipts_object_of_a_live_group() {
    let mut s = Signer::new([0x61; 32], 1, 10, 11, 64, "alice".into(), Some(100));
    let first = s.sign(0, b"WAI1 a live object", "2026-10-18T12:00:00.000Z", "2026-10-18T12:00:00.000Z").unwrap();
    let second = s.sign(1, b"WAI1 a live object", "2026-10-18T12:00:00.020Z", "2026-10-18T12:00:00.020Z").unwrap();
    for g in [&first, &second] {
        let payload = bundle_payload(&g.receipts_stream);
        let bundle = ReceiptBundle::from_payload(&payload).unwrap();
        let json = json_form_len(&bundle);
        println!("receipts object: {} bytes in the binary form, {json} bytes as JSON", payload.len());
        assert!((payload.len() as u64) < json);
    }
}

/// A page's own rating claim (WAI age-assurance §3.1, §3.6): about its track by its digest,
/// signed by its key, carried as `WAI_RATING` Track Properties of a PUBLISH and read back from
/// a SUBSCRIBE_OK's properties.
#[test]
fn a_pages_rating_claim_rides_its_publish() {
    let seed = [0x2b; 32];
    let me = hex::encode(SigningKey::from_bytes(&seed).verifying_key().to_bytes());
    let room = "ab".repeat(32);
    let fields = [room.as_str(), me.as_str(), "video"];
    let text = rating_claim_text(seed, &fields, "0", 5, "https://rating-a.example/scheme", "18", 18, "2026-10-21T00:00:00Z").expect("a claim");
    let c = joule_moq::rating::read(text.as_bytes()).expect("it reads");
    let digest = joule_moq::rating::track_digest(&TrackNamespace::from_strs(fields), &TrackName::from_str("0"));
    assert_eq!((c.age(), c.signer.to_vec()), (18, hex::decode(&me).unwrap()));
    assert!(c.covers(&digest, None, 5) && !c.covers(&digest, None, 4));
    assert!(rating_claim_text(seed, &fields, "0", 0, "not a uri", "18", 18, "2026-10-21T00:00:00Z").is_err(), "a scheme that is not an absolute URI");
    // The PUBLISH carries it, and a SUBSCRIBE_OK's properties give it back.
    let desc = serde_json::json!({"message": "PUBLISH", "request_id": 0, "namespace": fields, "name": "0", "track_alias": 3, "track_properties": {"publisher_priority": 96, "ratings": [text]}});
    let b = control_bytes(&desc.to_string()).expect("a PUBLISH");
    let ControlMessage::Publish(p) = parse(&b) else { panic!("a PUBLISH") };
    assert_eq!(joule_moq::rating::claims_in(&p.track_properties), vec![text.clone().into_bytes()]);
    let props = joule_moq::draft22::json::properties(&p.track_properties).to_string();
    assert_eq!(ratings_of(&props).unwrap(), serde_json::to_string(&vec![text]).unwrap());
}