Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! NGSI-LD HTTP binding: axum routers, thin
3 : //! handlers per spec operation.
4 : // a panic on a request path is a remote DoS. Outside tests
5 : // this crate may not unwrap, expect, panic, index or overflow.
6 : #![cfg_attr(
7 : not(test),
8 : deny(
9 : clippy::unwrap_used,
10 : clippy::expect_used,
11 : clippy::panic,
12 : clippy::todo,
13 : clippy::unimplemented,
14 : clippy::dbg_macro,
15 : clippy::print_stdout,
16 : clippy::print_stderr,
17 : clippy::indexing_slicing,
18 : clippy::arithmetic_side_effects
19 : )
20 : )]
21 : #![cfg_attr(test, allow(clippy::unwrap_used))]
22 : // the Send proof of the spawned snapshot fill walks temporal → entity map
23 : // futures deeper than the default 128 (nightly: recursion_depth_exceeding_limit)
24 : #![recursion_limit = "256"]
25 :
26 : /// Ask the policy engine about one operation (ADR-0020), where the request
27 : /// enters its handler:
28 : ///
29 : /// ```ignore
30 : /// gate!(st, &tenant, headers, "5.6.6", ids: &[&id]).await?;
31 : /// ```
32 : ///
33 : /// The named members go into the [`policy::Operation`]; every other member
34 : /// keeps the empty default of `Operation::new`. The awaited value is the
35 : /// `Filter` the engine narrowed the operation to, shared (`Arc`) so a
36 : /// view's compiled narrowing reaches the handler uncopied, or the refusal that
37 : /// becomes a 403. It is a macro because a gate reads as one line at the
38 : /// call site and the alternative is the same eight lines in every handler.
39 : ///
40 : /// With no engine attached there is nothing to ask, so nothing is built:
41 : /// the operation, the subject and its headers, the boxed future and the
42 : /// [`policy::TIMEOUT`] timer all belong to the arm that has an engine to
43 : /// ask. A broker running the default pays one branch per operation.
44 : macro_rules! gate {
45 : ($st:expr, $tenant:expr, $headers:expr, $clause:expr $(, $field:ident: $value:expr)* $(,)?) => {
46 53622 : async {
47 53622 : let answer = match &$st.policy {
48 50246 : None => Ok($crate::policy::unnarrowed()),
49 3376 : Some(engine) => {
50 3376 : $crate::policy::gate(
51 3376 : engine.as_ref(),
52 3376 : &$crate::snapshots::asking_tenant(&$st, $tenant).await,
53 : $headers,
54 3376 : &$crate::policy::Operation {
55 3376 : $($field: $value,)*
56 3376 : ..$crate::policy::Operation::new($clause)
57 3376 : },
58 : )
59 3376 : .await
60 3376 : .map(std::sync::Arc::new)
61 : }
62 : };
63 53622 : answer
64 53622 : }
65 : };
66 : }
67 :
68 : pub mod admin_auth;
69 : pub(crate) mod attrs;
70 : pub(crate) mod batch;
71 : pub mod bounds;
72 : pub mod conformance;
73 : pub(crate) mod contexts;
74 : pub(crate) mod csource;
75 : pub(crate) mod distsub;
76 : pub mod egress;
77 : pub(crate) mod entities;
78 : pub(crate) mod entity_map;
79 : pub(crate) mod entity_maps;
80 : pub(crate) mod federation;
81 : // The request parsers for the fuzz targets; never in a broker build.
82 : #[cfg(feature = "fuzzing")]
83 : pub mod fuzzing;
84 : pub mod history;
85 : pub mod mirror;
86 : pub mod negotiate;
87 : pub mod notify;
88 : pub(crate) mod paging;
89 : pub mod policy;
90 : pub mod provenance;
91 : pub(crate) mod registry;
92 : pub(crate) mod repr;
93 : /// The one GeoJSON geometry rule, for the surfaces that write GeoJSON.
94 : pub use repr::{geometry_of, rewound};
95 : mod ops;
96 : mod router;
97 : pub mod secret;
98 : pub(crate) mod snapshots;
99 : pub(crate) mod stamp;
100 : pub mod state;
101 : pub(crate) mod subscriptions;
102 : pub(crate) mod temporal;
103 : pub(crate) mod temporalq;
104 : #[cfg(test)]
105 : mod tests;
106 : pub(crate) mod types_attrs;
107 :
108 : pub use antares_notifier::DeliveryPolicy;
109 : // The sink contract, so a surface reaches it through this crate and names no
110 : // antares-notifier (rule M2).
111 : pub use antares_notifier::{
112 : build_message, DeliveryError, DeliveryFuture, NotificationSink, Outbound,
113 : };
114 : pub use antares_store::{HistoryScope, ShortHistory, HISTORY_RETENTION, SHORT_HISTORY};
115 : pub use contexts::public_base;
116 : pub use federation::{
117 : AdmitFuture, BuiltForward, ForwardAnswer, ForwardFuture, ForwardTransport, NotifyFuture,
118 : };
119 : pub use ops::*;
120 : pub use router::*;
121 : pub use state::{AppState, HistoryRules, OnChange, Profile, RecordMode, Recording, TemporalRecord};
122 :
123 : /// Spawn for both targets — tokio natively, the JS microtask queue on
124 : /// wasm32 (no tokio runtime exists in a browser). Call sites are identical;
125 : /// only the executor differs. Send is required natively (worker threads) and
126 : /// meaningless on single-threaded wasm.
127 : #[cfg(not(target_arch = "wasm32"))]
128 6164 : pub fn spawn<F>(fut: F)
129 6164 : where
130 6164 : F: std::future::Future<Output = ()> + Send + 'static,
131 : {
132 6164 : BACKGROUND.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
133 6164 : let counted = Counted;
134 6164 : tokio::spawn(async move {
135 5906 : let _counted = counted;
136 5906 : fut.await;
137 5886 : });
138 6164 : }
139 :
140 : /// One [`BACKGROUND`] task, uncounted when it ends however it ends: a task
141 : /// that panicked or was cancelled left the count up and every later
142 : /// shutdown drain waiting for its whole deadline.
143 : #[cfg(not(target_arch = "wasm32"))]
144 : struct Counted;
145 :
146 : #[cfg(not(target_arch = "wasm32"))]
147 : impl Drop for Counted {
148 6164 : fn drop(&mut self) {
149 6164 : BACKGROUND.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
150 6164 : }
151 : }
152 :
153 : /// Work a request left behind after its response — the remote leg of a
154 : /// distributed subscription, an initial Context Source notification, a
155 : /// forwarded notification, a retry. Counted so a shutdown drain waits for
156 : /// it: a rolling update that killed these tasks left the chain half-built.
157 : static BACKGROUND: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
158 :
159 : /// Request-born tasks still running.
160 44 : pub fn background_tasks() -> usize {
161 44 : BACKGROUND.load(std::sync::atomic::Ordering::SeqCst)
162 44 : }
163 :
164 : /// A task that lives as long as the process (the matcher drain, the
165 : /// interval tick): never counted, or a drain would wait on it forever.
166 : #[cfg(not(target_arch = "wasm32"))]
167 2568 : pub(crate) fn spawn_loop<F>(fut: F)
168 2568 : where
169 2568 : F: std::future::Future<Output = ()> + Send + 'static,
170 : {
171 2568 : tokio::spawn(fut);
172 2568 : }
173 :
174 : #[cfg(target_arch = "wasm32")]
175 : pub(crate) fn spawn_loop<F>(fut: F)
176 : where
177 : F: std::future::Future<Output = ()> + 'static,
178 : {
179 : wasm_bindgen_futures::spawn_local(fut);
180 : }
181 :
182 : #[cfg(target_arch = "wasm32")]
183 : pub fn spawn<F>(fut: F)
184 : where
185 : F: std::future::Future<Output = ()> + 'static,
186 : {
187 : wasm_bindgen_futures::spawn_local(fut);
188 : }
189 :
190 : /// The browser build's notification channel: re-exported from the HTTP
191 : /// binding, where it belongs — the page sink is how that binding delivers
192 : /// when the runtime has no inbound socket.
193 : #[cfg(target_arch = "wasm32")]
194 : pub use antares_notifier::http::page_sink;
195 :
196 : use antares_model::{NgsiError, TenantId, API_ROOT};
197 : use axum::http::{HeaderMap, StatusCode};
198 : use axum::response::{IntoResponse, Response};
199 : use axum::routing::{delete, get, patch, post, MethodRouter};
200 : use axum::Router;
201 : use negotiate::{echo_tenant, respond, tenant_from, ApiError};
202 :
203 : /// 5.5.4 Fragment applied to a stored document resource — a Subscription
204 : /// (5.8.2), a Context Source Registration (5.9.3) or a Context Source
205 : /// Subscription (5.11.4). Each member of the already-normalized Fragment
206 : /// replaces the target's; a member that normalization left as JSON null is
207 : /// the removal form and takes the member out; `id` is never touched, and the
208 : /// write stamps `modifiedAt` (4.8). Entities do NOT go through here: 5.5.12
209 : /// Merge Patch descends into Attribute instances and value objects, which is
210 : /// `entities::merge_instance`.
211 226 : pub(crate) fn apply_doc_fragment(
212 226 : target: &mut serde_json::Map<String, serde_json::Value>,
213 226 : fragment: &serde_json::Map<String, serde_json::Value>,
214 226 : ts: &str,
215 226 : ) {
216 264 : for (k, v) in fragment {
217 264 : if k == "id" {
218 0 : continue;
219 264 : }
220 264 : if v.is_null() {
221 10 : target.remove(k);
222 254 : } else {
223 254 : target.insert(k.clone(), v.clone());
224 254 : }
225 : }
226 226 : target.insert(
227 226 : "modifiedAt".into(),
228 226 : serde_json::Value::String(ts.to_owned()),
229 : );
230 226 : }
231 :
232 : pub(crate) mod surface;
233 : pub use surface::ApiSurface;
234 :
235 : pub use antares_ql::scope::scope_matches;
236 :
237 : /// Installs the in-process pipeline on a state: the subscription mirror and
238 : /// matcher (5.8.6), the interval firing, and the consumer half of a
239 : /// distributed subscription (5.8.1.4) — a notification arriving on the
240 : /// internal endpoint is handed to `distsub` from here, so the delivery path
241 : /// never names it. Every root that serves requests calls this once.
242 1284 : pub async fn wire(state: &mut AppState) {
243 1284 : notify::wire_matcher(state).await;
244 1284 : install_csource_notification(state);
245 1284 : }
246 :
247 : /// The 5.8.1.4 consumer half on its own, for a root that wires its matcher
248 : /// some other way. A distributed Subscription forwards no copy until this
249 : /// handler exists: the internal Context Source Registration Subscription
250 : /// notifies to `urn:antares:distsub:…`, and the delivery path drops that
251 : /// notification when nothing is installed to take it.
252 1296 : pub fn install_csource_notification(state: &mut AppState) {
253 1296 : state.csource_notification = Some(std::sync::Arc::new(|st, tenant, own_id, reason, regs| {
254 104 : Box::pin(distsub::on_csource_notification(
255 104 : st, tenant, own_id, reason, regs,
256 104 : ))
257 104 : }));
258 1296 : }
259 :
260 : /// 4.22 for the documents that carry their own `expiresAt`: Registrations,
261 : /// Snapshots and EntityMaps, and the leased Subscriptions of socket
262 : /// connections. The store's own sweep reaps Entities, which is
263 : /// what 4.22 names; these three are refused by every read once their expiry
264 : /// passes but nothing collected them, so they stayed for the life of the
265 : /// broker. Each kind is reaped by the predicate its own reader uses, so the
266 : /// sweep removes exactly what a read already hides — never more.
267 : ///
268 : /// Opportunistic by construction: a driver that cannot enumerate its
269 : /// tenants reaps nothing and the reads keep hiding what is left, and a
270 : /// tenant whose walk fails is skipped rather than failing the tick.
271 : ///
272 : /// The walk is the client tenants. A Snapshot's frozen copy carries its own
273 : /// expiry on the Snapshot document, and expiring it drops the whole synthetic
274 : /// tenant (`snap_remove`), so there is nothing left inside for a per-document
275 : /// reaper to find.
276 1972 : pub async fn sweep_expired_docs(st: &AppState) -> usize {
277 1972 : let Ok(tenants) = st.store.tenant_ids().await else {
278 0 : return 0;
279 : };
280 1972 : let mut n: usize = 0;
281 145058 : for name in tenants {
282 145058 : let Ok(tenant) = antares_model::TenantId::new(&name) else {
283 0 : continue;
284 : };
285 : // Each sweep counts documents it deleted, far below `usize::MAX`.
286 145058 : n = n
287 145058 : .saturating_add(csource::sweep_expired_registrations(st, &tenant).await)
288 145058 : .saturating_add(snapshots::sweep_expired_snapshots(st, &tenant).await)
289 145058 : .saturating_add(entity_map::sweep_expired_maps(st, &tenant).await)
290 145058 : .saturating_add(subscriptions::sweep_expired_socket_subscriptions(st, &tenant).await);
291 : }
292 1972 : n
293 1972 : }
294 :
295 : /// An `AppState` with the notification pipeline installed, for tests that
296 : /// exercise a route which notifies. The router does not wire it: a caller
297 : /// that only reads never pays for the mirror seed.
298 : #[cfg(any(test, feature = "test-kit"))]
299 84 : pub async fn wired_state(host_alias: &str) -> AppState {
300 84 : let mut st = AppState::new(host_alias.to_owned());
301 84 : wire(&mut st).await;
302 84 : st
303 84 : }
304 :
305 : #[cfg(test)]
306 : mod clause_5_2_8 {
307 : use super::*;
308 : use axum::body::Body;
309 : use axum::http::{Request, StatusCode};
310 : use tower::ServiceExt;
311 :
312 20 : async fn post_reg(entities: serde_json::Value) -> StatusCode {
313 20 : let app = router(AppState::new("t528".into()));
314 20 : let body = serde_json::json!({
315 20 : "id": format!("urn:ngsi-ld:ContextSourceRegistration:528-{}",
316 20 : entities.to_string().len()),
317 20 : "type": "ContextSourceRegistration",
318 20 : "information": [{"entities": entities}],
319 20 : "endpoint": "http://cs.example.org:1026"
320 : })
321 20 : .to_string();
322 20 : let req = Request::post("/ngsi-ld/v1/csourceRegistrations")
323 20 : .header("Content-Type", "application/json")
324 20 : .header("Content-Length", body.len())
325 20 : .body(Body::from(body))
326 20 : .expect("req");
327 20 : app.oneshot(req).await.expect("resp").status()
328 20 : }
329 :
330 : /// Table 5.2.8-1: type is "String or String[]" — the array form must be
331 : /// accepted; id must be a URI; idPattern must be a valid regex.
332 : #[tokio::test]
333 4 : async fn entity_info_type_accepts_the_array_form() {
334 4 : assert_eq!(
335 4 : post_reg(serde_json::json!([{"type": ["Building", "Vehicle"]}])).await,
336 : StatusCode::CREATED,
337 : "String[] type is legal"
338 : );
339 4 : assert_eq!(
340 4 : post_reg(serde_json::json!([{"type": "Building"}])).await,
341 : StatusCode::CREATED
342 : );
343 4 : assert_eq!(
344 4 : post_reg(serde_json::json!([{"type": [] }])).await,
345 : StatusCode::BAD_REQUEST,
346 : "an empty type array names no Entity Type"
347 : );
348 4 : assert_eq!(
349 4 : post_reg(serde_json::json!([{"type": "Building", "id": "not a uri"}])).await,
350 : StatusCode::BAD_REQUEST
351 : );
352 4 : assert_eq!(
353 4 : post_reg(serde_json::json!([{"type": "Building", "idPattern": "urn:[" }])).await,
354 4 : StatusCode::BAD_REQUEST
355 4 : );
356 4 : }
357 : }
358 :
359 : #[cfg(test)]
360 : mod clause_5_2_9 {
361 : use super::*;
362 : use axum::body::Body;
363 : use axum::http::{Request, StatusCode};
364 : use tower::ServiceExt;
365 :
366 68 : async fn post_reg(extra: serde_json::Value) -> StatusCode {
367 68 : let app = router(AppState::new("t529".into()));
368 68 : let mut doc = serde_json::json!({
369 68 : "id": format!("urn:ngsi-ld:ContextSourceRegistration:529-{}",
370 68 : extra.to_string().len()),
371 68 : "type": "ContextSourceRegistration",
372 68 : "information": [{"entities": [{"type": "Building"}]}],
373 68 : "endpoint": "http://cs.example.org:1026"
374 : });
375 68 : for (k, v) in extra.as_object().expect("obj") {
376 68 : doc[k] = v.clone();
377 68 : }
378 68 : let body = doc.to_string();
379 68 : let req = Request::post("/ngsi-ld/v1/csourceRegistrations")
380 68 : .header("Content-Type", "application/json")
381 68 : .header("Content-Length", body.len())
382 68 : .body(Body::from(body))
383 68 : .expect("req");
384 68 : app.oneshot(req).await.expect("resp").status()
385 68 : }
386 :
387 : /// Table 5.2.9-1 value spaces: operations limited to the 4.20 names and
388 : /// groups; localOnly boolean; contextSourceAlias a non-empty RFC 7230
389 : /// pseudonym token; refreshRate an ISO 8601 duration; datasetId URIs or
390 : /// @none; scope per the 4.18 grammar; geometries per 4.7; description
391 : /// and registrationName non-empty strings.
392 : #[tokio::test]
393 4 : async fn registration_member_value_spaces() {
394 : use serde_json::json;
395 4 : let cases: &[(serde_json::Value, StatusCode)] = &[
396 4 : (json!({"operations": ["bogusOp"]}), StatusCode::BAD_REQUEST),
397 4 : (
398 4 : json!({"operations": ["updateOps", "retrieveEntity"]}),
399 4 : StatusCode::CREATED,
400 4 : ),
401 4 : (json!({"localOnly": "yes"}), StatusCode::BAD_REQUEST),
402 4 : (json!({"localOnly": true}), StatusCode::CREATED),
403 4 : (json!({"contextSourceAlias": ""}), StatusCode::BAD_REQUEST),
404 4 : (
405 4 : json!({"contextSourceAlias": "has space"}),
406 4 : StatusCode::BAD_REQUEST,
407 4 : ),
408 4 : (json!({"contextSourceAlias": "cs1"}), StatusCode::CREATED),
409 4 : (json!({"refreshRate": "5 minutes"}), StatusCode::BAD_REQUEST),
410 4 : (json!({"refreshRate": "PT5M"}), StatusCode::CREATED),
411 4 : (
412 4 : json!({"datasetId": ["urn:ds:1", "@none"]}),
413 4 : StatusCode::CREATED,
414 4 : ),
415 4 : (json!({"datasetId": ["not a uri"]}), StatusCode::BAD_REQUEST),
416 4 : (json!({"scope": "9bad"}), StatusCode::BAD_REQUEST),
417 4 : (json!({"scope": ["/Madrid", "/A/B_2"]}), StatusCode::CREATED),
418 4 : (json!({"description": ""}), StatusCode::BAD_REQUEST),
419 4 : (json!({"registrationName": ""}), StatusCode::BAD_REQUEST),
420 4 : (json!({"location": 5}), StatusCode::BAD_REQUEST),
421 4 : (
422 4 : json!({"location": {"type": "Point", "coordinates": [8, 40]}}),
423 4 : StatusCode::CREATED,
424 4 : ),
425 4 : ];
426 68 : for (extra, want) in cases {
427 68 : let got = post_reg(extra.clone()).await;
428 68 : assert_eq!(got, *want, "extra={extra}");
429 4 : }
430 4 : }
431 :
432 : /// 5.2.9, after Table 5.2.9-1: the Table 5.2.9-2 members "are read-only
433 : /// and shall be automatically generated by NGSI-LD implementations. In
434 : /// the event that they are provided (in update or create operations)
435 : /// NGSI-LD implementations shall ignore them." A tolerant reader that
436 : /// stores an unknown member verbatim would let a client dictate its own
437 : /// forward history, and every one of these is served back.
438 : #[tokio::test]
439 4 : async fn table_5_2_9_2_members_are_ignored_on_create_and_update() {
440 : use serde_json::json;
441 : const READ_ONLY: [&str; 5] = [
442 : "timesSent",
443 : "timesFailed",
444 : "lastSuccess",
445 : "lastFailure",
446 : "status",
447 : ];
448 4 : let id = "urn:ngsi-ld:ContextSourceRegistration:529-readonly";
449 4 : let st = AppState::new("t5292".into());
450 4 : let claimed = json!({
451 4 : "timesSent": 9_000,
452 4 : "timesFailed": 0,
453 4 : "lastSuccess": "2035-01-01T00:00:00Z",
454 4 : "lastFailure": "2035-01-01T00:00:00Z",
455 4 : "status": "ok",
456 : });
457 :
458 4 : let mut doc = json!({
459 4 : "id": id,
460 4 : "type": "ContextSourceRegistration",
461 4 : "information": [{"entities": [{"type": "Building"}]}],
462 4 : "endpoint": "http://cs.example.org:1026",
463 : });
464 20 : for (k, v) in claimed.as_object().expect("obj") {
465 20 : doc[k] = v.clone();
466 20 : }
467 4 : let body = doc.to_string();
468 4 : let res = router(st.clone())
469 4 : .oneshot(
470 4 : Request::post("/ngsi-ld/v1/csourceRegistrations")
471 4 : .header("Content-Type", "application/json")
472 4 : .header("Content-Length", body.len())
473 4 : .body(Body::from(body))
474 4 : .expect("req"),
475 4 : )
476 4 : .await
477 4 : .expect("resp");
478 4 : assert_eq!(res.status(), StatusCode::CREATED, "create");
479 :
480 8 : let served = |st: AppState| async move {
481 8 : let res = router(st)
482 8 : .oneshot(
483 8 : Request::get(format!("/ngsi-ld/v1/csourceRegistrations/{id}"))
484 8 : .body(Body::empty())
485 8 : .expect("req"),
486 8 : )
487 8 : .await
488 8 : .expect("resp");
489 8 : assert_eq!(res.status(), StatusCode::OK, "retrieve");
490 8 : let bytes = axum::body::to_bytes(res.into_body(), usize::MAX)
491 8 : .await
492 8 : .expect("body");
493 8 : serde_json::from_slice::<serde_json::Value>(&bytes).expect("json")
494 16 : };
495 :
496 4 : let got = served(st.clone()).await;
497 20 : for m in READ_ONLY {
498 20 : assert!(
499 20 : got.get(m).is_none(),
500 : "create must ignore the read-only member {m}: {got}"
501 : );
502 : }
503 :
504 : // 5.9.3 update: the same members, through the merge patch.
505 4 : let patch = claimed.to_string();
506 4 : let res = router(st.clone())
507 4 : .oneshot(
508 4 : Request::patch(format!("/ngsi-ld/v1/csourceRegistrations/{id}"))
509 4 : .header("Content-Type", "application/json")
510 4 : .header("Content-Length", patch.len())
511 4 : .body(Body::from(patch))
512 4 : .expect("req"),
513 4 : )
514 4 : .await
515 4 : .expect("resp");
516 4 : assert_eq!(res.status(), StatusCode::NO_CONTENT, "update");
517 :
518 4 : let got = served(st).await;
519 20 : for m in READ_ONLY {
520 20 : assert!(
521 20 : got.get(m).is_none(),
522 4 : "update must ignore the read-only member {m}: {got}"
523 4 : );
524 4 : }
525 4 : }
526 : }
527 :
528 : #[cfg(test)]
529 : mod clause_5_2_10 {
530 : use super::*;
531 : use axum::body::Body;
532 : use axum::http::{Request, StatusCode};
533 : use tower::ServiceExt;
534 :
535 16 : async fn post_info(info: serde_json::Value) -> StatusCode {
536 16 : let app = router(AppState::new("t5210".into()));
537 16 : let body = serde_json::json!({
538 16 : "id": format!("urn:ngsi-ld:ContextSourceRegistration:5210-{}",
539 16 : info.to_string().len()),
540 16 : "type": "ContextSourceRegistration",
541 16 : "information": [info],
542 16 : "endpoint": "http://cs.example.org:1026"
543 : })
544 16 : .to_string();
545 16 : let req = Request::post("/ngsi-ld/v1/csourceRegistrations")
546 16 : .header("Content-Type", "application/json")
547 16 : .header("Content-Length", body.len())
548 16 : .body(Body::from(body))
549 16 : .expect("req");
550 16 : app.oneshot(req).await.expect("resp").status()
551 16 : }
552 :
553 : /// Table 5.2.10-1: empty arrays are not allowed for entities,
554 : /// propertyNames or relationshipNames; non-empty name lists are fine.
555 : #[tokio::test]
556 4 : async fn registration_info_empty_arrays_are_rejected() {
557 : use serde_json::json;
558 4 : assert_eq!(
559 4 : post_info(json!({"entities": []})).await,
560 : StatusCode::BAD_REQUEST
561 : );
562 4 : assert_eq!(
563 4 : post_info(json!({"propertyNames": []})).await,
564 : StatusCode::BAD_REQUEST,
565 : "empty propertyNames"
566 : );
567 4 : assert_eq!(
568 4 : post_info(json!({"relationshipNames": []})).await,
569 : StatusCode::BAD_REQUEST,
570 : "empty relationshipNames"
571 : );
572 4 : assert_eq!(
573 4 : post_info(json!({"entities": [{"type": "Building"}],
574 4 : "propertyNames": ["speed"],
575 4 : "relationshipNames": ["isParked"]}))
576 4 : .await,
577 4 : StatusCode::CREATED
578 4 : );
579 4 : }
580 : }
581 :
582 : #[cfg(test)]
583 : mod clause_5_2_11 {
584 : use super::*;
585 : use axum::body::Body;
586 : use axum::http::{Request, StatusCode};
587 : use tower::ServiceExt;
588 :
589 20 : async fn post_interval(iv: serde_json::Value) -> StatusCode {
590 20 : let app = router(AppState::new("t5211".into()));
591 20 : let body = serde_json::json!({
592 20 : "id": format!("urn:ngsi-ld:ContextSourceRegistration:5211-{}",
593 20 : iv.to_string().len()),
594 20 : "type": "ContextSourceRegistration",
595 20 : "information": [{"entities": [{"type": "Building"}]}],
596 20 : "endpoint": "http://cs.example.org:1026",
597 20 : "observationInterval": iv
598 : })
599 20 : .to_string();
600 20 : let req = Request::post("/ngsi-ld/v1/csourceRegistrations")
601 20 : .header("Content-Type", "application/json")
602 20 : .header("Content-Length", body.len())
603 20 : .body(Body::from(body))
604 20 : .expect("req");
605 20 : app.oneshot(req).await.expect("resp").status()
606 20 : }
607 :
608 : /// Table 5.2.11-1: startAt is a mandatory DateTime; endAt optional but a
609 : /// DateTime when present (absent = open interval).
610 : #[tokio::test]
611 4 : async fn time_interval_member_rules() {
612 : use serde_json::json;
613 4 : assert_eq!(
614 4 : post_interval(json!({"endAt": "2030-01-01T00:00:00Z"})).await,
615 : StatusCode::BAD_REQUEST,
616 : "startAt mandatory"
617 : );
618 4 : assert_eq!(
619 4 : post_interval(json!({"startAt": "2020-01-01"})).await,
620 : StatusCode::BAD_REQUEST,
621 : "a Date is not a DateTime"
622 : );
623 4 : assert_eq!(
624 4 : post_interval(json!({"startAt": "2020-01-01T00:00:00Z",
625 4 : "endAt": "not a date"}))
626 4 : .await,
627 : StatusCode::BAD_REQUEST
628 : );
629 4 : assert_eq!(
630 4 : post_interval(json!({"startAt": "2020-01-01T00:00:00Z"})).await,
631 : StatusCode::CREATED,
632 : "open interval"
633 : );
634 4 : assert_eq!(
635 4 : post_interval(json!({"startAt": "2020-01-01T00:00:00Z",
636 4 : "endAt": "2030-01-01T00:00:00Z"}))
637 4 : .await,
638 4 : StatusCode::CREATED
639 4 : );
640 4 : }
641 : }
642 :
643 : #[cfg(test)]
644 : mod clause_5_2_12 {
645 : use super::*;
646 : use axum::body::Body;
647 : use axum::http::{Request, StatusCode};
648 : use tower::ServiceExt;
649 :
650 16 : async fn post_sub(doc: serde_json::Value) -> StatusCode {
651 16 : let app = router(AppState::new("t5212".into()));
652 16 : let body = doc.to_string();
653 16 : let req = Request::post("/ngsi-ld/v1/subscriptions")
654 16 : .header("Content-Type", "application/json")
655 16 : .header("Content-Length", body.len())
656 16 : .body(Body::from(body))
657 16 : .expect("req");
658 16 : app.oneshot(req).await.expect("resp").status()
659 16 : }
660 :
661 : /// 5.2.12: "At least one of (a) entities or (b) watchedAttributes shall
662 : /// be present, unless the member localOnly is set to true".
663 : #[tokio::test]
664 4 : async fn local_only_waives_the_selector_requirement() {
665 : use serde_json::json;
666 16 : let base = |extra: serde_json::Value| {
667 16 : let mut d = json!({
668 16 : "id": format!("urn:ngsi-ld:Subscription:5212-{}", extra.to_string().len()),
669 16 : "type": "Subscription",
670 16 : "notification": {"endpoint": {"uri": "http://client.example.org/cb"}}
671 : });
672 16 : for (k, v) in extra.as_object().expect("obj") {
673 16 : d[k] = v.clone();
674 16 : }
675 16 : d
676 16 : };
677 4 : assert_eq!(
678 4 : post_sub(base(json!({}))).await,
679 : StatusCode::BAD_REQUEST,
680 : "no selector and no localOnly"
681 : );
682 4 : assert_eq!(
683 4 : post_sub(base(json!({"localOnly": true}))).await,
684 : StatusCode::CREATED,
685 : "localOnly=true waives entities/watchedAttributes"
686 : );
687 4 : assert_eq!(
688 4 : post_sub(base(json!({"localOnly": false}))).await,
689 : StatusCode::BAD_REQUEST
690 : );
691 : // the exclusions stay intact
692 4 : assert_eq!(
693 4 : post_sub(base(json!({"watchedAttributes": ["speed"],
694 4 : "timeInterval": 5})))
695 4 : .await,
696 4 : StatusCode::BAD_REQUEST
697 4 : );
698 4 : }
699 : }
700 :
701 : #[cfg(test)]
702 : mod clause_5_2_13 {
703 : use super::*;
704 : use axum::body::Body;
705 : use axum::http::{Request, StatusCode};
706 : use tower::ServiceExt;
707 :
708 20 : async fn post_geoq(geoq: serde_json::Value) -> StatusCode {
709 20 : let app = router(AppState::new("t5213".into()));
710 20 : let body = serde_json::json!({
711 20 : "id": format!("urn:ngsi-ld:Subscription:5213-{}", geoq.to_string().len()),
712 20 : "type": "Subscription",
713 20 : "entities": [{"type": "Vehicle"}],
714 20 : "geoQ": geoq,
715 20 : "notification": {"endpoint": {"uri": "http://client.example.org/cb"}}
716 : })
717 20 : .to_string();
718 20 : let req = Request::post("/ngsi-ld/v1/subscriptions")
719 20 : .header("Content-Type", "application/json")
720 20 : .header("Content-Length", body.len())
721 20 : .body(Body::from(body))
722 20 : .expect("req");
723 20 : app.oneshot(req).await.expect("resp").status()
724 20 : }
725 :
726 : /// Table 5.2.13-1: coordinates as JSON Array OR string form, geometry
727 : /// from the legal set (no GeometryCollection), georel per 4.10,
728 : /// geoproperty optional.
729 : #[tokio::test]
730 4 : async fn subscription_geoquery_member_rules() {
731 : use serde_json::json;
732 4 : assert_eq!(
733 4 : post_geoq(json!({"georel": "near;maxDistance==2000",
734 4 : "geometry": "Point", "coordinates": [8, 40]}))
735 4 : .await,
736 : StatusCode::CREATED
737 : );
738 4 : assert_eq!(
739 4 : post_geoq(json!({"georel": "within", "geometry": "Polygon",
740 4 : "coordinates": "[[[0,0],[4,0],[4,4],[0,4],[0,0]]]",
741 4 : "geoproperty": "observationSpace"}))
742 4 : .await,
743 : StatusCode::CREATED,
744 : "string-encoded coordinates (4.7.1) are legal"
745 : );
746 4 : assert_eq!(
747 4 : post_geoq(json!({"georel": "within",
748 4 : "geometry": "GeometryCollection", "coordinates": []}))
749 4 : .await,
750 : StatusCode::BAD_REQUEST
751 : );
752 4 : assert_eq!(
753 4 : post_geoq(json!({"georel": "touches", "geometry": "Point",
754 4 : "coordinates": [8, 40]}))
755 4 : .await,
756 : StatusCode::BAD_REQUEST
757 : );
758 4 : assert_eq!(
759 4 : post_geoq(json!({"geometry": "Point", "coordinates": [8, 40]})).await,
760 4 : StatusCode::BAD_REQUEST,
761 4 : "georel is mandatory"
762 4 : );
763 4 : }
764 : }
765 :
766 : #[cfg(test)]
767 : mod clause_5_2_14 {
768 : use super::*;
769 : use axum::body::Body;
770 : use axum::http::{Request, StatusCode};
771 : use tower::ServiceExt;
772 :
773 36 : async fn post_notif(n: serde_json::Value) -> StatusCode {
774 36 : let app = router(AppState::new("t5214".into()));
775 36 : let mut notif = serde_json::json!({"endpoint": {"uri": "http://client.example.org/cb"}});
776 40 : for (k, v) in n.as_object().expect("obj") {
777 40 : notif[k] = v.clone();
778 40 : }
779 36 : let body = serde_json::json!({
780 36 : "id": format!("urn:ngsi-ld:Subscription:5214-{}", n.to_string().len()),
781 36 : "type": "Subscription",
782 36 : "entities": [{"type": "Vehicle"}],
783 36 : "notification": notif
784 : })
785 36 : .to_string();
786 36 : let req = Request::post("/ngsi-ld/v1/subscriptions")
787 36 : .header("Content-Type", "application/json")
788 36 : .header("Content-Length", body.len())
789 36 : .body(Body::from(body))
790 36 : .expect("req");
791 36 : app.oneshot(req).await.expect("resp").status()
792 36 : }
793 :
794 : /// Table 5.2.14.1-1: join limited to flat/inline/@none, joinLevel a
795 : /// positive integer, sysAttrs/showChanges booleans, attributes may not
796 : /// name id/type/scope (it is "a synonym for pick, except that id, type,
797 : /// scope are not allowed").
798 : #[tokio::test]
799 4 : async fn notification_params_value_spaces() {
800 : use serde_json::json;
801 4 : assert_eq!(
802 4 : post_notif(json!({"join": "sideways"})).await,
803 : StatusCode::BAD_REQUEST
804 : );
805 4 : assert_eq!(
806 4 : post_notif(json!({"join": "inline"})).await,
807 : StatusCode::CREATED
808 : );
809 4 : assert_eq!(
810 4 : post_notif(json!({"joinLevel": 0})).await,
811 : StatusCode::BAD_REQUEST
812 : );
813 4 : assert_eq!(
814 4 : post_notif(json!({"joinLevel": 1.5})).await,
815 : StatusCode::BAD_REQUEST
816 : );
817 4 : assert_eq!(
818 4 : post_notif(json!({"join": "flat", "joinLevel": 2})).await,
819 : StatusCode::CREATED
820 : );
821 4 : assert_eq!(
822 4 : post_notif(json!({"sysAttrs": "yes"})).await,
823 : StatusCode::BAD_REQUEST
824 : );
825 4 : assert_eq!(
826 4 : post_notif(json!({"showChanges": "yes"})).await,
827 : StatusCode::BAD_REQUEST
828 : );
829 4 : assert_eq!(
830 4 : post_notif(json!({"attributes": ["id"]})).await,
831 : StatusCode::BAD_REQUEST,
832 : "attributes may not name id/type/scope"
833 : );
834 4 : assert_eq!(
835 4 : post_notif(json!({"attributes": ["speed"]})).await,
836 4 : StatusCode::CREATED
837 4 : );
838 4 : }
839 : }
840 :
841 : #[cfg(test)]
842 : mod clause_5_2_14_2 {
843 : use super::*;
844 : use axum::body::Body;
845 : use axum::http::{Request, StatusCode};
846 : use http_body_util::BodyExt;
847 : use tower::ServiceExt;
848 :
849 : /// Table 5.2.14.2-1: lastFailure/lastNotification/lastSuccess/timesSent
850 : /// are output-only — provided ones "shall ignore them" on create, and
851 : /// retrieval never echoes fabricated values.
852 : #[tokio::test]
853 4 : async fn output_only_members_are_ignored_on_input() {
854 4 : let st = AppState::new("t52142".into());
855 4 : let app = router(st);
856 4 : let body = serde_json::json!({
857 4 : "id": "urn:ngsi-ld:Subscription:52142",
858 4 : "type": "Subscription",
859 4 : "entities": [{"type": "Vehicle"}],
860 4 : "notification": {
861 4 : "endpoint": {"uri": "http://client.example.org/cb"},
862 4 : "timesSent": 999,
863 4 : "timesFailed": 999,
864 4 : "lastNotification": "1999-01-01T00:00:00Z",
865 4 : "lastSuccess": "1999-01-01T00:00:00Z",
866 4 : "lastFailure": "1999-01-01T00:00:00Z"
867 : }
868 : })
869 4 : .to_string();
870 4 : let req = Request::post("/ngsi-ld/v1/subscriptions")
871 4 : .header("Content-Type", "application/json")
872 4 : .header("Content-Length", body.len())
873 4 : .body(Body::from(body))
874 4 : .expect("req");
875 4 : let resp = app.clone().oneshot(req).await.expect("resp");
876 4 : assert_eq!(
877 4 : resp.status(),
878 : StatusCode::CREATED,
879 : "providing them is not an error"
880 : );
881 4 : let resp = app
882 4 : .oneshot(
883 4 : Request::get("/ngsi-ld/v1/subscriptions/urn:ngsi-ld:Subscription:52142")
884 4 : .body(Body::empty())
885 4 : .expect("req"),
886 4 : )
887 4 : .await
888 4 : .expect("resp");
889 4 : assert_eq!(resp.status(), StatusCode::OK);
890 4 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
891 4 : let doc: serde_json::Value = serde_json::from_slice(&bytes).expect("json");
892 4 : let n = &doc["notification"];
893 20 : for k in [
894 4 : "timesSent",
895 4 : "timesFailed",
896 4 : "lastNotification",
897 4 : "lastSuccess",
898 4 : "lastFailure",
899 4 : ] {
900 20 : assert!(
901 20 : n.get(k).is_none(),
902 4 : "client-fabricated {k} must be ignored, got {}",
903 4 : n[k]
904 4 : );
905 4 : }
906 4 : }
907 : }
908 :
909 : #[cfg(test)]
910 : mod clause_5_2_15 {
911 : use super::*;
912 : use axum::body::Body;
913 : use axum::http::{Request, StatusCode};
914 : use tower::ServiceExt;
915 :
916 28 : async fn post_ep(ep: serde_json::Value) -> StatusCode {
917 28 : let app = router(AppState::new("t5215".into()));
918 28 : let body = serde_json::json!({
919 28 : "id": format!("urn:ngsi-ld:Subscription:5215-{}", ep.to_string().len()),
920 28 : "type": "Subscription",
921 28 : "entities": [{"type": "Vehicle"}],
922 28 : "notification": {"endpoint": ep}
923 : })
924 28 : .to_string();
925 28 : let req = Request::post("/ngsi-ld/v1/subscriptions")
926 28 : .header("Content-Type", "application/json")
927 28 : .header("Content-Length", body.len())
928 28 : .body(Body::from(body))
929 28 : .expect("req");
930 28 : app.oneshot(req).await.expect("resp").status()
931 28 : }
932 :
933 : /// Table 5.2.15-1: uri mandatory, accept value space, cooldown/timeout
934 : /// > 0, receiverInfo/notifierInfo as KeyValuePair[] (5.2.22).
935 : #[tokio::test]
936 4 : async fn endpoint_member_rules() {
937 : use serde_json::json;
938 4 : let uri = "http://client.example.org/cb";
939 4 : assert_eq!(
940 4 : post_ep(json!({"accept": "application/json"})).await,
941 : StatusCode::BAD_REQUEST,
942 : "uri mandatory"
943 : );
944 4 : assert_eq!(
945 4 : post_ep(json!({"uri": uri, "accept": "text/plain"})).await,
946 : StatusCode::BAD_REQUEST
947 : );
948 4 : assert_eq!(
949 4 : post_ep(json!({"uri": uri, "cooldown": 0})).await,
950 : StatusCode::BAD_REQUEST
951 : );
952 4 : assert_eq!(
953 4 : post_ep(json!({"uri": uri, "timeout": -1})).await,
954 : StatusCode::BAD_REQUEST
955 : );
956 4 : assert_eq!(
957 4 : post_ep(json!({"uri": uri,
958 4 : "receiverInfo": [{"key": "Authorization", "value": "Bearer x"}],
959 4 : "cooldown": 500, "timeout": 3000}))
960 4 : .await,
961 : StatusCode::CREATED
962 : );
963 4 : assert_eq!(
964 4 : post_ep(json!({"uri": uri, "receiverInfo": ["junk"]})).await,
965 : StatusCode::BAD_REQUEST,
966 : "receiverInfo entries must be {{key, value}} pairs"
967 : );
968 4 : assert_eq!(
969 4 : post_ep(json!({"uri": uri, "notifierInfo": [{"novalue": true}]})).await,
970 4 : StatusCode::BAD_REQUEST,
971 4 : "notifierInfo entries must be {{key, value}} pairs"
972 4 : );
973 4 : }
974 : }
975 :
976 : #[cfg(test)]
977 : mod clause_5_2_16 {
978 : use super::*;
979 : use axum::body::Body;
980 : use axum::http::{Request, StatusCode};
981 : use http_body_util::BodyExt;
982 : use tower::ServiceExt;
983 :
984 : /// Tables 5.2.16-1/5.2.17-1: a partial batch failure answers 207 with
985 : /// success = URI array and errors = BatchEntityError[] (entityId +
986 : /// RFC 7807 ProblemDetails).
987 : #[tokio::test]
988 4 : async fn batch_result_and_entity_error_shapes() {
989 4 : let app = router(AppState::new("t5216".into()));
990 4 : let body = serde_json::json!([
991 4 : {"id": "urn:ngsi-ld:V:ok", "type": "Vehicle"},
992 4 : {"id": "not a uri", "type": "Vehicle"}
993 : ])
994 4 : .to_string();
995 4 : let req = Request::post("/ngsi-ld/v1/entityOperations/create")
996 4 : .header("Content-Type", "application/json")
997 4 : .header("Content-Length", body.len())
998 4 : .body(Body::from(body))
999 4 : .expect("req");
1000 4 : let resp = app.oneshot(req).await.expect("resp");
1001 4 : assert_eq!(resp.status(), StatusCode::MULTI_STATUS);
1002 4 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1003 4 : let doc: serde_json::Value = serde_json::from_slice(&bytes).expect("json");
1004 4 : assert_eq!(doc["success"], serde_json::json!(["urn:ngsi-ld:V:ok"]));
1005 4 : let errs = doc["errors"].as_array().expect("errors array");
1006 4 : assert_eq!(errs.len(), 1);
1007 4 : let e = &errs[0];
1008 4 : assert!(e.get("entityId").is_some());
1009 4 : let pd = &e["error"];
1010 12 : for k in ["type", "title", "status"] {
1011 12 : assert!(
1012 12 : pd.get(k).is_some(),
1013 4 : "ProblemDetails member {k} missing: {pd}"
1014 4 : );
1015 4 : }
1016 4 : assert!(
1017 4 : pd["type"].as_str().unwrap_or("").contains("errors/"),
1018 4 : "error.type is the NGSI-LD error URI"
1019 4 : );
1020 4 : }
1021 : }
1022 :
1023 : #[cfg(test)]
1024 : mod clause_5_2_18 {
1025 : use super::*;
1026 : use axum::body::Body;
1027 : use axum::http::{Request, StatusCode};
1028 : use http_body_util::BodyExt;
1029 : use tower::ServiceExt;
1030 :
1031 : /// Tables 5.2.18-1/5.2.19-1: partial attribute update answers 207 with
1032 : /// updated = names and notUpdated = {attributeName, reason}[].
1033 : #[tokio::test]
1034 4 : async fn update_result_and_not_updated_details_shape() {
1035 4 : let app = router(AppState::new("t5218".into()));
1036 4 : let body = serde_json::json!({"id": "urn:ngsi-ld:V:5218", "type": "Vehicle",
1037 4 : "speed": {"type": "Property", "value": 1}})
1038 4 : .to_string();
1039 4 : let req = Request::post("/ngsi-ld/v1/entities")
1040 4 : .header("Content-Type", "application/json")
1041 4 : .header("Content-Length", body.len())
1042 4 : .body(Body::from(body))
1043 4 : .expect("req");
1044 4 : assert_eq!(
1045 4 : app.clone().oneshot(req).await.expect("r").status(),
1046 : StatusCode::CREATED
1047 : );
1048 : // noOverwrite append: speed exists (skipped), brand is new (applied)
1049 4 : let frag = serde_json::json!({
1050 4 : "speed": {"type": "Property", "value": 2},
1051 4 : "brand": {"type": "Property", "value": "x"}})
1052 4 : .to_string();
1053 4 : let req =
1054 4 : Request::post("/ngsi-ld/v1/entities/urn:ngsi-ld:V:5218/attrs?options=noOverwrite")
1055 4 : .header("Content-Type", "application/json")
1056 4 : .header("Content-Length", frag.len())
1057 4 : .body(Body::from(frag))
1058 4 : .expect("req");
1059 4 : let resp = app.oneshot(req).await.expect("resp");
1060 4 : assert_eq!(resp.status(), StatusCode::MULTI_STATUS);
1061 4 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1062 4 : let doc: serde_json::Value = serde_json::from_slice(&bytes).expect("json");
1063 4 : let updated = doc["updated"].as_array().expect("updated");
1064 4 : assert!(
1065 4 : updated
1066 4 : .iter()
1067 4 : .any(|u| u.as_str().unwrap_or("").contains("brand")),
1068 : "brand applied: {doc}"
1069 : );
1070 4 : let nu = doc["notUpdated"].as_array().expect("notUpdated");
1071 4 : assert_eq!(nu.len(), 1, "{doc}");
1072 4 : assert!(nu[0]["attributeName"]
1073 4 : .as_str()
1074 4 : .unwrap_or("")
1075 4 : .contains("speed"));
1076 4 : assert!(!nu[0]["reason"].as_str().unwrap_or("").is_empty());
1077 4 : assert!(
1078 4 : nu[0].get("registrationId").is_none(),
1079 4 : "local failure carries no registrationId"
1080 4 : );
1081 4 : }
1082 : }
1083 :
1084 : #[cfg(test)]
1085 : mod clause_5_2_21 {
1086 : use super::*;
1087 : use axum::body::Body;
1088 : use axum::http::{Request, StatusCode};
1089 : use http_body_util::BodyExt;
1090 : use serde_json::json;
1091 : use tower::ServiceExt;
1092 :
1093 24 : async fn temporal_query(qs: &str) -> StatusCode {
1094 24 : let app = router(AppState::new("t5221".into()));
1095 24 : let req = Request::get(format!(
1096 : "/ngsi-ld/v1/temporal/entities?type=Vehicle&timerel=after&timeAt=2020-01-01T00:00:00Z{qs}"
1097 : ))
1098 24 : .body(Body::empty())
1099 24 : .expect("req");
1100 24 : app.oneshot(req).await.expect("resp").status()
1101 24 : }
1102 :
1103 : /// Table 5.2.21-1: lastN is a POSITIVE integer; aggrMethods entries are
1104 : /// limited to the 4.5.19 methods; endTimeAt is mandatory for between.
1105 : #[tokio::test]
1106 4 : async fn temporal_query_member_value_spaces() {
1107 4 : assert_eq!(
1108 4 : temporal_query("&lastN=0").await,
1109 : StatusCode::BAD_REQUEST,
1110 : "lastN=0"
1111 : );
1112 4 : assert_eq!(temporal_query("&lastN=-3").await, StatusCode::BAD_REQUEST);
1113 4 : assert_eq!(temporal_query("&lastN=5").await, StatusCode::OK);
1114 4 : assert_eq!(
1115 4 : temporal_query("&aggrMethods=bogus").await,
1116 : StatusCode::BAD_REQUEST
1117 : );
1118 4 : assert_eq!(
1119 4 : temporal_query("&options=aggregatedValues").await,
1120 : StatusCode::BAD_REQUEST,
1121 : "aggregatedValues without aggrMethods"
1122 : );
1123 4 : assert_eq!(
1124 4 : temporal_query("&options=aggregatedValues&aggrMethods=avg,max").await,
1125 : StatusCode::OK
1126 : );
1127 4 : let app = router(AppState::new("t5221b".into()));
1128 4 : let req = Request::get(
1129 : "/ngsi-ld/v1/temporal/entities?type=V&timerel=between&timeAt=2020-01-01T00:00:00Z",
1130 : )
1131 4 : .body(Body::empty())
1132 4 : .expect("req");
1133 4 : assert_eq!(
1134 4 : app.oneshot(req).await.expect("resp").status(),
1135 4 : StatusCode::BAD_REQUEST,
1136 4 : "between without endTimeAt"
1137 4 : );
1138 4 : }
1139 :
1140 : /// POST Query (5.2.23) carrying a temporalQ object (5.2.21 JSON form).
1141 56 : async fn post_tq(tq: serde_json::Value, qs: &str) -> (StatusCode, String) {
1142 56 : let app = router(AppState::new("t5221j".into()));
1143 56 : let body = json!({
1144 56 : "type": "Query",
1145 56 : "entities": [{"type": "Vehicle"}],
1146 56 : "temporalQ": tq
1147 : })
1148 56 : .to_string();
1149 56 : let req = Request::post(format!("/ngsi-ld/v1/temporal/entityOperations/query{qs}"))
1150 56 : .header("Content-Type", "application/json")
1151 56 : .header("Content-Length", body.len())
1152 56 : .body(Body::from(body))
1153 56 : .expect("req");
1154 56 : let resp = app.oneshot(req).await.expect("resp");
1155 56 : let status = resp.status();
1156 56 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1157 56 : (status, String::from_utf8_lossy(&bytes).into_owned())
1158 56 : }
1159 :
1160 : /// Table 5.2.21-1 (JSON form): lastN is a positive INTEGER — zero,
1161 : /// negative, fractional and string values are all outside the value
1162 : /// space; timerel is limited to before/after/between; endTimeAt is
1163 : /// mandatory for between; timeproperty is limited to the four 4.8 names.
1164 : #[tokio::test]
1165 4 : async fn temporal_query_json_member_value_spaces() {
1166 4 : let ok = json!({"timerel": "after", "timeAt": "2020-01-01T00:00:00Z"});
1167 36 : let with = |k: &str, v: serde_json::Value| {
1168 36 : let mut t = ok.clone();
1169 36 : t[k] = v;
1170 36 : t
1171 36 : };
1172 4 : let (st, body) = post_tq(with("lastN", json!(5)), "").await;
1173 4 : assert_eq!(st, StatusCode::OK, "{body}");
1174 4 : assert!(!body.contains("BadRequestData"), "{body}");
1175 16 : for bad in [json!(0), json!(-3), json!(2.5), json!("5")] {
1176 16 : let (st, body) = post_tq(with("lastN", bad.clone()), "").await;
1177 16 : assert_eq!(st, StatusCode::BAD_REQUEST, "lastN={bad}");
1178 16 : assert!(body.contains("lastN"), "{body}");
1179 : }
1180 4 : let (st, _) = post_tq(with("timerel", json!("bogus")), "").await;
1181 4 : assert_eq!(st, StatusCode::BAD_REQUEST);
1182 4 : let (st, _) = post_tq(with("timerel", json!("between")), "").await;
1183 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "between without endTimeAt");
1184 4 : let (st, _) = post_tq(with("timeproperty", json!("expiresAt")), "").await;
1185 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "timeproperty outside 4.8 set");
1186 4 : let (st, _) = post_tq(with("timeAt", json!(20200101)), "").await;
1187 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "timeAt must be a string");
1188 4 : }
1189 :
1190 : /// Table 5.2.21-1: aggrMethods (comma separated list of string — both
1191 : /// the string and string-array spellings) and aggrPeriodDuration are
1192 : /// carried by the JSON TemporalQuery and honoured when
1193 : /// aggregatedValues is requested via format/options.
1194 : #[tokio::test]
1195 4 : async fn temporal_query_json_aggregation_members() {
1196 4 : let base = json!({"timerel": "after", "timeAt": "2020-01-01T00:00:00Z"});
1197 16 : let with = |k: &str, v: serde_json::Value| {
1198 16 : let mut t = base.clone();
1199 16 : t[k] = v;
1200 16 : t
1201 16 : };
1202 4 : let (st, body) = post_tq(
1203 4 : with("aggrMethods", json!(["avg", "max"])),
1204 4 : "?format=aggregatedValues",
1205 : )
1206 4 : .await;
1207 4 : assert_eq!(st, StatusCode::OK, "array aggrMethods honoured: {body}");
1208 4 : let (st, body) = post_tq(
1209 4 : with("aggrMethods", json!("avg,max")),
1210 4 : "?format=aggregatedValues",
1211 : )
1212 4 : .await;
1213 4 : assert_eq!(st, StatusCode::OK, "string aggrMethods honoured: {body}");
1214 4 : let (st, body) = post_tq(
1215 4 : with("aggrMethods", json!(["bogus"])),
1216 4 : "?format=aggregatedValues",
1217 : )
1218 4 : .await;
1219 4 : assert_eq!(st, StatusCode::BAD_REQUEST);
1220 4 : assert!(body.contains("aggrMethods"), "{body}");
1221 4 : let (st, _) = post_tq(with("aggrMethods", json!(42)), "?format=aggregatedValues").await;
1222 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "aggrMethods wrong JSON type");
1223 4 : let mut t = base.clone();
1224 4 : t["aggrMethods"] = json!(["avg"]);
1225 4 : t["aggrPeriodDuration"] = json!("bogus");
1226 4 : let (st, body) = post_tq(t, "?format=aggregatedValues").await;
1227 4 : assert_eq!(st, StatusCode::BAD_REQUEST);
1228 4 : assert!(body.contains("aggrPeriodDuration"), "{body}");
1229 4 : }
1230 :
1231 : /// 5.2.21 used from Subscription.temporalQ (5.2.12, CSR subscriptions):
1232 : /// timerel and timeAt are cardinality 1 — a temporalQ violating the
1233 : /// data type is rejected at subscription creation.
1234 : #[tokio::test]
1235 4 : async fn subscription_temporal_q_validated() {
1236 20 : async fn post_sub(tq: serde_json::Value) -> StatusCode {
1237 20 : let app = router(AppState::new("t5221s".into()));
1238 20 : let body = json!({
1239 20 : "id": format!("urn:ngsi-ld:Subscription:5221-{}", tq.to_string().len()),
1240 20 : "type": "Subscription",
1241 20 : "entities": [{"type": "Building"}],
1242 20 : "notification": {"endpoint": {"uri": "http://client.example.org/cb"}},
1243 20 : "temporalQ": tq
1244 : })
1245 20 : .to_string();
1246 20 : let req = Request::post("/ngsi-ld/v1/subscriptions")
1247 20 : .header("Content-Type", "application/json")
1248 20 : .header("Content-Length", body.len())
1249 20 : .body(Body::from(body))
1250 20 : .expect("req");
1251 20 : app.oneshot(req).await.expect("resp").status()
1252 20 : }
1253 4 : assert_eq!(
1254 4 : post_sub(json!({
1255 4 : "timerel": "after",
1256 4 : "timeAt": "2020-06-01T22:07:00Z",
1257 4 : "timeproperty": "createdAt"
1258 4 : }))
1259 4 : .await,
1260 : StatusCode::CREATED,
1261 : "official fixture shape stays creatable"
1262 : );
1263 4 : assert_eq!(
1264 4 : post_sub(json!({"timerel": "after"})).await,
1265 : StatusCode::BAD_REQUEST,
1266 : "timeAt is cardinality 1"
1267 : );
1268 4 : assert_eq!(
1269 4 : post_sub(json!({"timeAt": "2020-06-01T22:07:00Z"})).await,
1270 : StatusCode::BAD_REQUEST,
1271 : "timerel is cardinality 1"
1272 : );
1273 4 : assert_eq!(
1274 4 : post_sub(json!({"timerel": "bogus", "timeAt": "2020-06-01T22:07:00Z"})).await,
1275 : StatusCode::BAD_REQUEST
1276 : );
1277 4 : assert_eq!(
1278 4 : post_sub(json!("after")).await,
1279 4 : StatusCode::BAD_REQUEST,
1280 4 : "temporalQ must be an object"
1281 4 : );
1282 4 : }
1283 : }
1284 :
1285 : #[cfg(test)]
1286 : mod clause_5_2_22 {
1287 : use super::*;
1288 : use axum::body::Body;
1289 : use axum::http::{Request, StatusCode};
1290 : use serde_json::json;
1291 : use tower::ServiceExt;
1292 :
1293 52 : async fn post(path: &str, doc: serde_json::Value, tenant: &str) -> StatusCode {
1294 52 : let app = router(AppState::new(tenant.into()));
1295 52 : let body = doc.to_string();
1296 52 : let req = Request::post(format!("/ngsi-ld/v1/{path}"))
1297 52 : .header("Content-Type", "application/json")
1298 52 : .header("Content-Length", body.len())
1299 52 : .body(Body::from(body))
1300 52 : .expect("req");
1301 52 : app.oneshot(req).await.expect("resp").status()
1302 52 : }
1303 :
1304 : /// Table 5.2.22-1: key AND value are Strings, both cardinality 1 —
1305 : /// enforced on the notification endpoint's receiverInfo/notifierInfo.
1306 : #[tokio::test]
1307 4 : async fn endpoint_info_values_must_be_strings() {
1308 28 : let sub = |info_key: &str, entries: serde_json::Value| {
1309 28 : let mut d = json!({
1310 28 : "type": "Subscription",
1311 28 : "entities": [{"type": "Building"}],
1312 28 : "notification": {"endpoint": {"uri": "http://client.example.org/cb"}}
1313 : });
1314 28 : d["notification"]["endpoint"][info_key] = entries;
1315 28 : d
1316 28 : };
1317 4 : assert_eq!(
1318 4 : post(
1319 4 : "subscriptions",
1320 4 : sub(
1321 4 : "receiverInfo",
1322 4 : json!([{"key": "Authorization", "value": "Bearer x"}])
1323 4 : ),
1324 4 : "t5222a"
1325 4 : )
1326 4 : .await,
1327 : StatusCode::CREATED,
1328 : "string values stay creatable"
1329 : );
1330 20 : for bad in [
1331 4 : json!(42),
1332 4 : json!({"a": 1}),
1333 4 : json!(["x"]),
1334 4 : json!(null),
1335 4 : json!(true),
1336 4 : ] {
1337 20 : assert_eq!(
1338 20 : post(
1339 20 : "subscriptions",
1340 20 : sub("receiverInfo", json!([{"key": "K", "value": bad}])),
1341 20 : "t5222a"
1342 20 : )
1343 20 : .await,
1344 4 : StatusCode::BAD_REQUEST,
1345 4 : "receiverInfo value {bad} is not a String"
1346 4 : );
1347 4 : }
1348 4 : assert_eq!(
1349 4 : post(
1350 4 : "subscriptions",
1351 4 : sub("notifierInfo", json!([{"key": "K", "value": 7}])),
1352 4 : "t5222a"
1353 4 : )
1354 4 : .await,
1355 4 : StatusCode::BAD_REQUEST,
1356 4 : "notifierInfo value must be a String"
1357 4 : );
1358 4 : }
1359 :
1360 : /// Table 5.2.22-1 via 5.2.9 contextSourceInfo: every pair's value is a
1361 : /// String — a non-string value on a custom key is rejected at
1362 : /// registration, not at first forward.
1363 : #[tokio::test]
1364 4 : async fn context_source_info_values_must_be_strings() {
1365 24 : let csr = |info: serde_json::Value| {
1366 24 : json!({
1367 24 : "type": "ContextSourceRegistration",
1368 24 : "endpoint": "http://peer.example/ngsi-ld/v1",
1369 24 : "information": [{"entities": [{"type": "Building"}]}],
1370 24 : "contextSourceInfo": info
1371 : })
1372 24 : };
1373 4 : assert_eq!(
1374 4 : post(
1375 4 : "csourceRegistrations",
1376 4 : csr(json!([{"key": "X-Auth-Token", "value": "abc"}])),
1377 4 : "t5222b"
1378 4 : )
1379 4 : .await,
1380 : StatusCode::CREATED,
1381 : "string values stay registrable"
1382 : );
1383 16 : for bad in [json!(123), json!(["a"]), json!({"v": 1}), json!(null)] {
1384 16 : assert_eq!(
1385 16 : post(
1386 16 : "csourceRegistrations",
1387 16 : csr(json!([{"key": "X-Custom", "value": bad}])),
1388 16 : "t5222b"
1389 16 : )
1390 16 : .await,
1391 4 : StatusCode::BAD_REQUEST,
1392 4 : "contextSourceInfo value {bad} is not a String"
1393 4 : );
1394 4 : }
1395 4 : assert_eq!(
1396 4 : post(
1397 4 : "csourceRegistrations",
1398 4 : csr(json!([{"key": 5, "value": "v"}])),
1399 4 : "t5222b"
1400 4 : )
1401 4 : .await,
1402 4 : StatusCode::BAD_REQUEST,
1403 4 : "key must be a String"
1404 4 : );
1405 4 : }
1406 : }
1407 :
1408 : #[cfg(test)]
1409 : mod clause_5_2_44 {
1410 : use super::*;
1411 : use axum::body::Body;
1412 : use axum::http::{Request, StatusCode};
1413 : use http_body_util::BodyExt;
1414 : use serde_json::json;
1415 : use tower::ServiceExt;
1416 :
1417 : /// Table 5.2.44-1: aggrParams carries aggrMethods (comma separated list
1418 : /// of strings — string and string-array spellings) + aggrPeriodDuration,
1419 : /// honoured on the temporal operation when aggregatedValues is requested.
1420 : #[tokio::test]
1421 4 : async fn aggregation_params_members() {
1422 4 : let app = router(AppState::new("t5244".into()));
1423 24 : let send = |doc: serde_json::Value, qs: &'static str| {
1424 24 : let app = app.clone();
1425 24 : async move {
1426 24 : let body = doc.to_string();
1427 24 : let req = Request::post(format!("/ngsi-ld/v1/temporal/entityOperations/query{qs}"))
1428 24 : .header("Content-Type", "application/json")
1429 24 : .header("Content-Length", body.len())
1430 24 : .body(Body::from(body))
1431 24 : .expect("req");
1432 24 : let resp = app.oneshot(req).await.expect("resp");
1433 24 : let status = resp.status();
1434 24 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1435 24 : (status, String::from_utf8_lossy(&bytes).into_owned())
1436 24 : }
1437 24 : };
1438 24 : let with_ap = |ap: serde_json::Value| {
1439 24 : json!({"type": "Query", "entities": [{"type": "Vehicle"}],
1440 24 : "temporalQ": {"timerel": "after", "timeAt": "2020-01-01T00:00:00Z"},
1441 24 : "aggrParams": ap})
1442 24 : };
1443 4 : let (st, body) = send(
1444 4 : with_ap(json!({"aggrMethods": ["avg"], "aggrPeriodDuration": "PT1H"})),
1445 4 : "?format=aggregatedValues",
1446 4 : )
1447 4 : .await;
1448 4 : assert_eq!(st, StatusCode::OK, "array spelling honoured: {body}");
1449 4 : assert!(!body.contains("BadRequestData"), "{body}");
1450 4 : let (st, _) = send(
1451 4 : with_ap(json!({"aggrMethods": "avg,max"})),
1452 4 : "?format=aggregatedValues",
1453 4 : )
1454 4 : .await;
1455 4 : assert_eq!(st, StatusCode::OK, "string spelling honoured");
1456 4 : let (st, body) = send(
1457 4 : with_ap(json!({"aggrMethods": ["bogus"]})),
1458 4 : "?format=aggregatedValues",
1459 4 : )
1460 4 : .await;
1461 4 : assert_eq!(st, StatusCode::BAD_REQUEST);
1462 4 : assert!(body.contains("aggrMethods"), "{body}");
1463 4 : let (st, body) = send(
1464 4 : with_ap(json!({"aggrMethods": ["avg"], "aggrPeriodDuration": "bogus"})),
1465 4 : "?format=aggregatedValues",
1466 4 : )
1467 4 : .await;
1468 4 : assert_eq!(st, StatusCode::BAD_REQUEST);
1469 4 : assert!(body.contains("aggrPeriodDuration"), "{body}");
1470 4 : let (st, _) = send(
1471 4 : with_ap(json!({"aggrMethods": 42})),
1472 4 : "?format=aggregatedValues",
1473 4 : )
1474 4 : .await;
1475 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "aggrMethods wrong JSON type");
1476 4 : let (st, _) = send(with_ap(json!("avg")), "?format=aggregatedValues").await;
1477 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "aggrParams must be an object");
1478 4 : }
1479 : }
1480 :
1481 : #[cfg(test)]
1482 : mod clause_5_2_43 {
1483 : use super::*;
1484 : use axum::body::Body;
1485 : use axum::http::{Request, StatusCode};
1486 : use http_body_util::BodyExt;
1487 : use serde_json::json;
1488 : use tower::ServiceExt;
1489 :
1490 36 : async fn send(
1491 36 : app: &axum::Router,
1492 36 : method: &str,
1493 36 : path: &str,
1494 36 : doc: Option<serde_json::Value>,
1495 36 : ) -> (StatusCode, String) {
1496 36 : let mut b = Request::builder()
1497 36 : .method(method)
1498 36 : .uri(format!("/ngsi-ld/v1/{path}"));
1499 36 : let body = match doc {
1500 32 : Some(d) => {
1501 32 : let s = d.to_string();
1502 32 : b = b
1503 32 : .header("Content-Type", "application/json")
1504 32 : .header("Content-Length", s.len());
1505 32 : Body::from(s)
1506 : }
1507 4 : None => Body::empty(),
1508 : };
1509 36 : let resp = app
1510 36 : .clone()
1511 36 : .oneshot(b.body(body).expect("req"))
1512 36 : .await
1513 36 : .expect("resp");
1514 36 : let status = resp.status();
1515 36 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1516 36 : (status, String::from_utf8_lossy(&bytes).into_owned())
1517 36 : }
1518 :
1519 : /// Table 5.2.43-1: orderBy maps to the 4.23 keys, coordinates (JSON
1520 : /// array, mandatory for dist ordering) + geometry (default Point) form
1521 : /// the reference geometry; collation is rejected loudly while only
1522 : /// codepoint order is offered (4.23.1 named gap).
1523 : #[tokio::test]
1524 4 : async fn ordering_params_members() {
1525 4 : let app = router(AppState::new("t5243".into()));
1526 8 : for (id, lon, lat) in [
1527 4 : ("urn:ngsi-ld:Vehicle:near", 8.01, 40.01),
1528 4 : ("urn:ngsi-ld:Vehicle:far", 10.0, 45.0),
1529 4 : ] {
1530 8 : let (st, body) = send(
1531 8 : &app,
1532 8 : "POST",
1533 8 : "entities",
1534 8 : Some(json!({"id": id, "type": "Vehicle",
1535 8 : "location": {"type": "GeoProperty",
1536 8 : "value": {"type": "Point", "coordinates": [lon, lat]}}})),
1537 : )
1538 8 : .await;
1539 8 : assert_eq!(st, StatusCode::CREATED, "{body}");
1540 : }
1541 4 : let q = "entityOperations/query";
1542 24 : let with_ordering = |o: serde_json::Value| json!({"type": "Query", "entities": [{"type": "Vehicle"}], "ordering": o});
1543 4 : let (st, body) = send(
1544 4 : &app,
1545 4 : "POST",
1546 4 : q,
1547 4 : Some(with_ordering(json!({"orderBy": ["location;dist-asc"],
1548 4 : "coordinates": [8, 40], "geometry": "Point"}))),
1549 : )
1550 4 : .await;
1551 4 : assert_eq!(st, StatusCode::OK, "{body}");
1552 4 : let near = body.find("urn:ngsi-ld:Vehicle:near").expect("near in body");
1553 4 : let far = body.find("urn:ngsi-ld:Vehicle:far").expect("far in body");
1554 4 : assert!(near < far, "dist-asc must order near before far: {body}");
1555 4 : let (st, _) = send(
1556 4 : &app,
1557 4 : "POST",
1558 4 : q,
1559 4 : Some(with_ordering(json!({"orderBy": ["location;dist-asc"]}))),
1560 : )
1561 4 : .await;
1562 4 : assert_eq!(
1563 : st,
1564 : StatusCode::BAD_REQUEST,
1565 : "dist ordering without coordinates"
1566 : );
1567 4 : let (st, _) = send(
1568 4 : &app,
1569 4 : "POST",
1570 4 : q,
1571 4 : Some(with_ordering(json!({"orderBy": ["location;dist-asc"],
1572 4 : "coordinates": "8,40"}))),
1573 : )
1574 4 : .await;
1575 4 : assert_eq!(
1576 : st,
1577 : StatusCode::BAD_REQUEST,
1578 : "coordinates must be a JSON array"
1579 : );
1580 4 : let (st, _) = send(
1581 4 : &app,
1582 4 : "POST",
1583 4 : q,
1584 4 : Some(with_ordering(json!({"orderBy": ["id;asc"], "geometry": 5}))),
1585 : )
1586 4 : .await;
1587 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "geometry must be a string");
1588 : // 4.23.3 EXAMPLE 7: a named ICU collation is honoured (200), and a
1589 : // non-string member stays a loud 400
1590 4 : let (st, body) = send(
1591 4 : &app,
1592 4 : "POST",
1593 4 : q,
1594 4 : Some(with_ordering(json!({"orderBy": ["id;asc"],
1595 4 : "collation": "de-u-co-phonebk"}))),
1596 : )
1597 4 : .await;
1598 4 : assert_eq!(st, StatusCode::OK, "named collation is honoured: {body}");
1599 4 : let (st, body) = send(
1600 4 : &app,
1601 4 : "POST",
1602 4 : q,
1603 4 : Some(with_ordering(
1604 4 : json!({"orderBy": ["id;asc"], "collation": 5}),
1605 4 : )),
1606 : )
1607 4 : .await;
1608 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "{body}");
1609 4 : assert!(body.contains("collation"), "{body}");
1610 : // GET twin: orderGeometry is a legal query parameter
1611 4 : let (st, body) = send(
1612 4 : &app,
1613 4 : "GET",
1614 4 : "entities?type=Vehicle&orderBy=location;dist-asc&orderFrom=[8,40]&orderGeometry=Point",
1615 4 : None,
1616 : )
1617 4 : .await;
1618 4 : assert_eq!(st, StatusCode::OK, "{body}");
1619 4 : }
1620 : }
1621 :
1622 : #[cfg(test)]
1623 : mod clause_5_2_34 {
1624 : use super::*;
1625 : use axum::body::Body;
1626 : use axum::http::{Request, StatusCode};
1627 : use serde_json::json;
1628 : use tower::ServiceExt;
1629 :
1630 32 : async fn post_csr(management: serde_json::Value) -> StatusCode {
1631 32 : let app = router(AppState::new("t5234".into()));
1632 32 : let body = json!({
1633 32 : "type": "ContextSourceRegistration",
1634 32 : "endpoint": "http://peer.example/ngsi-ld/v1",
1635 32 : "information": [{"entities": [{"type": "Building"}]}],
1636 32 : "management": management
1637 : })
1638 32 : .to_string();
1639 32 : let req = Request::post("/ngsi-ld/v1/csourceRegistrations")
1640 32 : .header("Content-Type", "application/json")
1641 32 : .header("Content-Length", body.len())
1642 32 : .body(Body::from(body))
1643 32 : .expect("req");
1644 32 : app.oneshot(req).await.expect("resp").status()
1645 32 : }
1646 :
1647 : /// Table 5.2.34-1: cacheDuration is an ISO 8601 duration, cooldown and
1648 : /// timeout are numbers greater than 0, localOnly is a boolean, and the
1649 : /// member itself is an object.
1650 : #[tokio::test]
1651 4 : async fn registration_management_info_value_spaces() {
1652 4 : assert_eq!(
1653 4 : post_csr(json!({"cacheDuration": "PT5M", "cooldown": 500,
1654 4 : "timeout": 3000, "localOnly": true}))
1655 4 : .await,
1656 : StatusCode::CREATED,
1657 : "conformant management info stays registrable"
1658 : );
1659 28 : for (label, m) in [
1660 4 : ("non-object", json!("yes")),
1661 4 : ("bad cacheDuration", json!({"cacheDuration": "bogus"})),
1662 4 : ("cacheDuration wrong type", json!({"cacheDuration": 300})),
1663 4 : ("cooldown zero", json!({"cooldown": 0})),
1664 4 : ("timeout negative", json!({"timeout": -5})),
1665 4 : ("timeout wrong type", json!({"timeout": "3000"})),
1666 4 : ("localOnly wrong type", json!({"localOnly": "yes"})),
1667 4 : ] {
1668 28 : assert_eq!(
1669 28 : post_csr(m).await,
1670 4 : StatusCode::BAD_REQUEST,
1671 4 : "management {label}"
1672 4 : );
1673 4 : }
1674 4 : }
1675 : }
1676 :
1677 : #[cfg(test)]
1678 : mod clause_5_2_33 {
1679 : use super::*;
1680 : use axum::body::Body;
1681 : use axum::http::{Request, StatusCode};
1682 : use http_body_util::BodyExt;
1683 : use serde_json::json;
1684 : use tower::ServiceExt;
1685 :
1686 40 : async fn send(app: &axum::Router, path: &str, doc: &serde_json::Value) -> (StatusCode, String) {
1687 40 : let body = doc.to_string();
1688 40 : let req = Request::post(format!("/ngsi-ld/v1/{path}"))
1689 40 : .header("Content-Type", "application/json")
1690 40 : .header("Content-Length", body.len())
1691 40 : .body(Body::from(body))
1692 40 : .expect("req");
1693 40 : let resp = app.clone().oneshot(req).await.expect("resp");
1694 40 : let status = resp.status();
1695 40 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1696 40 : (status, String::from_utf8_lossy(&bytes).into_owned())
1697 40 : }
1698 :
1699 : /// Table 5.2.33-1: id is "String or String[]" of valid URIs, type is
1700 : /// mandatory, and "id takes precedence over idPattern".
1701 : #[tokio::test]
1702 4 : async fn entity_selector_id_forms_and_precedence() {
1703 4 : let app = router(AppState::new("t5233a".into()));
1704 12 : for id in [
1705 4 : "urn:ngsi-ld:Vehicle:A1",
1706 4 : "urn:ngsi-ld:Vehicle:A2",
1707 4 : "urn:ngsi-ld:Vehicle:B1",
1708 4 : ] {
1709 12 : let (st, body) = send(
1710 12 : &app,
1711 12 : "entities",
1712 12 : &json!({"id": id, "type": "Vehicle",
1713 12 : "speed": {"type": "Property", "value": 1}}),
1714 : )
1715 12 : .await;
1716 12 : assert_eq!(st, StatusCode::CREATED, "{body}");
1717 : }
1718 4 : let q = "entityOperations/query";
1719 20 : let sel = |e: serde_json::Value| json!({"type": "Query", "entities": [e]});
1720 4 : let (st, body) = send(
1721 4 : &app,
1722 4 : q,
1723 4 : &sel(json!({"type": "Vehicle",
1724 4 : "id": ["urn:ngsi-ld:Vehicle:A1", "urn:ngsi-ld:Vehicle:A2"]})),
1725 : )
1726 4 : .await;
1727 4 : assert_eq!(st, StatusCode::OK, "id array form: {body}");
1728 4 : assert!(body.contains("urn:ngsi-ld:Vehicle:A1"), "{body}");
1729 4 : assert!(body.contains("urn:ngsi-ld:Vehicle:A2"), "{body}");
1730 4 : assert!(!body.contains("urn:ngsi-ld:Vehicle:B1"), "{body}");
1731 4 : let (st, body) = send(
1732 4 : &app,
1733 4 : q,
1734 4 : &sel(json!({"type": "Vehicle", "id": "urn:ngsi-ld:Vehicle:A1",
1735 4 : "idPattern": "^urn:ngsi-ld:Vehicle:B.*$"})),
1736 : )
1737 4 : .await;
1738 4 : assert_eq!(st, StatusCode::OK, "{body}");
1739 4 : assert!(
1740 4 : body.contains("urn:ngsi-ld:Vehicle:A1"),
1741 : "id takes precedence over idPattern: {body}"
1742 : );
1743 4 : let (st, _) = send(&app, q, &sel(json!({"id": "urn:ngsi-ld:Vehicle:A1"}))).await;
1744 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "type is mandatory (5.2.33)");
1745 4 : let (st, _) = send(
1746 4 : &app,
1747 4 : q,
1748 4 : &sel(json!({"type": "Vehicle", "id": ["urn:ngsi-ld:Vehicle:A1", 5]})),
1749 : )
1750 4 : .await;
1751 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "id entries must be strings");
1752 4 : let (st, _) = send(&app, q, &sel(json!({"type": "Vehicle", "id": "not a uri"}))).await;
1753 4 : assert_eq!(st, StatusCode::BAD_REQUEST, "id must be a valid URI");
1754 4 : }
1755 :
1756 : /// 5.2.33 in Subscription.entities: the String[] id form is accepted at
1757 : /// creation.
1758 : #[tokio::test]
1759 4 : async fn subscription_selector_id_array() {
1760 4 : let app = router(AppState::new("t5233b".into()));
1761 8 : let sub = |e: serde_json::Value| {
1762 8 : json!({"type": "Subscription", "entities": [e],
1763 8 : "notification": {"endpoint": {"uri": "http://client.example.org/cb"}}})
1764 8 : };
1765 4 : let (st, body) = send(
1766 4 : &app,
1767 4 : "subscriptions",
1768 4 : &sub(json!({"type": "Building",
1769 4 : "id": ["urn:ngsi-ld:Building:a", "urn:ngsi-ld:Building:b"]})),
1770 : )
1771 4 : .await;
1772 4 : assert_eq!(st, StatusCode::CREATED, "{body}");
1773 4 : let (st, _) = send(
1774 4 : &app,
1775 4 : "subscriptions",
1776 4 : &sub(json!({"type": "Building", "id": ["urn:ngsi-ld:Building:a", 5]})),
1777 : )
1778 4 : .await;
1779 4 : assert_eq!(st, StatusCode::BAD_REQUEST);
1780 4 : }
1781 : }
1782 :
1783 : #[cfg(test)]
1784 : mod clause_5_2_23 {
1785 : use super::*;
1786 : use axum::body::Body;
1787 : use axum::http::{Request, StatusCode};
1788 : use http_body_util::BodyExt;
1789 : use serde_json::json;
1790 : use tower::ServiceExt;
1791 :
1792 116 : async fn send(
1793 116 : app: &axum::Router,
1794 116 : method: &str,
1795 116 : path: &str,
1796 116 : doc: &serde_json::Value,
1797 116 : ) -> (StatusCode, String) {
1798 116 : let body = doc.to_string();
1799 116 : let req = Request::builder()
1800 116 : .method(method)
1801 116 : .uri(format!("/ngsi-ld/v1/{path}"))
1802 116 : .header("Content-Type", "application/json")
1803 116 : .header("Content-Length", body.len())
1804 116 : .body(Body::from(body))
1805 116 : .expect("req");
1806 116 : let resp = app.clone().oneshot(req).await.expect("resp");
1807 116 : let status = resp.status();
1808 116 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1809 116 : (status, String::from_utf8_lossy(&bytes).into_owned())
1810 116 : }
1811 :
1812 80 : fn query(extra: serde_json::Value) -> serde_json::Value {
1813 80 : let mut d = json!({"type": "Query", "entities": [{"type": "Vehicle"}]});
1814 84 : for (k, v) in extra.as_object().expect("obj") {
1815 84 : d[k] = v.clone();
1816 84 : }
1817 80 : d
1818 80 : }
1819 :
1820 : /// Table 5.2.23-1: member value spaces — empty arrays are not allowed
1821 : /// (entities/attrs/pick/omit), string members must be strings, joinLevel
1822 : /// is a positive integer, geoQ/ordering are objects, and
1823 : /// temporalQ/aggrParams are only allowed on the temporal operation.
1824 : #[tokio::test]
1825 4 : async fn query_body_member_value_spaces() {
1826 4 : let app = router(AppState::new("t5223a".into()));
1827 4 : let q = "entityOperations/query";
1828 4 : let (st, body) = send(&app, "POST", q, &query(json!({}))).await;
1829 4 : assert_eq!(st, StatusCode::OK, "control: {body}");
1830 4 : assert!(!body.contains("BadRequestData"), "{body}");
1831 72 : for (label, doc) in [
1832 4 : ("type not Query", json!({"type": "NotQuery"})),
1833 4 : ("entities empty", json!({"type": "Query", "entities": []})),
1834 4 : (
1835 4 : "entities not array",
1836 4 : query(json!({"entities": "Vehicle"})).clone(),
1837 4 : ),
1838 4 : ("attrs empty", query(json!({"attrs": []}))),
1839 4 : ("attrs non-string entry", query(json!({"attrs": [7]}))),
1840 4 : ("pick empty", query(json!({"pick": []}))),
1841 4 : ("omit empty", query(json!({"omit": []}))),
1842 4 : ("q not a string", query(json!({"q": 42}))),
1843 4 : ("csf not a string", query(json!({"csf": 42}))),
1844 4 : (
1845 4 : "datasetId not an array",
1846 4 : query(json!({"datasetId": "urn:x"})),
1847 4 : ),
1848 4 : (
1849 4 : "joinLevel zero",
1850 4 : query(json!({"join": "inline", "joinLevel": 0})),
1851 4 : ),
1852 4 : (
1853 4 : "joinLevel fractional",
1854 4 : query(json!({"join": "inline", "joinLevel": 2.5})),
1855 4 : ),
1856 4 : ("geoQ not an object", query(json!({"geoQ": "near"}))),
1857 4 : ("ordering not an object", query(json!({"ordering": "asc"}))),
1858 4 : (
1859 4 : "splitEntities not a boolean",
1860 4 : query(json!({"splitEntities": "yes"})),
1861 4 : ),
1862 4 : (
1863 4 : "entityMap not a boolean",
1864 4 : query(json!({"entityMap": "yes"})),
1865 4 : ),
1866 4 : (
1867 4 : "temporalQ only for the temporal operation",
1868 4 : query(json!({"temporalQ": {"timerel": "after", "timeAt": "2020-01-01T00:00:00Z"}})),
1869 4 : ),
1870 4 : (
1871 4 : "aggrParams only for the temporal operation",
1872 4 : query(json!({"aggrParams": {"aggrMethods": "avg"}})),
1873 4 : ),
1874 4 : ] {
1875 72 : let (st, body) = send(&app, "POST", q, &doc).await;
1876 72 : assert_eq!(st, StatusCode::BAD_REQUEST, "{label}: {body}");
1877 : }
1878 : // entities selector: non-object entries violate 5.2.33 EntitySelector[]
1879 4 : let (st, _) = send(
1880 4 : &app,
1881 4 : "POST",
1882 4 : q,
1883 4 : &json!({"type": "Query", "entities": ["Vehicle"]}),
1884 : )
1885 4 : .await;
1886 4 : assert_eq!(
1887 4 : st,
1888 4 : StatusCode::BAD_REQUEST,
1889 4 : "entities entries must be objects"
1890 4 : );
1891 4 : }
1892 :
1893 : /// Table 5.2.23-1: q/pick/omit conveyed in the body are honoured exactly
1894 : /// like their 6.3.7 query-parameter twins.
1895 : #[tokio::test]
1896 4 : async fn query_body_members_are_honoured() {
1897 4 : let app = router(AppState::new("t5223b".into()));
1898 8 : for (id, speed) in [
1899 4 : ("urn:ngsi-ld:Vehicle:A1", 80),
1900 4 : ("urn:ngsi-ld:Vehicle:A2", 120),
1901 4 : ] {
1902 8 : let (st, body) = send(
1903 8 : &app,
1904 8 : "POST",
1905 8 : "entities",
1906 8 : &json!({"id": id, "type": "Vehicle",
1907 8 : "speed": {"type": "Property", "value": speed},
1908 8 : "brand": {"type": "Property", "value": "Mercedes"}}),
1909 : )
1910 8 : .await;
1911 8 : assert_eq!(st, StatusCode::CREATED, "{body}");
1912 : }
1913 4 : let q = "entityOperations/query";
1914 4 : let (st, body) = send(&app, "POST", q, &query(json!({"q": "speed>100"}))).await;
1915 4 : assert_eq!(st, StatusCode::OK, "{body}");
1916 4 : assert!(body.contains("urn:ngsi-ld:Vehicle:A2"), "{body}");
1917 4 : assert!(
1918 4 : !body.contains("urn:ngsi-ld:Vehicle:A1"),
1919 : "q must filter: {body}"
1920 : );
1921 4 : let (st, body) = send(&app, "POST", q, &query(json!({"pick": ["speed"]}))).await;
1922 4 : assert_eq!(st, StatusCode::OK, "{body}");
1923 4 : assert!(body.contains("speed"), "{body}");
1924 4 : assert!(!body.contains("brand"), "pick must project: {body}");
1925 4 : let (st, body) = send(&app, "POST", q, &query(json!({"omit": ["speed"]}))).await;
1926 4 : assert_eq!(st, StatusCode::OK, "{body}");
1927 4 : assert!(body.contains("brand"), "{body}");
1928 4 : assert!(!body.contains("speed"), "omit must remove: {body}");
1929 4 : }
1930 :
1931 : /// Table 5.2.23-1 on the temporal operation (5.7.4): the entities
1932 : /// selector's id member selects, temporalQ is accepted, and containedBy
1933 : /// is "Only applicable for the Retrieve Entity and Query Entities
1934 : /// operations".
1935 : #[tokio::test]
1936 4 : async fn temporal_query_body_entity_selector() {
1937 4 : let app = router(AppState::new("t5223c".into()));
1938 8 : for id in ["urn:ngsi-ld:Vehicle:T1", "urn:ngsi-ld:Vehicle:T2"] {
1939 8 : let (st, body) = send(
1940 8 : &app,
1941 8 : "POST",
1942 8 : "temporal/entities",
1943 8 : &json!({"id": id, "type": "Vehicle",
1944 8 : "speed": [{"type": "Property", "value": 1,
1945 8 : "observedAt": "2020-08-01T12:00:00Z"}]}),
1946 : )
1947 8 : .await;
1948 8 : assert!(st.is_success(), "{st} {body}");
1949 : }
1950 4 : let tq = json!({"timerel": "after", "timeAt": "2020-01-01T00:00:00Z"});
1951 4 : let doc = json!({"type": "Query",
1952 4 : "entities": [{"type": "Vehicle", "id": "urn:ngsi-ld:Vehicle:T1"}],
1953 4 : "temporalQ": tq});
1954 4 : let (st, body) = send(&app, "POST", "temporal/entityOperations/query", &doc).await;
1955 4 : assert_eq!(st, StatusCode::OK, "{body}");
1956 4 : assert!(body.contains("urn:ngsi-ld:Vehicle:T1"), "{body}");
1957 4 : assert!(
1958 4 : !body.contains("urn:ngsi-ld:Vehicle:T2"),
1959 : "entities id selector must narrow the temporal query: {body}"
1960 : );
1961 4 : let doc = json!({"type": "Query", "entities": [{"type": "Vehicle"}],
1962 4 : "temporalQ": tq, "containedBy": ["urn:ngsi-ld:Vehicle:T2"]});
1963 4 : let (st, _) = send(&app, "POST", "temporal/entityOperations/query", &doc).await;
1964 4 : assert_eq!(
1965 4 : st,
1966 4 : StatusCode::BAD_REQUEST,
1967 4 : "containedBy is not applicable to 5.7.4"
1968 4 : );
1969 4 : }
1970 : }
1971 :
1972 : #[cfg(test)]
1973 : mod clause_6_3_11 {
1974 : use super::*;
1975 : use axum::body::Body;
1976 : use axum::http::{Request, StatusCode};
1977 : use axum::response::Response;
1978 : use http_body_util::BodyExt;
1979 : use tower::ServiceExt;
1980 :
1981 16 : async fn body_json(resp: Response) -> serde_json::Value {
1982 16 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1983 16 : serde_json::from_slice(&bytes).expect("json body")
1984 16 : }
1985 :
1986 : /// 6.3.11 Table 6.3.11-1: `expiresAt` is included in response payloads
1987 : /// only when `options=sysAttrs` — entity level, attribute level and
1988 : /// temporal instances alike (same gate as createdAt/modifiedAt).
1989 : #[tokio::test]
1990 4 : async fn clause_6_3_11_expires_at_gated_by_sysattrs() {
1991 4 : let mut st = AppState::new("antares-test".into());
1992 4 : crate::wire(&mut st).await;
1993 4 : let app = router(st);
1994 4 : let entity = serde_json::json!({
1995 4 : "id": "urn:ngsi-ld:Building:exp6311", "type": "Building",
1996 4 : "expiresAt": "2100-01-01T00:00:00Z",
1997 4 : "name": {"type": "Property", "value": "x",
1998 4 : "observedAt": "2026-08-13T00:00:00Z",
1999 4 : "expiresAt": "2100-01-01T00:00:00Z"}
2000 : });
2001 4 : let body = entity.to_string();
2002 4 : let resp = app
2003 4 : .clone()
2004 4 : .oneshot(
2005 4 : Request::post("/ngsi-ld/v1/entities")
2006 4 : .header("Content-Type", "application/json")
2007 4 : .header("Content-Length", body.len())
2008 4 : .body(Body::from(body))
2009 4 : .expect("req"),
2010 4 : )
2011 4 : .await
2012 4 : .expect("resp");
2013 4 : assert_eq!(resp.status(), StatusCode::CREATED);
2014 :
2015 4 : let plain = body_json(
2016 4 : app.clone()
2017 4 : .oneshot(
2018 4 : Request::get("/ngsi-ld/v1/entities/urn:ngsi-ld:Building:exp6311")
2019 4 : .body(Body::empty())
2020 4 : .expect("req"),
2021 4 : )
2022 4 : .await
2023 4 : .expect("resp"),
2024 : )
2025 4 : .await;
2026 4 : assert!(
2027 4 : plain.get("expiresAt").is_none(),
2028 : "entity expiresAt must be sysAttrs-gated (6.3.11): {plain}"
2029 : );
2030 4 : assert!(
2031 4 : plain["name"].get("expiresAt").is_none(),
2032 : "attribute expiresAt must be sysAttrs-gated (6.3.11): {plain}"
2033 : );
2034 :
2035 4 : let sys = body_json(
2036 4 : app.clone()
2037 4 : .oneshot(
2038 4 : Request::get(
2039 4 : "/ngsi-ld/v1/entities/urn:ngsi-ld:Building:exp6311?options=sysAttrs",
2040 4 : )
2041 4 : .body(Body::empty())
2042 4 : .expect("req"),
2043 4 : )
2044 4 : .await
2045 4 : .expect("resp"),
2046 : )
2047 4 : .await;
2048 4 : assert_eq!(sys["expiresAt"], "2100-01-01T00:00:00Z");
2049 4 : assert_eq!(sys["name"]["expiresAt"], "2100-01-01T00:00:00Z");
2050 :
2051 8 : let inst_of = |body: &serde_json::Value| -> serde_json::Value {
2052 8 : let a = &body["name"];
2053 8 : if a.is_array() {
2054 8 : a[0].clone()
2055 : } else {
2056 0 : a.clone()
2057 : }
2058 8 : };
2059 4 : let t_plain = body_json(
2060 4 : app.clone()
2061 4 : .oneshot(
2062 4 : Request::get(
2063 4 : "/ngsi-ld/v1/temporal/entities/urn:ngsi-ld:Building:exp6311?timerel=after&timeAt=2020-01-01T00:00:00Z",
2064 4 : )
2065 4 : .body(Body::empty())
2066 4 : .expect("req"),
2067 4 : )
2068 4 : .await
2069 4 : .expect("resp"),
2070 4 : ).await
2071 : ;
2072 4 : assert!(
2073 4 : inst_of(&t_plain).get("expiresAt").is_none(),
2074 : "temporal instance expiresAt must be sysAttrs-gated (6.3.11): {t_plain}"
2075 : );
2076 4 : let t_sys = body_json(
2077 4 : app.clone()
2078 4 : .oneshot(
2079 4 : Request::get(
2080 4 : "/ngsi-ld/v1/temporal/entities/urn:ngsi-ld:Building:exp6311?timerel=after&timeAt=2020-01-01T00:00:00Z&options=sysAttrs",
2081 4 : )
2082 4 : .body(Body::empty())
2083 4 : .expect("req"),
2084 4 : )
2085 4 : .await
2086 4 : .expect("resp"),
2087 4 : ).await
2088 : ;
2089 4 : assert_eq!(inst_of(&t_sys)["expiresAt"], "2100-01-01T00:00:00Z");
2090 4 : }
2091 : }
2092 :
2093 : /// The programmatic egress override, not `ANTARES_EGRESS_ALLOW_PRIVATE`: a
2094 : /// sibling test reading the environment while another rewrote it saw the
2095 : /// policy missing and refused the loopback forward. An atomic store carries
2096 : /// the same switch with no write for a reader to land in the middle of.
2097 : #[cfg(test)]
2098 152 : pub(crate) fn allow_private() {
2099 152 : antares_jsonld::allow_private_egress(true);
2100 152 : }
2101 :
2102 : #[cfg(test)]
2103 : mod f_ant2_019_background_count {
2104 : /// A request-born task that panics is still uncounted when it ends: 64
2105 : /// of them left the count 64 up for good, and a shutdown drain waited
2106 : /// its whole deadline for them. Other tests' tasks are few and short.
2107 : #[tokio::test(flavor = "multi_thread")]
2108 4 : async fn a_panicked_task_leaves_the_count() {
2109 : const N: usize = 64;
2110 4 : let (tx, mut rx) = tokio::sync::mpsc::channel::<()>(N);
2111 4 : for _ in 0..N {
2112 256 : let tx = tx.clone();
2113 256 : super::spawn(async move {
2114 256 : drop(tx);
2115 256 : std::panic::resume_unwind(Box::new("background task fails"));
2116 : });
2117 : }
2118 4 : drop(tx);
2119 : // every sender gone: every task ran to its panic
2120 4 : assert!(rx.recv().await.is_none());
2121 4 : tokio::time::sleep(std::time::Duration::from_millis(50)).await;
2122 4 : assert!(
2123 4 : super::background_tasks() < N,
2124 4 : "{}",
2125 4 : super::background_tasks()
2126 4 : );
2127 4 : }
2128 : }
|
