Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Version negotiation (6.3.6/6.3.21 + 5.8.6 `ngsildConformance`): amend
3 : //! response payloads to conform to an earlier NGSI-LD version per the
4 : //! backwards-compatibility fallbacks of clause 4.3.6.8 (Tables 4.3.6.8-1/2/3).
5 : //!
6 : //! `Prefer: ngsi-ld=<major.minor>` ⇒ apply the fallbacks, answer with
7 : //! `Preference-Applied: ngsi-ld=<conformant-version>`, and 203 Non-Authoritative
8 : //! instead of 200 when the payload was actually altered (the response tables'
9 : //! "altered Entity" rows). A subscription's `ngsildConformance` applies the
10 : //! same amendment to every notification (5.8.6).
11 :
12 : use axum::body::Body;
13 : use axum::http::{header, Request, StatusCode};
14 : use axum::middleware::Next;
15 : use axum::response::Response;
16 : use serde_json::Value;
17 :
18 : /// Members of an entity that are NOT attributes.
19 : /// `"major.minor"` (4.3.6.8; a patch part as in `1.9.1` is tolerated).
20 152 : pub(crate) fn parse_version(s: &str) -> Option<(u32, u32)> {
21 152 : let mut it = s.trim().split('.');
22 152 : let major = it.next()?.parse().ok()?;
23 104 : let minor = it.next()?.parse().ok()?;
24 84 : Some((major, minor))
25 152 : }
26 :
27 : /// The version this broker natively conforms to.
28 : const NATIVE: (u32, u32) = (1, 9);
29 :
30 : /// Amend one compacted entity document in place; true when anything changed.
31 66 : pub(crate) fn amend_entity(doc: &mut Value, ver: (u32, u32)) -> bool {
32 66 : if ver >= NATIVE {
33 4 : return false;
34 62 : }
35 62 : let Some(obj) = doc.as_object_mut() else {
36 0 : return false;
37 : };
38 62 : let mut changed = false;
39 : // Table 4.3.6.8-1 entity-level members.
40 62 : if ver < (1, 9) && obj.remove("expiresAt").is_some() {
41 16 : changed = true;
42 46 : }
43 62 : if ver < (1, 4) && obj.remove("scope").is_some() {
44 12 : changed = true;
45 50 : }
46 62 : if ver < (1, 3) {
47 : // Note 1: 1.0 knows a single entity type — keep the first.
48 24 : if let Some(Value::Array(types)) = obj.get("type") {
49 4 : if let Some(first) = types.first().cloned() {
50 4 : obj.insert("type".into(), first);
51 4 : changed = true;
52 4 : }
53 20 : }
54 38 : }
55 62 : let attr_names: Vec<String> = obj
56 62 : .keys()
57 244 : .filter(|k| !crate::repr::ENTITY_META.contains(&k.as_str()))
58 62 : .cloned()
59 62 : .collect();
60 116 : for name in attr_names {
61 116 : if let Some(v) = obj.get_mut(&name) {
62 116 : changed |= amend_attr(v, ver);
63 116 : }
64 : }
65 62 : changed
66 66 : }
67 :
68 : /// One attribute node (object or multi-instance array), recursively.
69 152 : fn amend_attr(v: &mut Value, ver: (u32, u32)) -> bool {
70 152 : let mut changed = false;
71 152 : if let Value::Array(instances) = v {
72 : // Notes 2/3: 1.0 has no datasetId multi-instances — keep the default
73 : // instance (the one without datasetId), else the first. Selection runs
74 : // BEFORE the <1.3 datasetId removal erases the tiebreaker.
75 : // A temporal instance array (clause 5.2.5) is a different shape —
76 : // every instance carries instanceId — and Notes 2/3 do not cover it,
77 : // so its history survives intact instead of collapsing to one point.
78 28 : let temporal = instances.iter().any(|i| i.get("instanceId").is_some());
79 16 : if ver < (1, 3) && !temporal {
80 4 : let pick = instances
81 4 : .iter()
82 8 : .position(|i| i.get("datasetId").is_none())
83 4 : .unwrap_or(0);
84 4 : if let Some(mut one) = instances.get(pick).cloned() {
85 4 : amend_attr(&mut one, ver);
86 4 : *v = one;
87 4 : return true;
88 0 : }
89 12 : }
90 24 : for i in instances.iter_mut() {
91 24 : changed |= amend_attr(i, ver);
92 24 : }
93 12 : return changed;
94 136 : }
95 136 : let Some(obj) = v.as_object_mut() else {
96 0 : return false;
97 : };
98 136 : let ty = obj.get("type").and_then(Value::as_str).unwrap_or("");
99 8 : match ty {
100 : // Table 4.3.6.8-1 attribute-type fallbacks.
101 136 : "LanguageProperty" if ver < (1, 4) => {
102 12 : obj.insert("type".into(), Value::String("Property".into()));
103 12 : if let Some(lm) = obj.remove("languageMap") {
104 12 : obj.insert("value".into(), lm);
105 12 : }
106 12 : changed = true;
107 : }
108 116 : "JsonProperty" if ver < (1, 8) => {
109 20 : obj.insert("type".into(), Value::String("Property".into()));
110 20 : if let Some(j) = obj.remove("json") {
111 20 : obj.insert("value".into(), j);
112 20 : }
113 20 : changed = true;
114 : }
115 96 : "VocabProperty" if ver < (1, 8) => {
116 8 : obj.insert("type".into(), Value::String("Property".into()));
117 8 : if let Some(vv) = obj.remove("vocab") {
118 8 : obj.insert("value".into(), vv);
119 8 : }
120 8 : changed = true;
121 : }
122 88 : "ListProperty" if ver < (1, 8) => {
123 8 : obj.insert("type".into(), Value::String("Property".into()));
124 8 : if let Some(vl) = obj.remove("valueList") {
125 8 : obj.insert("value".into(), vl);
126 8 : }
127 8 : changed = true;
128 : }
129 80 : "ListRelationship" if ver < (1, 8) => {
130 8 : obj.insert("type".into(), Value::String("Relationship".into()));
131 8 : if let Some(ol) = obj.remove("objectList") {
132 8 : obj.insert("object".into(), ol);
133 8 : }
134 8 : changed = true;
135 : }
136 80 : _ => {}
137 : }
138 : // Tables 4.3.6.8-2/3 sub-member removals (shared version boundaries).
139 136 : if ver < (1, 3) {
140 132 : for k in ["datasetId", "observedAt", "unitCode"] {
141 132 : if obj.remove(k).is_some() {
142 20 : changed = true;
143 112 : }
144 : }
145 92 : }
146 136 : if ver < (1, 8) && obj.remove("objectType").is_some() {
147 12 : changed = true;
148 124 : }
149 136 : if ver < (1, 9) {
150 272 : for k in ["valueType", "expiresAt"] {
151 272 : if obj.remove(k).is_some() {
152 16 : changed = true;
153 256 : }
154 : }
155 0 : }
156 : // Sub-attributes (properties-of-properties) get the same treatment.
157 : // Everything the Attribute itself is made of is excluded, and that
158 : // includes the 5.2.14 `previous*` members a showChanges notification
159 : // carries: each holds the VALUE the Attribute held, "any JSON value as
160 : // defined by IETF RFC 8259" (Table 4.3.6.8-2) or a String[] object
161 : // (Table 4.3.6.8-3). Walking one as an Attribute node applies Notes 2/3
162 : // to a plain JSON array and hands the subscriber a single element where
163 : // its own previous value had several.
164 136 : let subs: Vec<String> = obj
165 136 : .keys()
166 344 : .filter(|k| {
167 336 : !matches!(
168 344 : k.as_str(),
169 344 : "type"
170 208 : | "value"
171 108 : | "object"
172 80 : | "languageMap"
173 72 : | "json"
174 72 : | "vocab"
175 72 : | "valueList"
176 72 : | "objectList"
177 72 : | "previousValue"
178 60 : | "previousObject"
179 56 : | "previousLanguageMap"
180 52 : | "previousJson"
181 52 : | "previousVocab"
182 52 : | "previousValueList"
183 52 : | "previousObjectList"
184 52 : | "datasetId"
185 44 : | "observedAt"
186 32 : | "unitCode"
187 20 : | "objectType"
188 16 : | "valueType"
189 16 : | "expiresAt"
190 16 : | "createdAt"
191 16 : | "modifiedAt"
192 16 : | "instanceId"
193 : )
194 344 : })
195 136 : .cloned()
196 136 : .collect();
197 136 : for name in subs {
198 8 : if let Some(sub) = obj.get_mut(&name) {
199 8 : if sub.is_object() || sub.is_array() {
200 8 : changed |= amend_attr(sub, ver);
201 8 : }
202 0 : }
203 : }
204 136 : changed
205 152 : }
206 :
207 : /// Amend whatever entity-bearing payload shape a response carries.
208 : /// The 4.3.6.8 fallback tables describe Entity data. The other NGSI-LD
209 : /// resources are served with the same id + type shape but have no version
210 : /// fallbacks of their own, so they pass through untouched and keep their 200
211 : /// rather than the 203 that marks an altered Entity. They are recognised by
212 : /// their reserved clause 5.2 data-type name.
213 26 : fn is_reserved_resource(o: &serde_json::Map<String, Value>) -> bool {
214 : const RESERVED: [&str; 6] = [
215 : "Subscription",
216 : "ContextSourceRegistration",
217 : "CSourceRegistration",
218 : "Notification",
219 : "EntityMap",
220 : "Snapshot",
221 : ];
222 26 : o.get("type")
223 26 : .and_then(Value::as_str)
224 26 : .is_some_and(|t| RESERVED.contains(&t))
225 26 : }
226 :
227 50 : pub(crate) fn amend_payload(doc: &mut Value, ver: (u32, u32)) -> bool {
228 50 : match doc {
229 8 : Value::Array(items) => {
230 8 : let mut changed = false;
231 8 : for i in items {
232 4 : changed |= amend_payload(i, ver);
233 4 : }
234 8 : changed
235 : }
236 38 : Value::Object(o) => {
237 38 : if o.get("type").and_then(Value::as_str) == Some("Notification") {
238 8 : match o.get_mut("data") {
239 4 : Some(d) => amend_payload(d, ver),
240 4 : None => false,
241 : }
242 30 : } else if o.contains_key("id") && o.contains_key("type") && !is_reserved_resource(o) {
243 22 : amend_entity(doc, ver)
244 : } else {
245 8 : false
246 : }
247 : }
248 4 : _ => false,
249 : }
250 50 : }
251 :
252 : /// `Prefer: ngsi-ld=<version>` from a raw Prefer header value (RFC 7240 —
253 : /// preferences are comma-separated `token=value` pairs).
254 112 : pub(crate) fn preferred_version(prefer: &str) -> Option<(u32, u32)> {
255 120 : prefer.split(',').find_map(|p| {
256 120 : let (k, v) = p.split_once('=')?;
257 : // RFC 9110 clause 5.6.2: a field name is case-insensitive, and RFC
258 : // 7240 lets a preference carry its own parameters after ';'.
259 108 : if !k.trim().eq_ignore_ascii_case("ngsi-ld") {
260 28 : return None;
261 80 : }
262 80 : parse_version(v.split(';').next()?.trim().trim_matches('"'))
263 120 : })
264 112 : }
265 :
266 : /// Router middleware (6.3.6): honour `Prefer: ngsi-ld=` on JSON responses.
267 26106 : pub async fn prefer_version_layer(req: Request<Body>, next: Next) -> Response {
268 : // RFC 9110 clause 5.3: repeated field lines carry the same meaning as one
269 : // comma-separated list, so the preference is looked for on every line.
270 26106 : let requested = req
271 26106 : .headers()
272 26106 : .get_all("prefer")
273 26106 : .iter()
274 26106 : .filter_map(|h| h.to_str().ok())
275 26106 : .find_map(preferred_version);
276 26106 : let Some(ver) = requested else {
277 26062 : return next.run(req).await;
278 : };
279 44 : let resp = next.run(req).await;
280 44 : let is_json = resp
281 44 : .headers()
282 44 : .get(header::CONTENT_TYPE)
283 44 : .and_then(|h| h.to_str().ok())
284 44 : .is_some_and(|ct| {
285 44 : ct.starts_with("application/json") || ct.starts_with("application/ld+json")
286 44 : });
287 44 : if resp.status() != StatusCode::OK || !is_json {
288 4 : return resp;
289 40 : }
290 : // 6.3.6 owes `Preference-Applied` to every `Prefer: ngsi-ld=` request,
291 : // but at or above the version this broker serves natively there is
292 : // nothing to amend — every rule in Table 4.3.6.8-1 is guarded by
293 : // `ver < …` and `amend_entity` returns immediately on `ver >= NATIVE`.
294 : // So the header is the whole answer, and the handler's body reaches the
295 : // client as the handler wrote it: never buffered whole, never parsed,
296 : // never re-serialized, and (unlike the amend path below) with its own
297 : // Content-Length intact.
298 40 : if ver >= NATIVE {
299 16 : let mut resp = resp;
300 16 : if let Ok(v) = format!("ngsi-ld={}.{}", NATIVE.0, NATIVE.1).parse() {
301 16 : resp.headers_mut().insert("Preference-Applied", v);
302 16 : }
303 16 : return resp;
304 24 : }
305 24 : let (mut parts, body) = resp.into_parts();
306 : // Honouring the preference is optional (RFC 7240 section 2), so the
307 : // buffer is bounded by the same cap the request wall advertises: a
308 : // response bigger than MAX_BODY_BYTES passes through byte-identical,
309 : // unamended and without Preference-Applied.
310 : use futures_util::StreamExt;
311 24 : let mut stream = body.into_data_stream();
312 24 : let mut buf: Vec<u8> = Vec::new();
313 : loop {
314 42 : match stream.next().await {
315 18 : None => break,
316 : // The upstream body failed mid-stream, so the prefix already read
317 : // is not the response and must not be served under the handler's
318 : // 200. The transport error text stays server-side (5.5.6).
319 : Some(Err(_)) => {
320 4 : parts.status = StatusCode::INTERNAL_SERVER_ERROR;
321 4 : parts.headers.remove(header::CONTENT_LENGTH);
322 4 : return Response::from_parts(parts, Body::empty());
323 : }
324 20 : Some(Ok(chunk)) => {
325 20 : if buf.len() + chunk.len() > *crate::bounds::MAX_BODY_BYTES {
326 : // stitch the already-read prefix back in front of the
327 : // untouched remainder of the stream
328 2 : let read = futures_util::stream::iter([
329 2 : Ok::<_, axum::Error>(axum::body::Bytes::from(buf)),
330 2 : Ok(chunk),
331 2 : ]);
332 2 : return Response::from_parts(parts, Body::from_stream(read.chain(stream)));
333 18 : }
334 18 : buf.extend_from_slice(&chunk);
335 : }
336 : }
337 : }
338 18 : let bytes = axum::body::Bytes::from(buf);
339 18 : let conformant = ver;
340 18 : let mut altered = false;
341 18 : let bytes = match serde_json::from_slice::<Value>(&bytes) {
342 18 : Ok(mut doc) => {
343 18 : altered = amend_payload(&mut doc, ver);
344 : // Re-serialize in the egress key order every other response path
345 : // uses (id and type first), not serde_json's own map order.
346 18 : crate::negotiate::ordered_vec(&doc).into()
347 : }
348 0 : Err(_) => bytes,
349 : };
350 : // Two integers and a dot are always a legal header value; if that ever
351 : // stopped being true, omitting the header beats taking the response down.
352 18 : if let Ok(v) = format!("ngsi-ld={}.{}", conformant.0, conformant.1).parse() {
353 18 : parts.headers.insert("Preference-Applied", v);
354 18 : }
355 18 : if altered {
356 8 : // The response tables' "altered Entity" rows: 203 Non-Authoritative.
357 8 : parts.status = StatusCode::NON_AUTHORITATIVE_INFORMATION;
358 10 : }
359 18 : parts.headers.remove(header::CONTENT_LENGTH);
360 18 : let mut resp = Response::from_parts(parts, Body::from(bytes));
361 18 : resp.headers_mut().remove(header::TRANSFER_ENCODING);
362 18 : resp
363 26106 : }
364 :
365 : #[cfg(test)]
366 : mod tests {
367 : use super::*;
368 : use serde_json::json;
369 :
370 : #[test]
371 4 : fn version_parsing() {
372 4 : assert_eq!(parse_version("1.5"), Some((1, 5)));
373 4 : assert_eq!(parse_version("1.9.1"), Some((1, 9)));
374 4 : assert_eq!(parse_version("junk"), None);
375 4 : assert_eq!(preferred_version("ngsi-ld=1.6"), Some((1, 6)));
376 4 : assert_eq!(preferred_version("body=json, ngsi-ld=1.4"), Some((1, 4)));
377 4 : assert_eq!(preferred_version("body=json"), None);
378 4 : }
379 :
380 : /// 5.2.14: with `showChanges` an Attribute carries the value it HELD, in
381 : /// a `previous*` member. That member is an NGSI-LD Value (Table
382 : /// 4.3.6.8-2: "Any JSON value as defined by IETF RFC 8259") or, for a
383 : /// Relationship, a `String or String[]` object — not a sub-Attribute. The
384 : /// 4.3.6.8 tables say nothing about it, so the amender must carry it
385 : /// through untouched. Walking into it as if it were an Attribute node
386 : /// applies Notes 2/3 to a plain JSON array and hands the subscriber one
387 : /// element where its own previous value had several.
388 : #[test]
389 4 : fn show_changes_previous_members_are_values_not_sub_attributes() {
390 4 : let mk = || {
391 4 : json!({"id": "urn:a", "type": "T",
392 4 : "speed": {"type": "Property", "value": [1, 2, 3],
393 4 : "previousValue": [9, 8, 7]},
394 4 : "where": {"type": "GeoProperty",
395 4 : "value": {"type": "LineString",
396 4 : "coordinates": [[1.0, 2.0], [3.0, 4.0]]},
397 4 : "previousValue": {"type": "LineString",
398 4 : "coordinates": [[5.0, 6.0], [7.0, 8.0]]}},
399 4 : "near": {"type": "Relationship", "object": ["urn:b", "urn:c"],
400 4 : "previousObject": ["urn:d", "urn:e"]},
401 4 : "label": {"type": "LanguageProperty", "languageMap": {"en": "now"},
402 4 : "previousLanguageMap": {"en": "then"}}})
403 4 : };
404 : // 1.2 is below every Notes 2/3 boundary — the harshest amendment.
405 4 : let mut d = mk();
406 4 : amend_entity(&mut d, (1, 2));
407 4 : assert_eq!(
408 4 : d["speed"]["previousValue"],
409 4 : json!([9, 8, 7]),
410 : "a previous Value array is a value, not a multi-instance Attribute"
411 : );
412 4 : assert_eq!(
413 4 : d["where"]["previousValue"],
414 4 : json!({"type": "LineString", "coordinates": [[5.0, 6.0], [7.0, 8.0]]}),
415 : "a previous geometry keeps every position"
416 : );
417 4 : assert_eq!(
418 4 : d["near"]["previousObject"],
419 4 : json!(["urn:d", "urn:e"]),
420 : "a Relationship object is String or String[] (Table 4.3.6.8-3)"
421 : );
422 4 : assert_eq!(
423 4 : d["label"]["previousLanguageMap"],
424 4 : json!({"en": "then"}),
425 : "a previous languageMap is not walked as an Attribute either"
426 : );
427 : // and the members the tables DO govern still fall back
428 4 : let mut d = json!({"id": "urn:a", "type": "T",
429 4 : "speed": {"type": "Property", "value": 1, "observedAt": "2026-01-01T00:00:00Z",
430 4 : "previousValue": [9, 8, 7]}});
431 4 : assert!(amend_entity(&mut d, (1, 2)));
432 4 : assert!(
433 4 : d["speed"].get("observedAt").is_none(),
434 : "observedAt is a 1.3 member and still goes"
435 : );
436 4 : assert_eq!(d["speed"]["previousValue"], json!([9, 8, 7]));
437 4 : }
438 :
439 : #[test]
440 4 : fn new_attribute_types_fall_back_per_version() {
441 12 : let mk = || {
442 12 : json!({"id": "urn:a", "type": "T",
443 12 : "lp": {"type": "LanguageProperty", "languageMap": {"en": "hi"}},
444 12 : "jp": {"type": "JsonProperty", "json": {"k": 1}},
445 12 : "vp": {"type": "VocabProperty", "vocab": "V"},
446 12 : "list": {"type": "ListProperty", "valueList": [1, 2]},
447 12 : "lr": {"type": "ListRelationship", "objectList": ["urn:b"]}})
448 12 : };
449 : // 1.8 understands Json/Vocab/List*, not... everything stays but nothing
450 : // else: only versions below the introduction boundary reformat.
451 4 : let mut d = mk();
452 4 : assert!(!amend_entity(&mut d, (1, 9)), "native version: unchanged");
453 4 : let mut d = mk();
454 4 : assert!(amend_entity(&mut d, (1, 4)));
455 4 : assert_eq!(d["jp"], json!({"type": "Property", "value": {"k": 1}}));
456 4 : assert_eq!(d["vp"], json!({"type": "Property", "value": "V"}));
457 4 : assert_eq!(d["list"], json!({"type": "Property", "value": [1, 2]}));
458 4 : assert_eq!(
459 4 : d["lr"],
460 4 : json!({"type": "Relationship", "object": ["urn:b"]})
461 : );
462 4 : assert_eq!(
463 4 : d["lp"],
464 4 : json!({"type": "LanguageProperty", "languageMap": {"en": "hi"}}),
465 : "LanguageProperty exists since 1.4"
466 : );
467 4 : let mut d = mk();
468 4 : assert!(amend_entity(&mut d, (1, 3)));
469 4 : assert_eq!(
470 4 : d["lp"],
471 4 : json!({"type": "Property", "value": {"en": "hi"}}),
472 : "1.3 predates LanguageProperty"
473 : );
474 4 : }
475 :
476 : #[test]
477 4 : fn one_dot_zero_single_type_and_default_instance() {
478 4 : let mut d = json!({"id": "urn:a", "type": ["A", "B"],
479 4 : "speed": [
480 4 : {"type": "Property", "value": 1, "datasetId": "urn:ds:1"},
481 4 : {"type": "Property", "value": 2}
482 : ]});
483 4 : assert!(amend_entity(&mut d, (1, 0)));
484 4 : assert_eq!(d["type"], "A", "note 1: single type, first wins");
485 4 : assert_eq!(
486 4 : d["speed"],
487 4 : json!({"type": "Property", "value": 2}),
488 : "note 2: default instance preferred, datasetId gone (<1.3)"
489 : );
490 4 : }
491 :
492 : #[test]
493 4 : fn sub_member_removals_by_boundary() {
494 4 : let mut d = json!({"id": "urn:a", "type": "T", "expiresAt": "2030-01-01T00:00:00Z",
495 4 : "scope": "/a",
496 4 : "r": {"type": "Relationship", "object": "urn:b", "objectType": "B",
497 4 : "observedAt": "2020-01-01T00:00:00Z",
498 4 : "nested": {"type": "Property", "value": 1, "unitCode": "C"}}});
499 4 : let mut d13 = d.clone();
500 4 : assert!(amend_entity(&mut d13, (1, 3)));
501 4 : assert!(d13.get("scope").is_none(), "scope is 1.4");
502 4 : assert!(d13.get("expiresAt").is_none(), "entity expiresAt is 1.9");
503 4 : assert!(d13["r"].get("objectType").is_none(), "objectType is 1.8");
504 4 : assert!(
505 4 : d13["r"].get("observedAt").is_some(),
506 : "observedAt fine at 1.3"
507 : );
508 4 : assert!(
509 4 : d13["r"]["nested"].get("unitCode").is_some(),
510 : "unitCode fine at 1.3"
511 : );
512 4 : assert!(amend_entity(&mut d, (1, 0)));
513 4 : assert!(d["r"].get("observedAt").is_none());
514 4 : assert!(d["r"]["nested"].get("unitCode").is_none());
515 4 : }
516 :
517 : #[test]
518 4 : fn notification_data_is_amended() {
519 4 : let mut n = json!({"id": "urn:n:1", "type": "Notification",
520 4 : "data": [{"id": "urn:a", "type": "T",
521 4 : "jp": {"type": "JsonProperty", "json": 1}}]});
522 4 : assert!(amend_payload(&mut n, (1, 6)));
523 4 : assert_eq!(n["data"][0]["jp"], json!({"type": "Property", "value": 1}));
524 4 : }
525 :
526 : /// A version string that is not `major.minor` yields no version at all —
527 : /// the preference is then ignored rather than guessed at.
528 : #[test]
529 4 : fn version_parsing_rejects_malformed_input() {
530 40 : for bad in [
531 4 : "", "1", "1.", ".", ".5", "1.x", "-1.2", "1.-2", "v1.5", "1 . 5",
532 4 : ] {
533 40 : assert_eq!(parse_version(bad), None, "{bad:?} is not major.minor");
534 : }
535 : // out of u32 range on either part
536 4 : assert_eq!(parse_version("4294967296.0"), None);
537 4 : assert_eq!(parse_version("1.4294967296"), None);
538 4 : assert_eq!(parse_version(" 1.5 "), Some((1, 5)), "outer space trimmed");
539 4 : }
540 :
541 : /// RFC 7240: preference tokens are case-insensitive and may carry
542 : /// `;`-separated parameters; a preference that is not `ngsi-ld` (or whose
543 : /// value is not a version) leaves version negotiation off.
544 : #[test]
545 4 : fn prefer_header_forms() {
546 4 : assert_eq!(preferred_version("NGSI-LD=1.5"), Some((1, 5)));
547 4 : assert_eq!(preferred_version("Ngsi-Ld=\"1.5\""), Some((1, 5)));
548 4 : assert_eq!(preferred_version("ngsi-ld=1.5; foo=bar"), Some((1, 5)));
549 4 : assert_eq!(
550 4 : preferred_version("body=json, ngsi-ld=1.5;q=0.1"),
551 : Some((1, 5))
552 : );
553 24 : for none in [
554 4 : "",
555 4 : "ngsi-ld",
556 4 : "ngsi-ld=",
557 4 : "ngsi-ld=junk",
558 4 : "body=json",
559 4 : "respond-async",
560 4 : ] {
561 24 : assert_eq!(preferred_version(none), None, "{none:?}");
562 : }
563 4 : }
564 :
565 : /// Table 4.3.6.8-1/2/3 "Version Introduced" column: a member is only
566 : /// dropped BELOW the version that introduced it — at that version it must
567 : /// still be present.
568 : #[test]
569 4 : fn members_survive_at_their_introduction_version() {
570 8 : let mk = || {
571 8 : json!({"id": "urn:a", "type": ["A", "B"], "scope": "/s",
572 8 : "expiresAt": "2030-01-01T00:00:00Z",
573 8 : "p": [{"type": "Property", "value": 1, "datasetId": "urn:ds:1",
574 8 : "observedAt": "2020-01-01T00:00:00Z", "unitCode": "C",
575 8 : "valueType": "http://x/T", "expiresAt": "2030-01-01T00:00:00Z"},
576 8 : {"type": "Property", "value": 2}],
577 8 : "lp": {"type": "LanguageProperty", "languageMap": {"en": "hi"}},
578 8 : "r": {"type": "Relationship", "object": "urn:b", "objectType": "B"}})
579 8 : };
580 4 : let mut d = mk();
581 4 : assert!(amend_entity(&mut d, (1, 8)));
582 4 : assert!(d["p"][0].get("datasetId").is_some(), "datasetId is 1.3");
583 4 : assert!(d["p"][0].get("objectType").is_none(), "attr has none");
584 4 : assert!(d["r"].get("objectType").is_some(), "objectType is 1.8");
585 4 : assert!(d["p"][0].get("valueType").is_none(), "valueType is 1.9");
586 4 : assert!(
587 4 : d["p"][0].get("expiresAt").is_none(),
588 : "attr expiresAt is 1.9"
589 : );
590 4 : assert!(d.get("expiresAt").is_none(), "entity expiresAt is 1.9");
591 4 : assert!(d.get("scope").is_some(), "scope is 1.4");
592 4 : assert_eq!(d["type"], json!(["A", "B"]), "multi-type is 1.3");
593 4 : assert!(d["p"].is_array(), "multi-instance is 1.3");
594 4 : assert_eq!(
595 4 : d["lp"]["type"], "LanguageProperty",
596 : "LanguageProperty is 1.4"
597 : );
598 :
599 4 : let mut d = mk();
600 4 : assert!(amend_entity(&mut d, (1, 3)));
601 4 : assert!(d["p"][0].get("observedAt").is_some(), "observedAt is 1.3");
602 4 : assert!(d["p"][0].get("unitCode").is_some(), "unitCode is 1.3");
603 4 : assert!(d.get("scope").is_none(), "scope only from 1.4");
604 4 : assert!(
605 4 : d["r"].get("objectType").is_none(),
606 : "objectType only from 1.8"
607 : );
608 4 : assert_eq!(
609 4 : d["lp"]["type"], "Property",
610 : "LanguageProperty only from 1.4"
611 : );
612 4 : assert!(d["lp"].get("languageMap").is_none(), "reformatted away");
613 4 : }
614 :
615 : /// Notes 2/3 collapse the datasetId-separated instances of an Entity
616 : /// attribute. The temporal representation's instance array (clause 5.2.5,
617 : /// every instance carrying `instanceId`) is a different shape and is not
618 : /// covered by the table — it must survive intact.
619 : #[test]
620 4 : fn temporal_instance_arrays_are_not_collapsed() {
621 4 : let mut d = json!({"id": "urn:a", "type": "T",
622 4 : "speed": [
623 4 : {"type": "Property", "value": 1, "observedAt": "2020-01-01T00:00:00Z",
624 4 : "instanceId": "urn:ngsi-ld:Instance:1"},
625 4 : {"type": "Property", "value": 2, "observedAt": "2020-01-02T00:00:00Z",
626 4 : "instanceId": "urn:ngsi-ld:Instance:2"}
627 : ]});
628 4 : amend_entity(&mut d, (1, 0));
629 4 : assert_eq!(
630 4 : d["speed"].as_array().map(Vec::len),
631 : Some(2),
632 : "temporal instances must not collapse to one"
633 : );
634 4 : assert!(d["speed"][0].get("observedAt").is_none(), "observedAt <1.3");
635 4 : }
636 :
637 : /// Payload shapes that carry no Entity data are left alone: a document
638 : /// without both `id` and `type`, a scalar, and a Notification with no
639 : /// `data` member.
640 : #[test]
641 4 : fn non_entity_payloads_are_untouched() {
642 4 : let mut v = json!({"title": "BadRequestData", "status": 400});
643 4 : assert!(!amend_payload(&mut v, (1, 0)));
644 4 : let mut v = json!("just a string");
645 4 : assert!(!amend_payload(&mut v, (1, 0)));
646 4 : let mut v = json!({"id": "urn:n", "type": "Notification"});
647 4 : assert!(!amend_payload(&mut v, (1, 0)));
648 4 : let mut v = json!([]);
649 4 : assert!(!amend_payload(&mut v, (1, 0)));
650 : // an already-conformant entity reports no change
651 4 : let mut v = json!({"id": "urn:a", "type": "T", "p": {"type": "Property", "value": 1}});
652 4 : assert!(!amend_payload(&mut v, (1, 0)));
653 4 : }
654 : }
655 :
656 : /// Middleware behaviour of the `Prefer: ngsi-ld=` layer (6.3.6): what it
657 : /// amends, what it must leave byte-identical, and the status it answers with.
658 : #[cfg(test)]
659 : mod prefer_layer {
660 : use super::*;
661 : use axum::http::HeaderValue;
662 : use http_body_util::BodyExt;
663 : use tower::ServiceExt;
664 :
665 36 : fn app(path: &'static str, payload: &'static str) -> axum::Router {
666 36 : axum::Router::new()
667 36 : .route(
668 36 : path,
669 36 : axum::routing::get(move || async move {
670 36 : ([(header::CONTENT_TYPE, "application/json")], payload)
671 72 : }),
672 : )
673 36 : .layer(axum::middleware::from_fn(prefer_version_layer))
674 36 : }
675 :
676 40 : async fn get(
677 40 : app: axum::Router,
678 40 : uri: &str,
679 40 : prefer: Option<&str>,
680 40 : ) -> (StatusCode, String, String) {
681 40 : let mut b = Request::builder().uri(uri);
682 40 : if let Some(p) = prefer {
683 36 : b = b.header("Prefer", p);
684 36 : }
685 40 : let resp = app
686 40 : .oneshot(b.body(Body::empty()).expect("request"))
687 40 : .await
688 40 : .expect("response");
689 40 : let status = resp.status();
690 40 : let applied = resp
691 40 : .headers()
692 40 : .get("Preference-Applied")
693 40 : .and_then(|v| v.to_str().ok())
694 40 : .unwrap_or("")
695 40 : .to_owned();
696 40 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
697 40 : (
698 40 : status,
699 40 : applied,
700 40 : String::from_utf8_lossy(&bytes).into_owned(),
701 40 : )
702 40 : }
703 :
704 : /// The 4.3.6.8 fallbacks describe Entity data. A Subscription (5.2.12)
705 : /// has no version fallbacks: its members — including `expiresAt`, which
706 : /// is a Subscription member since 1.0 — and its string arrays must come
707 : /// back exactly as served, with 200 rather than the 203 that marks an
708 : /// altered Entity.
709 : #[tokio::test(flavor = "multi_thread")]
710 4 : async fn subscription_payload_is_never_amended() {
711 4 : let sub = r#"{"id":"urn:ngsi-ld:Subscription:1","type":"Subscription","expiresAt":"2030-01-01T00:00:00Z","notification":{"attributes":["a","b"],"format":"normalized"}}"#;
712 4 : let (status, applied, body) = get(
713 4 : app("/ngsi-ld/v1/subscriptions/{id}", sub),
714 4 : "/ngsi-ld/v1/subscriptions/urn:ngsi-ld:Subscription:1",
715 4 : Some("ngsi-ld=1.0"),
716 : )
717 4 : .await;
718 4 : assert_eq!(status, StatusCode::OK, "not an altered Entity: {body}");
719 4 : assert_eq!(applied, "ngsi-ld=1.0");
720 4 : assert_eq!(body, sub, "subscription served byte-identical");
721 4 : }
722 :
723 : /// An amended Entity keeps the egress key order (`id`/`type` first) and
724 : /// answers 203; an Entity that needed no amendment stays byte-identical
725 : /// and 200.
726 : #[tokio::test(flavor = "multi_thread")]
727 4 : async fn amended_entity_keeps_key_order_and_reports_203() {
728 4 : let ent = r#"{"id":"urn:a","type":"T","attr":{"type":"JsonProperty","json":{"k":1}}}"#;
729 4 : let (status, applied, body) = get(
730 4 : app("/ngsi-ld/v1/entities/{id}", ent),
731 4 : "/ngsi-ld/v1/entities/urn:a",
732 4 : Some("ngsi-ld=1.4"),
733 : )
734 4 : .await;
735 4 : assert_eq!(status, StatusCode::NON_AUTHORITATIVE_INFORMATION);
736 4 : assert_eq!(applied, "ngsi-ld=1.4");
737 4 : assert!(body.starts_with(r#"{"id":"urn:a","type":"T","#), "{body}");
738 4 : assert!(!body.contains("JsonProperty"), "reformatted away: {body}");
739 4 : assert!(!body.contains("\"json\""), "the 1.8 member is gone: {body}");
740 :
741 : // native/newer preference: nothing to amend, nothing to reorder
742 4 : let (status, applied, body) = get(
743 4 : app("/ngsi-ld/v1/entities/{id}", ent),
744 4 : "/ngsi-ld/v1/entities/urn:a",
745 4 : Some("ngsi-ld=2.0"),
746 : )
747 4 : .await;
748 4 : assert_eq!(status, StatusCode::OK);
749 4 : assert_eq!(applied, "ngsi-ld=1.9", "the version actually conformed to");
750 4 : assert_eq!(body, ent);
751 4 : }
752 :
753 : /// Without the preference the layer is inert, and it never touches a
754 : /// non-JSON body or a non-200 response.
755 : #[tokio::test(flavor = "multi_thread")]
756 4 : async fn layer_is_inert_without_a_usable_preference() {
757 4 : let ent = r#"{"id":"urn:a","type":"T","attr":{"type":"JsonProperty","json":1}}"#;
758 12 : for prefer in [None, Some("body=json"), Some("ngsi-ld=junk")] {
759 12 : let (status, applied, body) = get(
760 12 : app("/ngsi-ld/v1/entities/{id}", ent),
761 12 : "/ngsi-ld/v1/entities/urn:a",
762 12 : prefer,
763 : )
764 12 : .await;
765 12 : assert_eq!(status, StatusCode::OK);
766 12 : assert!(applied.is_empty(), "no preference was applied: {prefer:?}");
767 12 : assert_eq!(body, ent);
768 : }
769 :
770 4 : let app = axum::Router::new()
771 4 : .route(
772 4 : "/ngsi-ld/v1/entities/{id}",
773 4 : axum::routing::get(|| async {
774 4 : (
775 4 : StatusCode::CREATED,
776 4 : [(header::CONTENT_TYPE, "text/plain")],
777 4 : "not json",
778 4 : )
779 8 : }),
780 : )
781 4 : .layer(axum::middleware::from_fn(prefer_version_layer));
782 4 : let (status, applied, body) =
783 4 : get(app, "/ngsi-ld/v1/entities/urn:a", Some("ngsi-ld=1.0")).await;
784 4 : assert_eq!(status, StatusCode::CREATED);
785 4 : assert!(applied.is_empty(), "non-JSON non-200 is passed through");
786 4 : assert_eq!(body, "not json");
787 4 : }
788 :
789 : /// A body that fails mid-stream must not be served as a complete 200.
790 : #[tokio::test(flavor = "multi_thread")]
791 4 : async fn broken_body_stream_is_not_a_success() {
792 4 : let app = axum::Router::new()
793 4 : .route(
794 4 : "/ngsi-ld/v1/entities/{id}",
795 4 : axum::routing::get(|| async {
796 4 : let s = futures_util::stream::iter(vec![Err::<axum::body::Bytes, _>(
797 4 : std::io::Error::other("upstream gone"),
798 4 : )]);
799 4 : Response::builder()
800 4 : .header(header::CONTENT_TYPE, "application/json")
801 4 : .body(Body::from_stream(s))
802 4 : .expect("response")
803 8 : }),
804 : )
805 4 : .layer(axum::middleware::from_fn(prefer_version_layer));
806 4 : let (status, _, body) = get(app, "/ngsi-ld/v1/entities/urn:a", Some("ngsi-ld=1.0")).await;
807 4 : assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
808 4 : assert!(
809 4 : !body.contains("upstream gone"),
810 4 : "the transport error must not reach the client: {body}"
811 4 : );
812 4 : }
813 :
814 : /// RFC 9110 5.3: repeated field lines are equivalent to one
815 : /// comma-separated list, so the version preference is honoured whichever
816 : /// line carries it.
817 : #[tokio::test(flavor = "multi_thread")]
818 4 : async fn version_preference_found_on_a_second_prefer_line() {
819 4 : let ent = r#"{"id":"urn:a","type":"T"}"#;
820 4 : let mut req = Request::builder().uri("/ngsi-ld/v1/entities/urn:a");
821 4 : if let Some(h) = req.headers_mut() {
822 4 : h.append("Prefer", HeaderValue::from_static("body=json"));
823 4 : h.append("Prefer", HeaderValue::from_static("ngsi-ld=1.5"));
824 4 : }
825 4 : let resp = app("/ngsi-ld/v1/entities/{id}", ent)
826 4 : .oneshot(req.body(Body::empty()).expect("request"))
827 4 : .await
828 4 : .expect("response");
829 4 : assert_eq!(
830 4 : resp.headers()
831 4 : .get("Preference-Applied")
832 4 : .and_then(|v| v.to_str().ok()),
833 4 : Some("ngsi-ld=1.5")
834 4 : );
835 4 : }
836 :
837 : /// 6.3.6: a `Prefer: ngsi-ld=` request is owed a `Preference-Applied`
838 : /// header whatever version it names, including the one the broker serves
839 : /// natively. At that version Table 4.3.6.8-1 has nothing to amend —
840 : /// `amend_entity` returns on `ver >= NATIVE` — so the header is all that
841 : /// is owed: the handler's bytes reach the client as they were written,
842 : /// never buffered, parsed and re-serialized on the way.
843 : #[tokio::test(flavor = "multi_thread")]
844 4 : async fn the_native_version_is_stamped_and_the_body_passes_through() {
845 : // key order the egress serializer would rewrite, so a body that comes
846 : // back reordered proves the layer round-tripped it through serde
847 4 : let payload = r#"{"type":"T","id":"urn:a"}"#;
848 8 : for asked in ["ngsi-ld=1.9", "ngsi-ld=2.0"] {
849 8 : let (status, applied, body) = get(app("/e", payload), "/e", Some(asked)).await;
850 8 : assert_eq!(status, StatusCode::OK, "{asked}");
851 8 : assert_eq!(applied, "ngsi-ld=1.9", "{asked}");
852 8 : assert_eq!(body, payload, "{asked}: the body is passed through");
853 4 : }
854 4 : }
855 : }
|
