b247c5a701
All checks were successful
standards workspaces build + test / which workspaces this change touches (pull_request) Successful in 6s
repository checks / lockfiles, tests in CI, workflow hygiene, spec and section references, public text (pull_request) Successful in 46s
standards workspaces build + test / joule-code-rs (joule-code/joule-code-rs, joule-code-rs, linux-arm64, true) (pull_request) Successful in 1m27s
joule-code conformance gate / Build + round-trip C1-C4 vectors (pull_request) Successful in 2m20s
standards workspaces build + test / macOS energy backends (pull_request) Successful in 5m3s
Co-Authored-By: Claude Opus 5.5 <[email protected]>
768 lines
43 KiB
Rust
768 lines
43 KiB
Rust
//! The `draft-ietf-moq-transport-22` vector corpus (`tests/vectors/draft-22/`).
|
|
//!
|
|
//! Every vector is built here from typed values, deterministically: the bytes
|
|
//! are what this crate's encoder makes of the value (or, for a negative vector
|
|
//! and the few forms the encoder never makes, bytes written out by hand), and
|
|
//! the expected reading is the value itself in the JSON view of
|
|
//! `joule_moq::draft22::json`, or, for a negative vector, the refusal it must
|
|
//! meet and the draft's consequence. Expectations are stated here, never read
|
|
//! back from the decoder.
|
|
//!
|
|
//! This test then holds the crate to the corpus:
|
|
//!
|
|
//! - the committed `vectors.json` is exactly what this file generates (run with
|
|
//! `JOULE_MOQ_WRITE_VECTORS=1` to rewrite it);
|
|
//! - each good vector decodes to its stated reading, and, where it is the form
|
|
//! the encoder makes, re-encodes to its bytes;
|
|
//! - each negative vector is refused with its stated kind and consequence, by
|
|
//! the whole-stream decoder and, for streams, by the incremental one too.
|
|
//!
|
|
//! `vectors/draft-22/moqt22_check.py` decodes the same corpus independently.
|
|
|
|
use joule_moq::carriage::WaiReceiptProperty;
|
|
use joule_moq::draft22::control::*;
|
|
use joule_moq::draft22::data::{
|
|
encode_control_stream, encode_fetch_stream, encode_padding_stream, encode_subgroup_stream, FetchCodec, FetchObject, Limits, SubgroupCodec,
|
|
};
|
|
use joule_moq::draft22::kvp::{opt, prop, KeyValuePair, Properties, SetupOptions};
|
|
use joule_moq::draft22::params::{param, Range, RangeFilter};
|
|
use joule_moq::draft22::{
|
|
decode_uni_stream, json, varint, Datagram, DatagramContent, FetchEntry, FetchHeader, GroupOrder, Location, LocationFilter, ObjectDatagram,
|
|
ObjectStatus, Parameter, Parameters, PublishDoneCode, RangeEnd, RequestErrorCode, StreamHeader, SubgroupHeader, SubgroupId, SubgroupObject,
|
|
Token, UniEvent, UniStreamDecoder, WireError,
|
|
};
|
|
use joule_moq::track::{TrackName, TrackNamespace};
|
|
use serde_json::{json, Value};
|
|
|
|
const CORPUS: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/vectors/draft-22/vectors.json");
|
|
|
|
fn hx(b: &[u8]) -> String {
|
|
json::hex(b)
|
|
}
|
|
|
|
fn ns(fields: &[&str]) -> TrackNamespace {
|
|
TrackNamespace::from_strs(fields)
|
|
}
|
|
|
|
fn name(s: &str) -> TrackName {
|
|
TrackName::from_str(s)
|
|
}
|
|
|
|
fn kv(ty: u64, v: u64) -> KeyValuePair {
|
|
KeyValuePair::varint(ty, v)
|
|
}
|
|
|
|
fn kb(ty: u64, b: &[u8]) -> KeyValuePair {
|
|
KeyValuePair::bytes(ty, b.to_vec())
|
|
}
|
|
|
|
/// A property list in wire order (by type, stably), as a decoder reads it.
|
|
fn props(mut ps: Vec<KeyValuePair>) -> Properties {
|
|
ps.sort_by_key(|k| k.ty);
|
|
Properties(ps)
|
|
}
|
|
|
|
/// Properties with an Immutable Properties pair holding `inner`.
|
|
fn with_immutable(mut outer: Vec<KeyValuePair>, inner: Vec<KeyValuePair>) -> Properties {
|
|
let mut b = Vec::new();
|
|
joule_moq::draft22::kvp::encode_pairs(&inner, &mut b).unwrap();
|
|
outer.push(KeyValuePair::bytes(prop::IMMUTABLE_PROPERTIES, b));
|
|
let mut p = Properties(outer);
|
|
p.0.sort_by_key(|k| k.ty);
|
|
p
|
|
}
|
|
|
|
fn rf(ty: u64, set_id: u8, property_type: Option<u64>, ranges: &[(u64, Option<u64>)]) -> Parameter {
|
|
Parameter::RangeFilter {
|
|
ty,
|
|
filter: Some(RangeFilter { set_id, property_type, ranges: ranges.iter().map(|&(start, end)| Range { start, end }).collect() }),
|
|
}
|
|
}
|
|
|
|
/// Sort a parameter list as the wire orders it (by type, stably), so the
|
|
/// stated reading is the one a decoder gives.
|
|
fn sorted(ps: Vec<Parameter>) -> Parameters {
|
|
let mut v = ps;
|
|
v.sort_by_key(|p| p.type_code());
|
|
Parameters(v)
|
|
}
|
|
|
|
struct Corpus {
|
|
vectors: Vec<Value>,
|
|
}
|
|
|
|
impl Corpus {
|
|
fn push(&mut self, name: &str, kind: &str, bytes: &[u8], canonical: bool, context: Option<Value>, expect: Value) {
|
|
let mut v = json!({"name": name, "kind": kind, "hex": hx(bytes), "canonical": canonical, "expect": expect});
|
|
if let Some(c) = context {
|
|
v["context"] = c;
|
|
}
|
|
self.vectors.push(v);
|
|
}
|
|
|
|
fn control(&mut self, name: &str, m: ControlMessage) {
|
|
let b = m.encode().expect(name);
|
|
self.push(name, "control", &b, true, None, json::control(&m));
|
|
}
|
|
|
|
fn stream(&mut self, name: &str, bytes: Vec<u8>, order: GroupOrder, events: Vec<UniEvent>) {
|
|
let expect = Value::Array(events.iter().map(json::uni_event).collect());
|
|
self.push(name, "uni-stream", &bytes, true, Some(json!({"group_order": order as u8})), expect);
|
|
}
|
|
|
|
fn datagram(&mut self, name: &str, d: Datagram, bytes: Vec<u8>, canonical: bool) {
|
|
self.push(name, "datagram", &bytes, canonical, None, json::datagram(&d));
|
|
}
|
|
|
|
fn bad(&mut self, name: &str, kind: &str, bytes: &[u8], context: Option<Value>, error: &str, consequence: &str) {
|
|
self.push(name, kind, bytes, false, context, json!({"error": error, "consequence": consequence}));
|
|
}
|
|
}
|
|
|
|
const PV: &str = "session:PROTOCOL_VIOLATION";
|
|
|
|
/// Frame a control message body by hand: type, a 16-bit length, the body.
|
|
fn frame(ty: u64, body: &[u8]) -> Vec<u8> {
|
|
let mut out = varint::to_vec(ty);
|
|
out.extend_from_slice(&(body.len() as u16).to_be_bytes());
|
|
out.extend_from_slice(body);
|
|
out
|
|
}
|
|
|
|
fn v(x: u64) -> Vec<u8> {
|
|
varint::to_vec(x)
|
|
}
|
|
|
|
fn cat(parts: &[&[u8]]) -> Vec<u8> {
|
|
parts.concat()
|
|
}
|
|
|
|
fn wai_receipt(content: u8, joules: u64, class: &str, evidence: Vec<u8>, group: u8) -> WaiReceiptProperty {
|
|
WaiReceiptProperty { content_hash: [content; 32], joules_micro: joules, acquisition: class.into(), acquisition_evidence: evidence, group_receipt: [group; 32] }
|
|
}
|
|
|
|
fn wai_json(p: &WaiReceiptProperty) -> Value {
|
|
json!({
|
|
"content_hash": hx(&p.content_hash),
|
|
"joules_micro": p.joules_micro,
|
|
"acquisition": p.acquisition,
|
|
"acquisition_evidence": hx(&p.acquisition_evidence),
|
|
"group_receipt": hx(&p.group_receipt),
|
|
})
|
|
}
|
|
|
|
fn build() -> Vec<Value> {
|
|
let mut c = Corpus { vectors: Vec::new() };
|
|
|
|
// ── §8.1 the variable-length integer: Table 4 of the draft ──
|
|
for (h, val, shortest) in [
|
|
("25", 37u64, true),
|
|
("8025", 37, false),
|
|
("bbbd", 15_293, true),
|
|
("ed7f3e7d", 226_442_877, true),
|
|
("faa1a0e403d8", 2_893_212_287_960, true),
|
|
("fc8998abc66bc0", 151_288_809_941_952, true),
|
|
("fefa318fa8e3ca11", 70_423_237_261_249_041, true),
|
|
("ffffffffffffffffff", u64::MAX, true),
|
|
] {
|
|
let b: Vec<u8> = (0..h.len()).step_by(2).map(|i| u8::from_str_radix(&h[i..i + 2], 16).unwrap()).collect();
|
|
c.push(&format!("varint {h}"), "varint", &b, shortest, None, json!({"value": val, "length": b.len()}));
|
|
}
|
|
c.push("varint 0 in three bytes", "varint", &[0xc0, 0x00, 0x00], false, None, json!({"value": 0, "length": 3}));
|
|
c.bad("varint cut short", "varint", &[0xe0, 0x01], None, "truncated", PV);
|
|
|
|
// ── §9.1 SETUP ──
|
|
let mut client = SetupOptions::new();
|
|
client
|
|
.push(kb(opt::PATH, b"/live?room=1"))
|
|
.push(kb(opt::AUTHORIZATION_TOKEN, &Token::Register { alias: 1, token_type: 0, value: b"tok".to_vec() }.to_bytes()))
|
|
.push(kv(opt::MAX_AUTH_TOKEN_CACHE_SIZE, 1024))
|
|
.push(kb(opt::AUTHORITY, b"relay.example:443"))
|
|
.push(kv(opt::MAX_FILTER_RANGES, 4))
|
|
.push(kb(opt::MOQT_IMPLEMENTATION, b"example-impl/1.0"))
|
|
.push(kv(opt::MAX_REQUEST_UPDATES, 2))
|
|
.push(kb(0x9d, b"grease"))
|
|
.push(kv(0x3a46, 2));
|
|
let mut client_sorted = client.clone();
|
|
client_sorted.0.sort_by_key(|p| p.ty);
|
|
c.push("SETUP from a client", "control", &ControlMessage::Setup(Setup { options: client }).encode().unwrap(), true, None,
|
|
json::control(&ControlMessage::Setup(Setup { options: client_sorted })));
|
|
let mut server = SetupOptions::new();
|
|
server.push(kb(opt::MOQT_IMPLEMENTATION, b"example-impl/1.0"));
|
|
c.control("SETUP from a server", ControlMessage::Setup(Setup { options: server }));
|
|
c.control("SETUP with no options", ControlMessage::Setup(Setup::default()));
|
|
|
|
// ── §9.2 GOAWAY ──
|
|
c.control("GOAWAY with a new session URI", ControlMessage::Goaway(Goaway { new_session_uri: b"moqt://relay.example/next".to_vec(), timeout_ms: 5000 }));
|
|
c.control("GOAWAY with no URI and no timeout", ControlMessage::Goaway(Goaway::default()));
|
|
|
|
// ── §9.6 SUBSCRIBE, with every parameter it may carry ──
|
|
let every = sorted(vec![
|
|
Parameter::ObjectDeliveryTimeout(250),
|
|
Parameter::AuthorizationToken(Token::UseValue { token_type: 0, value: b"jwt".to_vec() }),
|
|
Parameter::RendezvousTimeout(1000),
|
|
Parameter::SubgroupDeliveryTimeout(500),
|
|
Parameter::Forward(true),
|
|
Parameter::SubscriberPriority(10),
|
|
Parameter::LocationFilter(LocationFilter::NextObject),
|
|
Parameter::GroupOrder(GroupOrder::Descending),
|
|
Parameter::FillParameters(sorted(vec![Parameter::LocationFilter(LocationFilter::RelativeStart { start_group: 1 }), Parameter::FillTimeout(100)])),
|
|
rf(param::SUBGROUP_FILTER, 0, None, &[(0, Some(2))]),
|
|
rf(param::OBJECTID_FILTER, 0, None, &[(3, Some(5)), (10, None)]),
|
|
rf(param::PRIORITY_FILTER, 1, None, &[(0, Some(127))]),
|
|
rf(param::OBJECT_PROPERTY_FILTER, 0, Some(0x3a40), &[(0, Some(2))]),
|
|
Parameter::NewGroupRequest(0),
|
|
Parameter::IncludeProperties(true),
|
|
]);
|
|
c.control("SUBSCRIBE with every parameter", ControlMessage::Subscribe(TrackRequest { request_id: 0, namespace: ns(&["example", "cam1"]), name: name("video"), params: every }));
|
|
c.control("SUBSCRIBE with no parameters, an empty namespace", ControlMessage::Subscribe(TrackRequest { request_id: 2, namespace: TrackNamespace(vec![]), name: name("t"), params: Parameters::new() }));
|
|
// A Request ID in a longer form than it needs: the reading is the same.
|
|
let sub = ControlMessage::Subscribe(TrackRequest { request_id: 5, namespace: ns(&["a"]), name: name("b"), params: Parameters::new() });
|
|
let mut body = Vec::new();
|
|
varint::put_with_len(&mut body, 5, 3).unwrap();
|
|
body.extend_from_slice(&cat(&[&v(1), &v(1), b"a", &v(1), b"b", &v(0)]));
|
|
c.push("SUBSCRIBE whose Request ID takes three bytes", "control", &frame(msg::SUBSCRIBE, &body), false, None, json::control(&sub));
|
|
|
|
// ── §9.7 SUBSCRIBE_OK ──
|
|
c.control(
|
|
"SUBSCRIBE_OK with track properties",
|
|
ControlMessage::SubscribeOk(SubscribeOk {
|
|
track_alias: 7,
|
|
params: sorted(vec![Parameter::Expires(30_000), Parameter::LargestObject(Location::new(5, 2))]),
|
|
track_properties: with_immutable(
|
|
vec![
|
|
kv(prop::OBJECT_DELIVERY_TIMEOUT, 200),
|
|
kv(prop::MAX_CACHE_DURATION, 10_000),
|
|
kv(prop::DEFAULT_PUBLISHER_PRIORITY, 64),
|
|
kv(prop::DEFAULT_PUBLISHER_GROUP_ORDER, 1),
|
|
kv(prop::DYNAMIC_GROUPS, 1),
|
|
kb(0x3a45, b"x"),
|
|
],
|
|
vec![kv(0x78, 5)],
|
|
),
|
|
}),
|
|
);
|
|
// ── §9.10 PUBLISH_STATE_NOTIFY ──
|
|
c.control(
|
|
"PUBLISH_STATE_NOTIFY",
|
|
ControlMessage::PublishStateNotify(sorted(vec![
|
|
Parameter::LargestObject(Location::new(9, 0)),
|
|
Parameter::Forward(false),
|
|
Parameter::LocationFilter(LocationFilter::AbsoluteStart { start: Location::new(9, 0) }),
|
|
])),
|
|
);
|
|
// ── §9.8 PUBLISH ──
|
|
c.control(
|
|
"PUBLISH",
|
|
ControlMessage::Publish(Publish {
|
|
request_id: 1,
|
|
namespace: ns(&["example", "cam1"]),
|
|
name: name("audio"),
|
|
track_alias: 3,
|
|
params: sorted(vec![
|
|
Parameter::Expires(60_000),
|
|
Parameter::LargestObject(Location::new(4, 9)),
|
|
Parameter::Forward(true),
|
|
Parameter::GroupOrder(GroupOrder::Ascending),
|
|
Parameter::LocationFilter(LocationFilter::AbsoluteStartGroupEnd { start: Location::new(4, 0), end_group_delta: 2 }),
|
|
]),
|
|
track_properties: props(vec![kv(prop::DYNAMIC_GROUPS, 0)]),
|
|
}),
|
|
);
|
|
// ── §9.9 PUBLISH_DONE ──
|
|
c.control("PUBLISH_DONE", ControlMessage::PublishDone(PublishDone { status: PublishDoneCode::TRACK_ENDED, stream_count: 3, reason: b"done".to_vec() }));
|
|
c.control("PUBLISH_DONE with a stream count of 2^64-1", ControlMessage::PublishDone(PublishDone { status: PublishDoneCode::TOO_FAR_BEHIND, stream_count: u64::MAX, reason: vec![] }));
|
|
// ── §9.11 FETCH, §9.12 FETCH_OK ──
|
|
c.control(
|
|
"FETCH",
|
|
ControlMessage::Fetch(TrackRequest {
|
|
request_id: 4,
|
|
namespace: ns(&["example", "cam1"]),
|
|
name: name("video"),
|
|
params: sorted(vec![
|
|
Parameter::AuthorizationToken(Token::UseAlias { alias: 1 }),
|
|
Parameter::FillTimeout(0),
|
|
Parameter::SubscriberPriority(200),
|
|
Parameter::LocationFilter(LocationFilter::AbsoluteRange { start: Location::new(1, 0), end_group_delta: 3, end_object: 7 }),
|
|
Parameter::GroupOrder(GroupOrder::Descending),
|
|
Parameter::IncludeProperties(false),
|
|
]),
|
|
}),
|
|
);
|
|
c.control(
|
|
"FETCH_OK at the end of the track",
|
|
ControlMessage::FetchOk(FetchOk { end_of_track: true, end_location: Location::new(4, 7), params: Parameters::new(), track_properties: props(vec![kv(prop::DEFAULT_PUBLISHER_PRIORITY, 128)]) }),
|
|
);
|
|
// ── §9.13 TRACK_STATUS ──
|
|
c.control(
|
|
"TRACK_STATUS",
|
|
ControlMessage::TrackStatus(TrackRequest {
|
|
request_id: 6,
|
|
namespace: ns(&["example"]),
|
|
name: name("cam1"),
|
|
params: sorted(vec![Parameter::AuthorizationToken(Token::Register { alias: 2, token_type: 0x9d, value: b"abc".to_vec() }), Parameter::IncludeProperties(true)]),
|
|
}),
|
|
);
|
|
// ── §9.14, §9.15, §9.18 namespace requests ──
|
|
c.control(
|
|
"PUBLISH_NAMESPACE",
|
|
ControlMessage::PublishNamespace(NamespaceRequest { request_id: 3, prefix: ns(&["example", "cam1"]), params: sorted(vec![Parameter::AuthorizationToken(Token::UseValue { token_type: 1, value: vec![0xde, 0xad] })]) }),
|
|
);
|
|
c.control("SUBSCRIBE_NAMESPACE", ControlMessage::SubscribeNamespace(NamespaceRequest { request_id: 8, prefix: ns(&["example"]), params: Parameters::new() }));
|
|
c.control(
|
|
"SUBSCRIBE_TRACKS for every track",
|
|
ControlMessage::SubscribeTracks(NamespaceRequest {
|
|
request_id: 10,
|
|
prefix: TrackNamespace(vec![]),
|
|
params: sorted(vec![
|
|
Parameter::Forward(false),
|
|
Parameter::GroupOrder(GroupOrder::Ascending),
|
|
rf(param::TRACK_PROPERTY_FILTER, 0, Some(prop::DYNAMIC_GROUPS), &[(1, Some(1))]),
|
|
Parameter::IncludeProperties(false),
|
|
]),
|
|
}),
|
|
);
|
|
// ── §9.16, §9.17, §9.19 ──
|
|
c.control("NAMESPACE", ControlMessage::Namespace(ns(&["42"])));
|
|
c.control("NAMESPACE_DONE", ControlMessage::NamespaceDone(ns(&["123"])));
|
|
c.control("PUBLISH_SKIPPED", ControlMessage::PublishSkipped(PublishSkipped { suffix: ns(&["cam2"]), name: name("video") }));
|
|
// ── §9.5 REQUEST_UPDATE ──
|
|
c.control(
|
|
"REQUEST_UPDATE for a subscription",
|
|
ControlMessage::RequestUpdate(RequestUpdate {
|
|
request_id: 12,
|
|
params: sorted(vec![
|
|
Parameter::Forward(false),
|
|
Parameter::SubscriberPriority(5),
|
|
Parameter::LocationFilter(LocationFilter::None),
|
|
Parameter::RangeFilter { ty: param::OBJECTID_FILTER, filter: None },
|
|
Parameter::NewGroupRequest(12),
|
|
]),
|
|
}),
|
|
);
|
|
c.control(
|
|
"REQUEST_UPDATE for a namespace subscription",
|
|
ControlMessage::RequestUpdate(RequestUpdate { request_id: 14, params: sorted(vec![Parameter::TrackNamespacePrefix(ns(&["example", "b"]))]) }),
|
|
);
|
|
// ── §9.3 REQUEST_OK, §9.4 REQUEST_ERROR ──
|
|
c.control(
|
|
"REQUEST_OK answering TRACK_STATUS",
|
|
ControlMessage::RequestOk(RequestOk { params: sorted(vec![Parameter::LargestObject(Location::new(3, 9))]), track_properties: props(vec![kv(prop::DEFAULT_PUBLISHER_GROUP_ORDER, 2)]) }),
|
|
);
|
|
c.control("REQUEST_OK answering PUBLISH", ControlMessage::RequestOk(RequestOk { params: sorted(vec![Parameter::Expires(1000)]), track_properties: Properties::new() }));
|
|
c.control(
|
|
"REQUEST_ERROR",
|
|
ControlMessage::RequestError(RequestError { code: RequestErrorCode::DOES_NOT_EXIST, retry_interval: 0, reason: b"no such track".to_vec(), redirect: None }),
|
|
);
|
|
c.control(
|
|
"REQUEST_ERROR with a redirect",
|
|
ControlMessage::RequestError(RequestError {
|
|
code: RequestErrorCode::REDIRECT,
|
|
retry_interval: 1,
|
|
reason: vec![],
|
|
redirect: Some(Redirect { connect_uri: b"moqt://other.example/live".to_vec(), namespace: ns(&["example"]), name: name("cam1") }),
|
|
}),
|
|
);
|
|
c.control(
|
|
"REQUEST_ERROR with a code this draft does not register",
|
|
ControlMessage::RequestError(RequestError { code: RequestErrorCode(0x9d), retry_interval: 30_001, reason: b"grease".to_vec(), redirect: None }),
|
|
);
|
|
|
|
// ── §6.3 a control stream: SETUP, then GOAWAY ──
|
|
let setup = Setup { options: { let mut o = SetupOptions::new(); o.push(kb(opt::MOQT_IMPLEMENTATION, b"example-impl/1.0")); o } };
|
|
let goaway = ControlMessage::Goaway(Goaway { new_session_uri: vec![], timeout_ms: 1000 });
|
|
c.stream(
|
|
"control stream: SETUP then GOAWAY",
|
|
encode_control_stream(&setup, std::slice::from_ref(&goaway)).unwrap(),
|
|
GroupOrder::Ascending,
|
|
vec![UniEvent::Header(StreamHeader::Control), UniEvent::Control(ControlMessage::Setup(setup.clone())), UniEvent::Control(goaway.clone())],
|
|
);
|
|
|
|
// ── §11.3 subgroup streams ──
|
|
let sg = |h: SubgroupHeader, objs: Vec<SubgroupObject>| -> (Vec<u8>, Vec<UniEvent>) {
|
|
let b = encode_subgroup_stream(&h, &objs).unwrap();
|
|
let mut ev = vec![UniEvent::Header(StreamHeader::Subgroup(h))];
|
|
ev.extend(objs.into_iter().map(UniEvent::SubgroupObject));
|
|
(b, ev)
|
|
};
|
|
let (b, ev) = sg(
|
|
SubgroupHeader { track_alias: 2, group_id: 0, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: false, end_of_group: true, first_object: true },
|
|
vec![
|
|
SubgroupObject::normal(0, b"key".to_vec()),
|
|
SubgroupObject::normal(1, b"delta".to_vec()),
|
|
SubgroupObject { object_id: 2, properties: Properties::new(), status: ObjectStatus::EndOfGroup, payload: vec![] },
|
|
],
|
|
);
|
|
c.stream("subgroup: Subgroup ID 0, a priority, End of Group", b, GroupOrder::Ascending, ev);
|
|
let wai = wai_receipt(0x11, 0, "", vec![], 0x22);
|
|
let (b, ev) = sg(
|
|
SubgroupHeader { track_alias: 9, group_id: 7, subgroup_id: SubgroupId::FirstObjectId, publisher_priority: None, properties: true, end_of_group: false, first_object: false },
|
|
vec![
|
|
SubgroupObject { object_id: 5, properties: props(vec![kv(prop::PRIOR_OBJECT_ID_GAP, 2), kv(prop::OBJECT_DELIVERY_TIMEOUT, 40)]), status: ObjectStatus::Normal, payload: b"five".to_vec() },
|
|
SubgroupObject { object_id: 6, properties: Properties::new(), status: ObjectStatus::Normal, payload: vec![] },
|
|
SubgroupObject {
|
|
object_id: 8,
|
|
properties: with_immutable(vec![], vec![kv(0x3a40, 1), kb(0x3a41, &wai.to_bytes()), kv(0x3a42, 3), kv(0x3a44, 1)]),
|
|
status: ObjectStatus::Normal,
|
|
payload: b"WAI1".to_vec(),
|
|
},
|
|
],
|
|
);
|
|
c.stream("subgroup: Subgroup ID from the first object, default priority, properties", b, GroupOrder::Ascending, ev);
|
|
let (b, ev) = sg(
|
|
SubgroupHeader { track_alias: 1, group_id: 300, subgroup_id: SubgroupId::Explicit(3), publisher_priority: Some(0), properties: false, end_of_group: false, first_object: true },
|
|
vec![SubgroupObject::normal(0, b"x".to_vec()), SubgroupObject { object_id: 1, properties: Properties::new(), status: ObjectStatus::EndOfTrack, payload: vec![] }],
|
|
);
|
|
c.stream("subgroup: an explicit Subgroup ID, End of Track", b, GroupOrder::Ascending, ev);
|
|
let h = SubgroupHeader { track_alias: 4, group_id: 1, subgroup_id: SubgroupId::Zero, publisher_priority: Some(9), properties: false, end_of_group: false, first_object: true };
|
|
let mut hb = Vec::new();
|
|
h.encode(&mut hb);
|
|
c.stream("subgroup: a header and no object", hb, GroupOrder::Ascending, vec![UniEvent::Header(StreamHeader::Subgroup(h))]);
|
|
|
|
// ── §11.4 fetch streams ──
|
|
let fo = |g: u64, o: u64, s: Option<u64>, p: u8, pr: Option<Properties>, pl: &[u8]| FetchEntry::Object(FetchObject { group_id: g, object_id: o, subgroup_id: s, publisher_priority: p, properties: pr, payload: pl.to_vec() });
|
|
for (order, label, gs) in [(GroupOrder::Ascending, "Ascending", [1u64, 4, 6, 9]), (GroupOrder::Descending, "Descending", [9, 6, 4, 1])] {
|
|
let entries = vec![
|
|
fo(gs[0], 0, Some(0), 7, None, b"a"),
|
|
fo(gs[0], 1, Some(0), 7, None, b"b"),
|
|
fo(gs[0], 5, Some(1), 9, Some(props(vec![kv(prop::PRIOR_OBJECT_ID_GAP, 3)])), b""),
|
|
FetchEntry::EndOfRange { kind: RangeEnd::NonExistent, location: Location::new(gs[1], 3) },
|
|
fo(gs[1], 4, None, 9, None, b"d"),
|
|
FetchEntry::EndOfRange { kind: RangeEnd::Unknown, location: Location::new(gs[2], 0) },
|
|
FetchEntry::EndOfRange { kind: RangeEnd::TimedOut, location: Location::new(gs[3], 2) },
|
|
fo(gs[3], 3, Some(7), 9, None, b"e"),
|
|
fo(gs[3], 4, Some(8), 2, None, b"f"),
|
|
];
|
|
let b = encode_fetch_stream(&FetchHeader { request_id: 4 }, order, &entries).unwrap();
|
|
let mut ev = vec![UniEvent::Header(StreamHeader::Fetch(FetchHeader { request_id: 4 }))];
|
|
ev.extend(entries.into_iter().map(UniEvent::FetchEntry));
|
|
c.stream(&format!("fetch in {label} order, with each End of Range"), b, order, ev);
|
|
}
|
|
let first_eor = vec![FetchEntry::EndOfRange { kind: RangeEnd::Unknown, location: Location::new(2, 5) }, fo(2, 6, Some(0), 1, None, b"z")];
|
|
let b = encode_fetch_stream(&FetchHeader { request_id: 2 }, GroupOrder::Ascending, &first_eor).unwrap();
|
|
let mut ev = vec![UniEvent::Header(StreamHeader::Fetch(FetchHeader { request_id: 2 }))];
|
|
ev.extend(first_eor.into_iter().map(UniEvent::FetchEntry));
|
|
c.stream("fetch beginning with an End of Range", b, GroupOrder::Ascending, ev);
|
|
let mut fb = Vec::new();
|
|
FetchHeader { request_id: 10 }.encode(&mut fb);
|
|
c.stream("fetch with no objects", fb, GroupOrder::Ascending, vec![UniEvent::Header(StreamHeader::Fetch(FetchHeader { request_id: 10 }))]);
|
|
|
|
// ── §11.5 padding ──
|
|
c.stream("padding stream", encode_padding_stream(8), GroupOrder::Ascending, vec![UniEvent::Header(StreamHeader::Padding), UniEvent::Padding(8)]);
|
|
|
|
// ── §11.2 datagrams ──
|
|
let dg = |d: ObjectDatagram| (Datagram::Object(d.clone()), d.encode().unwrap());
|
|
for (label, d) in [
|
|
("datagram: Object ID 0, default priority", ObjectDatagram { track_alias: 2, group_id: 5, object_id: 0, publisher_priority: None, properties: None, end_of_group: false, content: DatagramContent::Payload(b"opus".to_vec()) }),
|
|
(
|
|
"datagram: a priority, properties, End of Group",
|
|
ObjectDatagram { track_alias: 2, group_id: 5, object_id: 9, publisher_priority: Some(3), properties: Some(props(vec![kv(prop::OBJECT_DELIVERY_TIMEOUT, 20)])), end_of_group: true, content: DatagramContent::Payload(b"last".to_vec()) },
|
|
),
|
|
("datagram: End of Track status", ObjectDatagram { track_alias: 2, group_id: 6, object_id: 0, publisher_priority: Some(3), properties: None, end_of_group: false, content: DatagramContent::Status(ObjectStatus::EndOfTrack) }),
|
|
(
|
|
"datagram: Normal status with properties",
|
|
ObjectDatagram { track_alias: 3, group_id: 1, object_id: 1, publisher_priority: None, properties: Some(props(vec![kb(0x3a45, b"p")])), end_of_group: false, content: DatagramContent::Status(ObjectStatus::Normal) },
|
|
),
|
|
("datagram: an empty payload", ObjectDatagram { track_alias: 3, group_id: 1, object_id: 2, publisher_priority: None, properties: None, end_of_group: false, content: DatagramContent::Payload(vec![]) }),
|
|
] {
|
|
let (d, b) = dg(d);
|
|
c.datagram(label, d, b, true);
|
|
}
|
|
// Object ID 0 written out rather than flagged: the same object.
|
|
let d = ObjectDatagram { track_alias: 2, group_id: 5, object_id: 0, publisher_priority: None, properties: None, end_of_group: false, content: DatagramContent::Payload(b"x".to_vec()) };
|
|
c.datagram("datagram: Object ID 0 written out", Datagram::Object(d), cat(&[&v(0x08), &v(2), &v(5), &v(0), b"x"]), false);
|
|
c.datagram("padding datagram", Datagram::Padding(4), Datagram::padding(4), true);
|
|
|
|
// ── the WAI_RECEIPT property's value (carriage) ──
|
|
let mut on_chip = 10_000u32.to_be_bytes().to_vec();
|
|
on_chip.extend_from_slice(&1_000_000u64.to_be_bytes());
|
|
on_chip.push(1);
|
|
let mut calibrated = 2_000u32.to_be_bytes().to_vec();
|
|
calibrated.extend_from_slice(&250_000u64.to_be_bytes());
|
|
calibrated.extend_from_slice(&(b"cal-2026-07".len() as u32).to_be_bytes());
|
|
calibrated.extend_from_slice(b"cal-2026-07");
|
|
for (label, p) in [
|
|
("WAI_RECEIPT: unmetered", wai_receipt(0x11, 0, "", vec![], 0x22)),
|
|
("WAI_RECEIPT: OnChipCounter", wai_receipt(0x33, 1_200_000, "OnChipCounter", on_chip, 0x44)),
|
|
("WAI_RECEIPT: CalibratedInstrument", wai_receipt(0x55, 4_000_000, "CalibratedInstrument", calibrated, 0x66)),
|
|
("WAI_RECEIPT: ModelBased", wai_receipt(0x77, 300, "ModelBased", vec![], 0x88)),
|
|
] {
|
|
c.push(label, "wai-receipt", &p.to_bytes(), true, None, wai_json(&p));
|
|
}
|
|
let mut extra = wai_receipt(0x11, 0, "", vec![], 0x22).to_bytes();
|
|
extra.push(0);
|
|
c.bad("WAI_RECEIPT: a byte after the last field", "wai-receipt", &extra, None, "invalid-property", "property_invalid");
|
|
c.bad("WAI_RECEIPT: cut short", "wai-receipt", &wai_receipt(0x11, 0, "", vec![], 0x22).to_bytes()[..60], None, "invalid-property", "property_invalid");
|
|
|
|
// ── negative vectors: control messages ──
|
|
let good_sub = ControlMessage::Subscribe(TrackRequest { request_id: 0, namespace: ns(&["a"]), name: name("b"), params: sorted(vec![Parameter::Forward(true)]) }).encode().unwrap();
|
|
c.bad("control: the Length runs past the bytes", "control", &good_sub[..good_sub.len() - 1], None, "truncated", PV);
|
|
// Length one short: the last parameter is cut inside the body.
|
|
let mut short = good_sub.clone();
|
|
short[2] -= 1;
|
|
short.pop();
|
|
c.bad("control: a body shorter than its fields", "control", &short, None, "truncated", PV);
|
|
let body = cat(&[&v(4), &v(0), &[0]]);
|
|
c.bad("control: a byte after the last field", "control", &frame(msg::REQUEST_UPDATE, &body), None, "trailing-bytes", PV);
|
|
c.bad("control: a varint that runs past its body", "control", &frame(msg::GOAWAY, &[0x00, 0xff, 0x01]), None, "truncated", PV);
|
|
c.bad("control: type 0x40, reserved (CLIENT_SETUP up to -10)", "control", &frame(0x40, &[]), None, "unknown-message-type", PV);
|
|
c.bad("control: type 0x21, reserved (SERVER_SETUP up to -16)", "control", &frame(0x21, &[]), None, "unknown-message-type", PV);
|
|
c.bad("control: an unknown type", "control", &frame(0x3f, &[]), None, "unknown-message-type", PV);
|
|
let sub_body = |params: &[u8], n: u64| cat(&[&v(0), &v(1), &v(1), b"a", &v(1), b"b", &v(n), params]);
|
|
c.bad("SUBSCRIBE: FORWARD twice", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x10, 1, 0x00, 0], 2)), None, "duplicate", PV);
|
|
c.bad("SUBSCRIBE: one OBJECTID_FILTER key twice", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x26, 2, 0, 1, 0x00, 2, 0, 4], 2)), None, "duplicate-filter", "request:INVALID_FILTER");
|
|
c.bad("SUBSCRIBE: an unknown parameter", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x05, 0], 1)), None, "unknown-parameter", PV);
|
|
c.bad("SUBSCRIBE: FILL_PARAMETERS holding FORWARD", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x23, 2, 0x10, 1], 1)), None, "parameter-not-allowed", PV);
|
|
c.bad("SUBSCRIBE: FORWARD of 2", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x10, 2], 1)), None, "invalid-value", PV);
|
|
c.bad("SUBSCRIBE: GROUP_ORDER of 0", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x22, 0], 1)), None, "invalid-value", PV);
|
|
c.bad("SUBSCRIBE: a Location Filter Type of 6", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x21, 6], 1)), None, "invalid-value", PV);
|
|
let mut eg = vec![0x21, 3];
|
|
eg.extend(v(u64::MAX));
|
|
eg.extend(v(0));
|
|
eg.extend(v(1));
|
|
c.bad("SUBSCRIBE: StartGroup + EndGroupDelta past 2^64-1", "control", &frame(msg::SUBSCRIBE, &sub_body(&eg, 1)), None, "overflow", PV);
|
|
let mut ov = vec![0x35, 1];
|
|
ov.extend(v(u64::MAX));
|
|
ov.push(1);
|
|
c.bad("SUBSCRIBE: a parameter type past 2^64-1", "control", &frame(msg::SUBSCRIBE, &sub_body(&ov, 2)), None, "type-overflow", PV);
|
|
c.bad("SUBSCRIBE: a PRIORITY_FILTER above 255", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x27, 4, 0, 0x81, 0x00, 0], 1)), None, "invalid-filter", "request:INVALID_FILTER");
|
|
c.bad("SUBSCRIBE: an OBJECT_PROPERTY_FILTER on an odd type", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x28, 3, 0, 0x0b, 0], 1)), None, "invalid-filter", "request:INVALID_FILTER");
|
|
let mut rd = vec![0x26];
|
|
let inner = cat(&[&[0], &v(5), &v(u64::MAX)]);
|
|
rd.extend(v(inner.len() as u64));
|
|
rd.extend(inner);
|
|
c.bad("SUBSCRIBE: a range end past 2^64-1", "control", &frame(msg::SUBSCRIBE, &sub_body(&rd, 1)), None, "invalid-filter", "request:INVALID_FILTER");
|
|
c.bad("SUBSCRIBE: a token of an unknown alias type", "control", &frame(msg::SUBSCRIBE, &sub_body(&[0x03, 2, 0x04, 0x00], 1)), None, "malformed-token", "session:KEY_VALUE_FORMATTING_ERROR");
|
|
c.bad("SUBSCRIBE_OK: a parameter it may not carry (FORWARD)", "control", &frame(msg::SUBSCRIBE_OK, &[7, 1, 0x10, 1]), None, "parameter-not-allowed", PV);
|
|
let mut many = v(33);
|
|
for _ in 0..33 {
|
|
many.extend(cat(&[&v(1), b"x"]));
|
|
}
|
|
c.bad("SUBSCRIBE_NAMESPACE: 33 namespace fields", "control", &frame(msg::SUBSCRIBE_NAMESPACE, &cat(&[&v(0), &many, &v(0)])), None, "too-many-namespace-fields", PV);
|
|
c.bad("SUBSCRIBE_NAMESPACE: an empty namespace field", "control", &frame(msg::SUBSCRIBE_NAMESPACE, &cat(&[&v(0), &v(1), &v(0), &v(0)])), None, "empty-namespace-field", PV);
|
|
let long_field = vec![b'n'; 4000];
|
|
let fb = cat(&[&v(0), &v(1), &v(4000), &long_field, &v(200)]);
|
|
c.bad("SUBSCRIBE: a full track name past 4,096 bytes", "control", &frame(msg::SUBSCRIBE, &fb), None, "length-exceeds-limit", PV);
|
|
let reason = vec![b'r'; 1025];
|
|
c.bad("REQUEST_ERROR: a reason phrase of 1,025 bytes", "control", &frame(msg::REQUEST_ERROR, &cat(&[&v(0), &v(0), &v(1025), &reason])), None, "length-exceeds-limit", PV);
|
|
c.bad("GOAWAY: a URI of 8,193 bytes", "control", &frame(msg::GOAWAY, &v(8193)), None, "length-exceeds-limit", PV);
|
|
c.bad("SUBSCRIBE_OK: a property value of 65,536 bytes", "control", &frame(msg::SUBSCRIBE_OK, &cat(&[&v(1), &v(0), &v(0x3a45), &v(0x1_0000)])), None, "length-exceeds-limit", PV);
|
|
c.bad("FETCH_OK: End Of Track of 2", "control", &frame(msg::FETCH_OK, &[2, 0, 0, 0]), None, "invalid-value", PV);
|
|
c.bad("REQUEST_ERROR: REDIRECT cut short", "control", &frame(msg::REQUEST_ERROR, &cat(&[&v(0x34), &v(0), &v(0), &v(0)])), None, "truncated", PV);
|
|
c.bad("SETUP: MAX_FILTER_RANGES twice", "control", &frame(msg::SETUP, &[0x06, 1, 0x00, 2]), None, "duplicate", PV);
|
|
let mut sd = v(u64::MAX);
|
|
sd.extend(v(0));
|
|
sd.extend(v(1));
|
|
sd.extend(v(0));
|
|
c.bad("SETUP: an option type past 2^64-1", "control", &frame(msg::SETUP, &sd), None, "type-overflow", PV);
|
|
c.bad("SETUP: a token that does not read", "control", &frame(msg::SETUP, &[0x03, 1, 0x07]), None, "malformed-token", "session:KEY_VALUE_FORMATTING_ERROR");
|
|
c.bad("SUBSCRIBE_OK: Immutable Properties inside Immutable Properties", "control", &frame(msg::SUBSCRIBE_OK, &[1, 0, 0x0b, 3, 0x0b, 1, 0x00]), None, "malformed-track", "malformed-track");
|
|
c.bad("SUBSCRIBE_OK: DYNAMIC_GROUPS of 2", "control", &frame(msg::SUBSCRIBE_OK, &[1, 0, 0x30, 2]), None, "invalid-value", PV);
|
|
|
|
// ── negative vectors: streams and datagrams ──
|
|
let asc = Some(json!({"group_order": 1}));
|
|
let setup_bytes = ControlMessage::Setup(setup.clone()).encode().unwrap();
|
|
let g = goaway.encode().unwrap();
|
|
c.bad("control stream: GOAWAY twice", "uni-stream", &cat(&[&setup_bytes, &g, &g]), asc.clone(), "unexpected-message", PV);
|
|
let sub_msg = ControlMessage::Subscribe(TrackRequest { request_id: 0, namespace: ns(&["a"]), name: name("b"), params: Parameters::new() }).encode().unwrap();
|
|
c.bad("control stream: SUBSCRIBE after SETUP", "uni-stream", &cat(&[&setup_bytes, &sub_msg]), asc.clone(), "unexpected-message", PV);
|
|
c.bad("stream: type 0x06", "uni-stream", &[0x06, 0], asc.clone(), "unknown-stream-type", PV);
|
|
c.bad("stream: a subgroup type of 0x91, two bytes", "uni-stream", &[0x80, 0x91, 0, 0, 0], asc.clone(), "unknown-stream-type", PV);
|
|
c.bad("subgroup: SUBGROUP_ID_MODE 0b11", "uni-stream", &[0x16, 0, 0, 0, 0], asc.clone(), "invalid-flags", PV);
|
|
let sgh = |flags: u8| vec![flags, 1, 2, 7];
|
|
c.bad("subgroup: Object Status 0x1", "uni-stream", &cat(&[&sgh(0x10), &[0, 0, 1]]), asc.clone(), "unknown-object-status", PV);
|
|
c.bad("subgroup: properties on End of Group", "uni-stream", &cat(&[&sgh(0x11), &[0, 2, 0x02, 5, 0, 3]]), asc.clone(), "properties-on-non-normal", PV);
|
|
c.bad("subgroup: a FIN inside an object", "uni-stream", &cat(&[&sgh(0x10), &[0, 5, b'a', b'b']]), asc.clone(), "truncated", PV);
|
|
let mut big = sgh(0x10);
|
|
big.extend(v(u64::MAX));
|
|
big.extend(&[1, b'a', 0, 1, b'b']);
|
|
c.bad("subgroup: an Object ID past 2^64-1", "uni-stream", &big, asc.clone(), "overflow", PV);
|
|
c.bad("subgroup: Immutable Properties inside Immutable Properties", "uni-stream", &cat(&[&sgh(0x11), &[0, 5, 0x0b, 3, 0x0b, 1, 0x00, 1, b'x']]), asc.clone(), "malformed-track", "malformed-track");
|
|
let mut mand = sgh(0x11);
|
|
mand.extend(&[0]);
|
|
let mp = cat(&[&v(0x4000), &v(1)]);
|
|
mand.extend(v(mp.len() as u64));
|
|
mand.extend(mp);
|
|
mand.extend(&[1, b'x']);
|
|
c.bad("subgroup: a Mandatory Track Property on an object", "uni-stream", &mand, asc.clone(), "malformed-track", "malformed-track");
|
|
c.bad("subgroup: a Prior Object ID Gap past the Object ID", "uni-stream", &cat(&[&sgh(0x11), &[1, 2, 0x3e, 3, 1, b'x']]), asc.clone(), "malformed-track", "malformed-track");
|
|
let mut fh = Vec::new();
|
|
FetchHeader { request_id: 0 }.encode(&mut fh);
|
|
c.bad("fetch: a first object that refers back", "uni-stream", &cat(&[&fh, &[0x1d, 0, 0, 7, 1, b'a']]), asc.clone(), "no-prior-object", PV);
|
|
c.bad("fetch: Serialization Flags of 0x80", "uni-stream", &cat(&[&fh, &v(0x80), &[0, 0]]), asc.clone(), "invalid-flags", PV);
|
|
c.bad("fetch: a Group ID below 0 in Descending order", "uni-stream", &cat(&[&fh, &[0x1c, 1, 0, 7, 1, b'a', 0x08, 1, 0, 1, b'b']]), Some(json!({"group_order": 2})), "overflow", PV);
|
|
c.bad("fetch: a FIN inside an object", "uni-stream", &cat(&[&fh, &[0x1c, 1, 0, 7, 3, b'a']]), asc.clone(), "truncated", PV);
|
|
c.bad("datagram: STATUS and END_OF_GROUP", "datagram", &[0x22, 1, 2, 3, 0], None, "invalid-flags", PV);
|
|
c.bad("datagram: bit 4 set", "datagram", &[0x10, 1, 2, 3, b'x'], None, "invalid-flags", PV);
|
|
c.bad("datagram: an undefined bit (0x40)", "datagram", &[0x40, 1, 2, 3, b'x'], None, "invalid-flags", PV);
|
|
c.bad("datagram: PROPERTIES with a length of 0", "datagram", &[0x0d, 1, 2, 0, b'x'], None, "empty-datagram-properties", PV);
|
|
c.bad("datagram: properties beside End of Track", "datagram", &[0x2d, 1, 2, 2, 0x02, 5, 4], None, "properties-on-non-normal", PV);
|
|
c.bad("datagram: an unknown status", "datagram", &[0x2c, 1, 2, 2], None, "unknown-object-status", PV);
|
|
c.bad("datagram: a byte after the status", "datagram", &[0x2c, 1, 2, 0, 0], None, "trailing-bytes", PV);
|
|
c.bad("datagram: cut short", "datagram", &[0x00, 1], None, "truncated", PV);
|
|
|
|
c.vectors
|
|
}
|
|
|
|
fn corpus_doc(vectors: Vec<Value>) -> Value {
|
|
json!({
|
|
"about": "draft-ietf-moq-transport-22 wire vectors: varints, control messages, unidirectional streams (control, subgroup, fetch, padding), datagrams, and the value of the WAI_RECEIPT object property of this crate's carriage. Each vector gives its bytes in hex and either its reading, in the JSON view of joule_moq::draft22::json described in the crate README, or the refusal it must meet with the draft's consequence. A uni-stream vector is the stream's bytes up to its FIN; its context gives the Group Order of the request a fetch stream answers (1 Ascending, 2 Descending). A vector marked canonical is the form this crate's encoder makes. Generated by tests/draft22_vectors.rs; regenerate with JOULE_MOQ_WRITE_VECTORS=1.",
|
|
"draft": "draft-ietf-moq-transport-22",
|
|
"vectors": vectors,
|
|
})
|
|
}
|
|
|
|
fn unhex(s: &str) -> Vec<u8> {
|
|
(0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap()).collect()
|
|
}
|
|
|
|
fn order_of(v: &Value) -> GroupOrder {
|
|
match v["context"]["group_order"].as_u64() {
|
|
Some(2) => GroupOrder::Descending,
|
|
_ => GroupOrder::Ascending,
|
|
}
|
|
}
|
|
|
|
/// Decode a vector: its reading, or its refusal.
|
|
fn decode(v: &Value) -> Result<(Value, Option<Vec<u8>>), Value> {
|
|
let b = unhex(v["hex"].as_str().unwrap());
|
|
let err = |e: WireError| json::error(&e);
|
|
match v["kind"].as_str().unwrap() {
|
|
"varint" => {
|
|
let (val, n) = varint::get(&b, "varint").map_err(err)?;
|
|
if n != b.len() {
|
|
return Err(json!({"error": "trailing-bytes"}));
|
|
}
|
|
Ok((json!({"value": val, "length": n}), Some(varint::to_vec(val))))
|
|
}
|
|
"control" => {
|
|
let m = ControlMessage::decode_exact(&b).map_err(err)?;
|
|
Ok((json::control(&m), Some(m.encode().unwrap())))
|
|
}
|
|
"uni-stream" => {
|
|
let order = order_of(v);
|
|
let ev = decode_uni_stream(&b, order).map_err(err)?;
|
|
Ok((Value::Array(ev.iter().map(json::uni_event).collect()), reencode(&ev, order)))
|
|
}
|
|
"datagram" => {
|
|
let d = Datagram::decode(&b).map_err(err)?;
|
|
let again = match &d {
|
|
Datagram::Object(o) => o.encode().unwrap(),
|
|
Datagram::Padding(n) => Datagram::padding(*n),
|
|
};
|
|
Ok((json::datagram(&d), Some(again)))
|
|
}
|
|
"wai-receipt" => match WaiReceiptProperty::from_bytes(&b) {
|
|
Ok(p) => Ok((wai_json(&p), Some(p.to_bytes()))),
|
|
Err(_) => Err(json!({"error": "invalid-property", "consequence": "property_invalid"})),
|
|
},
|
|
k => panic!("unknown kind {k}"),
|
|
}
|
|
}
|
|
|
|
/// The bytes this crate's encoder makes of a decoded stream.
|
|
fn reencode(ev: &[UniEvent], order: GroupOrder) -> Option<Vec<u8>> {
|
|
let mut out = Vec::new();
|
|
let mut sub: Option<SubgroupCodec> = None;
|
|
let mut fetch: Option<FetchCodec> = None;
|
|
for e in ev {
|
|
match e {
|
|
UniEvent::Header(StreamHeader::Control) => {}
|
|
UniEvent::Header(StreamHeader::Padding) => out.extend(varint::to_vec(joule_moq::draft22::data::stream_type::PADDING)),
|
|
UniEvent::Header(StreamHeader::Subgroup(h)) => {
|
|
h.encode(&mut out);
|
|
sub = Some(SubgroupCodec::new(*h));
|
|
}
|
|
UniEvent::Header(StreamHeader::Fetch(h)) => {
|
|
h.encode(&mut out);
|
|
fetch = Some(FetchCodec::new(order));
|
|
}
|
|
UniEvent::Control(m) => out.extend(m.encode().ok()?),
|
|
UniEvent::SubgroupObject(o) => sub.as_mut()?.encode_object(o, &mut out).ok()?,
|
|
UniEvent::FetchEntry(f) => fetch.as_mut()?.encode(f, &mut out).ok()?,
|
|
UniEvent::Padding(n) => out.resize(out.len() + n, 0),
|
|
}
|
|
}
|
|
Some(out)
|
|
}
|
|
|
|
/// The incremental decoder over the same bytes, fed in pieces of `step`.
|
|
fn decode_incrementally(b: &[u8], order: GroupOrder, step: usize) -> Result<Vec<UniEvent>, WireError> {
|
|
let mut d = UniStreamDecoder::new(order, Limits { max_payload: u64::MAX, max_properties: u64::MAX });
|
|
let mut out = Vec::new();
|
|
for chunk in b.chunks(step.max(1)) {
|
|
d.push(chunk);
|
|
while let Some(e) = d.next()? {
|
|
out.push(e);
|
|
}
|
|
}
|
|
d.finish()?;
|
|
Ok(out)
|
|
}
|
|
|
|
#[test]
|
|
fn the_draft22_corpus_is_current_and_holds() {
|
|
let doc = corpus_doc(build());
|
|
let text = serde_json::to_string_pretty(&doc).unwrap() + "\n";
|
|
if std::env::var("JOULE_MOQ_WRITE_VECTORS").as_deref() == Ok("1") {
|
|
std::fs::create_dir_all(std::path::Path::new(CORPUS).parent().unwrap()).unwrap();
|
|
std::fs::write(CORPUS, &text).unwrap();
|
|
}
|
|
let committed = std::fs::read_to_string(CORPUS).unwrap_or_else(|e| panic!("read {CORPUS}: {e}"));
|
|
assert!(committed == text, "{CORPUS} is not what tests/draft22_vectors.rs generates; regenerate it with JOULE_MOQ_WRITE_VECTORS=1");
|
|
|
|
let mut failures = Vec::new();
|
|
let (mut good, mut bad) = (0, 0);
|
|
for v in doc["vectors"].as_array().unwrap() {
|
|
let name = v["name"].as_str().unwrap();
|
|
let expect = &v["expect"];
|
|
let b = unhex(v["hex"].as_str().unwrap());
|
|
match (expect.get("error"), decode(v)) {
|
|
(None, Ok((got, again))) => {
|
|
good += 1;
|
|
if &got != expect {
|
|
failures.push(format!("{name}: read as {got}"));
|
|
}
|
|
if v["canonical"] == true && again.as_deref() != Some(&b[..]) {
|
|
failures.push(format!("{name}: re-encodes to other bytes"));
|
|
}
|
|
}
|
|
(None, Err(e)) => failures.push(format!("{name}: refused ({e}); the corpus reads it")),
|
|
(Some(_), Ok((got, _))) => failures.push(format!("{name}: read as {got}; the corpus refuses it")),
|
|
(Some(_), Err(e)) => {
|
|
bad += 1;
|
|
if &e != expect {
|
|
failures.push(format!("{name}: refused as {e}, the corpus says {expect}"));
|
|
}
|
|
}
|
|
}
|
|
// The incremental decoder agrees on every stream: the same events for a good
|
|
// one, a refusal for a negative one, whatever the pieces.
|
|
if v["kind"] == "uni-stream" {
|
|
for step in [1, 2, 7, b.len().max(1)] {
|
|
let inc = decode_incrementally(&b, order_of(v), step);
|
|
match (expect.get("error"), inc) {
|
|
(None, Ok(ev)) => {
|
|
// Padding arrives in whatever pieces the bytes did; count it whole.
|
|
let mut merged: Vec<UniEvent> = Vec::new();
|
|
for e in ev {
|
|
match (merged.last_mut(), e) {
|
|
(Some(UniEvent::Padding(n)), UniEvent::Padding(k)) => *n += k,
|
|
(_, e) => merged.push(e),
|
|
}
|
|
}
|
|
let got = Value::Array(merged.iter().map(json::uni_event).collect());
|
|
if &got != expect {
|
|
failures.push(format!("{name}: in pieces of {step}, read as {got}"));
|
|
}
|
|
}
|
|
(None, Err(e)) => failures.push(format!("{name}: in pieces of {step}, refused ({e})")),
|
|
(Some(_), Ok(_)) => failures.push(format!("{name}: in pieces of {step}, accepted")),
|
|
(Some(_), Err(_)) => {}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
assert!(failures.is_empty(), "{} vector(s) disagree:\n{}", failures.len(), failures.join("\n"));
|
|
assert!(good > 0 && bad > 0, "the corpus holds both kinds");
|
|
}
|