Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The driver contract, as executable code.
3 : //!
4 : //! Every rule here is one a caller in `antares-api` relies on and no
5 : //! backend may decide for itself. A driver that passes both functions can be
6 : //! dropped into the broker; one that does not will break requests in ways
7 : //! its own unit tests are free to miss, because a backend's tests assert
8 : //! what that backend does, not what the seam promises.
9 : //!
10 : //! Both functions PANIC on the first violation, naming the rule. The caller
11 : //! supplies two tenants that already exist (a backend may require a tenant
12 : //! row before it accepts writes) and a prefix that makes the ids of this run
13 : //! unique, so a shared database can host several runs at once.
14 : //!
15 : //! Both functions are `async`: a driver is awaited, never blocked on.
16 :
17 : use crate::{CurrentStateDriver, CurrentStateDriverExt, Kind, TemporalDriver, TemporalDriverExt};
18 : use antares_model::TenantId;
19 : use serde_json::{json, Value};
20 :
21 114 : fn doc(id: &str) -> Value {
22 114 : json!({"id": id, "type": "https://uri.etsi.org/ngsi-ld/default-context/T"})
23 114 : }
24 :
25 : /// Hold a current-state driver to the contract `antares-api` writes against.
26 : ///
27 : /// `a` and `b` are two existing tenants; `prefix` namespaces the ids this
28 : /// run creates and deletes. Panics on the first rule broken.
29 6 : pub async fn run_current_state_contract(
30 6 : d: &dyn CurrentStateDriver,
31 6 : a: &TenantId,
32 6 : b: &TenantId,
33 6 : prefix: &str,
34 6 : ) {
35 6 : let e1 = format!("urn:ngsi-ld:{prefix}:1");
36 6 : let e2 = format!("urn:ngsi-ld:{prefix}:2");
37 6 : let gone = format!("urn:ngsi-ld:{prefix}:absent");
38 :
39 : // An absent row is absent, not an error and not an empty document.
40 6 : assert!(
41 6 : d.get(a, Kind::Entity, &gone).await.expect("get").is_none(),
42 : "get of an absent row must answer None"
43 : );
44 6 : assert!(
45 6 : !d.delete(a, Kind::Entity, &gone).await.expect("delete"),
46 : "delete of an absent row must answer false"
47 : );
48 :
49 : // ADR-0005 / ETSI 047_06: a mutate is a read-modify-write under the row
50 : // lock, and a missing row is None — NEVER an insert. A get+upsert
51 : // implementation lets a bookkeeping writeback racing a DELETE resurrect
52 : // the deleted row.
53 6 : let missed = d
54 6 : .mutate::<(), ()>(a, Kind::Entity, &gone, |_| Ok(()))
55 6 : .await
56 6 : .expect("mutate");
57 6 : assert!(missed.is_none(), "mutate of an absent row must answer None");
58 6 : assert!(
59 6 : d.get(a, Kind::Entity, &gone).await.expect("get").is_none(),
60 : "mutate of an absent row must not insert it (047_06)"
61 : );
62 :
63 : // create is create-if-absent: the second one is refused and changes
64 : // nothing.
65 6 : assert!(
66 6 : d.create(a, Kind::Entity, &e1, doc(&e1))
67 6 : .await
68 6 : .expect("create"),
69 : "the first create must report true"
70 : );
71 6 : assert!(
72 6 : !d.create(a, Kind::Entity, &e1, doc("urn:overwritten"))
73 6 : .await
74 6 : .expect("create"),
75 : "a create over an existing id must report false"
76 : );
77 6 : let row = d
78 6 : .get(a, Kind::Entity, &e1)
79 6 : .await
80 6 : .expect("get")
81 6 : .expect("row");
82 6 : assert_eq!(
83 6 : row["id"],
84 6 : e1.as_str(),
85 : "the refused create must not overwrite"
86 : );
87 :
88 : // A mutate that returns Ok commits; one that returns Err commits nothing.
89 6 : let applied = d
90 6 : .mutate::<(), ()>(a, Kind::Entity, &e1, |v| {
91 6 : v["marker"] = json!(1);
92 6 : Ok(())
93 6 : })
94 6 : .await
95 6 : .expect("mutate");
96 6 : assert_eq!(applied, Some(Ok(())), "a mutate over a present row applies");
97 6 : assert_eq!(
98 6 : d.get(a, Kind::Entity, &e1)
99 6 : .await
100 6 : .expect("get")
101 6 : .expect("row")["marker"],
102 6 : json!(1),
103 : "an accepted mutate must be visible to the next read"
104 : );
105 6 : let rejected = d
106 6 : .mutate::<(), &'static str>(a, Kind::Entity, &e1, |v| {
107 6 : v["marker"] = json!(2);
108 6 : Err("rejected")
109 6 : })
110 6 : .await
111 6 : .expect("mutate");
112 6 : assert_eq!(
113 : rejected,
114 : Some(Err("rejected")),
115 : "the closure's error crosses back"
116 : );
117 6 : assert_eq!(
118 6 : d.get(a, Kind::Entity, &e1)
119 6 : .await
120 6 : .expect("get")
121 6 : .expect("row")["marker"],
122 6 : json!(1),
123 : "a rejected mutate must commit nothing"
124 : );
125 :
126 : // A type list, where the driver keeps one, names the type of a live
127 : // Entity: `/types` answers from it without a fold.
128 6 : if let Some(names) = d.entity_type_names(a).await.expect("type list") {
129 6 : assert!(
130 6 : names
131 6 : .iter()
132 6 : .any(|t| t == "https://uri.etsi.org/ngsi-ld/default-context/T"),
133 : "the type list must name the type of a live Entity: {names:?}"
134 : );
135 0 : }
136 :
137 : // An attribute catalogue, where the driver keeps one, lists every
138 : // member a live Entity carries under each of its types, the
139 : // member a write added included.
140 6 : if let Some(pairs) = d.entity_attr_pairs(a).await.expect("attribute list") {
141 6 : let t = "https://uri.etsi.org/ngsi-ld/default-context/T";
142 6 : assert!(
143 12 : pairs.iter().any(|(ty, attr)| ty == t && attr == "marker"),
144 : "the attribute list must name a live Entity's member: {pairs:?}"
145 : );
146 0 : }
147 :
148 : // Tenant isolation, on every read and every write path.
149 6 : assert!(
150 6 : d.get(b, Kind::Entity, &e1).await.expect("get").is_none(),
151 : "another tenant must not read the row"
152 : );
153 6 : assert!(
154 6 : !d.delete(b, Kind::Entity, &e1).await.expect("delete"),
155 : "another tenant must not delete the row"
156 : );
157 6 : assert!(
158 6 : d.mutate::<(), ()>(b, Kind::Entity, &e1, |_| Ok(()))
159 6 : .await
160 6 : .expect("mutate")
161 6 : .is_none(),
162 : "another tenant must not mutate the row"
163 : );
164 6 : assert!(
165 6 : d.list(b, Kind::Entity)
166 6 : .await
167 6 : .expect("list")
168 6 : .iter()
169 6 : .all(|r| r["id"] != e1.as_str()),
170 : "another tenant must not list the row"
171 : );
172 :
173 : // Kinds are separate namespaces: the same id under two kinds is two rows.
174 6 : assert!(
175 6 : d.create(a, Kind::Subscription, &e1, doc(&e1))
176 6 : .await
177 6 : .expect("create"),
178 : "the same id under another kind is a different row"
179 : );
180 6 : assert!(d.delete(a, Kind::Subscription, &e1).await.expect("delete"));
181 6 : assert!(
182 6 : d.get(a, Kind::Entity, &e1).await.expect("get").is_some(),
183 : "deleting one kind must not touch another"
184 : );
185 :
186 : // upsert answers whether a document was ALREADY there, and batch_upsert
187 : // answers the opposite polarity — created-flags, which the batch path
188 : // needs to split 201 from 204 (5.6.8). Getting one of the two backwards
189 : // is invisible until a batch reports every create as an update.
190 6 : assert!(
191 6 : !d.upsert(a, Kind::Entity, &e2, doc(&e2))
192 6 : .await
193 6 : .expect("upsert"),
194 : "upsert that created the row reports false"
195 : );
196 6 : assert!(
197 6 : d.upsert(a, Kind::Entity, &e2, doc(&e2))
198 6 : .await
199 6 : .expect("upsert"),
200 : "upsert over an existing row reports true"
201 : );
202 6 : let e3 = format!("urn:ngsi-ld:{prefix}:3");
203 6 : let flags = d
204 6 : .batch_upsert(a, vec![(e3.clone(), doc(&e3)), (e2.clone(), doc(&e2))])
205 6 : .await
206 6 : .expect("batch_upsert");
207 6 : assert_eq!(
208 : flags,
209 6 : vec![true, false],
210 : "batch_upsert answers created-flags in input order, the opposite \
211 : polarity of upsert"
212 : );
213 6 : assert!(d.delete(a, Kind::Entity, &e3).await.expect("delete"));
214 6 : let ids = vec![e2.clone(), gone.clone(), e1.clone()];
215 6 : let batch = d
216 12 : .batch_mutate::<()>(a, &ids, |_, v| {
217 12 : v["batched"] = json!(true);
218 12 : Ok(())
219 12 : })
220 6 : .await
221 6 : .expect("batch_mutate");
222 6 : assert_eq!(batch.len(), ids.len(), "one result per input id");
223 6 : assert!(batch[0].is_some(), "results align with ids: {ids:?}");
224 6 : assert!(batch[1].is_none(), "an absent id answers None, in place");
225 6 : assert!(batch[2].is_some(), "results align with ids: {ids:?}");
226 :
227 : // A member whose closure refuses commits nothing while the others
228 : // commit: a batch write conditioned on each stored Entity
229 : // refuses member by member, and a refused member stays as it was.
230 6 : let pair = vec![e1.clone(), e2.clone()];
231 6 : let judged = d
232 12 : .batch_mutate::<&'static str>(a, &pair, |id, v| {
233 12 : v["judged"] = json!(true);
234 12 : if id == e2.as_str() {
235 6 : Err("refused")
236 : } else {
237 6 : Ok(())
238 : }
239 12 : })
240 6 : .await
241 6 : .expect("batch_mutate");
242 6 : assert_eq!(
243 : judged,
244 6 : vec![Some(Ok(())), Some(Err("refused"))],
245 : "each member's own answer, in place"
246 : );
247 6 : let accepted = d.get(a, Kind::Entity, &e1).await.expect("get");
248 6 : assert_eq!(
249 6 : accepted.expect("row")["judged"],
250 : json!(true),
251 : "an accepted member commits"
252 : );
253 6 : let refused = d.get(a, Kind::Entity, &e2).await.expect("get");
254 6 : assert!(
255 6 : refused.expect("row").get("judged").is_none(),
256 : "a refused member must commit nothing"
257 : );
258 :
259 : // A query may over-return (the caller re-checks whatever the backend
260 : // could not decide) but must never drop a matching row, and must never
261 : // cross a tenant.
262 6 : let ids_ref: Vec<&str> = vec![&e1, &e2];
263 6 : let mine = d
264 6 : .query_entities(
265 6 : a,
266 6 : &crate::filter::EntityFilter {
267 6 : ids: Some(&ids_ref),
268 6 : ..Default::default()
269 6 : },
270 6 : )
271 6 : .await
272 6 : .expect("query_entities");
273 12 : for want in [&e1, &e2] {
274 12 : assert!(
275 18 : mine.rows.iter().any(|r| r["id"] == want.as_str()),
276 : "query must not drop a matching row: {want}"
277 : );
278 : }
279 6 : let theirs = d
280 6 : .query_entities(
281 6 : b,
282 6 : &crate::filter::EntityFilter {
283 6 : ids: Some(&ids_ref),
284 6 : ..Default::default()
285 6 : },
286 6 : )
287 6 : .await
288 6 : .expect("query_entities");
289 6 : assert!(
290 6 : theirs.rows.is_empty(),
291 : "a query must never cross a tenant: {:?}",
292 : theirs.rows
293 : );
294 :
295 : // Paging pushdown is optional; claiming it and not doing it is not.
296 6 : let paged = d
297 6 : .query_entities(
298 6 : a,
299 6 : &crate::filter::EntityFilter {
300 6 : ids: Some(&ids_ref),
301 6 : page: Some(crate::filter::Page {
302 6 : offset: 0,
303 6 : limit: 1,
304 6 : count: true,
305 6 : }),
306 6 : ..Default::default()
307 6 : },
308 6 : )
309 6 : .await
310 6 : .expect("query_entities");
311 6 : if paged.paged {
312 6 : assert!(
313 6 : paged.rows.len() <= 1,
314 : "a driver reporting paged=true has applied the LIMIT"
315 : );
316 6 : assert!(
317 6 : paged.decided,
318 : "paged implies decided: a LIMIT over an undecided set pages the wrong rows"
319 : );
320 0 : }
321 : // Keyset: `after_id` is optional too, but a driver that pages
322 : // must honour it, or a discovery walk reads its first page forever.
323 6 : let first = std::cmp::min(e1.as_str(), e2.as_str());
324 6 : let after = d
325 6 : .query_entities(
326 6 : a,
327 6 : &crate::filter::EntityFilter {
328 6 : ids: Some(&ids_ref),
329 6 : after_id: Some(first),
330 6 : page: Some(crate::filter::Page {
331 6 : offset: 0,
332 6 : limit: 10,
333 6 : count: false,
334 6 : }),
335 6 : ..Default::default()
336 6 : },
337 6 : )
338 6 : .await
339 6 : .expect("query_entities");
340 6 : if after.paged {
341 6 : assert!(
342 6 : after.rows.iter().all(|r| r["id"].as_str() > Some(first)),
343 : "a paging driver returns only ids after `after_id`: {:?}",
344 : after.rows
345 : );
346 6 : assert_eq!(after.rows.len(), 1, "the one id after {first} is kept");
347 0 : }
348 :
349 : // `subscription_tenants` is the iteration domain the interval sweep and
350 : // both mirror hydrations walk. A backend may return MORE tenants than
351 : // hold a subscription — every caller lists per tenant afterwards and an
352 : // empty list costs nothing. It may never return FEWER: a tenant missing
353 : // here is a tenant whose periodic notifications never fire and whose
354 : // subscriptions never reach the mirror, with no error anywhere.
355 6 : let sub = format!("urn:ngsi-ld:Subscription:{prefix}:1");
356 6 : let sub_doc = json!({
357 6 : "id": &sub,
358 6 : "type": "Subscription",
359 6 : "status": "active",
360 6 : "notification": {"endpoint": {"uri": "http://sink.invalid/n"}},
361 : });
362 6 : d.upsert(a, Kind::Subscription, &sub, sub_doc)
363 6 : .await
364 6 : .expect("upsert subscription");
365 6 : let domain = d
366 6 : .subscription_tenants()
367 6 : .await
368 6 : .expect("subscription_tenants");
369 6 : assert!(
370 26 : domain.iter().any(|t| t == a.as_str()),
371 : "a tenant holding a subscription must appear in subscription_tenants: {domain:?}"
372 : );
373 : // A REGISTRATION puts a tenant in the domain too. One of the hydrations
374 : // that walks this domain fills the registration mirror, and the
375 : // federation path reads that mirror alone once it is installed — so a
376 : // tenant holding registrations and no subscription is a tenant that
377 : // silently forwards to no Context Source.
378 6 : let reg_only = TenantId::new(&format!("{prefix}regonly")).expect("tenant");
379 6 : let reg = format!("urn:ngsi-ld:ContextSourceRegistration:{prefix}:1");
380 6 : d.upsert(
381 6 : ®_only,
382 6 : Kind::Registration,
383 6 : ®,
384 6 : json!({
385 6 : "id": ®,
386 6 : "type": "ContextSourceRegistration",
387 6 : "endpoint": "http://cs.invalid/ngsi-ld/v1",
388 6 : "information": [{"entities": [{"type": "Vehicle"}]}],
389 6 : }),
390 6 : )
391 6 : .await
392 6 : .expect("upsert registration");
393 6 : let domain = d
394 6 : .subscription_tenants()
395 6 : .await
396 6 : .expect("subscription_tenants");
397 6 : assert!(
398 52 : domain.iter().any(|t| t == reg_only.as_str()),
399 : "a tenant holding only a registration must appear in the mirror-hydration \
400 : domain, or its registration mirror hydrates empty: {domain:?}"
401 : );
402 :
403 : // Table 5.2.9-2 forward bookkeeping, pinned here for the same reason as
404 : // 5.2.14.2's: a backend may write it as one statement instead of a
405 : // read-modify-write, and the result must not depend on which it chose.
406 6 : let f1 = "2020-02-01T00:00:00.000Z";
407 6 : let f2 = "2020-02-02T00:00:00.000Z";
408 6 : let ok1 = d
409 6 : .record_forward(®_only, ®, f1, true)
410 6 : .await
411 6 : .expect("record_forward")
412 6 : .expect("the registration is there");
413 6 : assert_eq!(
414 6 : ok1["timesSent"], 1,
415 : "the first forward moves timesSent to 1"
416 : );
417 6 : assert_eq!(ok1["lastSuccess"], f1, "lastSuccess takes the stamp");
418 6 : assert_eq!(ok1["status"], "ok", "a 2xx leaves the registration ok");
419 6 : assert!(
420 6 : ok1.get("timesFailed").is_none() && ok1.get("lastFailure").is_none(),
421 : "a registration that has only succeeded carries no failure members: {ok1}"
422 : );
423 :
424 6 : let bad = d
425 6 : .record_forward(®_only, ®, f2, false)
426 6 : .await
427 6 : .expect("record_forward")
428 6 : .expect("the registration is still there");
429 6 : assert_eq!(
430 6 : bad["timesSent"], 2,
431 : "timesSent counts the failed attempt too (Table 5.2.9-2)"
432 : );
433 6 : assert_eq!(bad["timesFailed"], 1, "the failure moves timesFailed");
434 6 : assert_eq!(bad["lastFailure"], f2, "lastFailure takes the stamp");
435 6 : assert_eq!(bad["status"], "failed", "status names the LAST attempt");
436 6 : assert_eq!(
437 6 : bad["lastSuccess"], f1,
438 : "a failure never rewinds the last success"
439 : );
440 :
441 : // A registration deleted while its forward was in flight has no row to
442 : // book against, and the writeback must not resurrect it.
443 6 : let vanished = format!("urn:ngsi-ld:ContextSourceRegistration:{prefix}:gone");
444 6 : assert!(
445 6 : d.record_forward(®_only, &vanished, f1, true)
446 6 : .await
447 6 : .expect("record_forward")
448 6 : .is_none(),
449 : "an absent registration books nothing"
450 : );
451 6 : assert!(
452 6 : d.get(®_only, Kind::Registration, &vanished)
453 6 : .await
454 6 : .expect("get")
455 6 : .is_none(),
456 : "a bookkeeping writeback must never insert the row it missed"
457 : );
458 :
459 6 : d.delete(®_only, Kind::Registration, ®)
460 6 : .await
461 6 : .expect("delete");
462 : // `list_page` is how the readers that must see EVERY document read —
463 : // the mirror seed above all — so a backend may not refuse it for volume
464 : // the way `list` may (5.5.6 licenses TooManyResults for a query
465 : // operation, which an internal bootstrap is not). Every arm must agree
466 : // on the walk, including the one that takes the trait default: ids
467 : // strictly greater than `after`, id-ordered, at most `limit`, a short
468 : // page means the end, and every stored document served exactly once.
469 6 : let mut ids: Vec<String> = (0..7)
470 42 : .map(|i| format!("urn:ngsi-ld:Subscription:{prefix}:page:{i}"))
471 6 : .collect();
472 42 : for id in &ids {
473 42 : d.upsert(
474 42 : a,
475 42 : Kind::Subscription,
476 42 : id,
477 42 : json!({"id": id, "type": "Subscription"}),
478 42 : )
479 42 : .await
480 42 : .expect("upsert page subscription");
481 : }
482 6 : ids.push(sub.clone());
483 6 : ids.sort();
484 :
485 6 : let mut walked: Vec<String> = Vec::new();
486 6 : let mut after: Option<String> = None;
487 : loop {
488 18 : let page = d
489 18 : .list_page(a, Kind::Subscription, after.as_deref(), 3)
490 18 : .await
491 18 : .expect("list_page");
492 18 : assert!(page.len() <= 3, "a page may not exceed its limit: {page:?}");
493 18 : let short = page.len() < 3;
494 48 : for doc in &page {
495 48 : let id = doc["id"].as_str().expect("a stored doc keeps its id");
496 48 : assert!(
497 48 : after.as_deref().is_none_or(|prev| id > prev),
498 : "`after` is exclusive and the walk ascends: {id} after {after:?}"
499 : );
500 48 : after = Some(id.to_owned());
501 48 : walked.push(id.to_owned());
502 : }
503 18 : if short {
504 6 : break;
505 12 : }
506 : }
507 6 : assert_eq!(
508 : walked, ids,
509 : "the walk must serve every subscription of the tenant exactly once, in id order"
510 : );
511 : // The other tenant's documents are not on this walk. 4.14: operations
512 : // "only apply to the information of the specified `Tenant` in isolation".
513 6 : assert!(
514 6 : d.list_page(b, Kind::Subscription, None, 100)
515 6 : .await
516 6 : .expect("list_page")
517 6 : .is_empty(),
518 : "a tenant with no subscriptions pages empty, whatever another tenant holds"
519 : );
520 : // Entities page like every other kind. 5.9.2.4's registration-vs-entity
521 : // conflict check reads them this way — it must see every Entity of the
522 : // tenant and has no TooManyResults to raise — so a backend may not refuse
523 : // this walk for volume either, and may not build the page by
524 : // materializing the tenant first.
525 : //
526 : // 4.22 applies INSIDE the page: an expired Entity is not there to be
527 : // served, and dropping one after the limit was applied would hand back a
528 : // short page, which every walker reads as the end of the tenant.
529 6 : let stem = format!("urn:ngsi-ld:{prefix}:page:");
530 24 : let live: Vec<String> = (0..4).map(|i| format!("{stem}live{i}")).collect();
531 : // sorts between live1 and live2, so it falls inside the first page below
532 6 : let expired = format!("{stem}live1x");
533 24 : for id in &live {
534 24 : d.upsert(a, Kind::Entity, id, doc(id))
535 24 : .await
536 24 : .expect("upsert page entity");
537 : }
538 6 : let mut gone_doc = doc(&expired);
539 6 : gone_doc["expiresAt"] = json!("2000-01-01T00:00:00.000Z");
540 6 : d.upsert(a, Kind::Entity, &expired, gone_doc)
541 6 : .await
542 6 : .expect("upsert expired entity");
543 :
544 6 : let page = d
545 6 : .list_page(a, Kind::Entity, Some(&stem), 3)
546 6 : .await
547 6 : .expect("list_page entities");
548 18 : let served: Vec<&str> = page.iter().filter_map(|d| d["id"].as_str()).collect();
549 6 : assert_eq!(
550 : served,
551 6 : [live[0].as_str(), live[1].as_str(), live[2].as_str()],
552 : "an expired entity may neither be served nor consume a slot of the page"
553 : );
554 :
555 6 : let mut walked: Vec<String> = Vec::new();
556 6 : let mut after = stem.clone();
557 : loop {
558 18 : let page = d
559 18 : .list_page(a, Kind::Entity, Some(&after), 2)
560 18 : .await
561 18 : .expect("list_page entities");
562 18 : let short = page.len() < 2;
563 18 : let mut moved = false;
564 24 : for doc in &page {
565 24 : let id = doc["id"].as_str().expect("a stored entity keeps its id");
566 24 : assert!(
567 24 : id > after.as_str(),
568 : "`after` is exclusive: {id} after {after}"
569 : );
570 24 : after = id.to_owned();
571 24 : moved = true;
572 24 : if id.starts_with(stem.as_str()) {
573 24 : walked.push(id.to_owned());
574 24 : }
575 : }
576 18 : if short || !moved || !after.starts_with(stem.as_str()) {
577 6 : break;
578 12 : }
579 : }
580 6 : assert_eq!(
581 : walked, live,
582 : "the walk must serve every live entity of the tenant exactly once, in id \
583 : order, and never the expired one"
584 : );
585 30 : for id in live.iter().chain(std::iter::once(&expired)) {
586 30 : d.delete(a, Kind::Entity, id)
587 30 : .await
588 30 : .expect("delete page entity");
589 : }
590 :
591 : // 4.22 at the read boundary, the part a backend must not decide for
592 : // itself. The stamp marks "a certain Entity, Property or Relationship"
593 : // invalid; it is not a store-wide rule that hides any document carrying
594 : // one, and it reaches every Attribute of an Entity, sub-Attributes
595 : // included.
596 6 : let exp_e = format!("urn:ngsi-ld:{prefix}:expiry:entity");
597 6 : let mut expired_entity = doc(&exp_e);
598 6 : expired_entity["expiresAt"] = json!("2000-01-01T00:00:00.000Z");
599 6 : d.upsert(a, Kind::Entity, &exp_e, expired_entity)
600 6 : .await
601 6 : .expect("upsert expired entity");
602 6 : assert!(
603 6 : d.get(a, Kind::Entity, &exp_e).await.expect("get").is_none(),
604 : "an Entity past its expiresAt reads absent, not just out of a page"
605 : );
606 6 : d.delete(a, Kind::Entity, &exp_e).await.expect("delete");
607 :
608 : // An Attribute whose every instance expired leaves; the Entity carrying
609 : // it does not, and neither does a live sibling.
610 6 : let exp_a = format!("urn:ngsi-ld:{prefix}:expiry:attr");
611 78 : let iri = |n: &str| format!("https://uri.etsi.org/ngsi-ld/default-context/{n}");
612 6 : let mut with_attrs = doc(&exp_a);
613 6 : with_attrs[iri("gone")] = json!([{"value": 1, "expiresAt": "2000-01-01T00:00:00.000Z"}]);
614 6 : with_attrs[iri("kept")] = json!([{
615 6 : "value": 2,
616 6 : iri("subgone"): [{"value": 3, "expiresAt": "2000-01-01T00:00:00.000Z"}],
617 6 : iri("subkept"): [{"value": 4}],
618 : }]);
619 : // 4.6.3 leaves the seconds-fraction separator open, and ',' (0x2C) sorts
620 : // before both '.' and 'Z', so the two stamps have to be read as instants:
621 : // a backend that compares them as bytes serves this live Attribute as
622 : // expired. Every backend answers the same here or it is not the same
623 : // store.
624 6 : with_attrs[iri("commakept")] = json!([{"value": 5, "expiresAt": "2999-01-01T00:00:00,500Z"}]);
625 6 : d.upsert(a, Kind::Entity, &exp_a, with_attrs)
626 6 : .await
627 6 : .expect("upsert entity with an expired attribute");
628 6 : let served = d
629 6 : .get(a, Kind::Entity, &exp_a)
630 6 : .await
631 6 : .expect("get")
632 6 : .expect("an Entity outlives the expiry of one of its Attributes");
633 6 : assert!(
634 6 : served.get(iri("gone")).is_none(),
635 : "an Attribute whose only instance expired is not served"
636 : );
637 6 : assert_eq!(
638 6 : served[iri("kept")][0]["value"],
639 6 : json!(2),
640 : "a live Attribute survives its sibling's expiry"
641 : );
642 6 : assert!(
643 6 : served[iri("kept")][0].get(iri("subgone")).is_none(),
644 : "a sub-Attribute is a Property or Relationship too: past its stamp it \
645 : is not served"
646 : );
647 6 : assert_eq!(
648 6 : served[iri("kept")][0][iri("subkept")][0]["value"],
649 6 : json!(4),
650 : "a live sub-Attribute survives its sibling's expiry"
651 : );
652 6 : assert_eq!(
653 6 : served[iri("commakept")][0]["value"],
654 6 : json!(5),
655 : "a comma seconds-fraction is an instant, not a byte string"
656 : );
657 : // attrs= asks for an Entity that holds one of them NOW: an Attribute
658 : // whose every instance expired is not held (4.22), so an answer that
659 : // claims to be decided must not name the Entity for it. An undecided
660 : // one is re-judged by the caller over the served (stripped) rows.
661 6 : let only_gone = [iri("gone")];
662 6 : let by_attr = d
663 6 : .query_entities(
664 6 : a,
665 6 : &crate::filter::EntityFilter {
666 6 : attrs: Some(&only_gone),
667 6 : ..Default::default()
668 6 : },
669 6 : )
670 6 : .await
671 6 : .expect("query_entities");
672 6 : assert!(
673 6 : !by_attr.decided || by_attr.rows.iter().all(|r| r["id"] != exp_a.as_str()),
674 : "attrs= over an Attribute whose only instance expired: a decided \
675 : answer named the Entity"
676 : );
677 6 : assert!(
678 6 : by_attr
679 6 : .rows
680 6 : .iter()
681 6 : .filter(|r| r["id"] == exp_a.as_str())
682 6 : .all(|r| r.get(iri("gone")).is_none()),
683 : "a query served an expired Attribute"
684 : );
685 6 : d.delete(a, Kind::Entity, &exp_a).await.expect("delete");
686 :
687 : // batch_upsert takes each id once: the API splits repeats into rounds
688 : // (5.6.8 / 5.6.10), and what a backend does with a repeat inside one
689 : // call is its own guess (Postgres refuses an ON CONFLICT row affected
690 : // twice). A repeat is refused on every arm, and nothing is written.
691 6 : let dup = format!("urn:ngsi-ld:{prefix}:dup");
692 6 : assert!(
693 6 : d.batch_upsert(a, vec![(dup.clone(), doc(&dup)), (dup.clone(), doc(&dup))])
694 6 : .await
695 6 : .is_err(),
696 : "batch_upsert with an id twice must be refused"
697 : );
698 6 : assert!(
699 6 : d.get(a, Kind::Entity, &dup).await.expect("get").is_none(),
700 : "a refused batch_upsert wrote nothing"
701 : );
702 :
703 : // 5.8.6: an expired SUBSCRIPTION is not deleted and stays retrievable —
704 : // the API turns the stamp into status "expired" and keeps it updatable.
705 : // A backend that hid every document with a past expiresAt would lose it.
706 6 : let exp_s = format!("urn:ngsi-ld:{prefix}:expiry:sub");
707 6 : d.upsert(
708 6 : a,
709 6 : Kind::Subscription,
710 6 : &exp_s,
711 6 : json!({
712 6 : "id": exp_s,
713 6 : "type": "Subscription",
714 6 : "expiresAt": "2000-01-01T00:00:00.000Z",
715 6 : }),
716 6 : )
717 6 : .await
718 6 : .expect("upsert expired subscription");
719 6 : assert!(
720 6 : d.get(a, Kind::Subscription, &exp_s)
721 6 : .await
722 6 : .expect("get")
723 6 : .is_some(),
724 : "an expired Subscription stays retrievable (5.8.6)"
725 : );
726 6 : d.delete(a, Kind::Subscription, &exp_s)
727 6 : .await
728 6 : .expect("delete");
729 :
730 : // `delete_entity_if` decides and deletes under one lock, and it decides
731 : // on the STORED document. 5.6.6.4: an Entity the caller's selector
732 : // excludes "is not known" for the operation, which is the same answer an
733 : // absent Entity gets — so a refusal and a miss are both `false`, and a
734 : // refusal must leave the document exactly where it was.
735 6 : let cond = format!("urn:ngsi-ld:{prefix}:cond");
736 6 : d.upsert(a, Kind::Entity, &cond, doc(&cond))
737 6 : .await
738 6 : .expect("upsert conditional-delete entity");
739 6 : assert!(
740 6 : !d.delete_entity_if(a, &cond, &|_| false)
741 6 : .await
742 6 : .expect("delete_entity_if"),
743 : "a refused predicate reports no deletion"
744 : );
745 6 : assert!(
746 6 : d.get(a, Kind::Entity, &cond).await.expect("get").is_some(),
747 : "a refused delete leaves the document in place"
748 : );
749 : // 4.14: the predicate never even runs for another tenant's document.
750 6 : assert!(
751 6 : !d.delete_entity_if(b, &cond, &|_| panic!(
752 : "another tenant's document reached the predicate"
753 : ))
754 6 : .await
755 6 : .expect("delete_entity_if"),
756 : "an Entity of another tenant is absent here"
757 : );
758 6 : assert!(
759 6 : d.get(a, Kind::Entity, &cond).await.expect("get").is_some(),
760 : "a delete addressed to another tenant may not touch this one"
761 : );
762 6 : assert!(
763 6 : d.delete_entity_if(a, &cond, &|v| v["id"] == cond.as_str())
764 6 : .await
765 6 : .expect("delete_entity_if"),
766 : "the predicate is handed the stored document, not the id"
767 : );
768 6 : assert!(
769 6 : d.get(a, Kind::Entity, &cond).await.expect("get").is_none(),
770 : "an accepted predicate deletes"
771 : );
772 6 : assert!(
773 6 : !d.delete_entity_if(a, &cond, &|_| true)
774 6 : .await
775 6 : .expect("delete_entity_if"),
776 : "an absent Entity is not deleted, whatever the predicate answers"
777 : );
778 :
779 : // `list_slice` is the window a 5.5.9.2 limit/offset listing serves from,
780 : // and it must agree with the walk above on both the order and the set:
781 : // the same ids, the same order, and a total that counts the whole match
782 : // set rather than the page. A backend may not refuse it for volume — the
783 : // window bounds the result by construction.
784 6 : let (page, total) = d
785 6 : .list_slice(a, Kind::Subscription, 0, 3)
786 6 : .await
787 6 : .expect("list_slice");
788 6 : assert_eq!(total, ids.len(), "the total counts the set, not the page");
789 6 : assert_eq!(page.len(), 3, "limit is a maximum, and 3 were available");
790 18 : for (i, doc) in page.iter().enumerate() {
791 18 : assert_eq!(
792 18 : doc["id"].as_str(),
793 18 : Some(ids[i].as_str()),
794 : "the window is the id-ordered prefix: {page:?}"
795 : );
796 : }
797 : // Every single-element window lands on the element the walk has there.
798 48 : for (i, want) in ids.iter().enumerate() {
799 48 : let (page, _) = d
800 48 : .list_slice(a, Kind::Subscription, i, 1)
801 48 : .await
802 48 : .expect("list_slice");
803 48 : assert_eq!(
804 48 : page.first().and_then(|d| d["id"].as_str()),
805 48 : Some(want.as_str()),
806 : "offset {i} served the wrong element: {page:?}"
807 : );
808 : }
809 : // limit 0 is legal (6.3.10, with count): the count without the page.
810 6 : let (page, total) = d
811 6 : .list_slice(a, Kind::Subscription, 0, 0)
812 6 : .await
813 6 : .expect("list_slice");
814 6 : assert!(page.is_empty(), "limit 0 returns no elements: {page:?}");
815 6 : assert_eq!(total, ids.len(), "limit 0 still counts the whole set");
816 : // An offset past the end is an empty page, not an error, and the total
817 : // is unchanged by where the client asked to start.
818 6 : let (page, total) = d
819 6 : .list_slice(a, Kind::Subscription, ids.len() + 10, 5)
820 6 : .await
821 6 : .expect("list_slice");
822 6 : assert!(page.is_empty(), "past the end is empty: {page:?}");
823 6 : assert_eq!(total, ids.len(), "an offset does not change the match set");
824 : // 4.14 again: the window is inside one tenant.
825 6 : let (page, total) = d
826 6 : .list_slice(b, Kind::Subscription, 0, 100)
827 6 : .await
828 6 : .expect("slice");
829 6 : assert!(
830 6 : page.is_empty() && total == 0,
831 : "another tenant's rows reached this window: {page:?} / {total}"
832 : );
833 :
834 48 : for id in ids.iter().filter(|i| *i != &sub) {
835 42 : d.delete(a, Kind::Subscription, id).await.expect("delete");
836 : }
837 :
838 : // 5.2.14.2 delivery bookkeeping. A backend may write this as one
839 : // statement instead of a read-modify-write; the result must not depend on
840 : // which it chose, so the rule is pinned here rather than in one backend's
841 : // own tests.
842 6 : let t1 = "2020-01-01T00:00:00.000Z";
843 6 : let t2 = "2020-01-02T00:00:00.000Z";
844 6 : let first = d
845 6 : .record_delivery(a, Kind::Subscription, &sub, t1)
846 6 : .await
847 6 : .expect("record_delivery")
848 6 : .expect("the subscription is there");
849 6 : let n1 = &first.doc["notification"];
850 6 : assert_eq!(n1["timesSent"], 1, "the first attempt moves timesSent to 1");
851 6 : assert_eq!(
852 6 : n1["lastNotification"], t1,
853 : "lastNotification takes the stamp"
854 : );
855 6 : assert_eq!(n1["lastSuccess"], t1, "lastSuccess takes the stamp");
856 6 : assert_eq!(n1["status"], "ok", "the attempt leaves the notification ok");
857 6 : assert!(
858 6 : first.doc.get("status").is_none(),
859 : "the top-level status is a rendered member, not a stored one: {}",
860 : first.doc
861 : );
862 6 : assert!(
863 6 : first.prev_success.is_none(),
864 : "a subscription that never succeeded has no previous lastSuccess"
865 : );
866 :
867 6 : let second = d
868 6 : .record_delivery(a, Kind::Subscription, &sub, t2)
869 6 : .await
870 6 : .expect("record_delivery")
871 6 : .expect("the subscription is still there");
872 6 : assert_eq!(
873 6 : second.doc["notification"]["timesSent"], 2,
874 : "timesSent counts attempts, it does not reset"
875 : );
876 6 : assert_eq!(
877 6 : second.prev_success.as_ref().and_then(Value::as_str),
878 6 : Some(t1),
879 : "the overwritten lastSuccess comes back, or a failed attempt cannot \
880 : roll it back"
881 : );
882 :
883 : // 5.8.6: a subscription deleted between matching and delivery has no row
884 : // to book against, and nothing may be sent.
885 6 : assert!(
886 6 : d.record_delivery(a, Kind::Subscription, &gone, t1)
887 6 : .await
888 6 : .expect("record_delivery")
889 6 : .is_none(),
890 : "an absent subscription books nothing"
891 : );
892 :
893 : // the batched stamp is the same rule, one answer per attempt in
894 : // input order, and a batch never books another tenant's row.
895 6 : let t3 = "2020-01-03T00:00:00.000Z";
896 6 : let batch = d
897 6 : .record_deliveries(
898 6 : a,
899 6 : Kind::Subscription,
900 6 : &[(gone.as_str(), t3), (sub.as_str(), t3)],
901 6 : )
902 6 : .await
903 6 : .expect("record_deliveries");
904 6 : assert_eq!(batch.len(), 2, "one answer per attempt: {batch:?}");
905 6 : assert!(
906 6 : batch.first().is_some_and(Option::is_none),
907 : "an absent subscription books nothing"
908 : );
909 6 : let third = batch
910 6 : .get(1)
911 6 : .and_then(Option::as_ref)
912 6 : .expect("the subscription is booked");
913 6 : assert_eq!(third.doc["notification"]["timesSent"], 3);
914 6 : assert_eq!(third.doc["notification"]["lastSuccess"], t3);
915 6 : assert_eq!(
916 6 : third.prev_success.as_ref().and_then(Value::as_str),
917 6 : Some(t2),
918 : "each attempt of a batch gets back the lastSuccess it overwrote"
919 : );
920 6 : let elsewhere = d
921 6 : .record_deliveries(b, Kind::Subscription, &[(sub.as_str(), t3)])
922 6 : .await
923 6 : .expect("record_deliveries");
924 6 : assert!(
925 6 : elsewhere.first().is_some_and(Option::is_none),
926 : "a batch in another tenant books nothing: {elsewhere:?}"
927 : );
928 :
929 6 : assert!(
930 6 : d.delete(a, Kind::Subscription, &sub).await.expect("delete"),
931 : "the contract's own subscription must be removable"
932 : );
933 :
934 : // ADR-0021: a stored @context is a Tenant's document. "Hosted" and
935 : // "ImplicitlyCreated" rows hold term mappings authored through one
936 : // Tenant's requests, and 5.5.7 makes those mappings decide what that
937 : // Tenant's payloads mean, so a backend that answered every caller would
938 : // hand one Tenant's meaning of its own data to another. "Cached" is a
939 : // copy of a public document (5.13.1) and belongs to no Tenant.
940 6 : let hosted_id = format!("{prefix}-ctx-hosted");
941 6 : let cached_id = format!("{prefix}-ctx-cached");
942 6 : let hosted = json!({"localId": hosted_id, "kind": "Hosted", "owner": a.as_str(),
943 6 : "body": {"@context": {"t": "https://example.org/t"}}});
944 6 : d.context_put(Some(a), &hosted_id, hosted.clone())
945 6 : .await
946 6 : .expect("the owning tenant stores its own @context");
947 6 : d.context_put(
948 6 : None,
949 6 : &cached_id,
950 6 : json!({"localId": cached_id, "kind": "Cached", "body": {"@context": {}}}),
951 6 : )
952 6 : .await
953 6 : .expect("a Cached copy belongs to no tenant");
954 :
955 6 : assert!(
956 6 : d.context_get(Some(a), &hosted_id)
957 6 : .await
958 6 : .expect("context_get")
959 6 : .is_some(),
960 : "the owning tenant must read its own @context"
961 : );
962 6 : assert!(
963 6 : d.context_get(Some(b), &hosted_id)
964 6 : .await
965 6 : .expect("context_get")
966 6 : .is_none(),
967 : "another tenant's Hosted @context must be as absent as one never stored"
968 : );
969 6 : assert!(
970 6 : d.context_get(None, &hosted_id)
971 6 : .await
972 6 : .expect("context_get")
973 6 : .is_none(),
974 : "no tenant in scope must reach a tenant's @context"
975 : );
976 6 : assert!(
977 6 : d.context_list_meta(Some(b))
978 6 : .await
979 6 : .expect("context_list_meta")
980 6 : .iter()
981 8 : .all(|r| r["localId"] != hosted_id.as_str()),
982 : "another tenant's @context must not be listed"
983 : );
984 6 : assert!(
985 6 : !d.context_delete(Some(b), &hosted_id)
986 6 : .await
987 6 : .expect("context_delete"),
988 : "another tenant must not delete it"
989 : );
990 6 : assert!(
991 6 : d.context_put(Some(b), &hosted_id, hosted).await.is_err(),
992 : "another tenant must not overwrite it either"
993 : );
994 6 : assert!(
995 6 : d.context_get(Some(a), &hosted_id)
996 6 : .await
997 6 : .expect("context_get")
998 6 : .is_some(),
999 : "the owner's @context must survive every foreign attempt"
1000 : );
1001 18 : for t in [Some(a), Some(b), None] {
1002 18 : assert!(
1003 18 : d.context_get(t, &cached_id)
1004 18 : .await
1005 18 : .expect("context_get")
1006 18 : .is_some(),
1007 : "a Cached copy is a public document every tenant reaches"
1008 : );
1009 : }
1010 6 : assert!(
1011 6 : d.context_delete(Some(a), &hosted_id)
1012 6 : .await
1013 6 : .expect("context_delete"),
1014 : "the contract's own @context must be removable by its owner"
1015 : );
1016 6 : assert!(
1017 6 : d.context_delete(None, &cached_id)
1018 6 : .await
1019 6 : .expect("context_delete"),
1020 : "and the Cached copy by anyone"
1021 : );
1022 :
1023 6 : context_rows_are_locked(d, a, b, prefix).await;
1024 :
1025 12 : for id in [&e1, &e2] {
1026 12 : assert!(
1027 12 : d.delete(a, Kind::Entity, id).await.expect("delete"),
1028 : "the contract's own rows must be removable"
1029 : );
1030 : }
1031 6 : }
1032 :
1033 : /// Drive two futures to completion together on the calling task, so their
1034 : /// store calls interleave (the crate carries no executor of its own).
1035 6 : async fn both<A: std::future::Future, B: std::future::Future>(
1036 6 : x: A,
1037 6 : y: B,
1038 6 : ) -> (A::Output, B::Output) {
1039 6 : let (mut x, mut y) = (Box::pin(x), Box::pin(y));
1040 6 : let (mut xo, mut yo) = (None, None);
1041 1686 : std::future::poll_fn(|cx| {
1042 1686 : if xo.is_none() {
1043 37 : if let std::task::Poll::Ready(v) = std::future::Future::poll(x.as_mut(), cx) {
1044 6 : xo = Some(v);
1045 31 : }
1046 1649 : }
1047 1686 : if yo.is_none() {
1048 1686 : if let std::task::Poll::Ready(v) = std::future::Future::poll(y.as_mut(), cx) {
1049 6 : yo = Some(v);
1050 1680 : }
1051 0 : }
1052 1686 : match (xo.take(), yo.take()) {
1053 6 : (Some(a), Some(b)) => std::task::Poll::Ready((a, b)),
1054 1680 : (a, b) => {
1055 1680 : xo = a;
1056 1680 : yo = b;
1057 1680 : std::task::Poll::Pending
1058 : }
1059 : }
1060 1686 : })
1061 6 : .await
1062 6 : }
1063 :
1064 : /// 5.13.3.5 + ADR-0021 for the @context rows: a row has a kind, and a
1065 : /// read-modify-write of one holds its lock, so a usage bump racing a
1066 : /// delete leaves it deleted and two racing bumps both count.
1067 6 : async fn context_rows_are_locked(
1068 6 : d: &dyn CurrentStateDriver,
1069 6 : a: &TenantId,
1070 6 : b: &TenantId,
1071 6 : prefix: &str,
1072 6 : ) {
1073 6 : let id = format!("{prefix}-ctx-bumped");
1074 6 : assert!(
1075 6 : d.context_put(None, &id, json!({"localId": id, "body": {"@context": {}}}))
1076 6 : .await
1077 6 : .is_err(),
1078 : "a row without a kind must be refused"
1079 : );
1080 30 : let bump = || -> crate::MutateFn<'static> {
1081 30 : Box::new(|row| {
1082 12 : let o = row.as_object_mut().ok_or(())?;
1083 12 : let n = o.get("numberOfHits").and_then(Value::as_u64).unwrap_or(0);
1084 12 : o.insert("numberOfHits".into(), json!(n.saturating_add(1)));
1085 12 : Ok(())
1086 12 : })
1087 30 : };
1088 6 : assert!(
1089 6 : d.context_mutate(None, &id, bump())
1090 6 : .await
1091 6 : .expect("context_mutate")
1092 6 : .is_none(),
1093 : "context_mutate of an absent row must answer None"
1094 : );
1095 6 : assert!(
1096 6 : d.context_get(None, &id)
1097 6 : .await
1098 6 : .expect("context_get")
1099 6 : .is_none(),
1100 : "context_mutate of an absent row must not insert it"
1101 : );
1102 6 : d.context_put(
1103 6 : None,
1104 6 : &id,
1105 6 : json!({"localId": id, "kind": "Cached", "numberOfHits": 0, "body": {"@context": {}}}),
1106 6 : )
1107 6 : .await
1108 6 : .expect("context_put");
1109 6 : let (x, y) = both(
1110 6 : d.context_mutate(Some(a), &id, bump()),
1111 6 : d.context_mutate(Some(b), &id, bump()),
1112 : )
1113 6 : .await;
1114 6 : assert!(
1115 6 : matches!((x, y), (Ok(Some(Ok(()))), Ok(Some(Ok(()))))),
1116 : "both bumps of a Cached row must apply"
1117 : );
1118 6 : let hits = d
1119 6 : .context_get(None, &id)
1120 6 : .await
1121 6 : .expect("context_get")
1122 6 : .map(|r| r["numberOfHits"].clone());
1123 6 : assert_eq!(hits, Some(json!(2)), "two racing bumps must both count");
1124 :
1125 6 : let hosted = format!("{prefix}-ctx-owned");
1126 6 : d.context_put(
1127 6 : Some(a),
1128 6 : &hosted,
1129 6 : json!({"localId": hosted, "kind": "Hosted", "owner": a.as_str(), "body": {"@context": {}}}),
1130 6 : )
1131 6 : .await
1132 6 : .expect("context_put");
1133 6 : assert!(
1134 6 : d.context_mutate(Some(b), &hosted, bump())
1135 6 : .await
1136 6 : .expect("context_mutate")
1137 6 : .is_none(),
1138 : "another tenant's @context must be as absent to a mutate as to a read"
1139 : );
1140 6 : assert!(
1141 6 : d.context_delete(Some(a), &hosted)
1142 6 : .await
1143 6 : .expect("context_delete"),
1144 : "the owner removes its @context"
1145 : );
1146 6 : assert!(
1147 6 : d.context_mutate(Some(a), &hosted, bump())
1148 6 : .await
1149 6 : .expect("context_mutate")
1150 6 : .is_none(),
1151 : "a bump after the delete must find nothing"
1152 : );
1153 6 : assert!(
1154 6 : d.context_get(Some(a), &hosted)
1155 6 : .await
1156 6 : .expect("context_get")
1157 6 : .is_none(),
1158 : "a bump racing a delete must leave the row deleted"
1159 : );
1160 6 : assert!(d.context_delete(None, &id).await.expect("context_delete"));
1161 6 : }
1162 :
1163 : /// Hold a temporal driver to the same shape. A driver that answers
1164 : /// `supported() == false` declares it keeps no history and is held to
1165 : /// nothing else.
1166 6 : pub async fn run_temporal_contract(
1167 6 : d: &dyn TemporalDriver,
1168 6 : a: &TenantId,
1169 6 : b: &TenantId,
1170 6 : prefix: &str,
1171 6 : ) {
1172 6 : if !d.supported() {
1173 0 : return;
1174 6 : }
1175 6 : let e1 = format!("urn:ngsi-ld:{prefix}:t1");
1176 6 : let gone = format!("urn:ngsi-ld:{prefix}:tabsent");
1177 :
1178 6 : assert!(
1179 6 : d.get(a, &gone).await.expect("get").is_none(),
1180 : "get of an absent temporal document must answer None"
1181 : );
1182 6 : assert!(
1183 6 : !d.delete(a, &gone).await.expect("delete"),
1184 : "delete of an absent temporal document must answer false"
1185 : );
1186 6 : let missed = d
1187 6 : .mutate::<(), ()>(a, &gone, |_| Ok(()))
1188 6 : .await
1189 6 : .expect("mutate");
1190 6 : assert!(
1191 6 : missed.is_none(),
1192 : "mutate of an absent document must answer None"
1193 : );
1194 6 : assert!(
1195 6 : d.get(a, &gone).await.expect("get").is_none(),
1196 : "mutate of an absent document must not insert it (047_06)"
1197 : );
1198 :
1199 6 : assert!(
1200 6 : d.create(a, &e1, doc(&e1)).await.expect("create"),
1201 : "the first create reports true"
1202 : );
1203 6 : assert!(
1204 6 : !d.create(a, &e1, doc(&e1)).await.expect("create"),
1205 : "a create over an existing id reports false"
1206 : );
1207 6 : let rejected = d
1208 6 : .mutate::<(), &'static str>(a, &e1, |v| {
1209 6 : v["marker"] = json!(2);
1210 6 : Err("rejected")
1211 6 : })
1212 6 : .await
1213 6 : .expect("mutate");
1214 6 : assert_eq!(rejected, Some(Err("rejected")));
1215 6 : assert!(
1216 6 : d.get(a, &e1)
1217 6 : .await
1218 6 : .expect("get")
1219 6 : .expect("row")
1220 6 : .get("marker")
1221 6 : .is_none(),
1222 : "a rejected mutate must commit nothing"
1223 : );
1224 :
1225 6 : assert!(
1226 6 : d.get(b, &e1).await.expect("get").is_none(),
1227 : "another tenant must not read the temporal document"
1228 : );
1229 6 : assert!(
1230 6 : !d.delete(b, &e1).await.expect("delete"),
1231 : "another tenant must not delete it"
1232 : );
1233 6 : assert!(
1234 6 : d.list(b)
1235 6 : .await
1236 6 : .expect("list")
1237 6 : .iter()
1238 6 : .all(|r| r["id"] != e1.as_str()),
1239 : "another tenant must not list it"
1240 : );
1241 :
1242 6 : assert!(
1243 6 : d.delete(a, &e1).await.expect("delete"),
1244 : "the contract's own rows must be removable"
1245 : );
1246 6 : }
1247 :
1248 : /// Hold a driver to tenant isolation: `a` and `b` each hold an
1249 : /// Entity, a Subscription, a Registration and a temporal document under the
1250 : /// SAME ids, and nothing done as `a` reads, changes or deletes `b`'s copy.
1251 : /// The walk covers every tenant-taking family the API reaches: one-row and
1252 : /// batch writes, the three list reads, the query, delivery bookkeeping, the
1253 : /// registration lookup, the temporal reads and the drain, and last a purge
1254 : /// of `a`, which must leave `b` whole.
1255 : ///
1256 : /// The caller picks the pair that matters for its backend: two names a
1257 : /// LIKE pattern would confuse (`a_b`, `axb`), the default Tenant beside a
1258 : /// named one, a name that extends the other. `a` is purged, so it must not
1259 : /// be a Tenant other data lives in. `cs` and `tp` may be the same store.
1260 24 : pub async fn run_tenant_isolation_contract(
1261 24 : cs: &dyn CurrentStateDriver,
1262 24 : tp: &dyn TemporalDriver,
1263 24 : a: &TenantId,
1264 24 : b: &TenantId,
1265 24 : prefix: &str,
1266 24 : ) {
1267 : use crate::filter::{EntityFilter, TemporalFilter};
1268 24 : assert_ne!(a, b, "the walk needs two Tenants");
1269 24 : let e = format!("urn:ngsi-ld:{prefix}:shared");
1270 24 : let sub = format!("urn:ngsi-ld:Subscription:{prefix}:shared");
1271 24 : let reg = format!("urn:ngsi-ld:ContextSourceRegistration:{prefix}:shared");
1272 24 : let ty = "https://uri.etsi.org/ngsi-ld/default-context/T";
1273 312 : let mine = |t: &TenantId| json!({"id": e, "type": ty, "who": t.as_str()});
1274 24 : let ids = [e.clone()];
1275 24 : let ids_ref = [e.as_str()];
1276 :
1277 48 : for t in [a, b] {
1278 48 : assert!(
1279 48 : cs.create(t, Kind::Entity, &e, mine(t))
1280 48 : .await
1281 48 : .expect("create"),
1282 : "{}: the shared id is new in each Tenant",
1283 0 : t.as_str()
1284 : );
1285 48 : cs.create(
1286 48 : t,
1287 48 : Kind::Subscription,
1288 48 : &sub,
1289 48 : json!({"id": sub, "type": "Subscription", "who": t.as_str()}),
1290 48 : )
1291 48 : .await
1292 48 : .expect("subscription");
1293 48 : cs.create(
1294 48 : t,
1295 48 : Kind::Registration,
1296 48 : ®,
1297 48 : json!({"id": reg, "type": "ContextSourceRegistration", "who": t.as_str(),
1298 48 : "information": [{"entities": [{"type": ty}]}], "endpoint": "http://cs.example"}),
1299 48 : )
1300 48 : .await
1301 48 : .expect("registration");
1302 : }
1303 : // `b`'s copies as they stand; every step below compares against these.
1304 24 : let b_entity = cs.get(b, Kind::Entity, &e).await.expect("get");
1305 24 : let b_sub = cs.get(b, Kind::Subscription, &sub).await.expect("get");
1306 : // A macro, not a closure: the check awaits, and an async closure that
1307 : // borrows the ids cannot be called twice.
1308 : macro_rules! b_unchanged {
1309 : ($step:literal) => {
1310 : assert_eq!(
1311 : cs.get(b, Kind::Entity, &e).await.expect("get"),
1312 : b_entity,
1313 : "{} as {} changed {}'s Entity",
1314 : $step,
1315 : a.as_str(),
1316 : b.as_str()
1317 : );
1318 : assert_eq!(
1319 : cs.get(b, Kind::Subscription, &sub).await.expect("get"),
1320 : b_sub,
1321 : "{} as {} changed {}'s Subscription",
1322 : $step,
1323 : a.as_str(),
1324 : b.as_str()
1325 : );
1326 : };
1327 : }
1328 96 : let only_a = |rows: &[Value], what: &str| {
1329 96 : let shared: Vec<&Value> = rows.iter().filter(|r| r["id"] == e.as_str()).collect();
1330 96 : assert_eq!(shared.len(), 1, "{what}: the shared id once, {rows:?}");
1331 96 : assert_eq!(
1332 96 : shared[0]["who"],
1333 96 : a.as_str(),
1334 : "{what} served another Tenant's copy"
1335 : );
1336 96 : };
1337 :
1338 : // Reads.
1339 24 : let got = cs
1340 24 : .get(a, Kind::Entity, &e)
1341 24 : .await
1342 24 : .expect("get")
1343 24 : .expect("row");
1344 24 : assert_eq!(got["who"], a.as_str(), "get served another Tenant's copy");
1345 24 : only_a(&cs.list(a, Kind::Entity).await.expect("list"), "list");
1346 24 : only_a(
1347 24 : &cs.list_page(a, Kind::Entity, None, 1000)
1348 24 : .await
1349 24 : .expect("page"),
1350 24 : "list_page",
1351 : );
1352 24 : only_a(
1353 24 : &cs.list_slice(a, Kind::Entity, 0, 1000)
1354 24 : .await
1355 24 : .expect("slice")
1356 : .0,
1357 24 : "list_slice",
1358 : );
1359 24 : let q = cs
1360 24 : .query_entities(
1361 24 : a,
1362 24 : &EntityFilter {
1363 24 : ids: Some(&ids_ref),
1364 24 : ..Default::default()
1365 24 : },
1366 24 : )
1367 24 : .await
1368 24 : .expect("query");
1369 24 : only_a(&q.rows, "query_entities");
1370 24 : let subs = cs.list(a, Kind::Subscription).await.expect("list");
1371 24 : assert!(
1372 24 : subs.iter()
1373 24 : .filter(|s| s["id"] == sub.as_str())
1374 24 : .all(|s| s["who"] == a.as_str()),
1375 : "the Subscription list served another Tenant's copy: {subs:?}"
1376 : );
1377 24 : let regs = cs
1378 24 : .matching_registrations(a, None, None)
1379 24 : .await
1380 24 : .expect("registrations");
1381 24 : let shared: Vec<&Value> = regs.iter().filter(|r| r["id"] == reg.as_str()).collect();
1382 24 : assert!(
1383 24 : shared.len() == 1 && shared[0]["who"] == a.as_str(),
1384 : "matching_registrations served another Tenant's copy: {regs:?}"
1385 : );
1386 :
1387 : // Writes, one by one and batched.
1388 24 : cs.mutate::<(), ()>(a, Kind::Entity, &e, |d| {
1389 24 : d["touched"] = json!(true);
1390 24 : Ok(())
1391 24 : })
1392 24 : .await
1393 24 : .expect("mutate");
1394 24 : b_unchanged!("mutate");
1395 24 : cs.upsert(
1396 24 : a,
1397 24 : Kind::Entity,
1398 24 : &e,
1399 24 : json!({"id": e, "type": ty, "who": a.as_str(), "n": 1}),
1400 24 : )
1401 24 : .await
1402 24 : .expect("upsert");
1403 24 : b_unchanged!("upsert");
1404 24 : assert_eq!(
1405 24 : cs.batch_create(a, vec![(e.clone(), mine(a))])
1406 24 : .await
1407 24 : .expect("batch_create"),
1408 : [false],
1409 : "a batch create sees only its own Tenant's row"
1410 : );
1411 24 : b_unchanged!("batch_create");
1412 24 : assert_eq!(
1413 24 : cs.batch_upsert(a, vec![(e.clone(), mine(a))])
1414 24 : .await
1415 24 : .expect("batch_upsert"),
1416 : [false],
1417 : "a batch upsert replaces its own Tenant's row"
1418 : );
1419 24 : b_unchanged!("batch_upsert");
1420 24 : cs.batch_mutate::<()>(a, &ids, |_, d| {
1421 24 : d["batched"] = json!(true);
1422 24 : Ok(())
1423 24 : })
1424 24 : .await
1425 24 : .expect("batch_mutate");
1426 24 : b_unchanged!("batch_mutate");
1427 24 : cs.record_delivery(a, Kind::Subscription, &sub, "2026-10-11T00:00:00Z")
1428 24 : .await
1429 24 : .expect("record_delivery")
1430 24 : .expect("a's Subscription");
1431 24 : b_unchanged!("record_delivery");
1432 24 : assert!(
1433 24 : cs.delete_entity_if(a, &e, &|_| true)
1434 24 : .await
1435 24 : .expect("delete_entity_if"),
1436 : "a conditional delete removes its own Tenant's row"
1437 : );
1438 24 : b_unchanged!("delete_entity_if");
1439 24 : cs.create(a, Kind::Entity, &e, mine(a))
1440 24 : .await
1441 24 : .expect("recreate");
1442 24 : assert_eq!(
1443 24 : cs.batch_delete(a, &ids).await.expect("batch_delete"),
1444 : [true],
1445 : "a batch delete removes its own Tenant's row"
1446 : );
1447 24 : b_unchanged!("batch_delete");
1448 24 : cs.create(a, Kind::Entity, &e, mine(a))
1449 24 : .await
1450 24 : .expect("recreate");
1451 24 : assert!(cs.delete(a, Kind::Entity, &e).await.expect("delete"));
1452 24 : b_unchanged!("delete");
1453 24 : cs.create(a, Kind::Entity, &e, mine(a))
1454 24 : .await
1455 24 : .expect("recreate");
1456 :
1457 : // History under the same id.
1458 24 : if tp.supported() {
1459 48 : let instance = |t: &TenantId| {
1460 48 : json!({"n": [{"type": "Property", "value": t.as_str(),
1461 48 : "observedAt": "2026-10-11T00:00:00Z",
1462 48 : "instanceId": format!("urn:ngsi-ld:{prefix}:{}", t.as_str())}]})
1463 48 : };
1464 48 : for t in [a, b] {
1465 48 : tp.create(t, &e, mine(t)).await.expect("temporal create");
1466 48 : tp.temporal_append(t, &e, &mine(t), &instance(t))
1467 48 : .await
1468 48 : .expect("append");
1469 : }
1470 24 : let b_history = tp.get(b, &e).await.expect("get");
1471 : macro_rules! history_unchanged {
1472 : ($step:literal) => {
1473 : assert_eq!(
1474 : tp.get(b, &e).await.expect("get"),
1475 : b_history,
1476 : "{} as {} changed {}'s history",
1477 : $step,
1478 : a.as_str(),
1479 : b.as_str()
1480 : );
1481 : };
1482 : }
1483 48 : let values = |doc: &Value| doc["n"].to_string();
1484 24 : let one = tp
1485 24 : .get_temporal(a, &e, &TemporalFilter::default())
1486 24 : .await
1487 24 : .expect("get_temporal")
1488 24 : .expect("a's history");
1489 24 : assert!(
1490 24 : !values(&one).contains(&format!("\"{}\"", b.as_str())),
1491 : "get_temporal served another Tenant's instances: {one}"
1492 : );
1493 24 : let rows = tp
1494 24 : .query_temporal(
1495 24 : a,
1496 24 : &TemporalFilter {
1497 24 : ids: Some(&ids_ref),
1498 24 : ..Default::default()
1499 24 : },
1500 24 : )
1501 24 : .await
1502 24 : .expect("query_temporal")
1503 : .rows;
1504 24 : assert_eq!(
1505 24 : rows.len(),
1506 : 1,
1507 : "query_temporal: the shared id once, {rows:?}"
1508 : );
1509 24 : assert!(
1510 24 : !values(&rows[0]).contains(&format!("\"{}\"", b.as_str())),
1511 : "query_temporal served another Tenant's instances: {rows:?}"
1512 : );
1513 24 : assert!(
1514 24 : tp.list(a)
1515 24 : .await
1516 24 : .expect("list")
1517 24 : .iter()
1518 24 : .filter(|r| r["id"] == e.as_str())
1519 24 : .count()
1520 : == 1,
1521 : "the temporal list holds the shared id once"
1522 : );
1523 24 : tp.event_list(&[crate::TemporalEvent {
1524 24 : op: crate::TemporalOp::AttrModified,
1525 24 : tenant: a.clone(),
1526 24 : entity_id: e.clone(),
1527 24 : shell: mine(a),
1528 24 : attr: "n".into(),
1529 24 : instance: json!({"type": "Property", "value": "drained",
1530 24 : "observedAt": "2026-10-11T00:01:00Z",
1531 24 : "instanceId": format!("urn:ngsi-ld:{prefix}:drained")}),
1532 24 : }])
1533 24 : .await
1534 24 : .expect("event_list");
1535 24 : history_unchanged!("event_list");
1536 24 : tp.mutate::<(), ()>(a, &e, |d| {
1537 24 : d["touched"] = json!(true);
1538 24 : Ok(())
1539 24 : })
1540 24 : .await
1541 24 : .expect("mutate");
1542 24 : history_unchanged!("temporal mutate");
1543 24 : assert!(tp.delete(a, &e).await.expect("delete"));
1544 24 : history_unchanged!("temporal delete");
1545 24 : tp.create(a, &e, mine(a)).await.expect("recreate");
1546 24 : tp.purge_tenant(a).await.expect("temporal purge");
1547 24 : assert!(
1548 24 : tp.get(a, &e).await.expect("get").is_none(),
1549 : "the purge empties a's history"
1550 : );
1551 24 : history_unchanged!("temporal purge");
1552 24 : assert!(tp.delete(b, &e).await.expect("cleanup"));
1553 0 : }
1554 :
1555 : // A purge of `a` leaves `b` whole.
1556 24 : match cs.purge_tenant(a).await {
1557 : Ok(_) => {
1558 24 : assert!(
1559 24 : cs.get(a, Kind::Entity, &e).await.expect("get").is_none(),
1560 : "the purge empties a"
1561 : );
1562 24 : b_unchanged!("purge_tenant");
1563 24 : assert!(
1564 24 : cs.get(b, Kind::Registration, ®)
1565 24 : .await
1566 24 : .expect("get")
1567 24 : .is_some(),
1568 : "the purge of {} took {}'s Registration",
1569 0 : a.as_str(),
1570 0 : b.as_str()
1571 : );
1572 : }
1573 : Err(antares_model::NgsiError::OperationNotSupported(_)) => {
1574 0 : for kind in [Kind::Entity, Kind::Subscription, Kind::Registration] {
1575 0 : let id = match kind {
1576 0 : Kind::Entity => &e,
1577 0 : Kind::Subscription => &sub,
1578 0 : _ => ®,
1579 : };
1580 0 : cs.delete(a, kind, id).await.expect("cleanup");
1581 : }
1582 : }
1583 0 : Err(other) => panic!("purge_tenant: {other:?}"),
1584 : }
1585 72 : for (kind, id) in [
1586 24 : (Kind::Entity, &e),
1587 24 : (Kind::Subscription, &sub),
1588 24 : (Kind::Registration, ®),
1589 24 : ] {
1590 72 : assert!(
1591 72 : cs.delete(b, kind, id).await.expect("cleanup"),
1592 : "{}'s {kind:?} survived",
1593 0 : b.as_str()
1594 : );
1595 : }
1596 24 : }
|
