mod common;
use common::TestCall;
use std::time::Duration;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use unb::{handler, Handler, HandlerError, Node, Operation, Reply, Request, State, Streaming};
use unb_client::pair;
use unb_core::Kind;
use unb_runtime::{ClientStream, Wire};
struct Greeting(&'static str);
#[derive(Deserialize, JsonSchema)]
struct Ask {
name: String,
}
#[derive(Serialize, JsonSchema)]
struct Answer {
text: String,
}
#[handler]
async fn greet(
prefix: State<Greeting>,
request: Request<Ask>,
) -> Result<Reply<Answer>, HandlerError> {
Ok(Reply::new(Answer {
text: format!("{} {}", prefix.0, request.payload().name),
}))
}
#[handler]
async fn watch(request: Request<Ask>) -> Result<Streaming<Answer, HandlerError>, HandlerError> {
let name = request.into_payload().name;
Ok(Streaming::new(futures_util::stream::iter(vec![
Ok(Answer {
text: format!("first {name}"),
}),
Ok(Answer {
text: format!("second {name}"),
}),
])))
}
#[handler]
async fn forecast(request: Request<Ask>) -> Result<Reply<Answer>, HandlerError> {
Ok(Reply::new(Answer {
text: request.into_payload().name,
}))
}
#[derive(Deserialize, JsonSchema)]
struct AddTodo {
title: String,
}
#[handler]
async fn create_todo(request: Request<AddTodo>) -> Result<Reply<Value>, HandlerError> {
Ok(Reply::new(json!({ "created": request.payload().title })))
}
#[derive(Deserialize, Serialize, JsonSchema)]
#[serde(transparent)]
struct AnyJson(Value);
async fn deliver_for(stream: &mut ClientStream) -> Vec<unb_core::Envelope> {
let mut frames = Vec::new();
loop {
let event = tokio::time::timeout(Duration::from_secs(5), stream.next())
.await
.expect("frames arrive promptly");
match event.unwrap() {
Some(envelope) => {
let terminal = matches!(
envelope.kind,
unb_core::Kind::Response | unb_core::Kind::Error
);
frames.push(envelope);
if terminal {
return frames;
}
}
None => return frames,
}
}
}
#[tokio::test(flavor = "multi_thread")]
async fn one_subject_serves_a_unary_and_a_streaming_operation() {
let node = Node::builder("world-1")
.state(Greeting("hello"))
.service(greet.at_subject("world"))
.service(watch.at_subject("world"))
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let (client_side, server) = pair();
node.serve_transport(server).await;
let client = Wire::open(client_side);
let mut call = common::stream(
&client,
"/world-1/world",
Kind::Request,
json!({ "name": "ada" }),
)
.await;
let frames = deliver_for(&mut call).await;
assert_eq!(frames.last().unwrap().kind, Kind::Response);
assert_eq!(frames.last().unwrap().payload_json()["text"], "hello ada");
let mut call = common::stream(
&client,
"/world-1/world",
Kind::Subscribe,
json!({ "name": "ada" }),
)
.await;
let frames = deliver_for(&mut call).await;
let events: Vec<String> = frames
.iter()
.filter(|frame| frame.kind == Kind::Event)
.map(|frame| frame.payload_json()["text"].as_str().unwrap().to_string())
.collect();
assert_eq!(events, vec!["first ada", "second ada"]);
assert_eq!(frames.last().unwrap().kind, Kind::Response);
}
#[tokio::test(flavor = "multi_thread")]
async fn duplicate_operations_on_one_subject_fail_construction() {
let error = Node::builder("world-2")
.state(Greeting("hi"))
.service(greet.at_subject("world"))
.service(forecast.at_subject("world"))
.insecure_accept_declared_peer_identities()
.build()
.map(|_| ())
.unwrap_err();
let message = error.to_string();
assert!(
message.contains("world") && message.contains("Unary"),
"the error names the subject and operation: {message}"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn scoped_registration_composes_the_effective_subject() {
let node = Node::builder("scoped-1")
.scope("world", |world| {
world.scope("todo", |todo| todo.service(create_todo))
})
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let value = node
.request(
"/scoped-1/world.todo.create_todo",
json!({ "title": "ship" }),
)
.await
.unwrap();
assert_eq!(value["created"], "ship");
}
#[tokio::test(flavor = "multi_thread")]
async fn missing_registered_state_fails_construction_naming_the_type() {
let error = Node::builder("stateless-1")
.service(greet)
.insecure_accept_declared_peer_identities()
.build()
.map(|_| ())
.unwrap_err();
assert!(
error.to_string().contains("Greeting"),
"the error names the missing state type: {error}"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn nearest_scope_state_wins_for_scoped_services() {
let node = Node::builder("stateful-1")
.state(Greeting("node"))
.scope("inner", |scope| {
scope.state(Greeting("scope")).service(greet)
})
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let value = node
.request("/stateful-1/inner.greet", json!({ "name": "ada" }))
.await
.unwrap();
assert_eq!(value["text"], "scope ada");
}
#[tokio::test(flavor = "multi_thread")]
async fn runtime_mutation_is_operation_aware_and_atomic() {
let node = Node::builder("mutable-1")
.state(Greeting("hey"))
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let before = node.catalog_revision();
node.add_service(greet.at_subject("world")).await.unwrap();
assert_eq!(node.catalog_revision(), before + 1);
node.add_service(watch.at_subject("world")).await.unwrap();
assert_eq!(node.catalog_revision(), before + 2);
assert_eq!(node.reachable_names(), ["mutable-1"]);
let operations = &node.local_catalog(true)["subjects"][0]["operations"];
assert!(operations["unary"].is_object());
assert!(operations["streaming"].is_object());
let value = node
.request("/mutable-1/world", json!({ "name": "ada" }))
.await
.unwrap();
assert_eq!(value["text"], "hey ada");
node.remove_operation("world", Operation::Unary)
.await
.unwrap();
assert_eq!(node.catalog_revision(), before + 3);
let operations = &node.local_catalog(true)["subjects"][0]["operations"];
assert!(operations["unary"].is_null());
assert!(operations["streaming"].is_object());
let error = node
.request("/mutable-1/world", json!({ "name": "ada" }))
.await
.unwrap_err();
assert!(
error.message.contains("does not serve"),
"the remaining subject teaches the missing operation: {}",
error.message
);
assert!(
node.reachable_names()
.iter()
.any(|name| name == "mutable-1"),
"feature mutation does not alter the local node route"
);
node.remove_operation("world", Operation::Streaming)
.await
.unwrap();
assert_eq!(node.catalog_revision(), before + 4);
assert!(node.local_catalog(true)["subjects"]
.as_array()
.unwrap()
.is_empty());
assert!(
node.reachable_names()
.iter()
.any(|name| name == "mutable-1"),
"removing the final operation does not retire the local node route"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn every_handler_publishes_generated_schemas_and_its_doc_one_line() {
let node = Node::builder("catalog-1")
.state(Greeting("hi"))
.service(greet)
.service(forecast)
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let catalog = node.local_catalog(true);
let subjects = catalog["subjects"].as_array().unwrap();
let entry = |name: &str| -> Value {
subjects
.iter()
.find(|entry| entry["subject"] == name)
.cloned()
.unwrap_or_else(|| panic!("subject {name} missing from the catalog"))
};
let documented = entry("greet");
assert_eq!(
documented["one_line"], "Greet a caller by name.",
"the doc comment becomes the catalog one_line: {documented}"
);
assert!(
documented["operations"]["unary"]["input_schema"]["properties"]
.get("name")
.is_some(),
"every handler publishes its generated input schema: {documented}"
);
let generated = entry("forecast");
assert!(
generated["one_line"].is_null(),
"an undocumented handler carries no one_line"
);
assert!(
generated["operations"]["unary"]["output_schema"]["properties"]
.get("text")
.is_some(),
"every handler publishes its generated output schema"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn an_invalid_subject_fails_registration_before_serving() {
let error = Node::builder("bad-subject-1")
.state(Greeting("hi"))
.service(greet.at_subject("world..bad"))
.insecure_accept_declared_peer_identities()
.build()
.map(|_| ())
.unwrap_err();
assert!(
error.to_string().contains("world..bad"),
"construction names the invalid subject: {error}"
);
let node = Node::builder("bad-subject-2")
.state(Greeting("hi"))
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let error = node
.add_service(greet.at_subject(".leading"))
.await
.map(|_| ())
.unwrap_err();
assert!(
error.to_string().contains(".leading"),
"runtime mutation names the invalid subject: {error}"
);
assert_eq!(node.reachable_names(), ["bad-subject-2"]);
}
#[derive(Serialize, JsonSchema)]
struct Unserializable {
#[serde(serialize_with = "always_fail")]
poison: u32,
}
fn always_fail<S: serde::Serializer>(_: &u32, _serializer: S) -> Result<S::Ok, S::Error> {
Err(serde::ser::Error::custom("poisoned"))
}
#[handler]
async fn poisoned(_request: Request<AnyJson>) -> Result<Reply<Unserializable>, HandlerError> {
Ok(Reply::new(Unserializable { poison: 1 }))
}
#[tokio::test(flavor = "multi_thread")]
async fn a_failed_response_serialization_is_a_loud_internal_error() {
let node = Node::builder("poison-1")
.service(poisoned)
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let error = node
.request("/poison-1/poisoned", json!({}))
.await
.unwrap_err();
assert_eq!(error.code, unb::ErrorCode::Internal);
assert!(
error.message.contains("serialization"),
"the failure is named, never a silent null: {}",
error.message
);
}
#[tokio::test(flavor = "multi_thread")]
async fn metadata_attaches_once_per_subject() {
let node = Node::builder("meta-1")
.state(Greeting("hi"))
.service(
greet
.at_subject("world")
.describe(json!({ "one_line": "the world profile" })),
)
.service(watch.at_subject("world"))
.insecure_accept_declared_peer_identities()
.build()
.unwrap();
let catalog = node.local_catalog(false);
let subjects = catalog["subjects"].as_array().unwrap();
let entry = subjects.iter().find(|e| e["subject"] == "world").unwrap();
assert_eq!(entry["one_line"], "the world profile");
let error = Node::builder("meta-2")
.state(Greeting("hi"))
.service(greet.at_subject("world").describe(json!({ "a": 1 })))
.service(watch.at_subject("world").describe(json!({ "b": 2 })))
.insecure_accept_declared_peer_identities()
.build()
.map(|_| ())
.unwrap_err();
assert!(
error.to_string().contains("metadata"),
"conflicting metadata fails registration: {error}"
);
}