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 102 : fn doc(id: &str) -> Value {
22 102 : json!({"id": id, "type": "https://uri.etsi.org/ngsi-ld/default-context/T"})
23 102 : }
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 : // Tenant isolation, on every read and every write path.
127 6 : assert!(
128 6 : d.get(b, Kind::Entity, &e1).await.expect("get").is_none(),
129 : "another tenant must not read the row"
130 : );
131 6 : assert!(
132 6 : !d.delete(b, Kind::Entity, &e1).await.expect("delete"),
133 : "another tenant must not delete the row"
134 : );
135 6 : assert!(
136 6 : d.mutate::<(), ()>(b, Kind::Entity, &e1, |_| Ok(()))
137 6 : .await
138 6 : .expect("mutate")
139 6 : .is_none(),
140 : "another tenant must not mutate the row"
141 : );
142 6 : assert!(
143 6 : d.list(b, Kind::Entity)
144 6 : .await
145 6 : .expect("list")
146 6 : .iter()
147 6 : .all(|r| r["id"] != e1.as_str()),
148 : "another tenant must not list the row"
149 : );
150 :
151 : // Kinds are separate namespaces: the same id under two kinds is two rows.
152 6 : assert!(
153 6 : d.create(a, Kind::Subscription, &e1, doc(&e1))
154 6 : .await
155 6 : .expect("create"),
156 : "the same id under another kind is a different row"
157 : );
158 6 : assert!(d.delete(a, Kind::Subscription, &e1).await.expect("delete"));
159 6 : assert!(
160 6 : d.get(a, Kind::Entity, &e1).await.expect("get").is_some(),
161 : "deleting one kind must not touch another"
162 : );
163 :
164 : // upsert answers whether a document was ALREADY there, and batch_upsert
165 : // answers the opposite polarity — created-flags, which the batch path
166 : // needs to split 201 from 204 (5.6.8). Getting one of the two backwards
167 : // is invisible until a batch reports every create as an update.
168 6 : assert!(
169 6 : !d.upsert(a, Kind::Entity, &e2, doc(&e2))
170 6 : .await
171 6 : .expect("upsert"),
172 : "upsert that created the row reports false"
173 : );
174 6 : assert!(
175 6 : d.upsert(a, Kind::Entity, &e2, doc(&e2))
176 6 : .await
177 6 : .expect("upsert"),
178 : "upsert over an existing row reports true"
179 : );
180 6 : let e3 = format!("urn:ngsi-ld:{prefix}:3");
181 6 : let flags = d
182 6 : .batch_upsert(a, vec![(e3.clone(), doc(&e3)), (e2.clone(), doc(&e2))])
183 6 : .await
184 6 : .expect("batch_upsert");
185 6 : assert_eq!(
186 : flags,
187 6 : vec![true, false],
188 : "batch_upsert answers created-flags in input order, the opposite \
189 : polarity of upsert"
190 : );
191 6 : assert!(d.delete(a, Kind::Entity, &e3).await.expect("delete"));
192 6 : let ids = vec![e2.clone(), gone.clone(), e1.clone()];
193 6 : let batch = d
194 12 : .batch_mutate::<()>(a, &ids, |_, v| {
195 12 : v["batched"] = json!(true);
196 12 : Ok(())
197 12 : })
198 6 : .await
199 6 : .expect("batch_mutate");
200 6 : assert_eq!(batch.len(), ids.len(), "one result per input id");
201 6 : assert!(batch[0].is_some(), "results align with ids: {ids:?}");
202 6 : assert!(batch[1].is_none(), "an absent id answers None, in place");
203 6 : assert!(batch[2].is_some(), "results align with ids: {ids:?}");
204 :
205 : // A query may over-return (the caller re-checks whatever the backend
206 : // could not decide) but must never drop a matching row, and must never
207 : // cross a tenant.
208 6 : let ids_ref: Vec<&str> = vec![&e1, &e2];
209 6 : let mine = d
210 6 : .query_entities(
211 6 : a,
212 6 : &crate::filter::EntityFilter {
213 6 : ids: Some(&ids_ref),
214 6 : ..Default::default()
215 6 : },
216 6 : )
217 6 : .await
218 6 : .expect("query_entities");
219 12 : for want in [&e1, &e2] {
220 12 : assert!(
221 18 : mine.rows.iter().any(|r| r["id"] == want.as_str()),
222 : "query must not drop a matching row: {want}"
223 : );
224 : }
225 6 : let theirs = d
226 6 : .query_entities(
227 6 : b,
228 6 : &crate::filter::EntityFilter {
229 6 : ids: Some(&ids_ref),
230 6 : ..Default::default()
231 6 : },
232 6 : )
233 6 : .await
234 6 : .expect("query_entities");
235 6 : assert!(
236 6 : theirs.rows.is_empty(),
237 : "a query must never cross a tenant: {:?}",
238 : theirs.rows
239 : );
240 :
241 : // Paging pushdown is optional; claiming it and not doing it is not.
242 6 : let paged = d
243 6 : .query_entities(
244 6 : a,
245 6 : &crate::filter::EntityFilter {
246 6 : ids: Some(&ids_ref),
247 6 : page: Some(crate::filter::Page {
248 6 : offset: 0,
249 6 : limit: 1,
250 6 : count: true,
251 6 : }),
252 6 : ..Default::default()
253 6 : },
254 6 : )
255 6 : .await
256 6 : .expect("query_entities");
257 6 : if paged.paged {
258 2 : assert!(
259 2 : paged.rows.len() <= 1,
260 : "a driver reporting paged=true has applied the LIMIT"
261 : );
262 2 : assert!(
263 2 : paged.decided,
264 : "paged implies decided: a LIMIT over an undecided set pages the wrong rows"
265 : );
266 4 : }
267 :
268 : // `subscription_tenants` is the iteration domain the interval sweep and
269 : // both mirror hydrations walk. A backend may return MORE tenants than
270 : // hold a subscription — every caller lists per tenant afterwards and an
271 : // empty list costs nothing. It may never return FEWER: a tenant missing
272 : // here is a tenant whose periodic notifications never fire and whose
273 : // subscriptions never reach the mirror, with no error anywhere.
274 6 : let sub = format!("urn:ngsi-ld:Subscription:{prefix}:1");
275 6 : let sub_doc = json!({
276 6 : "id": &sub,
277 6 : "type": "Subscription",
278 6 : "status": "active",
279 6 : "notification": {"endpoint": {"uri": "http://sink.invalid/n"}},
280 : });
281 6 : d.upsert(a, Kind::Subscription, &sub, sub_doc)
282 6 : .await
283 6 : .expect("upsert subscription");
284 6 : let domain = d
285 6 : .subscription_tenants()
286 6 : .await
287 6 : .expect("subscription_tenants");
288 6 : assert!(
289 22 : domain.iter().any(|t| t == a.as_str()),
290 : "a tenant holding a subscription must appear in subscription_tenants: {domain:?}"
291 : );
292 : // A REGISTRATION puts a tenant in the domain too. One of the hydrations
293 : // that walks this domain fills the registration mirror, and the
294 : // federation path reads that mirror alone once it is installed — so a
295 : // tenant holding registrations and no subscription is a tenant that
296 : // silently forwards to no Context Source.
297 6 : let reg_only = TenantId::new(&format!("{prefix}regonly")).expect("tenant");
298 6 : let reg = format!("urn:ngsi-ld:ContextSourceRegistration:{prefix}:1");
299 6 : d.upsert(
300 6 : ®_only,
301 6 : Kind::Registration,
302 6 : ®,
303 6 : json!({
304 6 : "id": ®,
305 6 : "type": "ContextSourceRegistration",
306 6 : "endpoint": "http://cs.invalid/ngsi-ld/v1",
307 6 : "information": [{"entities": [{"type": "Vehicle"}]}],
308 6 : }),
309 6 : )
310 6 : .await
311 6 : .expect("upsert registration");
312 6 : let domain = d
313 6 : .subscription_tenants()
314 6 : .await
315 6 : .expect("subscription_tenants");
316 6 : assert!(
317 34 : domain.iter().any(|t| t == reg_only.as_str()),
318 : "a tenant holding only a registration must appear in the mirror-hydration \
319 : domain, or its registration mirror hydrates empty: {domain:?}"
320 : );
321 :
322 : // Table 5.2.9-2 forward bookkeeping, pinned here for the same reason as
323 : // 5.2.14.2's: a backend may write it as one statement instead of a
324 : // read-modify-write, and the result must not depend on which it chose.
325 6 : let f1 = "2020-02-01T00:00:00.000Z";
326 6 : let f2 = "2020-02-02T00:00:00.000Z";
327 6 : let ok1 = d
328 6 : .record_forward(®_only, ®, f1, true)
329 6 : .await
330 6 : .expect("record_forward")
331 6 : .expect("the registration is there");
332 6 : assert_eq!(
333 6 : ok1["timesSent"], 1,
334 : "the first forward moves timesSent to 1"
335 : );
336 6 : assert_eq!(ok1["lastSuccess"], f1, "lastSuccess takes the stamp");
337 6 : assert_eq!(ok1["status"], "ok", "a 2xx leaves the registration ok");
338 6 : assert!(
339 6 : ok1.get("timesFailed").is_none() && ok1.get("lastFailure").is_none(),
340 : "a registration that has only succeeded carries no failure members: {ok1}"
341 : );
342 :
343 6 : let bad = d
344 6 : .record_forward(®_only, ®, f2, false)
345 6 : .await
346 6 : .expect("record_forward")
347 6 : .expect("the registration is still there");
348 6 : assert_eq!(
349 6 : bad["timesSent"], 2,
350 : "timesSent counts the failed attempt too (Table 5.2.9-2)"
351 : );
352 6 : assert_eq!(bad["timesFailed"], 1, "the failure moves timesFailed");
353 6 : assert_eq!(bad["lastFailure"], f2, "lastFailure takes the stamp");
354 6 : assert_eq!(bad["status"], "failed", "status names the LAST attempt");
355 6 : assert_eq!(
356 6 : bad["lastSuccess"], f1,
357 : "a failure never rewinds the last success"
358 : );
359 :
360 : // A registration deleted while its forward was in flight has no row to
361 : // book against, and the writeback must not resurrect it.
362 6 : let vanished = format!("urn:ngsi-ld:ContextSourceRegistration:{prefix}:gone");
363 6 : assert!(
364 6 : d.record_forward(®_only, &vanished, f1, true)
365 6 : .await
366 6 : .expect("record_forward")
367 6 : .is_none(),
368 : "an absent registration books nothing"
369 : );
370 6 : assert!(
371 6 : d.get(®_only, Kind::Registration, &vanished)
372 6 : .await
373 6 : .expect("get")
374 6 : .is_none(),
375 : "a bookkeeping writeback must never insert the row it missed"
376 : );
377 :
378 6 : d.delete(®_only, Kind::Registration, ®)
379 6 : .await
380 6 : .expect("delete");
381 : // `list_page` is how the readers that must see EVERY document read —
382 : // the mirror seed above all — so a backend may not refuse it for volume
383 : // the way `list` may (5.5.6 licenses TooManyResults for a query
384 : // operation, which an internal bootstrap is not). Every arm must agree
385 : // on the walk, including the one that takes the trait default: ids
386 : // strictly greater than `after`, id-ordered, at most `limit`, a short
387 : // page means the end, and every stored document served exactly once.
388 6 : let mut ids: Vec<String> = (0..7)
389 42 : .map(|i| format!("urn:ngsi-ld:Subscription:{prefix}:page:{i}"))
390 6 : .collect();
391 42 : for id in &ids {
392 42 : d.upsert(
393 42 : a,
394 42 : Kind::Subscription,
395 42 : id,
396 42 : json!({"id": id, "type": "Subscription"}),
397 42 : )
398 42 : .await
399 42 : .expect("upsert page subscription");
400 : }
401 6 : ids.push(sub.clone());
402 6 : ids.sort();
403 :
404 6 : let mut walked: Vec<String> = Vec::new();
405 6 : let mut after: Option<String> = None;
406 : loop {
407 18 : let page = d
408 18 : .list_page(a, Kind::Subscription, after.as_deref(), 3)
409 18 : .await
410 18 : .expect("list_page");
411 18 : assert!(page.len() <= 3, "a page may not exceed its limit: {page:?}");
412 18 : let short = page.len() < 3;
413 48 : for doc in &page {
414 48 : let id = doc["id"].as_str().expect("a stored doc keeps its id");
415 48 : assert!(
416 48 : after.as_deref().is_none_or(|prev| id > prev),
417 : "`after` is exclusive and the walk ascends: {id} after {after:?}"
418 : );
419 48 : after = Some(id.to_owned());
420 48 : walked.push(id.to_owned());
421 : }
422 18 : if short {
423 6 : break;
424 12 : }
425 : }
426 6 : assert_eq!(
427 : walked, ids,
428 : "the walk must serve every subscription of the tenant exactly once, in id order"
429 : );
430 : // The other tenant's documents are not on this walk. 4.14: operations
431 : // "only apply to the information of the specified `Tenant` in isolation".
432 6 : assert!(
433 6 : d.list_page(b, Kind::Subscription, None, 100)
434 6 : .await
435 6 : .expect("list_page")
436 6 : .is_empty(),
437 : "a tenant with no subscriptions pages empty, whatever another tenant holds"
438 : );
439 : // Entities page like every other kind. 5.9.2.4's registration-vs-entity
440 : // conflict check reads them this way — it must see every Entity of the
441 : // tenant and has no TooManyResults to raise — so a backend may not refuse
442 : // this walk for volume either, and may not build the page by
443 : // materializing the tenant first.
444 : //
445 : // 4.22 applies INSIDE the page: an expired Entity is not there to be
446 : // served, and dropping one after the limit was applied would hand back a
447 : // short page, which every walker reads as the end of the tenant.
448 6 : let stem = format!("urn:ngsi-ld:{prefix}:page:");
449 24 : let live: Vec<String> = (0..4).map(|i| format!("{stem}live{i}")).collect();
450 : // sorts between live1 and live2, so it falls inside the first page below
451 6 : let expired = format!("{stem}live1x");
452 24 : for id in &live {
453 24 : d.upsert(a, Kind::Entity, id, doc(id))
454 24 : .await
455 24 : .expect("upsert page entity");
456 : }
457 6 : let mut gone_doc = doc(&expired);
458 6 : gone_doc["expiresAt"] = json!("2000-01-01T00:00:00.000Z");
459 6 : d.upsert(a, Kind::Entity, &expired, gone_doc)
460 6 : .await
461 6 : .expect("upsert expired entity");
462 :
463 6 : let page = d
464 6 : .list_page(a, Kind::Entity, Some(&stem), 3)
465 6 : .await
466 6 : .expect("list_page entities");
467 18 : let served: Vec<&str> = page.iter().filter_map(|d| d["id"].as_str()).collect();
468 6 : assert_eq!(
469 : served,
470 6 : [live[0].as_str(), live[1].as_str(), live[2].as_str()],
471 : "an expired entity may neither be served nor consume a slot of the page"
472 : );
473 :
474 6 : let mut walked: Vec<String> = Vec::new();
475 6 : let mut after = stem.clone();
476 : loop {
477 18 : let page = d
478 18 : .list_page(a, Kind::Entity, Some(&after), 2)
479 18 : .await
480 18 : .expect("list_page entities");
481 18 : let short = page.len() < 2;
482 18 : let mut moved = false;
483 24 : for doc in &page {
484 24 : let id = doc["id"].as_str().expect("a stored entity keeps its id");
485 24 : assert!(
486 24 : id > after.as_str(),
487 : "`after` is exclusive: {id} after {after}"
488 : );
489 24 : after = id.to_owned();
490 24 : moved = true;
491 24 : if id.starts_with(stem.as_str()) {
492 24 : walked.push(id.to_owned());
493 24 : }
494 : }
495 18 : if short || !moved || !after.starts_with(stem.as_str()) {
496 6 : break;
497 12 : }
498 : }
499 6 : assert_eq!(
500 : walked, live,
501 : "the walk must serve every live entity of the tenant exactly once, in id \
502 : order, and never the expired one"
503 : );
504 30 : for id in live.iter().chain(std::iter::once(&expired)) {
505 30 : d.delete(a, Kind::Entity, id)
506 30 : .await
507 30 : .expect("delete page entity");
508 : }
509 :
510 : // 4.22 at the read boundary, the part a backend must not decide for
511 : // itself. The stamp marks "a certain Entity, Property or Relationship"
512 : // invalid; it is not a store-wide rule that hides any document carrying
513 : // one, and it reaches every Attribute of an Entity, sub-Attributes
514 : // included.
515 6 : let exp_e = format!("urn:ngsi-ld:{prefix}:expiry:entity");
516 6 : let mut expired_entity = doc(&exp_e);
517 6 : expired_entity["expiresAt"] = json!("2000-01-01T00:00:00.000Z");
518 6 : d.upsert(a, Kind::Entity, &exp_e, expired_entity)
519 6 : .await
520 6 : .expect("upsert expired entity");
521 6 : assert!(
522 6 : d.get(a, Kind::Entity, &exp_e).await.expect("get").is_none(),
523 : "an Entity past its expiresAt reads absent, not just out of a page"
524 : );
525 6 : d.delete(a, Kind::Entity, &exp_e).await.expect("delete");
526 :
527 : // An Attribute whose every instance expired leaves; the Entity carrying
528 : // it does not, and neither does a live sibling.
529 6 : let exp_a = format!("urn:ngsi-ld:{prefix}:expiry:attr");
530 72 : let iri = |n: &str| format!("https://uri.etsi.org/ngsi-ld/default-context/{n}");
531 6 : let mut with_attrs = doc(&exp_a);
532 6 : with_attrs[iri("gone")] = json!([{"value": 1, "expiresAt": "2000-01-01T00:00:00.000Z"}]);
533 6 : with_attrs[iri("kept")] = json!([{
534 6 : "value": 2,
535 6 : iri("subgone"): [{"value": 3, "expiresAt": "2000-01-01T00:00:00.000Z"}],
536 6 : iri("subkept"): [{"value": 4}],
537 : }]);
538 : // 4.6.3 leaves the seconds-fraction separator open, and ',' (0x2C) sorts
539 : // before both '.' and 'Z', so the two stamps have to be read as instants:
540 : // a backend that compares them as bytes serves this live Attribute as
541 : // expired. Every backend answers the same here or it is not the same
542 : // store.
543 6 : with_attrs[iri("commakept")] = json!([{"value": 5, "expiresAt": "2999-01-01T00:00:00,500Z"}]);
544 6 : d.upsert(a, Kind::Entity, &exp_a, with_attrs)
545 6 : .await
546 6 : .expect("upsert entity with an expired attribute");
547 6 : let served = d
548 6 : .get(a, Kind::Entity, &exp_a)
549 6 : .await
550 6 : .expect("get")
551 6 : .expect("an Entity outlives the expiry of one of its Attributes");
552 6 : assert!(
553 6 : served.get(iri("gone")).is_none(),
554 : "an Attribute whose only instance expired is not served"
555 : );
556 6 : assert_eq!(
557 6 : served[iri("kept")][0]["value"],
558 6 : json!(2),
559 : "a live Attribute survives its sibling's expiry"
560 : );
561 6 : assert!(
562 6 : served[iri("kept")][0].get(iri("subgone")).is_none(),
563 : "a sub-Attribute is a Property or Relationship too: past its stamp it \
564 : is not served"
565 : );
566 6 : assert_eq!(
567 6 : served[iri("kept")][0][iri("subkept")][0]["value"],
568 6 : json!(4),
569 : "a live sub-Attribute survives its sibling's expiry"
570 : );
571 6 : assert_eq!(
572 6 : served[iri("commakept")][0]["value"],
573 6 : json!(5),
574 : "a comma seconds-fraction is an instant, not a byte string"
575 : );
576 6 : d.delete(a, Kind::Entity, &exp_a).await.expect("delete");
577 :
578 : // 5.8.6: an expired SUBSCRIPTION is not deleted and stays retrievable —
579 : // the API turns the stamp into status "expired" and keeps it updatable.
580 : // A backend that hid every document with a past expiresAt would lose it.
581 6 : let exp_s = format!("urn:ngsi-ld:{prefix}:expiry:sub");
582 6 : d.upsert(
583 6 : a,
584 6 : Kind::Subscription,
585 6 : &exp_s,
586 6 : json!({
587 6 : "id": exp_s,
588 6 : "type": "Subscription",
589 6 : "expiresAt": "2000-01-01T00:00:00.000Z",
590 6 : }),
591 6 : )
592 6 : .await
593 6 : .expect("upsert expired subscription");
594 6 : assert!(
595 6 : d.get(a, Kind::Subscription, &exp_s)
596 6 : .await
597 6 : .expect("get")
598 6 : .is_some(),
599 : "an expired Subscription stays retrievable (5.8.6)"
600 : );
601 6 : d.delete(a, Kind::Subscription, &exp_s)
602 6 : .await
603 6 : .expect("delete");
604 :
605 : // `delete_entity_if` decides and deletes under one lock, and it decides
606 : // on the STORED document. 5.6.6.4: an Entity the caller's selector
607 : // excludes "is not known" for the operation, which is the same answer an
608 : // absent Entity gets — so a refusal and a miss are both `false`, and a
609 : // refusal must leave the document exactly where it was.
610 6 : let cond = format!("urn:ngsi-ld:{prefix}:cond");
611 6 : d.upsert(a, Kind::Entity, &cond, doc(&cond))
612 6 : .await
613 6 : .expect("upsert conditional-delete entity");
614 6 : assert!(
615 6 : !d.delete_entity_if(a, &cond, &|_| false)
616 6 : .await
617 6 : .expect("delete_entity_if"),
618 : "a refused predicate reports no deletion"
619 : );
620 6 : assert!(
621 6 : d.get(a, Kind::Entity, &cond).await.expect("get").is_some(),
622 : "a refused delete leaves the document in place"
623 : );
624 : // 4.14: the predicate never even runs for another tenant's document.
625 6 : assert!(
626 6 : !d.delete_entity_if(b, &cond, &|_| panic!(
627 : "another tenant's document reached the predicate"
628 : ))
629 6 : .await
630 6 : .expect("delete_entity_if"),
631 : "an Entity of another tenant is absent here"
632 : );
633 6 : assert!(
634 6 : d.get(a, Kind::Entity, &cond).await.expect("get").is_some(),
635 : "a delete addressed to another tenant may not touch this one"
636 : );
637 6 : assert!(
638 6 : d.delete_entity_if(a, &cond, &|v| v["id"] == cond.as_str())
639 6 : .await
640 6 : .expect("delete_entity_if"),
641 : "the predicate is handed the stored document, not the id"
642 : );
643 6 : assert!(
644 6 : d.get(a, Kind::Entity, &cond).await.expect("get").is_none(),
645 : "an accepted predicate deletes"
646 : );
647 6 : assert!(
648 6 : !d.delete_entity_if(a, &cond, &|_| true)
649 6 : .await
650 6 : .expect("delete_entity_if"),
651 : "an absent Entity is not deleted, whatever the predicate answers"
652 : );
653 :
654 : // `list_slice` is the window a 5.5.9.2 limit/offset listing serves from,
655 : // and it must agree with the walk above on both the order and the set:
656 : // the same ids, the same order, and a total that counts the whole match
657 : // set rather than the page. A backend may not refuse it for volume — the
658 : // window bounds the result by construction.
659 6 : let (page, total) = d
660 6 : .list_slice(a, Kind::Subscription, 0, 3)
661 6 : .await
662 6 : .expect("list_slice");
663 6 : assert_eq!(total, ids.len(), "the total counts the set, not the page");
664 6 : assert_eq!(page.len(), 3, "limit is a maximum, and 3 were available");
665 18 : for (i, doc) in page.iter().enumerate() {
666 18 : assert_eq!(
667 18 : doc["id"].as_str(),
668 18 : Some(ids[i].as_str()),
669 : "the window is the id-ordered prefix: {page:?}"
670 : );
671 : }
672 : // Every single-element window lands on the element the walk has there.
673 48 : for (i, want) in ids.iter().enumerate() {
674 48 : let (page, _) = d
675 48 : .list_slice(a, Kind::Subscription, i, 1)
676 48 : .await
677 48 : .expect("list_slice");
678 48 : assert_eq!(
679 48 : page.first().and_then(|d| d["id"].as_str()),
680 48 : Some(want.as_str()),
681 : "offset {i} served the wrong element: {page:?}"
682 : );
683 : }
684 : // limit 0 is legal (6.3.10, with count): the count without the page.
685 6 : let (page, total) = d
686 6 : .list_slice(a, Kind::Subscription, 0, 0)
687 6 : .await
688 6 : .expect("list_slice");
689 6 : assert!(page.is_empty(), "limit 0 returns no elements: {page:?}");
690 6 : assert_eq!(total, ids.len(), "limit 0 still counts the whole set");
691 : // An offset past the end is an empty page, not an error, and the total
692 : // is unchanged by where the client asked to start.
693 6 : let (page, total) = d
694 6 : .list_slice(a, Kind::Subscription, ids.len() + 10, 5)
695 6 : .await
696 6 : .expect("list_slice");
697 6 : assert!(page.is_empty(), "past the end is empty: {page:?}");
698 6 : assert_eq!(total, ids.len(), "an offset does not change the match set");
699 : // 4.14 again: the window is inside one tenant.
700 6 : let (page, total) = d
701 6 : .list_slice(b, Kind::Subscription, 0, 100)
702 6 : .await
703 6 : .expect("slice");
704 6 : assert!(
705 6 : page.is_empty() && total == 0,
706 : "another tenant's rows reached this window: {page:?} / {total}"
707 : );
708 :
709 48 : for id in ids.iter().filter(|i| *i != &sub) {
710 42 : d.delete(a, Kind::Subscription, id).await.expect("delete");
711 : }
712 :
713 : // 5.2.14.2 delivery bookkeeping. A backend may write this as one
714 : // statement instead of a read-modify-write; the result must not depend on
715 : // which it chose, so the rule is pinned here rather than in one backend's
716 : // own tests.
717 6 : let t1 = "2020-01-01T00:00:00.000Z";
718 6 : let t2 = "2020-01-02T00:00:00.000Z";
719 6 : let first = d
720 6 : .record_delivery(a, Kind::Subscription, &sub, t1)
721 6 : .await
722 6 : .expect("record_delivery")
723 6 : .expect("the subscription is there");
724 6 : let n1 = &first.doc["notification"];
725 6 : assert_eq!(n1["timesSent"], 1, "the first attempt moves timesSent to 1");
726 6 : assert_eq!(
727 6 : n1["lastNotification"], t1,
728 : "lastNotification takes the stamp"
729 : );
730 6 : assert_eq!(n1["lastSuccess"], t1, "lastSuccess takes the stamp");
731 6 : assert_eq!(n1["status"], "ok", "the attempt leaves the notification ok");
732 6 : assert!(
733 6 : first.doc.get("status").is_none(),
734 : "the top-level status is a rendered member, not a stored one: {}",
735 : first.doc
736 : );
737 6 : assert!(
738 6 : first.prev_success.is_none(),
739 : "a subscription that never succeeded has no previous lastSuccess"
740 : );
741 :
742 6 : let second = d
743 6 : .record_delivery(a, Kind::Subscription, &sub, t2)
744 6 : .await
745 6 : .expect("record_delivery")
746 6 : .expect("the subscription is still there");
747 6 : assert_eq!(
748 6 : second.doc["notification"]["timesSent"], 2,
749 : "timesSent counts attempts, it does not reset"
750 : );
751 6 : assert_eq!(
752 6 : second.prev_success.as_ref().and_then(Value::as_str),
753 6 : Some(t1),
754 : "the overwritten lastSuccess comes back, or a failed attempt cannot \
755 : roll it back"
756 : );
757 :
758 : // 5.8.6: a subscription deleted between matching and delivery has no row
759 : // to book against, and nothing may be sent.
760 6 : assert!(
761 6 : d.record_delivery(a, Kind::Subscription, &gone, t1)
762 6 : .await
763 6 : .expect("record_delivery")
764 6 : .is_none(),
765 : "an absent subscription books nothing"
766 : );
767 :
768 6 : assert!(
769 6 : d.delete(a, Kind::Subscription, &sub).await.expect("delete"),
770 : "the contract's own subscription must be removable"
771 : );
772 :
773 : // ADR-0021: a stored @context is a Tenant's document. "Hosted" and
774 : // "ImplicitlyCreated" rows hold term mappings authored through one
775 : // Tenant's requests, and 5.5.7 makes those mappings decide what that
776 : // Tenant's payloads mean, so a backend that answered every caller would
777 : // hand one Tenant's meaning of its own data to another. "Cached" is a
778 : // copy of a public document (5.13.1) and belongs to no Tenant.
779 6 : let hosted_id = format!("{prefix}-ctx-hosted");
780 6 : let cached_id = format!("{prefix}-ctx-cached");
781 6 : let hosted = json!({"localId": hosted_id, "kind": "Hosted", "owner": a.as_str(),
782 6 : "body": {"@context": {"t": "https://example.org/t"}}});
783 6 : d.context_put(Some(a), &hosted_id, hosted.clone())
784 6 : .await
785 6 : .expect("the owning tenant stores its own @context");
786 6 : d.context_put(
787 6 : None,
788 6 : &cached_id,
789 6 : json!({"localId": cached_id, "kind": "Cached", "body": {"@context": {}}}),
790 6 : )
791 6 : .await
792 6 : .expect("a Cached copy belongs to no tenant");
793 :
794 6 : assert!(
795 6 : d.context_get(Some(a), &hosted_id)
796 6 : .await
797 6 : .expect("context_get")
798 6 : .is_some(),
799 : "the owning tenant must read its own @context"
800 : );
801 6 : assert!(
802 6 : d.context_get(Some(b), &hosted_id)
803 6 : .await
804 6 : .expect("context_get")
805 6 : .is_none(),
806 : "another tenant's Hosted @context must be as absent as one never stored"
807 : );
808 6 : assert!(
809 6 : d.context_get(None, &hosted_id)
810 6 : .await
811 6 : .expect("context_get")
812 6 : .is_none(),
813 : "no tenant in scope must reach a tenant's @context"
814 : );
815 6 : assert!(
816 6 : d.context_list_meta(Some(b))
817 6 : .await
818 6 : .expect("context_list_meta")
819 6 : .iter()
820 6 : .all(|r| r["localId"] != hosted_id.as_str()),
821 : "another tenant's @context must not be listed"
822 : );
823 6 : assert!(
824 6 : !d.context_delete(Some(b), &hosted_id)
825 6 : .await
826 6 : .expect("context_delete"),
827 : "another tenant must not delete it"
828 : );
829 6 : assert!(
830 6 : d.context_put(Some(b), &hosted_id, hosted).await.is_err(),
831 : "another tenant must not overwrite it either"
832 : );
833 6 : assert!(
834 6 : d.context_get(Some(a), &hosted_id)
835 6 : .await
836 6 : .expect("context_get")
837 6 : .is_some(),
838 : "the owner's @context must survive every foreign attempt"
839 : );
840 18 : for t in [Some(a), Some(b), None] {
841 18 : assert!(
842 18 : d.context_get(t, &cached_id)
843 18 : .await
844 18 : .expect("context_get")
845 18 : .is_some(),
846 : "a Cached copy is a public document every tenant reaches"
847 : );
848 : }
849 6 : assert!(
850 6 : d.context_delete(Some(a), &hosted_id)
851 6 : .await
852 6 : .expect("context_delete"),
853 : "the contract's own @context must be removable by its owner"
854 : );
855 6 : assert!(
856 6 : d.context_delete(None, &cached_id)
857 6 : .await
858 6 : .expect("context_delete"),
859 : "and the Cached copy by anyone"
860 : );
861 :
862 12 : for id in [&e1, &e2] {
863 12 : assert!(
864 12 : d.delete(a, Kind::Entity, id).await.expect("delete"),
865 : "the contract's own rows must be removable"
866 : );
867 : }
868 6 : }
869 :
870 : /// Hold a temporal driver to the same shape. A driver that answers
871 : /// `supported() == false` declares it keeps no history and is held to
872 : /// nothing else.
873 6 : pub async fn run_temporal_contract(
874 6 : d: &dyn TemporalDriver,
875 6 : a: &TenantId,
876 6 : b: &TenantId,
877 6 : prefix: &str,
878 6 : ) {
879 6 : if !d.supported() {
880 0 : return;
881 6 : }
882 6 : let e1 = format!("urn:ngsi-ld:{prefix}:t1");
883 6 : let gone = format!("urn:ngsi-ld:{prefix}:tabsent");
884 :
885 6 : assert!(
886 6 : d.get(a, &gone).await.expect("get").is_none(),
887 : "get of an absent temporal document must answer None"
888 : );
889 6 : assert!(
890 6 : !d.delete(a, &gone).await.expect("delete"),
891 : "delete of an absent temporal document must answer false"
892 : );
893 6 : let missed = d
894 6 : .mutate::<(), ()>(a, &gone, |_| Ok(()))
895 6 : .await
896 6 : .expect("mutate");
897 6 : assert!(
898 6 : missed.is_none(),
899 : "mutate of an absent document must answer None"
900 : );
901 6 : assert!(
902 6 : d.get(a, &gone).await.expect("get").is_none(),
903 : "mutate of an absent document must not insert it (047_06)"
904 : );
905 :
906 6 : assert!(
907 6 : d.create(a, &e1, doc(&e1)).await.expect("create"),
908 : "the first create reports true"
909 : );
910 6 : assert!(
911 6 : !d.create(a, &e1, doc(&e1)).await.expect("create"),
912 : "a create over an existing id reports false"
913 : );
914 6 : let rejected = d
915 6 : .mutate::<(), &'static str>(a, &e1, |v| {
916 6 : v["marker"] = json!(2);
917 6 : Err("rejected")
918 6 : })
919 6 : .await
920 6 : .expect("mutate");
921 6 : assert_eq!(rejected, Some(Err("rejected")));
922 6 : assert!(
923 6 : d.get(a, &e1)
924 6 : .await
925 6 : .expect("get")
926 6 : .expect("row")
927 6 : .get("marker")
928 6 : .is_none(),
929 : "a rejected mutate must commit nothing"
930 : );
931 :
932 6 : assert!(
933 6 : d.get(b, &e1).await.expect("get").is_none(),
934 : "another tenant must not read the temporal document"
935 : );
936 6 : assert!(
937 6 : !d.delete(b, &e1).await.expect("delete"),
938 : "another tenant must not delete it"
939 : );
940 6 : assert!(
941 6 : d.list(b)
942 6 : .await
943 6 : .expect("list")
944 6 : .iter()
945 6 : .all(|r| r["id"] != e1.as_str()),
946 : "another tenant must not list it"
947 : );
948 :
949 6 : assert!(
950 6 : d.delete(a, &e1).await.expect("delete"),
951 : "the contract's own rows must be removable"
952 : );
953 6 : }
|
