Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Attribute-level operations (5.6.2–5.6.5, 5.6.19; resources 6.6/6.7).
3 :
4 : use crate::federation::path_segment;
5 : use crate::negotiate::*;
6 : use crate::policy::Condition;
7 : use crate::state::{now_iso, AppState};
8 : use antares_jsonld::{expand_entity, ExpandOpts};
9 : use antares_model::{check_attr_name, NgsiError};
10 : use antares_store::CurrentStateDriverExt;
11 : use antares_store::Kind;
12 : use axum::body::Bytes;
13 : use axum::extract::{Path, State};
14 : use axum::http::HeaderMap;
15 : use axum::response::{IntoResponse, Response};
16 : use serde_json::{Map, Value};
17 : use std::collections::HashMap;
18 :
19 : use crate::negotiate::CleanParams;
20 :
21 : /// The fully qualified name of the Entity `scope` member (core @context).
22 : /// The stored document keeps it under the short name, so the attribute
23 : /// operations that address it (5.6.4.4, 5.6.5.4, 5.6.19.4) compare against
24 : /// both spellings.
25 : const SCOPE_IRI: &str = "https://uri.etsi.org/ngsi-ld/scope";
26 :
27 : /// Outcome of a multi-attribute write: 204 when everything applied, else 207
28 : /// with an UpdateResult (5.2.18).
29 13472 : fn update_result(
30 13472 : tenant: &antares_model::TenantId,
31 13472 : updated: Vec<String>,
32 13472 : not_updated: Vec<(String, String)>,
33 13472 : ) -> Response {
34 13472 : if not_updated.is_empty() {
35 13448 : return no_content(tenant);
36 24 : }
37 : // Attribute names are the expanded IRIs the write worked on (5.5.7);
38 : // the Entity core members keep their reserved names `type` and `scope`.
39 24 : let payload = serde_json::json!({
40 24 : "updated": updated,
41 24 : "notUpdated": not_updated
42 24 : .iter()
43 24 : .map(|(a, r)| serde_json::json!({
44 24 : "attributeName": a,
45 24 : "reason": r,
46 : }))
47 24 : .collect::<Vec<_>>(),
48 : });
49 24 : multi_status(payload, tenant)
50 13472 : }
51 :
52 : // ---------- POST /entities/{id}/attrs/ — Append (5.6.3) ----------
53 :
54 234 : pub async fn append_attrs(
55 234 : State(st): State<AppState>,
56 234 : Path(id): Path<String>,
57 234 : CleanParams(params): CleanParams,
58 234 : headers: HeaderMap,
59 234 : body: Bytes,
60 234 : ) -> Response {
61 234 : match append_attrs_inner(&st, &id, ¶ms, &headers, &body).await {
62 184 : Ok(r) => r,
63 50 : Err(e) => e.into_response(),
64 : }
65 234 : }
66 :
67 : /// What separates 5.6.3 Append Attributes from 5.6.2 Update Attributes: the
68 : /// body kind each accepts, whether NGSI-LD Null may appear in it (5.5.4:
69 : /// only the update form carries deletions), and the operation name and
70 : /// method a forward to a registered Context Source carries. Everything else
71 : /// -- id and parameter validation, expansion, the registration plan, the
72 : /// loop guard, which half is served locally, the multi-status assembly --
73 : /// is one operation, written once in `write_attrs`.
74 : struct AttrWrite {
75 : body_kind: BodyKind,
76 : allow_null: bool,
77 : op: &'static str,
78 : method: reqwest::Method,
79 : /// The clause the policy seam is asked about, since one function serves
80 : /// both of these operations.
81 : clause: &'static str,
82 : }
83 :
84 : const APPEND: AttrWrite = AttrWrite {
85 : body_kind: BodyKind::Standard,
86 : allow_null: false,
87 : op: "appendAttrs",
88 : method: reqwest::Method::POST,
89 : clause: "5.6.3",
90 : };
91 :
92 : const UPDATE: AttrWrite = AttrWrite {
93 : body_kind: BodyKind::MergePatch,
94 : allow_null: true,
95 : op: "updateEntity",
96 : method: reqwest::Method::PATCH,
97 : clause: "5.6.2",
98 : };
99 :
100 : /// Why an attribute write never reached the store. `Untouched` is not a
101 : /// failure: 5.6.3.4 leaves an Attribute that may not be overwritten
102 : /// "untouched", and a write that applied nothing modified nothing -- there is
103 : /// no `modifiedAt` to move (4.8) and no change to hand to the store.
104 : enum Unwritten {
105 : Failed(NgsiError),
106 : Untouched,
107 : }
108 :
109 : impl From<NgsiError> for Unwritten {
110 26 : fn from(e: NgsiError) -> Self {
111 26 : Self::Failed(e)
112 26 : }
113 : }
114 :
115 : /// One entity-level attribute write. `merge` is the clause's own algorithm
116 : /// for the Entity Fragment's `scope` and its Attributes; the Entity Type
117 : /// union above it is the same rule in both clauses ("added to the list of
118 : /// Entity Type names of the target Entity").
119 13644 : async fn write_attrs(
120 13644 : st: &AppState,
121 13644 : id: &str,
122 13644 : params: &HashMap<String, String>,
123 13644 : headers: &HeaderMap,
124 13644 : body: &[u8],
125 13644 : mode: &AttrWrite,
126 13644 : merge: impl FnOnce(
127 13644 : &mut Map<String, Value>,
128 13644 : &Map<String, Value>,
129 13644 : &str,
130 13644 : &mut Vec<String>,
131 13644 : &mut Vec<(String, String)>,
132 13644 : ) + Send,
133 13644 : ) -> ApiResult<Response> {
134 13644 : let tenant = tenant_from(headers)?;
135 13644 : antares_model::EntityId::new(id)?;
136 13628 : check_params(params, &["options", "local", "type"])?;
137 13628 : let parsed = parse_body(st, headers, body, mode.body_kind).await?;
138 13620 : let obj = parsed.object(NgsiError::BadRequestData(
139 13620 : "fragment must be a JSON object".into(),
140 13620 : ))?;
141 13620 : let fragment = expand_entity(
142 13620 : obj,
143 13620 : &parsed.ctx,
144 13620 : ExpandOpts {
145 13620 : fragment: true,
146 13620 : allow_null: mode.allow_null,
147 13620 : temporal: false,
148 13620 : ..Default::default()
149 13620 : },
150 4 : )?;
151 13616 : let pre = crate::policy::pre_image(st, &tenant, id, Some(&fragment)).await?;
152 13614 : let granted = gate!(
153 : st, &tenant, headers, mode.clause,
154 : ids: &[id],
155 : types: &pre.types,
156 : stored: pre.stored.as_ref(),
157 : attrs: &attr_iris_of(&fragment),
158 : body: Some(&fragment),
159 : )
160 13614 : .await?;
161 13592 : let cond = Condition::of(&granted);
162 13592 : let (plan, all_attr_iris) =
163 13592 : attr_fed_plan(st, &tenant, id, &fragment, &parsed.ctx, params, headers).await?;
164 13592 : let regs = match plan {
165 4 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
166 13588 : crate::federation::WritePlan::Forward(regs) => regs,
167 : };
168 13588 : Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
169 13586 : let local_covered = proxies_cover_all(®s, &all_attr_iris);
170 13586 : let fragment = crate::federation::strip_covered_expanded(&fragment, ®s);
171 13586 : let local_iris = attr_iris_of(&fragment);
172 13586 : let ts = now_iso();
173 13586 : let mut updated = Vec::new();
174 13586 : let mut not_updated = Vec::new();
175 : // The Entity's change, judged and applied on the document it lands on:
176 : // inside the store's read-modify-write, or on a copy for a relay.
177 13586 : let apply = |doc: &mut Value| {
178 13498 : crate::history::restage(id);
179 13498 : Condition::pre(cond.as_ref(), id, doc)?;
180 : // 5.6.2.4 / 5.6.3.4: the ?type selector narrows the target — a
181 : // mismatch means the entity is not known for this operation.
182 13488 : if !matches_type_param(doc, params, &parsed.ctx) {
183 16 : return Err(NgsiError::ResourceNotFound(format!(
184 16 : "entity {id} does not match the type selector"
185 16 : ))
186 16 : .into());
187 13472 : }
188 13472 : let target = antares_store::stored_object(doc)?;
189 13472 : let frag = antares_jsonld::expanded_object(&fragment)?;
190 : // 5.6.2.4 / 5.6.3.4: Entity Type names not yet in the target are
191 : // added to its list
192 13472 : if let Some(new_types) = frag.get("type").and_then(Value::as_array) {
193 18 : let mut cur: Vec<Value> = target
194 18 : .get("type")
195 18 : .and_then(Value::as_array)
196 18 : .cloned()
197 18 : .unwrap_or_default();
198 18 : let known = cur.len();
199 26 : for t in new_types {
200 26 : if !cur.contains(t) {
201 16 : cur.push(t.clone());
202 16 : }
203 : }
204 18 : if cur.len() > known {
205 12 : target.insert("type".into(), Value::Array(cur));
206 12 : updated.push("type".into());
207 12 : }
208 13454 : }
209 13472 : merge(target, frag, &ts, &mut updated, &mut not_updated);
210 13472 : if updated.is_empty() {
211 18 : return Err(Unwritten::Untouched);
212 13454 : }
213 13454 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
214 13454 : Ok::<(), Unwritten>(())
215 13498 : };
216 13586 : let local_resp: Option<ApiResult<Response>> = if local_covered {
217 50 : None
218 : } else {
219 13536 : let res = st.store.mutate(&tenant, Kind::Entity, id, apply).await?;
220 13536 : crate::history::settle(id, matches!(res, Some(Ok(()))));
221 44 : Some(match res {
222 38 : None => Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into()),
223 26 : Some(Err(Unwritten::Failed(e))) => Err(e.into()),
224 : Some(Ok(())) | Some(Err(Unwritten::Untouched)) => {
225 13472 : Ok(update_result(&tenant, updated.clone(), not_updated.clone()))
226 : }
227 : })
228 : };
229 13586 : if regs.is_empty() {
230 : // the local half is skipped only while registrations cover the
231 : // attributes, so an empty list always leaves a local response
232 13508 : return local_resp
233 13508 : .unwrap_or_else(|| Err(NgsiError::InternalError("no local result".into()).into()));
234 78 : }
235 78 : let local_outcome = classify_local(&local_resp);
236 78 : let query = fwd_query(params, &["options", "type"]);
237 78 : let fed_parts = crate::federation::fed_attr_parts(
238 78 : st,
239 78 : headers,
240 78 : &tenant,
241 78 : &parsed.ctx,
242 78 : ®s,
243 78 : mode.op,
244 78 : mode.method.clone(),
245 78 : &format!("/entities/{}/attrs/", path_segment(id)),
246 78 : &query,
247 78 : Some(Value::Object(without_context_map(obj))),
248 78 : true,
249 78 : )
250 78 : .await;
251 78 : Ok(combine_attr_parts(
252 78 : &tenant,
253 78 : &all_attr_iris,
254 78 : &local_iris,
255 78 : local_outcome,
256 78 : updated,
257 78 : not_updated.into_iter().map(|(a, r)| (a, r, None)).collect(),
258 78 : ®s,
259 78 : &fed_parts,
260 : ))
261 13644 : }
262 :
263 : /// 5.6.3.4 Append Attributes: an Attribute the target does not have is
264 : /// appended; one it has is replaced datasetId-wise unless `noOverwrite` was
265 : /// asked for, in which case it is reported in `notUpdated`.
266 234 : async fn append_attrs_inner(
267 234 : st: &AppState,
268 234 : id: &str,
269 234 : params: &HashMap<String, String>,
270 234 : headers: &HeaderMap,
271 234 : body: &[u8],
272 234 : ) -> ApiResult<Response> {
273 234 : let no_overwrite = params
274 234 : .get("options")
275 234 : .is_some_and(|o| o.split(',').any(|s| s.trim() == "noOverwrite"));
276 234 : write_attrs(
277 234 : st,
278 234 : id,
279 234 : params,
280 234 : headers,
281 234 : body,
282 234 : &APPEND,
283 162 : |target, frag, ts, updated, not_updated| {
284 : // The Entity as a backfill names it.
285 162 : let shell = serde_json::json!({"id": target.get("id"), "type": target.get("type")});
286 : // appended scope: overwrite replaces; noOverwrite unions (010_07)
287 162 : if let Some(new_scope) = frag.get("scope") {
288 8 : if target.contains_key("scope") && no_overwrite {
289 4 : let mut cur: Vec<Value> = target
290 4 : .get("scope")
291 4 : .and_then(Value::as_array)
292 4 : .cloned()
293 4 : .unwrap_or_default();
294 4 : let known = cur.len();
295 4 : for sc in new_scope.as_array().cloned().unwrap_or_default() {
296 4 : if !cur.contains(&sc) {
297 4 : cur.push(sc);
298 4 : }
299 : }
300 4 : if cur.len() > known {
301 4 : target.insert("scope".into(), Value::Array(cur));
302 4 : updated.push("scope".into());
303 4 : }
304 4 : } else {
305 4 : target.insert("scope".into(), new_scope.clone());
306 4 : updated.push("scope".into());
307 4 : }
308 154 : }
309 190 : for (k, v) in frag {
310 190 : if is_fragment_meta(k) {
311 10 : continue;
312 180 : }
313 180 : let mut incoming = v.clone();
314 180 : crate::stamp::stamp_instances(&mut incoming, ts);
315 180 : match target.get_mut(k) {
316 132 : None => {
317 132 : target.insert(k.clone(), incoming);
318 132 : updated.push(k.clone());
319 132 : }
320 48 : Some(existing) => {
321 48 : let merged = merge_instance_sets(
322 48 : existing,
323 48 : &incoming,
324 48 : no_overwrite,
325 48 : Some((&shell, k)),
326 : );
327 48 : if merged {
328 28 : updated.push(k.clone());
329 28 : } else {
330 20 : not_updated
331 20 : .push((k.clone(), "attribute already exists (noOverwrite)".into()));
332 20 : }
333 : }
334 : }
335 : }
336 162 : },
337 : )
338 234 : .await
339 234 : }
340 :
341 : /// 5.6.2.4 Update Attributes: every Attribute of the Fragment is applied —
342 : /// an unknown one is appended silently (011_01_03), an NGSI-LD Null deletes
343 : /// the instance it matches (5.5.8), and an Attribute left with no instances
344 : /// leaves the Entity.
345 13410 : async fn update_attrs_inner(
346 13410 : st: &AppState,
347 13410 : id: &str,
348 13410 : params: &HashMap<String, String>,
349 13410 : headers: &HeaderMap,
350 13410 : body: &[u8],
351 13410 : ) -> ApiResult<Response> {
352 13410 : write_attrs(
353 13410 : st,
354 13410 : id,
355 13410 : params,
356 13410 : headers,
357 13410 : body,
358 13410 : &UPDATE,
359 13310 : |target, frag, ts, updated, not_updated| {
360 : // The Entity as a backfill names it.
361 13310 : let shell = serde_json::json!({"id": target.get("id"), "type": target.get("type")});
362 : // 5.6.2: scope updates only when the entity already has one
363 13310 : if let Some(new_scope) = frag.get("scope") {
364 10 : if !target.contains_key("scope") {
365 4 : not_updated.push(("scope".into(), "entity has no scope".into()));
366 6 : } else if antares_jsonld::is_ngsi_null_list(new_scope) {
367 2 : // 5.5.12: the NGSI-LD Null removes the member, as on merge
368 2 : target.remove("scope");
369 2 : updated.push("scope".into());
370 4 : } else {
371 4 : target.insert("scope".into(), new_scope.clone());
372 4 : updated.push("scope".into());
373 4 : }
374 13300 : }
375 51334 : for (k, v) in frag {
376 51334 : if is_fragment_meta(k) {
377 26 : continue;
378 51308 : }
379 51308 : let mut incoming = v.clone();
380 51308 : crate::stamp::stamp_instances(&mut incoming, ts);
381 51308 : match target.get_mut(k) {
382 : // 5.6.2 + 011_01_03: unknown attributes are appended silently
383 : None => {
384 58 : let live: Vec<Value> = incoming
385 58 : .as_array()
386 58 : .cloned()
387 58 : .unwrap_or_default()
388 58 : .into_iter()
389 58 : .filter(|i| !antares_jsonld::is_deletion_instance(i))
390 58 : .collect();
391 58 : if !live.is_empty() {
392 58 : target.insert(k.clone(), Value::Array(live));
393 58 : }
394 58 : updated.push(k.clone());
395 : }
396 51250 : Some(existing) => {
397 51250 : merge_instance_sets(existing, &incoming, false, Some((&shell, k)));
398 51250 : if existing.as_array().is_some_and(Vec::is_empty) {
399 32 : target.remove(k);
400 51218 : }
401 51250 : updated.push(k.clone());
402 : }
403 : }
404 : }
405 13310 : },
406 : )
407 13410 : .await
408 13410 : }
409 :
410 : /// The Entity Fragment members that are not Attributes: handled by the
411 : /// write itself (`id`, `type`, `scope`) or server-generated (4.8).
412 51524 : fn is_fragment_meta(k: &str) -> bool {
413 51524 : matches!(k, "id" | "type" | "scope" | "createdAt" | "modifiedAt")
414 51524 : }
415 :
416 : /// Shared federation plan for attribute writes: the matching non-aux
417 : /// registrations, plus the touched attribute IRIs they were matched on.
418 13592 : async fn attr_fed_plan(
419 13592 : st: &AppState,
420 13592 : tenant: &antares_model::TenantId,
421 13592 : id: &str,
422 13592 : fragment: &Value,
423 13592 : ctx: &antares_jsonld::Context,
424 13592 : params: &HashMap<String, String>,
425 13592 : headers: &axum::http::HeaderMap,
426 13592 : ) -> Result<(crate::federation::WritePlan, Vec<String>), NgsiError> {
427 13592 : let attr_iris = attr_iris_of(fragment);
428 13592 : let plan = attr_fed_plan_iris(st, tenant, id, &attr_iris, ctx, params, headers).await?;
429 13592 : Ok((plan, attr_iris))
430 13592 : }
431 :
432 14238 : async fn attr_fed_plan_iris(
433 14238 : st: &AppState,
434 14238 : tenant: &antares_model::TenantId,
435 14238 : id: &str,
436 14238 : attr_iris: &[String],
437 14238 : ctx: &antares_jsonld::Context,
438 14238 : params: &HashMap<String, String>,
439 14238 : headers: &axum::http::HeaderMap,
440 14238 : ) -> Result<crate::federation::WritePlan, NgsiError> {
441 14238 : let spec = crate::registry::CsrSpec {
442 14238 : ids: Some(vec![id.to_owned()]),
443 14238 : attrs: (!attr_iris.is_empty()).then(|| attr_iris.to_vec()),
444 14238 : ..Default::default()
445 : };
446 14238 : crate::federation::write_plan(st, tenant, &spec, ctx, params, headers).await
447 14238 : }
448 :
449 : /// The registrations an attribute-level operation has to consider, with the
450 : /// 6.3.18 loop guard already applied. A returned response is the loop
451 : /// chain's own answer (Table 6.3.18-2) and ends the operation before
452 : /// anything local happens.
453 646 : async fn attr_regs_or_loop(
454 646 : st: &AppState,
455 646 : tenant: &antares_model::TenantId,
456 646 : id: &str,
457 646 : attr_iri: &str,
458 646 : ctx: &antares_jsonld::Context,
459 646 : params: &HashMap<String, String>,
460 646 : headers: &HeaderMap,
461 646 : ) -> Result<(Vec<crate::federation::FedReg>, Option<Response>), NgsiError> {
462 : Ok(
463 646 : match attr_fed_plan_iris(st, tenant, id, &[attr_iri.to_owned()], ctx, params, headers)
464 646 : .await?
465 : {
466 4 : crate::federation::WritePlan::Answered(r) => (Vec::new(), Some(*r)),
467 642 : crate::federation::WritePlan::Forward(regs) => (regs, None),
468 : },
469 : )
470 646 : }
471 :
472 : /// The local half of a distributed attribute write is skipped only when every
473 : /// touched attribute is held by an exclusive/redirect registration (5.6.2.4
474 : /// and siblings). The decision is taken AFTER 6.3.18 loop handling: a `Via`
475 : /// chain naming this broker removes registrations from matching (Table
476 : /// 6.3.18-2), and the operation then has to be served locally.
477 14228 : fn proxies_cover_all(regs: &[crate::federation::FedReg], attr_iris: &[String]) -> bool {
478 14228 : !regs.is_empty()
479 166 : && !attr_iris.is_empty()
480 166 : && attr_iris
481 166 : .iter()
482 188 : .all(|a| regs.iter().any(|r| r.is_proxy() && r.covers_attr(a)))
483 14228 : }
484 :
485 : /// The URL parameters that travel with a forwarded attribute operation.
486 : /// `type` narrows the target Entity for the registered source exactly as it
487 : /// does locally — 5.6.5.4 identifies the target by its "id (URI), and where
488 : /// specified type", and the parameter is shall-support on each of these
489 : /// resources (Tables 6.6.3.1-1, 6.6.3.2-1, 6.7.3.1-1, 6.7.3.2-1, 6.7.3.3-1).
490 : /// `local` never travels: Table 6.3.18-1 scopes it to the receiving broker.
491 166 : fn fwd_query(params: &HashMap<String, String>, keys: &[&str]) -> Vec<(String, String)> {
492 166 : keys.iter()
493 328 : .filter_map(|k| params.get(*k).map(|v| ((*k).to_owned(), v.clone())))
494 166 : .collect()
495 166 : }
496 :
497 : /// Attribute IRIs of an expanded fragment: every key but the Entity's own
498 : /// members (5.2.4, [`antares_model::is_meta`]), so `expiresAt` (4.22) is no
499 : /// Attribute name. A client `deletedAt` never reaches here: expansion drops
500 : /// it (5.2.2 "shall ignore").
501 27338 : pub(crate) fn attr_iris_of(fragment: &Value) -> Vec<String> {
502 27338 : fragment
503 27338 : .as_object()
504 27338 : .map(|o| {
505 27338 : o.keys()
506 103498 : .filter(|k| !antares_model::is_meta(k))
507 27338 : .cloned()
508 27338 : .collect()
509 27338 : })
510 27338 : .unwrap_or_default()
511 27338 : }
512 :
513 : /// Fold the buffered local response into a `LocalOutcome` without losing the
514 : /// ProblemDetails reason.
515 166 : fn classify_local(resp: &Option<ApiResult<Response>>) -> LocalOutcome {
516 12 : match resp {
517 122 : None => LocalOutcome::Skipped,
518 32 : Some(Ok(_)) => LocalOutcome::Ok,
519 12 : Some(Err(ApiError::Ngsi(e))) => {
520 12 : let pd = e.to_problem_details();
521 12 : if pd.status == 404 {
522 12 : LocalOutcome::NotFound(pd.detail)
523 : } else {
524 0 : LocalOutcome::Failed(pd.detail)
525 : }
526 : }
527 0 : Some(Err(_)) => LocalOutcome::Failed("local operation failed".into()),
528 : }
529 166 : }
530 :
531 : /// How the LOCAL half of a distributed /attrs operation ended.
532 : enum LocalOutcome {
533 : /// proxies cover every touched attribute — no local write attempted
534 : Skipped,
535 : /// entity (or the addressed attribute) unknown locally
536 : NotFound(String),
537 : /// local write failed for another reason
538 : Failed(String),
539 : Ok,
540 : }
541 :
542 : /// 6.3.17: a distributed /attrs operation answers 204, 404, or **207 with an
543 : /// UpdateResult** (Tables 6.6.3.1-2, 6.6.3.2-2, 6.7.3.1-2, 6.7.3.2-2,
544 : /// 6.7.3.3-2; 5.2.18 applies "regardless of whether local or distributed")
545 : /// — never the batch {success, errors} shape. Per-registration failures are
546 : /// listed per covered attribute with `registrationId` (5.2.19).
547 : #[allow(clippy::too_many_arguments)] // one param per input of the 6.3.17 answer
548 206 : fn combine_attr_parts(
549 206 : tenant: &antares_model::TenantId,
550 206 : attr_iris: &[String],
551 206 : local_iris: &[String],
552 206 : local: LocalOutcome,
553 206 : mut updated: Vec<String>,
554 206 : mut not_updated: Vec<(String, String, Option<String>)>,
555 206 : regs: &[crate::federation::FedReg],
556 206 : fed_parts: &[crate::federation::Part],
557 206 : ) -> Response {
558 206 : match &local {
559 16 : LocalOutcome::NotFound(d) | LocalOutcome::Failed(d) => {
560 16 : for a in local_iris {
561 16 : not_updated.push((a.clone(), d.clone(), None));
562 16 : }
563 : }
564 190 : _ => {}
565 : }
566 206 : let mut any_fed_ok = false;
567 240 : for (reg, part) in regs.iter().zip(fed_parts) {
568 : // status 0: inclusive registration skipped for lack of operation
569 : // support (5.6.2.4) — neither a success nor a failure branch.
570 240 : if part.status == 0 {
571 2 : continue;
572 238 : }
573 284 : let covered: Vec<&String> = attr_iris.iter().filter(|a| reg.covers_attr(a)).collect();
574 : // 5.2.18: a source's own 207 UpdateResult names what it "appended
575 : // or updated"; only the Attributes this registration covers count
576 : // (4.3.6.1), each listed once.
577 238 : if let Some(u) = &part.update {
578 6 : for a in covered {
579 6 : if u.updated.contains(a) {
580 4 : any_fed_ok = true;
581 4 : if !updated.contains(a) {
582 4 : updated.push(a.clone());
583 4 : }
584 : } else {
585 2 : let reason = u
586 2 : .not_updated
587 2 : .iter()
588 2 : .find(|(n, _)| n == a)
589 2 : .map_or_else(|| part.detail.clone(), |(_, r)| r.clone());
590 2 : not_updated.push((a.clone(), reason, Some(reg.reg_id.clone())));
591 : }
592 : }
593 4 : continue;
594 234 : }
595 234 : if part.ok() {
596 176 : any_fed_ok = true;
597 188 : for a in covered {
598 188 : if !updated.iter().any(|u| u == a.as_str()) {
599 136 : updated.push(a.clone());
600 136 : }
601 : }
602 : } else {
603 70 : for a in covered {
604 70 : not_updated.push((a.clone(), part.detail.clone(), Some(reg.reg_id.clone())));
605 70 : }
606 : }
607 : }
608 : // 6.3.17: "In the case of an exclusive or redirect registration, where
609 : // all of the data is held outside of the Context Broker and held in a
610 : // single registered source ... 508 Loop Detected" — a proxy part's loop
611 : // verdict passes through instead of dissolving into 207/404.
612 206 : if updated.is_empty()
613 42 : && !any_fed_ok
614 42 : && !matches!(local, LocalOutcome::Ok)
615 38 : && regs
616 38 : .iter()
617 38 : .zip(fed_parts)
618 42 : .any(|(r, p)| r.is_proxy() && p.status == 508)
619 : {
620 0 : return crate::federation::loop_508(tenant);
621 206 : }
622 : // nothing was found anywhere → 404 ProblemDetails (6.6/6.7 tables)
623 206 : if matches!(&local, LocalOutcome::NotFound(_)) && !any_fed_ok && updated.is_empty() {
624 4 : if let LocalOutcome::NotFound(d) = local {
625 4 : return ApiError::from(NgsiError::ResourceNotFound(d)).into_response();
626 0 : }
627 202 : }
628 : // 5.6.2.4: unsupported operation on a proxy registration is "an error of
629 : // type Conflict if the complete update failed" — nothing updated
630 : // anywhere and at least one registration refused the operation.
631 202 : if updated.is_empty()
632 38 : && !matches!(local, LocalOutcome::Ok)
633 34 : && !any_fed_ok
634 34 : && fed_parts.iter().any(|p| p.conflict)
635 : {
636 6 : let mut resp = (
637 6 : axum::http::StatusCode::CONFLICT,
638 6 : [(axum::http::header::CONTENT_TYPE, "application/json")],
639 6 : axum::Json(serde_json::json!({
640 6 : "type": "https://uri.etsi.org/ngsi-ld/errors/Conflict",
641 6 : "title": "Conflict",
642 6 : "detail": "registration does not accept the operation",
643 6 : "status": 409,
644 6 : })),
645 6 : )
646 6 : .into_response();
647 6 : crate::negotiate::echo_tenant(tenant, &mut resp);
648 6 : return resp;
649 196 : }
650 : // 6.3.17: "In the case of an exclusive or redirect registration, where all
651 : // of the data is held outside of the Context Broker and held in a single
652 : // registered source, the following errors shall be returned: 508 Loop
653 : // Detected … 504 Gateway Timeout … 404 Not Found … 502 Bad Gateway."
654 : // 207 Multi Status is the answer for an entity distributed over multiple
655 : // endpoints, so a lone proxied source's failure passes through as itself.
656 196 : if let ([reg], [part]) = (regs, fed_parts) {
657 162 : if reg.is_proxy()
658 120 : && !part.ok()
659 28 : && part.update.is_none()
660 28 : && matches!(local, LocalOutcome::Skipped)
661 : {
662 24 : return crate::federation::part_error(part, tenant);
663 138 : }
664 34 : }
665 172 : if not_updated.is_empty() {
666 138 : return no_content(tenant);
667 34 : }
668 34 : let nu: Vec<Value> = not_updated
669 34 : .into_iter()
670 58 : .map(|(a, r, reg_id)| {
671 58 : let mut m = Map::new();
672 58 : m.insert("attributeName".into(), Value::String(a));
673 58 : m.insert("reason".into(), Value::String(r));
674 58 : if let Some(rid) = reg_id.filter(|r| !r.is_empty()) {
675 38 : m.insert("registrationId".into(), Value::String(rid));
676 38 : }
677 58 : Value::Object(m)
678 58 : })
679 34 : .collect();
680 34 : multi_status(
681 34 : serde_json::json!({"updated": updated, "notUpdated": nu}),
682 34 : tenant,
683 : )
684 206 : }
685 :
686 : /// The local half's answer for an operation that addresses exactly ONE
687 : /// Attribute (5.6.4 Partial Update, 5.6.5 Delete, 5.6.19 Replace): per
688 : /// Table 6.3.2-1 the Entity may be absent (404 naming the Entity), present
689 : /// without the addressed Attribute (404 naming the Attribute), or edited
690 : /// (204). `found` is what the edit itself reports.
691 570 : fn single_attr_local(
692 570 : res: Option<Result<(), NgsiError>>,
693 570 : found: bool,
694 570 : id: &str,
695 570 : attr: &str,
696 570 : tenant: &antares_model::TenantId,
697 570 : ) -> ApiResult<Response> {
698 462 : match res {
699 60 : None => Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into()),
700 48 : Some(Err(e)) => Err(e.into()),
701 416 : Some(Ok(())) if found => Ok(no_content(tenant)),
702 : Some(Ok(())) => {
703 46 : Err(NgsiError::ResourceNotFound(format!("attribute {attr} not found")).into())
704 : }
705 : }
706 570 : }
707 :
708 : /// The multi-status assembly for the same three operations: one Attribute is
709 : /// addressed, so the local and distributed halves are combined over a
710 : /// one-element attribute list, and the Attribute counts as updated only when
711 : /// the local half actually applied it. `local_covered` means an
712 : /// exclusive/redirect registration holds it and there was no local half.
713 88 : fn combine_single_attr(
714 88 : tenant: &antares_model::TenantId,
715 88 : attr_iri: &str,
716 88 : local_covered: bool,
717 88 : local_resp: &Option<ApiResult<Response>>,
718 88 : regs: &[crate::federation::FedReg],
719 88 : fed_parts: &[crate::federation::Part],
720 88 : ) -> Response {
721 88 : let local_outcome = classify_local(local_resp);
722 88 : let all_attr_iris = vec![attr_iri.to_owned()];
723 88 : let local_iris = if local_covered {
724 72 : Vec::new()
725 : } else {
726 16 : all_attr_iris.clone()
727 : };
728 88 : let updated = if matches!(local_outcome, LocalOutcome::Ok) {
729 8 : all_attr_iris.clone()
730 : } else {
731 80 : Vec::new()
732 : };
733 88 : combine_attr_parts(
734 88 : tenant,
735 88 : &all_attr_iris,
736 88 : &local_iris,
737 88 : local_outcome,
738 88 : updated,
739 88 : Vec::new(),
740 88 : regs,
741 88 : fed_parts,
742 : )
743 88 : }
744 :
745 : /// Merge incoming instances into an existing instance array by datasetId.
746 : /// Deletion-marker instances (urn:ngsi-ld:null) remove the matched instance.
747 : /// Returns false when nothing was applied (noOverwrite and all existed).
748 : ///
749 : /// `backfill` names the Entity (its id and type) and the attribute: an
750 : /// instance older than the one it would replace goes to history only then
751 : /// ([`crate::history::backfilled`]).
752 51310 : fn merge_instance_sets(
753 51310 : existing: &mut Value,
754 51310 : incoming: &Value,
755 51310 : no_overwrite: bool,
756 51310 : backfill: Option<(&Value, &str)>,
757 51310 : ) -> bool {
758 51310 : let (Some(cur), Some(inc)) = (existing.as_array_mut(), incoming.as_array()) else {
759 0 : return false;
760 : };
761 51310 : let mut any = false;
762 51318 : for ni in inc {
763 51318 : let ds = ni.get("datasetId").and_then(Value::as_str);
764 51318 : let pos = cur
765 51318 : .iter()
766 51362 : .position(|ci| ci.get("datasetId").and_then(Value::as_str) == ds);
767 51318 : if antares_jsonld::is_deletion_instance(ni) {
768 68 : if let Some(p) = pos {
769 68 : cur.remove(p);
770 68 : any = true;
771 68 : }
772 68 : continue;
773 51250 : }
774 51250 : match pos {
775 51226 : Some(p) => {
776 51226 : let older = backfill
777 51226 : .zip(cur.get(p))
778 51226 : .is_some_and(|((doc, attr), current)| {
779 51218 : crate::history::backfilled(doc, attr, current, ni)
780 51218 : });
781 51226 : if older {
782 4 : any = true;
783 51222 : } else if !no_overwrite {
784 : // keep original createdAt
785 51198 : if let Some(slot) = cur.get_mut(p) {
786 51198 : let created = slot.get("createdAt").cloned();
787 51198 : *slot = ni.clone();
788 51198 : if let (Some(o), Some(c)) = (slot.as_object_mut(), created) {
789 51198 : o.insert("createdAt".into(), c);
790 51198 : }
791 51198 : any = true;
792 0 : }
793 24 : }
794 : }
795 24 : None => {
796 24 : cur.push(ni.clone());
797 24 : any = true;
798 24 : }
799 : }
800 : }
801 51310 : any
802 51310 : }
803 :
804 : // ---------- PATCH /entities/{id}/attrs/ — Update (5.6.2) ----------
805 :
806 13410 : pub async fn update_attrs(
807 13410 : State(st): State<AppState>,
808 13410 : Path(id): Path<String>,
809 13410 : CleanParams(params): CleanParams,
810 13410 : headers: HeaderMap,
811 13410 : body: Bytes,
812 13410 : ) -> Response {
813 13410 : match update_attrs_inner(&st, &id, ¶ms, &headers, &body).await {
814 13346 : Ok(r) => r,
815 64 : Err(e) => e.into_response(),
816 : }
817 13410 : }
818 :
819 : // ---------- PATCH /entities/{id}/attrs/{attrId} — Partial update (5.6.4) ----------
820 :
821 298 : pub async fn partial_update_attr(
822 298 : State(st): State<AppState>,
823 298 : Path((id, attr)): Path<(String, String)>,
824 298 : CleanParams(params): CleanParams,
825 298 : headers: HeaderMap,
826 298 : body: Bytes,
827 298 : ) -> Response {
828 298 : match partial_update_inner(&st, &id, &attr, ¶ms, &headers, &body).await {
829 192 : Ok(r) => r,
830 106 : Err(e) => e.into_response(),
831 : }
832 298 : }
833 :
834 298 : async fn partial_update_inner(
835 298 : st: &AppState,
836 298 : id: &str,
837 298 : attr: &str,
838 298 : params: &HashMap<String, String>,
839 298 : headers: &HeaderMap,
840 298 : body: &[u8],
841 298 : ) -> ApiResult<Response> {
842 298 : let tenant = tenant_from(headers)?;
843 298 : antares_model::EntityId::new(id)?;
844 290 : check_attr_name(attr)?;
845 274 : check_params(params, &["local", "type"])?;
846 274 : let parsed = parse_body(st, headers, body, BodyKind::MergePatch).await?;
847 266 : let obj = parsed.object(NgsiError::BadRequestData(
848 266 : "fragment must be a JSON object".into(),
849 266 : ))?;
850 266 : let frag_inst = antares_jsonld::expand_attr_fragment(obj, &parsed.ctx)?;
851 : // 5.6.4.4: "Apply term expansion as mandated by clause 5.5.7, so that
852 : // the fully qualified name (URI) associated to the target Attribute is
853 : // properly obtained" — a path name with no fully qualified name is not a
854 : // valid Attribute name, and lands on a member of the stored document
855 : // that is not an Attribute.
856 252 : let attr_iri = antares_jsonld::expand_attr_name(attr, &parsed.ctx)?;
857 250 : let pre = crate::policy::pre_image(st, &tenant, id, None).await?;
858 250 : let granted = gate!(
859 : st, &tenant, headers, "5.6.4",
860 : ids: &[id],
861 : types: &pre.types,
862 : stored: pre.stored.as_ref(),
863 : attrs: std::slice::from_ref(&attr_iri),
864 : body: Some(&frag_inst),
865 : )
866 250 : .await?;
867 248 : let cond = Condition::of(&granted);
868 : // 5.6.4.4: "If the target Attribute is scope, then an error of type
869 : // BadRequestData shall be raised."
870 248 : if attr == "scope" || attr_iri == SCOPE_IRI {
871 8 : return Err(NgsiError::BadRequestData(
872 8 : "scope cannot be the target of a partial attribute update (5.6.4)".into(),
873 8 : )
874 8 : .into());
875 240 : }
876 240 : let (regs, loop_answer) =
877 240 : attr_regs_or_loop(st, &tenant, id, &attr_iri, &parsed.ctx, params, headers).await?;
878 240 : if let Some(r) = loop_answer {
879 0 : return Ok(r);
880 240 : }
881 240 : Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
882 240 : let local_covered = proxies_cover_all(®s, std::slice::from_ref(&attr_iri));
883 240 : let want_ds = frag_inst
884 240 : .get("datasetId")
885 240 : .and_then(Value::as_str)
886 240 : .map(String::from);
887 240 : let is_deletion = antares_jsonld::is_deletion_instance(&frag_inst);
888 240 : let ts = now_iso();
889 240 : let mut found = false;
890 240 : let local_resp: Option<ApiResult<Response>> = if local_covered {
891 22 : None
892 : } else {
893 218 : let res = st
894 218 : .store
895 218 : .mutate(&tenant, Kind::Entity, id, |doc| {
896 202 : Condition::pre(cond.as_ref(), id, doc)?;
897 : // 5.6.4.4: the ?type selector narrows the target entity
898 198 : if !matches_type_param(doc, params, &parsed.ctx) {
899 8 : return Err(NgsiError::ResourceNotFound(format!(
900 8 : "entity {id} does not match the type selector"
901 8 : )));
902 190 : }
903 190 : let target = antares_store::stored_object(doc)?;
904 190 : if let Some(existing) = target.get_mut(&attr_iri).and_then(Value::as_array_mut) {
905 208 : let pos = existing.iter().position(|ci| {
906 208 : ci.get("datasetId").and_then(Value::as_str) == want_ds.as_deref()
907 208 : });
908 184 : if let Some(p) = pos {
909 172 : found = true;
910 172 : if is_deletion {
911 60 : existing.remove(p);
912 112 : } else if let Some(slot) = existing.get_mut(p) {
913 : // 5.6.4.4: the fragment may not change the Attribute type
914 78 : if let (Some(ft), Some(et)) = (
915 112 : frag_inst.get("type").and_then(Value::as_str),
916 112 : slot.get("type").and_then(Value::as_str),
917 : ) {
918 78 : if ft != et {
919 4 : return Err(NgsiError::BadRequestData(format!(
920 4 : "attribute type mismatch: {ft} != {et} (5.6.4)"
921 4 : )));
922 74 : }
923 34 : }
924 108 : let t = antares_store::stored_object(slot)?;
925 226 : for (k, v) in antares_jsonld::expanded_object(&frag_inst)? {
926 226 : if matches!(k.as_str(), "createdAt" | "modifiedAt") {
927 0 : continue;
928 226 : }
929 : // a member set to the Null, or a sub-Attribute whose every
930 : // instance is a deletion, leaves the instance (10.2.5.4)
931 226 : if antares_jsonld::removes_member(v) {
932 4 : t.remove(k);
933 222 : } else {
934 222 : t.insert(k.clone(), v.clone());
935 222 : }
936 : }
937 : // 4.7.1: a GeoProperty's Value "shall be a GeoJSON
938 : // Geometry"; the fragment carries no type, so its value
939 : // is checked here, against the type it lands on.
940 108 : if t.get("type").and_then(Value::as_str) == Some("GeoProperty") {
941 4 : if let Some(g) = t.get("value") {
942 4 : antares_jsonld::expand::validate_geojson(attr, g)?;
943 0 : }
944 104 : }
945 106 : t.insert("modifiedAt".into(), Value::String(ts.clone()));
946 0 : }
947 12 : }
948 178 : if existing.is_empty() {
949 28 : target.remove(&attr_iri);
950 150 : }
951 6 : }
952 184 : if found {
953 166 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
954 166 : }
955 184 : Ok::<(), NgsiError>(())
956 202 : })
957 218 : .await?;
958 218 : Some(single_attr_local(res, found, id, attr, &tenant))
959 : };
960 240 : if regs.is_empty() {
961 : // the local half is skipped only while registrations cover the
962 : // attributes, so an empty list always leaves a local response
963 214 : return local_resp
964 214 : .unwrap_or_else(|| Err(NgsiError::InternalError("no local result".into()).into()));
965 26 : }
966 26 : let fed_parts = crate::federation::fed_attr_parts(
967 26 : st,
968 26 : headers,
969 26 : &tenant,
970 26 : &parsed.ctx,
971 26 : ®s,
972 26 : "updateAttrs",
973 26 : reqwest::Method::PATCH,
974 26 : &format!(
975 26 : "/entities/{}/attrs/{}",
976 26 : path_segment(id),
977 26 : path_segment(attr)
978 26 : ),
979 26 : &fwd_query(params, &["type"]),
980 26 : Some(Value::Object(without_context_map(obj))),
981 26 : false,
982 26 : )
983 26 : .await;
984 26 : Ok(combine_single_attr(
985 26 : &tenant,
986 26 : &attr_iri,
987 26 : local_covered,
988 26 : &local_resp,
989 26 : ®s,
990 26 : &fed_parts,
991 26 : ))
992 298 : }
993 :
994 238 : fn without_context_map(o: &Map<String, Value>) -> Map<String, Value> {
995 238 : let mut o = o.clone();
996 238 : o.remove("@context");
997 238 : o
998 238 : }
999 :
1000 : // ---------- PUT /entities/{id}/attrs/{attrId} — Replace attribute (5.6.19) ----------
1001 :
1002 162 : pub async fn replace_attr(
1003 162 : State(st): State<AppState>,
1004 162 : Path((id, attr)): Path<(String, String)>,
1005 162 : CleanParams(params): CleanParams,
1006 162 : headers: HeaderMap,
1007 162 : body: Bytes,
1008 162 : ) -> Response {
1009 162 : let go = async {
1010 162 : let tenant = tenant_from(&headers)?;
1011 162 : antares_model::EntityId::new(&id)?;
1012 158 : check_attr_name(&attr)?;
1013 138 : check_params(¶ms, &["local", "type"])?;
1014 138 : let parsed = parse_body(&st, &headers, &body, BodyKind::Standard).await?;
1015 : // 5.5.7 term expansion first, then 5.6.19.4: "If the target Attribute
1016 : // is scope, then an error of type BadRequestData shall be raised" —
1017 : // the target is the expanded name, so the IRI spelling counts too.
1018 138 : let attr_iri = antares_jsonld::expand_attr_name(&attr, &parsed.ctx)?;
1019 130 : if attr == "scope" || attr_iri == SCOPE_IRI {
1020 16 : return Err(NgsiError::BadRequestData(
1021 16 : "scope cannot be the target of a replace attribute (5.6.19)".into(),
1022 16 : )
1023 16 : .into());
1024 114 : }
1025 114 : let obj = parsed.object(NgsiError::BadRequestData(
1026 114 : "fragment must be a JSON object".into(),
1027 114 : ))?;
1028 114 : let mut wrapper = Map::new();
1029 114 : wrapper.insert(attr.clone(), Value::Object(without_context_map(obj)));
1030 114 : let fragment = expand_entity(
1031 114 : &wrapper,
1032 114 : &parsed.ctx,
1033 114 : ExpandOpts {
1034 114 : fragment: true,
1035 114 : allow_null: false,
1036 114 : temporal: false,
1037 114 : ..Default::default()
1038 114 : },
1039 0 : )?;
1040 114 : let pre = crate::policy::pre_image(&st, &tenant, &id, None).await?;
1041 114 : let granted = gate!(
1042 : st, &tenant, &headers, "5.6.19",
1043 : ids: &[&id],
1044 : types: &pre.types,
1045 : stored: pre.stored.as_ref(),
1046 : attrs: std::slice::from_ref(&attr_iri),
1047 : body: Some(&fragment),
1048 : )
1049 114 : .await?;
1050 112 : let cond = Condition::of(&granted);
1051 112 : let incoming_arr = fragment
1052 112 : .get(&attr_iri)
1053 112 : .cloned()
1054 112 : .ok_or_else(|| NgsiError::BadRequestData("invalid attribute fragment".into()))?;
1055 112 : let new_inst = incoming_arr
1056 112 : .as_array()
1057 112 : .and_then(|a| a.first())
1058 112 : .cloned()
1059 112 : .ok_or_else(|| NgsiError::BadRequestData("invalid attribute fragment".into()))?;
1060 112 : let want_ds = new_inst
1061 112 : .get("datasetId")
1062 112 : .and_then(Value::as_str)
1063 112 : .map(String::from);
1064 112 : let (regs, loop_answer) =
1065 112 : attr_regs_or_loop(&st, &tenant, &id, &attr_iri, &parsed.ctx, ¶ms, &headers).await?;
1066 112 : if let Some(r) = loop_answer {
1067 0 : return Ok(r);
1068 112 : }
1069 112 : Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
1070 112 : let local_covered = proxies_cover_all(®s, std::slice::from_ref(&attr_iri));
1071 112 : let ts = now_iso();
1072 112 : let mut found = false;
1073 112 : let local_resp: Option<ApiResult<Response>> = if local_covered {
1074 16 : None
1075 : } else {
1076 96 : let res = st
1077 96 : .store
1078 96 : .mutate(&tenant, Kind::Entity, &id, |doc| {
1079 82 : Condition::pre(cond.as_ref(), &id, doc)?;
1080 : // 5.6.19.4: the ?type selector narrows the target entity
1081 78 : if !matches_type_param(doc, ¶ms, &parsed.ctx) {
1082 4 : return Err(NgsiError::ResourceNotFound(format!(
1083 4 : "entity {id} does not match the type selector"
1084 4 : )));
1085 74 : }
1086 74 : let target = antares_store::stored_object(doc)?;
1087 74 : if let Some(existing) = target.get_mut(&attr_iri).and_then(Value::as_array_mut)
1088 : {
1089 : // 5.6.19: only the instance with the matching datasetId is
1090 : // replaced; its createdAt survives (055_01/055_02)
1091 72 : if let Some(slot) = existing.iter_mut().find(|ci| {
1092 72 : ci.get("datasetId").and_then(Value::as_str) == want_ds.as_deref()
1093 72 : }) {
1094 64 : found = true;
1095 64 : let created = slot.get("createdAt").cloned();
1096 64 : let mut ni = new_inst.clone();
1097 64 : if let Some(o) = ni.as_object_mut() {
1098 64 : if let Some(c) = created {
1099 64 : o.insert("createdAt".into(), c);
1100 64 : } else {
1101 0 : o.insert("createdAt".into(), Value::String(ts.clone()));
1102 0 : }
1103 64 : o.insert("modifiedAt".into(), Value::String(ts.clone()));
1104 0 : }
1105 64 : *slot = ni;
1106 4 : }
1107 6 : }
1108 74 : if found {
1109 64 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
1110 64 : }
1111 74 : Ok::<(), NgsiError>(())
1112 82 : })
1113 96 : .await?;
1114 96 : Some(single_attr_local(res, found, &id, &attr, &tenant))
1115 : };
1116 112 : if regs.is_empty() {
1117 : // the local half is skipped only while registrations cover the
1118 : // attributes, so an empty list always leaves a local response
1119 92 : return local_resp
1120 92 : .unwrap_or_else(|| Err(NgsiError::InternalError("no local result".into()).into()));
1121 20 : }
1122 20 : let fed_parts = crate::federation::fed_attr_parts(
1123 20 : &st,
1124 20 : &headers,
1125 20 : &tenant,
1126 20 : &parsed.ctx,
1127 20 : ®s,
1128 20 : "replaceAttrs",
1129 20 : reqwest::Method::PUT,
1130 20 : &format!(
1131 20 : "/entities/{}/attrs/{}",
1132 20 : path_segment(&id),
1133 20 : path_segment(&attr)
1134 20 : ),
1135 20 : &fwd_query(¶ms, &["type"]),
1136 20 : Some(Value::Object(without_context_map(obj))),
1137 20 : false,
1138 20 : )
1139 20 : .await;
1140 20 : Ok(combine_single_attr(
1141 20 : &tenant,
1142 20 : &attr_iri,
1143 20 : local_covered,
1144 20 : &local_resp,
1145 20 : ®s,
1146 20 : &fed_parts,
1147 20 : ))
1148 162 : };
1149 162 : go.await.unwrap_or_else(|e: ApiError| e.into_response())
1150 162 : }
1151 :
1152 : // ---------- PUT /entities/{id}/attrs/{attrId}/value (TS 104 175 10.2.6) ----------
1153 :
1154 : /// Set Attribute Value (TS 104 175 10.2.6, TS 104 176 7.6.3.2): the body
1155 : /// is the bare value. The stored instance keeps its type and every other
1156 : /// sub-Attribute; only the member that type carries its value in changes.
1157 : /// The policy engine judges it as Replace Attribute (5.6.19), the write it
1158 : /// is, so no rule written for that clause can be stepped around here.
1159 60 : pub async fn set_attr_value(
1160 60 : State(st): State<AppState>,
1161 60 : Path((id, attr)): Path<(String, String)>,
1162 60 : CleanParams(params): CleanParams,
1163 60 : headers: HeaderMap,
1164 60 : body: Bytes,
1165 60 : ) -> Response {
1166 60 : set_attr_value_inner(&st, &id, &attr, ¶ms, &headers, &body)
1167 60 : .await
1168 60 : .unwrap_or_else(|e: ApiError| e.into_response())
1169 60 : }
1170 :
1171 60 : async fn set_attr_value_inner(
1172 60 : st: &AppState,
1173 60 : id: &str,
1174 60 : attr: &str,
1175 60 : params: &HashMap<String, String>,
1176 60 : headers: &HeaderMap,
1177 60 : body: &Bytes,
1178 60 : ) -> ApiResult<Response> {
1179 60 : let tenant = tenant_from(headers)?;
1180 60 : antares_model::EntityId::new(id)?;
1181 58 : check_attr_name(attr)?;
1182 56 : check_params(params, &["datasetId", "observedAt", "local", "type"])?;
1183 56 : let ctx = request_context(&st.loader, headers).await?;
1184 56 : let attr_iri = antares_jsonld::expand_attr_name(attr, &ctx)?;
1185 56 : if attr == "scope" || attr_iri == SCOPE_IRI {
1186 2 : return Err(NgsiError::BadRequestData(
1187 2 : "scope cannot be the target of Set Attribute Value (10.2.6)".into(),
1188 2 : )
1189 2 : .into());
1190 54 : }
1191 54 : let observed = params.get("observedAt").map(String::as_str);
1192 54 : if observed.is_some_and(|t| !antares_jsonld::parse_datetime(t)) {
1193 2 : return Err(NgsiError::BadRequestData("observedAt must be a DateTime (4.8)".into()).into());
1194 52 : }
1195 52 : let want_ds = crate::repr::target_dataset_id(params).map(String::from);
1196 52 : if let Some(ds) = &want_ds {
1197 4 : antares_model::EntityId::new(ds)
1198 4 : .map_err(|_| NgsiError::BadRequestData(format!("datasetId {ds:?} is not a URI")))?;
1199 48 : }
1200 52 : let value: Value = serde_json::from_slice(body)
1201 52 : .map_err(|e| NgsiError::InvalidRequest(format!("the body is not a JSON value: {e}")))?;
1202 : // 10.2.6.4: null and the NGSI-LD Null are refused except in a
1203 : // JsonProperty; without a local instance the type is unknown, so a
1204 : // null is refused before it can be forwarded
1205 50 : let stored = st.store.get(&tenant, Kind::Entity, id).await?;
1206 50 : let stored_inst = stored
1207 50 : .as_ref()
1208 50 : .and_then(|d| d.get(&attr_iri))
1209 50 : .and_then(|a| instance_with_dataset(a, want_ds.as_deref()))
1210 50 : .cloned();
1211 50 : refuse_null(stored_inst.as_ref(), &value)?;
1212 : // what the engine judges is the instance as it would be stored
1213 44 : let judged = stored_inst
1214 44 : .as_ref()
1215 44 : .map(|si| with_value(si, attr, &attr_iri, &value, &ctx, observed))
1216 44 : .transpose()?
1217 42 : .map(|i| serde_json::json!({ attr_iri.clone(): [i] }));
1218 42 : let pre = crate::policy::pre_image(st, &tenant, id, None).await?;
1219 42 : let granted = gate!(
1220 : st, &tenant, headers, "5.6.19",
1221 : ids: &[id],
1222 : types: &pre.types,
1223 : stored: pre.stored.as_ref(),
1224 : attrs: std::slice::from_ref(&attr_iri),
1225 : body: judged.as_ref(),
1226 : )
1227 42 : .await?;
1228 40 : let cond = Condition::of(&granted);
1229 40 : let (regs, loop_answer) =
1230 40 : attr_regs_or_loop(st, &tenant, id, &attr_iri, &ctx, params, headers).await?;
1231 40 : if let Some(r) = loop_answer {
1232 0 : return Ok(r);
1233 40 : }
1234 40 : Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
1235 40 : let local_covered = proxies_cover_all(®s, std::slice::from_ref(&attr_iri));
1236 40 : let ts = now_iso();
1237 40 : let mut found = false;
1238 40 : let local_resp: Option<ApiResult<Response>> = if local_covered {
1239 2 : None
1240 : } else {
1241 38 : let res = st
1242 38 : .store
1243 38 : .mutate(&tenant, Kind::Entity, id, |doc| {
1244 26 : Condition::pre(cond.as_ref(), id, doc)?;
1245 22 : if !matches_type_param(doc, params, &ctx) {
1246 2 : return Err(NgsiError::ResourceNotFound(format!(
1247 2 : "entity {id} does not match the type selector"
1248 2 : )));
1249 20 : }
1250 20 : let target = antares_store::stored_object(doc)?;
1251 20 : let Some(existing) = target.get_mut(&attr_iri).and_then(Value::as_array_mut) else {
1252 2 : return Ok(());
1253 : };
1254 18 : let Some(slot) = existing
1255 18 : .iter_mut()
1256 22 : .find(|ci| ci.get("datasetId").and_then(Value::as_str) == want_ds.as_deref())
1257 : else {
1258 2 : return Ok(());
1259 : };
1260 : // from the instance under the lock, not the read above: a
1261 : // write in between must not be lost
1262 16 : refuse_null(Some(slot), &value)?;
1263 16 : let mut ni = with_value(slot, attr, &attr_iri, &value, &ctx, observed)?;
1264 16 : if let Some(o) = ni.as_object_mut() {
1265 16 : o.insert("modifiedAt".into(), Value::String(ts.clone()));
1266 16 : }
1267 16 : *slot = ni;
1268 16 : found = true;
1269 16 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
1270 16 : Ok::<(), NgsiError>(())
1271 26 : })
1272 38 : .await?;
1273 38 : Some(single_attr_local(res, found, id, attr, &tenant))
1274 : };
1275 40 : if regs.is_empty() {
1276 38 : return local_resp
1277 38 : .unwrap_or_else(|| Err(NgsiError::InternalError("no local result".into()).into()));
1278 2 : }
1279 : // Decided: 9.2 names no operation for 10.2.6, so a source is
1280 : // asked when it supports replaceAttrs, and on its own value resource
1281 2 : let fed_parts = crate::federation::fed_attr_parts(
1282 2 : st,
1283 2 : headers,
1284 2 : &tenant,
1285 2 : &ctx,
1286 2 : ®s,
1287 2 : "replaceAttrs",
1288 2 : reqwest::Method::PUT,
1289 2 : &format!(
1290 2 : "/entities/{}/attrs/{}/value",
1291 2 : path_segment(id),
1292 2 : path_segment(attr)
1293 2 : ),
1294 2 : &fwd_query(params, &["type", "datasetId", "observedAt"]),
1295 2 : Some(value),
1296 2 : false,
1297 2 : )
1298 2 : .await;
1299 2 : Ok(combine_single_attr(
1300 2 : &tenant,
1301 2 : &attr_iri,
1302 2 : local_covered,
1303 2 : &local_resp,
1304 2 : ®s,
1305 2 : &fed_parts,
1306 2 : ))
1307 60 : }
1308 :
1309 : /// The instance of a stored Attribute with this datasetId (`None` = the
1310 : /// default instance).
1311 34 : fn instance_with_dataset<'a>(attr: &'a Value, ds: Option<&str>) -> Option<&'a Value> {
1312 34 : attr.as_array()?
1313 34 : .iter()
1314 38 : .find(|i| i.get("datasetId").and_then(Value::as_str) == ds)
1315 34 : }
1316 :
1317 : /// 10.2.6.4: "the payload body contains "urn:ngsi-ld:null" or contains null
1318 : /// (excepting for Attribute fragments representing a JsonProperty)" is
1319 : /// InvalidRequest, wherever in the value it sits.
1320 66 : fn refuse_null(stored: Option<&Value>, value: &Value) -> Result<(), NgsiError> {
1321 98 : fn holds_null(v: &Value) -> bool {
1322 98 : match v {
1323 4 : Value::Null => true,
1324 22 : Value::String(s) => s == "urn:ngsi-ld:null",
1325 2 : Value::Array(a) => a.iter().any(holds_null),
1326 22 : Value::Object(o) => o.values().any(holds_null),
1327 48 : _ => false,
1328 : }
1329 98 : }
1330 66 : let json_property = stored.is_some_and(|i| i.get("type") == Some(&Value::from("JsonProperty")));
1331 66 : if !json_property && holds_null(value) {
1332 6 : return Err(NgsiError::InvalidRequest(
1333 6 : "a value may hold no null and no urn:ngsi-ld:null (10.2.6.4)".into(),
1334 6 : ));
1335 60 : }
1336 60 : Ok(())
1337 66 : }
1338 :
1339 : /// The stored instance `si` with its value replaced: the member its type
1340 : /// carries the value in, expanded and validated as that type would be in a
1341 : /// fragment (8.2.4); observedAt follows the request when the instance had
1342 : /// one, and goes otherwise (10.2.6.4).
1343 42 : fn with_value(
1344 42 : si: &Value,
1345 42 : attr: &str,
1346 42 : attr_iri: &str,
1347 42 : value: &Value,
1348 42 : ctx: &antares_jsonld::Context,
1349 42 : observed: Option<&str>,
1350 42 : ) -> Result<Value, NgsiError> {
1351 42 : let ty = si.get("type").and_then(Value::as_str).unwrap_or("Property");
1352 42 : let member = match ty {
1353 42 : "Relationship" => "object",
1354 36 : "LanguageProperty" => "languageMap",
1355 32 : "JsonProperty" => "json",
1356 28 : "VocabProperty" => "vocab",
1357 28 : "ListProperty" => "valueList",
1358 28 : "ListRelationship" => "objectList",
1359 28 : _ => "value",
1360 : };
1361 42 : let mut wrapper = Map::new();
1362 42 : wrapper.insert(
1363 42 : attr.to_owned(),
1364 42 : serde_json::json!({ "type": ty, member: value }),
1365 : );
1366 42 : let expanded = expand_entity(
1367 42 : &wrapper,
1368 42 : ctx,
1369 42 : ExpandOpts {
1370 42 : fragment: true,
1371 42 : allow_null: false,
1372 42 : temporal: false,
1373 42 : ..Default::default()
1374 42 : },
1375 2 : )?;
1376 40 : let new_value = expanded
1377 40 : .get(attr_iri)
1378 40 : .and_then(Value::as_array)
1379 40 : .and_then(|a| a.first())
1380 40 : .and_then(|i| i.get(member))
1381 40 : .cloned()
1382 40 : .ok_or_else(|| NgsiError::BadRequestData(format!("not a {ty} value")))?;
1383 40 : let mut out = si.clone();
1384 40 : if let Some(o) = out.as_object_mut() {
1385 40 : o.insert(member.into(), new_value);
1386 40 : if o.contains_key("observedAt") {
1387 10 : match observed {
1388 4 : Some(t) => o.insert("observedAt".into(), Value::String(t.to_owned())),
1389 6 : None => o.remove("observedAt"),
1390 : };
1391 30 : }
1392 0 : }
1393 40 : Ok(out)
1394 42 : }
1395 :
1396 : // ---------- DELETE /entities/{id}/attrs/{attrId} (5.6.5) ----------
1397 :
1398 284 : pub async fn delete_attr(
1399 284 : State(st): State<AppState>,
1400 284 : Path((id, attr)): Path<(String, String)>,
1401 284 : CleanParams(params): CleanParams,
1402 284 : headers: HeaderMap,
1403 284 : ) -> Response {
1404 284 : match delete_attr_inner(&st, &id, &attr, ¶ms, &headers).await {
1405 206 : Ok(r) => r,
1406 78 : Err(e) => e.into_response(),
1407 : }
1408 284 : }
1409 :
1410 284 : async fn delete_attr_inner(
1411 284 : st: &AppState,
1412 284 : id: &str,
1413 284 : attr: &str,
1414 284 : params: &HashMap<String, String>,
1415 284 : headers: &HeaderMap,
1416 284 : ) -> ApiResult<Response> {
1417 284 : let tenant = tenant_from(headers)?;
1418 284 : antares_model::EntityId::new(id)?;
1419 276 : check_attr_name(attr)?;
1420 260 : check_params(params, &["datasetId", "deleteAll", "local", "type"])?;
1421 260 : let ctx = request_context(&st.loader, headers).await?;
1422 : // 5.6.5.4 expands the path name the same way 5.6.4.4 does, and then
1423 : // addresses `scope` as itself rather than as an Attribute. The target is
1424 : // what the name expands to, so both spellings — the reserved short name
1425 : // and the fully qualified one a client @context can map a term to —
1426 : // address the member the document stores under its short name.
1427 260 : let attr_iri = if attr == "scope" {
1428 22 : "scope".to_owned()
1429 : } else {
1430 238 : match antares_jsonld::expand_attr_name(attr, &ctx)? {
1431 236 : iri if iri == SCOPE_IRI => "scope".to_owned(),
1432 234 : iri => iri,
1433 : }
1434 : };
1435 258 : let pre = crate::policy::pre_image(st, &tenant, id, None).await?;
1436 258 : let granted = gate!(
1437 : st, &tenant, headers, "5.6.5",
1438 : ids: &[id],
1439 : types: &pre.types,
1440 : stored: pre.stored.as_ref(),
1441 : attrs: std::slice::from_ref(&attr_iri),
1442 : )
1443 258 : .await?;
1444 254 : let cond = Condition::of(&granted);
1445 254 : let delete_all = params.get("deleteAll").map(String::as_str) == Some("true");
1446 254 : let want_ds = crate::repr::target_dataset_id(params).map(String::from);
1447 254 : let (regs, loop_answer) =
1448 254 : attr_regs_or_loop(st, &tenant, id, &attr_iri, &ctx, params, headers).await?;
1449 254 : if let Some(r) = loop_answer {
1450 4 : return Ok(r);
1451 250 : }
1452 250 : Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
1453 250 : let local_covered = proxies_cover_all(®s, std::slice::from_ref(&attr_iri));
1454 250 : let ts = now_iso();
1455 250 : let mut found = false;
1456 250 : let local_resp: Option<ApiResult<Response>> = if local_covered {
1457 32 : None
1458 : } else {
1459 218 : let res = st
1460 218 : .store
1461 218 : .mutate(&tenant, Kind::Entity, id, |doc| {
1462 200 : Condition::pre(cond.as_ref(), id, doc)?;
1463 : // 5.6.5.4: the ?type selector narrows the target entity
1464 196 : if !matches_type_param(doc, params, &ctx) {
1465 12 : return Err(NgsiError::ResourceNotFound(format!(
1466 12 : "entity {id} does not match the type selector"
1467 12 : )));
1468 184 : }
1469 184 : if attr_iri == "scope" {
1470 24 : let target = antares_store::stored_object(doc)?;
1471 24 : found = target.remove("scope").is_some();
1472 24 : if found {
1473 22 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
1474 22 : }
1475 24 : return Ok(());
1476 160 : }
1477 160 : let target = antares_store::stored_object(doc)?;
1478 160 : if let Some(existing) = target.get_mut(&attr_iri).and_then(Value::as_array_mut) {
1479 148 : if delete_all {
1480 12 : found = !existing.is_empty();
1481 12 : existing.clear();
1482 12 : } else {
1483 154 : let pos = existing.iter().position(|ci| {
1484 154 : ci.get("datasetId").and_then(Value::as_str) == want_ds.as_deref()
1485 154 : });
1486 136 : if let Some(p) = pos {
1487 136 : existing.remove(p);
1488 136 : found = true;
1489 136 : }
1490 : }
1491 148 : if existing.is_empty() {
1492 120 : target.remove(&attr_iri);
1493 120 : }
1494 12 : }
1495 160 : if found {
1496 148 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
1497 148 : }
1498 160 : Ok::<(), NgsiError>(())
1499 200 : })
1500 218 : .await?;
1501 : // The store's change hook records the deletion (4.8 deletedAt,
1502 : // notify::record_temporal_change). Only what the Entity held is
1503 : // deleted (5.6.5.4: "If the target Entity does not contain the target
1504 : // Attribute then an error of type ResourceNotFound shall be raised"),
1505 : // so an Entity held only temporally, an Attribute already deleted and
1506 : // a refused delete (the type selector did not match) write nothing
1507 : // and leave the history untouched.
1508 218 : Some(single_attr_local(res, found, id, attr, &tenant))
1509 : };
1510 250 : if regs.is_empty() {
1511 : // the local half is skipped only while registrations cover the
1512 : // attributes, so an empty list always leaves a local response
1513 210 : return local_resp
1514 210 : .unwrap_or_else(|| Err(NgsiError::InternalError("no local result".into()).into()));
1515 40 : }
1516 40 : let query = fwd_query(params, &["datasetId", "deleteAll", "type"]);
1517 40 : let fed_parts = crate::federation::fed_attr_parts(
1518 40 : st,
1519 40 : headers,
1520 40 : &tenant,
1521 40 : &ctx,
1522 40 : ®s,
1523 40 : "deleteAttrs",
1524 40 : reqwest::Method::DELETE,
1525 40 : &format!(
1526 40 : "/entities/{}/attrs/{}",
1527 40 : path_segment(id),
1528 40 : path_segment(attr)
1529 40 : ),
1530 40 : &query,
1531 40 : None,
1532 40 : false,
1533 40 : )
1534 40 : .await;
1535 40 : Ok(combine_single_attr(
1536 40 : &tenant,
1537 40 : &attr_iri,
1538 40 : local_covered,
1539 40 : &local_resp,
1540 40 : ®s,
1541 40 : &fed_parts,
1542 40 : ))
1543 284 : }
1544 :
1545 : #[cfg(test)]
1546 : mod attr_name_and_via_paths {
1547 : use crate::AppState;
1548 : use axum::body::Body;
1549 : use axum::http::{Request, StatusCode};
1550 : use std::io::{Read, Write};
1551 : use std::sync::atomic::{AtomicUsize, Ordering};
1552 : use std::sync::{Arc, Mutex};
1553 : use tower::ServiceExt;
1554 :
1555 : const ALIAS: &str = "antares1";
1556 :
1557 : /// A Context Source answering 204 to everything, counting the hits and
1558 : /// recording the request-target of each one (the wire truth a forwarded
1559 : /// operation is judged on).
1560 16 : fn mock_source() -> (u16, Arc<AtomicUsize>, Arc<Mutex<Vec<String>>>) {
1561 16 : let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
1562 16 : let port = listener.local_addr().expect("addr").port();
1563 16 : let hits: Arc<AtomicUsize> = Arc::default();
1564 16 : let seen = hits.clone();
1565 16 : let targets: Arc<Mutex<Vec<String>>> = Arc::default();
1566 16 : let seen_targets = targets.clone();
1567 16 : std::thread::spawn(move || {
1568 44 : for stream in listener.incoming() {
1569 44 : let Ok(mut s) = stream else { continue };
1570 44 : seen.fetch_add(1, Ordering::SeqCst);
1571 44 : let mut buf = [0u8; 8192];
1572 44 : let n = s.read(&mut buf).unwrap_or(0);
1573 44 : let head = String::from_utf8_lossy(&buf[..n]).into_owned();
1574 44 : let line = head.lines().next().unwrap_or_default().to_owned();
1575 44 : if let Some(t) = line.split_whitespace().nth(1) {
1576 44 : seen_targets.lock().expect("targets").push(t.to_owned());
1577 44 : }
1578 44 : let _ = s.write_all(
1579 44 : b"HTTP/1.1 204 No Content\r\nConnection: close\r\nContent-Length: 0\r\n\r\n",
1580 44 : );
1581 : }
1582 16 : });
1583 16 : (port, hits, targets)
1584 16 : }
1585 :
1586 : /// The request-targets the mock has seen so far, in arrival order.
1587 44 : fn seen(targets: &Arc<Mutex<Vec<String>>>) -> Vec<String> {
1588 44 : targets.lock().expect("targets").clone()
1589 44 : }
1590 :
1591 28 : fn state() -> AppState {
1592 : // the mock source is loopback, denied by the egress policy by default
1593 28 : crate::allow_private();
1594 28 : AppState::new(ALIAS.into())
1595 28 : }
1596 :
1597 176 : async fn send(st: &AppState, req: Request<Body>) -> axum::http::Response<Body> {
1598 176 : crate::router(st.clone())
1599 176 : .oneshot(req)
1600 176 : .await
1601 176 : .expect("response")
1602 176 : }
1603 :
1604 40 : async fn post(st: &AppState, uri: &str, body: String) -> StatusCode {
1605 40 : let req = Request::builder()
1606 40 : .method("POST")
1607 40 : .uri(uri)
1608 40 : .header("Content-Type", "application/json")
1609 40 : .header("Content-Length", body.len())
1610 40 : .body(Body::from(body))
1611 40 : .expect("request");
1612 40 : send(st, req).await.status()
1613 40 : }
1614 :
1615 : /// One registration for `entity` covering every attribute. The store is
1616 : /// process-wide, so every test uses its own ids. `mode` is a parameter
1617 : /// because 5.9.2 forbids two proxied (exclusive or redirect)
1618 : /// registrations from overlapping — a test that needs two matching
1619 : /// registrations has to register them inclusive.
1620 20 : async fn register(st: &AppState, port: u16, id: &str, entity: &str, mode: &str) {
1621 20 : let doc = serde_json::json!({
1622 20 : "id": format!("urn:ngsi-ld:ContextSourceRegistration:{id}"),
1623 20 : "type": "ContextSourceRegistration",
1624 20 : "mode": mode,
1625 20 : "operations": ["updateEntity", "updateAttrs", "replaceAttrs", "deleteAttrs", "appendAttrs"],
1626 20 : "information": [{"entities": [{"type": "Vehicle", "id": entity}]}],
1627 20 : "endpoint": format!("http://127.0.0.1:{port}"),
1628 : });
1629 20 : assert_eq!(
1630 20 : post(st, "/ngsi-ld/v1/csourceRegistrations", doc.to_string()).await,
1631 : StatusCode::CREATED,
1632 : "registration create"
1633 : );
1634 20 : }
1635 :
1636 12 : async fn create_entity(st: &AppState, entity: &str) {
1637 12 : let doc = serde_json::json!({
1638 12 : "id": entity, "type": "Vehicle",
1639 12 : "speed": {"type": "Property", "value": 1},
1640 : });
1641 12 : assert_eq!(
1642 12 : post(st, "/ngsi-ld/v1/entities", doc.to_string()).await,
1643 : StatusCode::CREATED,
1644 : "entity create"
1645 : );
1646 12 : }
1647 :
1648 : /// Antares serves the form it forwards: Append and Update
1649 : /// Entity Attributes (5.6.2, 5.6.3) go out as `…/attrs/`, the form the
1650 : /// ETSI mock sources expect, and the router itself answers that path,
1651 : /// not only the binary's layer. So a registration whose endpoint is
1652 : /// another Antares, or this one through the router in process or in the
1653 : /// browser build, lands. The peer is a second broker (its own alias, or
1654 : /// the Via chain reads the forward as a loop) served by the bare router
1655 : /// over HTTP; the Entity lives in its tenant `t3701` only.
1656 : #[tokio::test(flavor = "multi_thread")]
1657 4 : async fn a_forward_between_two_antares_tenants_lands() {
1658 4 : if std::env::var_os("ANTARES_TEST_SANITIZER").is_some() {
1659 0 : return;
1660 4 : }
1661 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-slash";
1662 4 : let st = state();
1663 4 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
1664 4 : .await
1665 4 : .expect("bind");
1666 4 : let port = listener.local_addr().expect("addr").port();
1667 4 : let peer = AppState::new("antares-peer".into());
1668 4 : let served = crate::router(peer.clone());
1669 4 : tokio::spawn(async move {
1670 4 : let _ = axum::serve(listener, served).await;
1671 0 : });
1672 20 : let call = |method: &str, uri: String, tenant: Option<&str>, body: Option<String>| {
1673 20 : let mut req = Request::builder().method(method).uri(uri);
1674 20 : if let Some(t) = tenant {
1675 12 : req = req.header("NGSILD-Tenant", t);
1676 12 : }
1677 20 : if let Some(b) = &body {
1678 16 : req = req
1679 16 : .header("Content-Type", "application/json")
1680 16 : .header("Content-Length", b.len());
1681 16 : }
1682 20 : req.body(body.map_or_else(Body::empty, Body::from))
1683 20 : .expect("request")
1684 20 : };
1685 4 : let entity = serde_json::json!({"id": ENTITY, "type": "Vehicle",
1686 4 : "speed": {"type": "Property", "value": 1}});
1687 4 : let req = call(
1688 4 : "POST",
1689 4 : "/ngsi-ld/v1/entities".into(),
1690 4 : Some("t3701"),
1691 4 : Some(entity.to_string()),
1692 4 : );
1693 4 : assert_eq!(send(&peer, req).await.status(), StatusCode::CREATED);
1694 4 : let reg = serde_json::json!({
1695 4 : "id": "urn:ngsi-ld:ContextSourceRegistration:attrs-slash",
1696 4 : "type": "ContextSourceRegistration", "mode": "redirect", "tenant": "t3701",
1697 4 : "operations": ["updateEntity", "updateAttrs", "replaceAttrs", "deleteAttrs", "appendAttrs"],
1698 4 : "information": [{"entities": [{"type": "Vehicle", "id": ENTITY}]}],
1699 4 : "endpoint": format!("http://127.0.0.1:{port}"),
1700 : });
1701 4 : assert_eq!(
1702 4 : post(&st, "/ngsi-ld/v1/csourceRegistrations", reg.to_string()).await,
1703 : StatusCode::CREATED
1704 : );
1705 4 : let attrs = format!("/ngsi-ld/v1/entities/{ENTITY}/attrs");
1706 4 : let update = r#"{"speed":{"type":"Property","value":2}}"#.to_owned();
1707 4 : let append = r#"{"colour":{"type":"Property","value":"red"}}"#.to_owned();
1708 8 : for (method, body) in [("PATCH", update), ("POST", append)] {
1709 8 : let resp = send(&st, call(method, attrs.clone(), None, Some(body))).await;
1710 8 : let status = resp.status();
1711 8 : let bytes = http_body_util::BodyExt::collect(resp.into_body())
1712 8 : .await
1713 8 : .expect("body")
1714 8 : .to_bytes();
1715 8 : assert_eq!(
1716 : status,
1717 : StatusCode::NO_CONTENT,
1718 : "{method} forwarded: {}",
1719 0 : String::from_utf8_lossy(&bytes)
1720 : );
1721 : }
1722 4 : let req = call(
1723 4 : "GET",
1724 4 : format!("/ngsi-ld/v1/entities/{ENTITY}?options=keyValues"),
1725 4 : Some("t3701"),
1726 4 : None,
1727 4 : );
1728 4 : let resp = send(&peer, req).await;
1729 4 : let bytes = http_body_util::BodyExt::collect(resp.into_body())
1730 4 : .await
1731 4 : .expect("body")
1732 4 : .to_bytes();
1733 4 : let got: serde_json::Value = serde_json::from_slice(&bytes).expect("json");
1734 4 : assert_eq!(
1735 4 : (&got["speed"], &got["colour"]),
1736 4 : (&serde_json::json!(2), &serde_json::json!("red")),
1737 : "{got}"
1738 : );
1739 : // and the router answers the slash form itself, whoever sends it
1740 4 : let direct = r#"{"speed":{"type":"Property","value":3}}"#.to_owned();
1741 4 : let req = call("PATCH", format!("{attrs}/"), Some("t3701"), Some(direct));
1742 4 : assert_eq!(send(&peer, req).await.status(), StatusCode::NO_CONTENT);
1743 4 : }
1744 :
1745 : /// 4.6.2: a name starts with a letter, so no valid Attribute name is a
1746 : /// relative-path dot segment (RFC 3986 clause 5.2.4). Such a name reaching
1747 : /// a forwarded request URL re-targets the registration endpoint —
1748 : /// `DELETE /entities/{id}/attrs/..` becomes a Delete Entity on the peer —
1749 : /// so it is refused with BadRequestData before anything is forwarded.
1750 : #[tokio::test(flavor = "multi_thread")]
1751 4 : async fn dot_segment_attribute_name_is_refused_and_never_forwarded() {
1752 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-traversal";
1753 4 : let st = state();
1754 4 : let (port, hits, targets) = mock_source();
1755 4 : register(&st, port, "csr-traversal", ENTITY, "redirect").await;
1756 4 : let frag = r#"{"type":"Property","value":1}"#;
1757 : // raw, decoded once by this broker, and decoded once more by the peer
1758 16 : for attr in ["..", "%2e%2e", "%252e%252e", "."] {
1759 48 : for (method, body) in [("DELETE", None), ("PATCH", Some(frag)), ("PUT", Some(frag))] {
1760 48 : let mut req = Request::builder()
1761 48 : .method(method)
1762 48 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}/attrs/{attr}"));
1763 48 : if let Some(b) = body {
1764 32 : req = req
1765 32 : .header("Content-Type", "application/json")
1766 32 : .header("Content-Length", b.len());
1767 32 : }
1768 48 : let req = req
1769 48 : .body(body.map_or_else(Body::empty, |b| Body::from(b.to_owned())))
1770 48 : .expect("request");
1771 48 : assert_eq!(
1772 48 : send(&st, req).await.status(),
1773 : StatusCode::BAD_REQUEST,
1774 : "{method} attribute name {attr:?}"
1775 : );
1776 : }
1777 : }
1778 4 : assert_eq!(
1779 4 : hits.load(Ordering::SeqCst),
1780 : 0,
1781 : "a rejected attribute name must never reach a registration endpoint"
1782 : );
1783 : // an absolute IRI carries '/' and '.' legitimately: it passes the check
1784 : // and is forwarded, so the guard rejects the shape, not the characters
1785 4 : let req = Request::builder()
1786 4 : .method("DELETE")
1787 4 : .uri(format!(
1788 : "/ngsi-ld/v1/entities/{ENTITY}/attrs/https%3A%2F%2Fexample.org%2Fv1.0%2Fspeed"
1789 : ))
1790 4 : .body(Body::empty())
1791 4 : .expect("request");
1792 4 : assert_eq!(
1793 4 : send(&st, req).await.status(),
1794 : StatusCode::NO_CONTENT,
1795 : "an absolute IRI is a valid attribute name"
1796 : );
1797 : // ':' is a legal path character (RFC 3986 clause 3.3 pchar), '/' is not
1798 4 : assert_eq!(
1799 4 : seen(&targets),
1800 4 : vec![format!(
1801 : "/ngsi-ld/v1/entities/{ENTITY}/attrs/https:%2F%2Fexample.org%2Fv1.0%2Fspeed"
1802 : )],
1803 : "the peer is asked for that attribute of that entity, nothing else"
1804 : );
1805 4 : assert_eq!(
1806 4 : hits.load(Ordering::SeqCst),
1807 4 : 1,
1808 4 : "a valid attribute name is still forwarded"
1809 4 : );
1810 4 : }
1811 :
1812 : /// Tables 6.6.3.1-1/6.6.3.2-1/6.7.3.1-1 make the entity id and the
1813 : /// attribute name single URI path variables, so a forwarded attribute
1814 : /// operation must address the SAME resource at the registered source.
1815 : /// RFC 3986 clause 3.3 ends a path segment at `#`, `?` or `/`: spliced
1816 : /// raw, an id carrying `#` truncates the forwarded path and Delete
1817 : /// Attribute (5.6.5) reaches the peer as Delete Entity (5.6.6).
1818 : #[tokio::test(flavor = "multi_thread")]
1819 4 : async fn forwarded_attribute_path_keeps_the_whole_id_and_name() {
1820 : // five forwards through a blocking mock, each under the 8 s forward
1821 : // cap; a sanitizer's slowdown turns them into 504s
1822 4 : if std::env::var_os("ANTARES_TEST_SANITIZER").is_some() {
1823 0 : return;
1824 4 : }
1825 : // both delimiters an id may legally carry (RFC 3986 clause 3.3)
1826 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-frag#1?x";
1827 : const ID: &str = "urn:ngsi-ld:Vehicle:attrs-frag%231%3Fx";
1828 : // an absolute IRI is a legal Attribute name (4.6.2) and carries '/'
1829 : const ATTR: &str = "https%3A%2F%2Fexample.org%2Fv1.0%2Fspeed";
1830 : // ':' is a legal path character (RFC 3986 clause 3.3 pchar), '/' is not
1831 : const SEG: &str = "https:%2F%2Fexample.org%2Fv1.0%2Fspeed";
1832 4 : let st = state();
1833 4 : let (port, hits, targets) = mock_source();
1834 4 : register(&st, port, "csr-frag", ENTITY, "redirect").await;
1835 4 : let multi = r#"{"speed":{"type":"Property","value":1}}"#;
1836 4 : let single = r#"{"type":"Property","value":2}"#;
1837 4 : let single_tail = format!("/attrs/{ATTR}");
1838 4 : let cases = [
1839 4 : ("POST", "/attrs".to_owned(), Some(multi), "/attrs/"),
1840 4 : ("PATCH", "/attrs".to_owned(), Some(multi), "/attrs/"),
1841 4 : ("PATCH", single_tail.clone(), Some(single), "single"),
1842 4 : ("PUT", single_tail.clone(), Some(single), "single"),
1843 4 : ("DELETE", single_tail.clone(), None, "single"),
1844 4 : ];
1845 20 : for (method, tail, body, want_tail) in cases {
1846 20 : let mut req = Request::builder()
1847 20 : .method(method)
1848 20 : .uri(format!("/ngsi-ld/v1/entities/{ID}{tail}"));
1849 20 : if let Some(b) = body {
1850 16 : req = req
1851 16 : .header("Content-Type", "application/json")
1852 16 : .header("Content-Length", b.len());
1853 16 : }
1854 20 : let req = req
1855 20 : .body(body.map_or_else(Body::empty, |b| Body::from(b.to_owned())))
1856 20 : .expect("request");
1857 20 : assert_eq!(
1858 20 : send(&st, req).await.status(),
1859 4 : StatusCode::NO_CONTENT,
1860 4 : "{method} {tail}"
1861 4 : );
1862 20 : let want = if want_tail == "single" {
1863 12 : format!("/ngsi-ld/v1/entities/{ID}/attrs/{SEG}")
1864 4 : } else {
1865 8 : format!("/ngsi-ld/v1/entities/{ID}{want_tail}")
1866 4 : };
1867 20 : let got = seen(&targets).pop().unwrap_or_default();
1868 20 : assert_eq!(got, want, "{method} {tail} forwarded target");
1869 4 : // the id must not have been cut short at its '#' or '?', and no
1870 4 : // delimiter may survive undecoded inside a segment
1871 20 : assert!(
1872 20 : !got.contains('#') && !got.contains('?'),
1873 4 : "{method} {tail}: raw delimiter in the forwarded path {got}"
1874 4 : );
1875 4 : }
1876 4 : assert_eq!(hits.load(Ordering::SeqCst), 5, "one forward per operation");
1877 4 : }
1878 :
1879 : /// `type` is a shall-support URL parameter of every attribute write
1880 : /// (Tables 6.6.3.1-1, 6.6.3.2-1, 6.7.3.1-1, 6.7.3.2-1, 6.7.3.3-1) and
1881 : /// 5.6.5.4 identifies the target by its "id (URI), and where specified
1882 : /// type" — so the selector has to travel with the forwarded operation,
1883 : /// or the registered source writes on entities the client scoped out.
1884 : #[tokio::test(flavor = "multi_thread")]
1885 4 : async fn type_selector_travels_with_the_forwarded_attribute_write() {
1886 : // five forwards through a blocking mock, each under the 8 s forward
1887 : // cap; a sanitizer's slowdown turns them into 504s
1888 4 : if std::env::var_os("ANTARES_TEST_SANITIZER").is_some() {
1889 0 : return;
1890 4 : }
1891 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-typesel";
1892 4 : let st = state();
1893 4 : let (port, hits, targets) = mock_source();
1894 4 : register(&st, port, "csr-typesel", ENTITY, "redirect").await;
1895 4 : let multi = r#"{"speed":{"type":"Property","value":1}}"#;
1896 4 : let single = r#"{"type":"Property","value":2}"#;
1897 4 : let cases = [
1898 4 : ("POST", "/attrs?type=Vehicle", Some(multi)),
1899 4 : ("PATCH", "/attrs?type=Vehicle", Some(multi)),
1900 4 : ("PATCH", "/attrs/speed?type=Vehicle", Some(single)),
1901 4 : ("PUT", "/attrs/speed?type=Vehicle", Some(single)),
1902 4 : ("DELETE", "/attrs/speed?type=Vehicle", None),
1903 4 : ];
1904 20 : for (method, tail, body) in cases {
1905 20 : let mut req = Request::builder()
1906 20 : .method(method)
1907 20 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}{tail}"));
1908 20 : if let Some(b) = body {
1909 16 : req = req
1910 16 : .header("Content-Type", "application/json")
1911 16 : .header("Content-Length", b.len());
1912 16 : }
1913 20 : let req = req
1914 20 : .body(body.map_or_else(Body::empty, |b| Body::from(b.to_owned())))
1915 20 : .expect("request");
1916 20 : assert_eq!(
1917 20 : send(&st, req).await.status(),
1918 4 : StatusCode::NO_CONTENT,
1919 4 : "{method} {tail}"
1920 4 : );
1921 20 : let got = seen(&targets).pop().unwrap_or_default();
1922 20 : assert!(
1923 20 : got.contains("type=Vehicle"),
1924 4 : "{method} {tail} forwarded without the type selector: {got}"
1925 4 : );
1926 20 : assert!(
1927 20 : !got.contains("local="),
1928 4 : "{method} {tail}: local is never forwarded"
1929 4 : );
1930 4 : }
1931 4 : assert_eq!(hits.load(Ordering::SeqCst), 5, "one forward per operation");
1932 4 : }
1933 :
1934 : /// 5.6.5.4: with a `?type` selector that the target Entity does not
1935 : /// match, the entity is "not known" for this operation and
1936 : /// ResourceNotFound is raised — nothing is deleted, so the temporal
1937 : /// representation must not gain a 4.8 `deletedAt` instance either.
1938 : #[tokio::test(flavor = "multi_thread")]
1939 4 : async fn refused_delete_writes_no_temporal_deletion() {
1940 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-refused";
1941 4 : let st = state();
1942 4 : create_entity(&st, ENTITY).await;
1943 : // 5.6.11: the temporal representation of the same entity
1944 4 : let history = serde_json::json!({
1945 4 : "id": ENTITY, "type": "Vehicle",
1946 4 : "speed": [{"type": "Property", "value": 1,
1947 4 : "observedAt": "2026-01-01T00:00:00Z"}],
1948 : });
1949 4 : assert_eq!(
1950 4 : post(&st, "/ngsi-ld/v1/temporal/entities", history.to_string()).await,
1951 : StatusCode::CREATED,
1952 : "temporal create"
1953 : );
1954 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1955 4 : let req = Request::builder()
1956 4 : .method("DELETE")
1957 4 : .uri(format!(
1958 : "/ngsi-ld/v1/entities/{ENTITY}/attrs/speed?type=Building"
1959 : ))
1960 4 : .body(Body::empty())
1961 4 : .expect("request");
1962 4 : assert_eq!(send(&st, req).await.status(), StatusCode::NOT_FOUND);
1963 4 : let doc = st
1964 4 : .store
1965 4 : .get(&tenant, antares_store::Kind::Temporal, ENTITY)
1966 4 : .await
1967 4 : .expect("store read")
1968 4 : .expect("temporal doc");
1969 4 : assert!(
1970 4 : !doc.to_string().contains("deletedAt"),
1971 : "a refused delete must not tombstone the attribute history: {doc}"
1972 : );
1973 4 : assert!(
1974 4 : !doc.to_string().contains("urn:ngsi-ld:null"),
1975 4 : "nor write a deletion instance"
1976 4 : );
1977 4 : }
1978 :
1979 : /// 5.6.19.4: "If the target Attribute is scope, then an error of type
1980 : /// BadRequestData shall be raised" — the target is the expanded name
1981 : /// (5.5.7), so the IRI spelling of scope is refused just like the term.
1982 : #[tokio::test(flavor = "multi_thread")]
1983 4 : async fn replace_attr_refuses_the_scope_iri_spelling() {
1984 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-scope-iri";
1985 4 : let st = state();
1986 4 : create_entity(&st, ENTITY).await;
1987 4 : let frag = r#"{"type":"Property","value":"/Madrid"}"#;
1988 8 : for attr in ["scope", "https%3A%2F%2Furi.etsi.org%2Fngsi-ld%2Fscope"] {
1989 8 : let req = Request::builder()
1990 8 : .method("PUT")
1991 8 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}/attrs/{attr}"))
1992 8 : .header("Content-Type", "application/json")
1993 8 : .header("Content-Length", frag.len())
1994 8 : .body(Body::from(frag))
1995 8 : .expect("request");
1996 8 : let resp = send(&st, req).await;
1997 8 : assert_eq!(
1998 8 : resp.status(),
1999 4 : StatusCode::BAD_REQUEST,
2000 4 : "scope target {attr}"
2001 4 : );
2002 8 : let bytes = http_body_util::BodyExt::collect(resp.into_body())
2003 8 : .await
2004 8 : .expect("body")
2005 8 : .to_bytes();
2006 8 : let body: serde_json::Value = serde_json::from_slice(&bytes).expect("json");
2007 8 : assert_eq!(
2008 8 : body["type"], "https://uri.etsi.org/ngsi-ld/errors/BadRequestData",
2009 4 : "scope target {attr}"
2010 4 : );
2011 4 : // the 5.6.19.4 guard, not an incidental fragment-shape rejection
2012 8 : assert!(
2013 8 : body["detail"]
2014 8 : .as_str()
2015 8 : .is_some_and(|d| d.contains("scope cannot be the target")),
2016 4 : "scope target {attr}: {body}"
2017 4 : );
2018 4 : }
2019 4 : }
2020 :
2021 : /// TS 104 175 9.7: a write whose Via chain names this broker is
2022 : /// a detected loop, 508 whatever its registrations, neither forwarded nor
2023 : /// applied locally; two registrations change nothing about that.
2024 : #[tokio::test(flavor = "multi_thread")]
2025 4 : async fn self_via_with_two_registrations_is_a_detected_loop() {
2026 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-via";
2027 4 : let st = state();
2028 4 : create_entity(&st, ENTITY).await;
2029 4 : let (port, hits, _targets) = mock_source();
2030 4 : register(&st, port, "csr-via-a", ENTITY, "inclusive").await;
2031 4 : register(&st, port, "csr-via-b", ENTITY, "inclusive").await;
2032 4 : let via = format!("1.1 {ALIAS}");
2033 :
2034 : // 5.6.2 Update Attributes — the multi-attribute fragment plan
2035 4 : let body = r#"{"speed":{"type":"Property","value":9}}"#;
2036 4 : let req = Request::builder()
2037 4 : .method("PATCH")
2038 4 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}/attrs"))
2039 4 : .header("Content-Type", "application/json")
2040 4 : .header("Content-Length", body.len())
2041 4 : .header("Via", via.as_str())
2042 4 : .body(Body::from(body))
2043 4 : .expect("request");
2044 4 : assert_eq!(send(&st, req).await.status(), StatusCode::LOOP_DETECTED);
2045 :
2046 : // 5.6.5 Delete Attribute — the single-attribute plan
2047 4 : let req = Request::builder()
2048 4 : .method("DELETE")
2049 4 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}/attrs/speed"))
2050 4 : .header("Via", via.as_str())
2051 4 : .body(Body::empty())
2052 4 : .expect("request");
2053 4 : assert_eq!(send(&st, req).await.status(), StatusCode::LOOP_DETECTED);
2054 :
2055 4 : assert_eq!(
2056 4 : hits.load(Ordering::SeqCst),
2057 : 0,
2058 : "registrations dropped by the Via chain must not be contacted"
2059 : );
2060 : // a read with the same chain is served locally: the write never landed
2061 4 : let req = Request::builder()
2062 4 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}"))
2063 4 : .header("Via", via.as_str())
2064 4 : .body(Body::empty())
2065 4 : .expect("request");
2066 4 : let resp = send(&st, req).await;
2067 4 : assert_eq!(resp.status(), StatusCode::OK);
2068 4 : let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
2069 4 : .await
2070 4 : .expect("body");
2071 4 : let entity: serde_json::Value = serde_json::from_slice(&bytes).expect("json");
2072 4 : assert_eq!(entity["speed"]["value"], 1, "{entity}");
2073 4 : assert_eq!(hits.load(Ordering::SeqCst), 0);
2074 4 : }
2075 : }
2076 :
2077 : #[cfg(test)]
2078 : mod clause_4_8 {
2079 : use serde_json::{json, Value};
2080 :
2081 : const TS: &str = "2026-08-30T10:00:00.000Z";
2082 :
2083 : /// 4.8: createdAt and modifiedAt are the times at which "the Entity,
2084 : /// Property or Relationship" entered and was last modified in an NGSI-LD
2085 : /// system. A sub-Property is a Property, so an Attribute arriving through
2086 : /// 5.6.2/5.6.3 is stamped to the same depth as one arriving through
2087 : /// 5.6.1 — the served representation cannot depend on which operation
2088 : /// wrote the Attribute.
2089 : #[test]
2090 4 : fn an_appended_attribute_is_stamped_to_the_same_depth_as_a_created_one() {
2091 4 : let attr = json!([{
2092 4 : "type": "Property",
2093 4 : "value": 21,
2094 4 : "unitCode": "CEL",
2095 4 : "https://example.org/accuracy": [{
2096 4 : "type": "Property",
2097 4 : "value": 0.5,
2098 4 : "https://example.org/basis": [{"type": "Property", "value": "spec"}]
2099 : }],
2100 4 : "https://example.org/provider": [{
2101 4 : "type": "Relationship",
2102 4 : "object": "urn:ngsi-ld:Sensor:1"
2103 : }]
2104 : }]);
2105 4 : let mut appended = attr.clone();
2106 4 : crate::stamp::stamp_instances(&mut appended, TS);
2107 :
2108 : // what 5.6.1 Create Entity produces for the same attribute
2109 4 : let mut created = json!({
2110 4 : "id": "urn:ngsi-ld:Room:1",
2111 4 : "type": "Room",
2112 4 : "https://example.org/temperature": attr
2113 : });
2114 4 : crate::stamp::stamp_new(&mut created, TS);
2115 4 : assert_eq!(
2116 4 : appended, created["https://example.org/temperature"],
2117 : "an appended attribute must carry the timestamps a created one carries"
2118 : );
2119 :
2120 : // and spelled out, so the shape is not merely equal to a wrong shape
2121 4 : let inst = &appended[0];
2122 4 : assert_eq!(inst["createdAt"], TS);
2123 4 : assert_eq!(inst["modifiedAt"], TS);
2124 4 : let acc = &inst["https://example.org/accuracy"][0];
2125 4 : assert_eq!(acc["createdAt"], TS, "sub-Property createdAt");
2126 4 : assert_eq!(acc["modifiedAt"], TS, "sub-Property modifiedAt");
2127 4 : assert_eq!(
2128 4 : acc["https://example.org/basis"][0]["createdAt"], TS,
2129 : "sub-Property of a sub-Property"
2130 : );
2131 4 : assert_eq!(
2132 4 : appended[0]["https://example.org/provider"][0]["createdAt"], TS,
2133 : "sub-Relationship createdAt"
2134 : );
2135 : // 4.8 NOTE 1: a TemporalProperty is not reified — the members that
2136 : // carry the Attribute's own value are left exactly as they arrived
2137 4 : assert_eq!(inst["value"], 21);
2138 4 : assert_eq!(inst["unitCode"], "CEL");
2139 4 : assert!(
2140 4 : !matches!(inst["unitCode"], Value::Array(_)),
2141 : "a reserved member is never walked as an instance array"
2142 : );
2143 4 : }
2144 : }
2145 :
2146 : #[cfg(test)]
2147 : mod clause_5_5_8 {
2148 : use super::merge_instance_sets;
2149 : use serde_json::{json, Value};
2150 :
2151 : /// 5.5.8 update algorithm: a Fragment member replaces the WHOLE matching
2152 : /// instance (unmapped sub-members are dropped — EXAMPLE 2), a different
2153 : /// datasetId is added as a new instance, an NGSI-LD Null deletes the
2154 : /// matching instance (EXAMPLE 3), and createdAt survives the replace.
2155 : #[test]
2156 4 : fn update_replaces_whole_instances_dataset_id_wise() {
2157 4 : let mut existing = json!([
2158 4 : {"type": "Property", "value": 25, "unitCode": "CEL",
2159 4 : "observedAt": "2022-03-14T01:59:26.535Z",
2160 4 : "createdAt": "2022-01-01T00:00:00Z"},
2161 4 : {"type": "Property", "value": 7, "datasetId": "urn:ngsi-ld:Dataset:a",
2162 4 : "createdAt": "2022-01-01T00:00:00Z"}
2163 : ]);
2164 : // default instance replaced wholesale; new datasetId appended
2165 4 : let incoming = json!([
2166 4 : {"type": "Property", "value": 100,
2167 4 : "observedAt": "2022-03-14T13:00:00.000Z"},
2168 4 : {"type": "Property", "value": 8, "datasetId": "urn:ngsi-ld:Dataset:b"}
2169 : ]);
2170 4 : assert!(merge_instance_sets(&mut existing, &incoming, false, None));
2171 4 : let arr = existing.as_array().unwrap();
2172 4 : assert_eq!(arr.len(), 3, "default replaced + a kept + b added");
2173 4 : let default = arr
2174 4 : .iter()
2175 4 : .find(|i| i.get("datasetId").is_none())
2176 4 : .expect("default instance");
2177 4 : assert_eq!(default["value"], 100);
2178 4 : assert!(
2179 4 : default.get("unitCode").is_none(),
2180 : "EXAMPLE 2: whole-attribute replace drops unitCode"
2181 : );
2182 4 : assert_eq!(
2183 4 : default["createdAt"], "2022-01-01T00:00:00Z",
2184 : "createdAt survives the replace"
2185 : );
2186 : // NGSI-LD Null deletes exactly the matching datasetId instance
2187 4 : let deletion = json!([
2188 4 : {"type": "Property", "value": "urn:ngsi-ld:null",
2189 4 : "datasetId": "urn:ngsi-ld:Dataset:a"}
2190 : ]);
2191 4 : assert!(merge_instance_sets(&mut existing, &deletion, false, None));
2192 4 : let arr = existing.as_array().unwrap();
2193 4 : assert_eq!(arr.len(), 2);
2194 4 : assert!(
2195 8 : !arr.iter().any(|i| {
2196 8 : i.get("datasetId").and_then(Value::as_str) == Some("urn:ngsi-ld:Dataset:a")
2197 8 : }),
2198 : "EXAMPLE 3: null deletes the instance"
2199 : );
2200 4 : assert!(
2201 4 : !existing.to_string().contains("urn:ngsi-ld:null"),
2202 : "the sentinel never persists"
2203 : );
2204 4 : }
2205 :
2206 : /// 5.5.8 (noOverwrite append, 5.6.3): an existing instance is NOT
2207 : /// replaced, an absent one is still added.
2208 : #[test]
2209 4 : fn no_overwrite_keeps_existing_instances() {
2210 4 : let mut existing = json!([{"type": "Property", "value": 1}]);
2211 4 : let incoming = json!([
2212 4 : {"type": "Property", "value": 2},
2213 4 : {"type": "Property", "value": 3, "datasetId": "urn:ngsi-ld:Dataset:n"}
2214 : ]);
2215 4 : assert!(merge_instance_sets(&mut existing, &incoming, true, None));
2216 4 : let arr = existing.as_array().unwrap();
2217 4 : assert_eq!(arr.len(), 2);
2218 4 : assert_eq!(
2219 4 : arr[0]["value"], 1,
2220 : "noOverwrite leaves the existing default instance"
2221 : );
2222 4 : assert_eq!(arr[1]["value"], 3);
2223 4 : }
2224 : }
2225 :
2226 : #[cfg(test)]
2227 : mod update_result_tests {
2228 : use super::*;
2229 : use crate::federation::{FedReg, Part};
2230 : use http_body_util::BodyExt;
2231 :
2232 40 : fn reg(reg_id: &str, attrs: Option<Vec<String>>) -> FedReg {
2233 40 : FedReg {
2234 40 : reg_id: reg_id.into(),
2235 40 : endpoint: "http://peer:9090".into(),
2236 40 : mode: "inclusive".into(),
2237 40 : attrs,
2238 40 : ..Default::default()
2239 40 : }
2240 40 : }
2241 :
2242 28 : async fn body_of(resp: Response) -> Value {
2243 28 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
2244 28 : serde_json::from_slice(&bytes).expect("json")
2245 28 : }
2246 :
2247 : /// Tables 6.6.3.1-2 / 6.7.3.1-2 — the /attrs 207 body is an
2248 : /// UpdateResult (updated: String[], notUpdated: NotUpdatedDetails[] with
2249 : /// mandatory attributeName+reason, optional registrationId), never the
2250 : /// batch {success, errors} shape.
2251 : #[tokio::test]
2252 4 : async fn distributed_attr_207_is_an_update_result() {
2253 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
2254 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed".to_owned();
2255 4 : let brand = "https://uri.etsi.org/ngsi-ld/default-context/brandName".to_owned();
2256 4 : let all = vec![speed.clone(), brand.clone()];
2257 4 : let regs = vec![reg(
2258 4 : "urn:ngsi-ld:ContextSourceRegistration:csr1",
2259 4 : Some(vec![brand.clone()]),
2260 : )];
2261 4 : let parts = vec![Part::local(504, "distributed operation timed out")];
2262 4 : let resp = combine_attr_parts(
2263 4 : &tenant,
2264 4 : &all,
2265 4 : std::slice::from_ref(&speed),
2266 4 : LocalOutcome::Ok,
2267 4 : vec![speed.clone()],
2268 4 : Vec::new(),
2269 4 : ®s,
2270 4 : &parts,
2271 : );
2272 4 : assert_eq!(resp.status().as_u16(), 207);
2273 4 : let body = body_of(resp).await;
2274 4 : assert_eq!(body["updated"], serde_json::json!([speed]));
2275 4 : assert_eq!(body["notUpdated"][0]["attributeName"], brand);
2276 4 : assert_eq!(
2277 4 : body["notUpdated"][0]["registrationId"],
2278 : "urn:ngsi-ld:ContextSourceRegistration:csr1"
2279 : );
2280 4 : assert!(body["notUpdated"][0]["reason"].is_string());
2281 4 : assert!(body.get("success").is_none(), "not the batch shape");
2282 4 : assert!(body.get("errors").is_none(), "not the batch shape");
2283 4 : }
2284 :
2285 : /// 6.3.17: "In the case of an exclusive or redirect registration, where
2286 : /// all of the data is held outside of the Context Broker and held in a
2287 : /// single registered source, the following errors shall be returned: 508
2288 : /// Loop Detected … 504 Gateway Timeout … 404 Not Found … 502 Bad
2289 : /// Gateway." 207 Multi-Status is reserved for an entity distributed over
2290 : /// multiple endpoints, so a single proxied source's failure passes
2291 : /// through as itself.
2292 : #[tokio::test]
2293 4 : async fn single_proxied_source_error_passes_through_instead_of_207() {
2294 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
2295 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed".to_owned();
2296 8 : for mode in ["exclusive", "redirect"] {
2297 24 : for (status, etype) in [
2298 8 : (404u16, "ResourceNotFound"),
2299 8 : (504, "InternalError"),
2300 8 : (502, "InternalError"),
2301 8 : ] {
2302 24 : let regs = vec![FedReg {
2303 24 : mode: mode.into(),
2304 24 : ..reg("urn:ngsi-ld:ContextSourceRegistration:csr1", None)
2305 24 : }];
2306 24 : let parts = vec![Part::local(
2307 24 : status,
2308 4 : "distributed operation to registration csr1 failed",
2309 4 : )];
2310 24 : let resp = combine_attr_parts(
2311 24 : &tenant,
2312 24 : std::slice::from_ref(&speed),
2313 24 : &[],
2314 4 : // every attribute is held by the proxy: no local half ran
2315 24 : LocalOutcome::Skipped,
2316 24 : Vec::new(),
2317 24 : Vec::new(),
2318 24 : ®s,
2319 24 : &parts,
2320 4 : );
2321 24 : assert_eq!(resp.status().as_u16(), status, "{mode} part {status}");
2322 24 : let body = body_of(resp).await;
2323 24 : assert_eq!(
2324 24 : body["type"],
2325 24 : format!("https://uri.etsi.org/ngsi-ld/errors/{etype}"),
2326 4 : "{mode} part {status}"
2327 4 : );
2328 24 : assert_eq!(body["status"], status);
2329 24 : assert!(
2330 24 : body.get("updated").is_none() && body.get("notUpdated").is_none(),
2331 4 : "a single-source failure is ProblemDetails, not an UpdateResult"
2332 4 : );
2333 4 : }
2334 4 : }
2335 4 : }
2336 :
2337 : /// The same single-source rule must not swallow a MULTI-endpoint failure:
2338 : /// two inclusive registrations over one entity stay 207 (6.3.17).
2339 : #[tokio::test]
2340 4 : async fn two_registrations_still_answer_207() {
2341 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
2342 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed".to_owned();
2343 4 : let regs = vec![
2344 4 : reg("urn:ngsi-ld:ContextSourceRegistration:csr1", None),
2345 4 : reg("urn:ngsi-ld:ContextSourceRegistration:csr2", None),
2346 : ];
2347 4 : let parts = vec![Part::local(204, "ok"), Part::local(404, "not found")];
2348 4 : let resp = combine_attr_parts(
2349 4 : &tenant,
2350 4 : std::slice::from_ref(&speed),
2351 4 : &[],
2352 4 : LocalOutcome::Skipped,
2353 4 : Vec::new(),
2354 4 : Vec::new(),
2355 4 : ®s,
2356 4 : &parts,
2357 : );
2358 4 : assert_eq!(resp.status().as_u16(), 207);
2359 4 : }
2360 :
2361 : /// All halves succeeded → 204; everything missing → 404.
2362 : #[tokio::test]
2363 4 : async fn distributed_attr_success_and_not_found_edges() {
2364 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
2365 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed".to_owned();
2366 4 : let regs = vec![reg("urn:ngsi-ld:ContextSourceRegistration:csr1", None)];
2367 4 : let ok_parts = vec![Part::local(204, "ok")];
2368 4 : let resp = combine_attr_parts(
2369 4 : &tenant,
2370 4 : std::slice::from_ref(&speed),
2371 4 : std::slice::from_ref(&speed),
2372 4 : LocalOutcome::Ok,
2373 4 : vec![speed.clone()],
2374 4 : Vec::new(),
2375 4 : ®s,
2376 4 : &ok_parts,
2377 : );
2378 4 : assert_eq!(resp.status().as_u16(), 204);
2379 : // entity unknown locally AND every forward failed → 404 ProblemDetails
2380 4 : let bad_parts = vec![Part::local(404, "not found")];
2381 4 : let resp = combine_attr_parts(
2382 4 : &tenant,
2383 4 : std::slice::from_ref(&speed),
2384 4 : std::slice::from_ref(&speed),
2385 4 : LocalOutcome::NotFound("entity urn:x not found".into()),
2386 4 : Vec::new(),
2387 4 : Vec::new(),
2388 4 : ®s,
2389 4 : &bad_parts,
2390 : );
2391 4 : assert_eq!(resp.status().as_u16(), 404);
2392 4 : }
2393 : }
|
