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