test(public-api): exercise scoped lifetimes across real HTTP
Construct example calls from fetched structured identity and scope_ref. Add exhaustive public DTO coverage and a transactional external-style provider over TCP to exercise root/self scopes, equal names in distinct scopes, optional and stale refs, deletion/republication, visibility, validator precedence and non-retry of unknown outcomes. Validation: cargo test -p wip-public-api-example (9 DTO/example tests, 4 real TCP conformance tests); Nix clippy --all-targets -D warnings.
This commit is contained in:
@@ -12,3 +12,7 @@ publish = false
|
||||
wip-host.workspace = true
|
||||
wip-interface.workspace = true
|
||||
wip-protocol.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
wip-client.workspace = true
|
||||
wip-http.workspace = true
|
||||
|
||||
@@ -229,7 +229,7 @@ pub fn run() -> Result<DemoReport, Box<dyn Error>> {
|
||||
|
||||
let published = dispatch_publish(
|
||||
&provider,
|
||||
catalog.reference(),
|
||||
&catalog_response,
|
||||
PublishRequest {
|
||||
title: "Public APIs".into(),
|
||||
body: Some("protocol values in, validated values out".into()),
|
||||
@@ -243,7 +243,7 @@ pub fn run() -> Result<DemoReport, Box<dyn Error>> {
|
||||
)?;
|
||||
let rejected = dispatch_publish(
|
||||
&provider,
|
||||
catalog.reference(),
|
||||
&catalog_response,
|
||||
PublishRequest {
|
||||
title: String::new(),
|
||||
body: None,
|
||||
@@ -256,7 +256,8 @@ pub fn run() -> Result<DemoReport, Box<dyn Error>> {
|
||||
&provider,
|
||||
request(
|
||||
OBJECT_PATH,
|
||||
catalog.reference(),
|
||||
&catalog_response.interface,
|
||||
catalog_response.scope_ref.as_deref(),
|
||||
Some(CATALOG_VALIDATOR),
|
||||
"status",
|
||||
BTreeMap::new(),
|
||||
@@ -266,7 +267,8 @@ pub fn run() -> Result<DemoReport, Box<dyn Error>> {
|
||||
&provider,
|
||||
request(
|
||||
OBJECT_PATH,
|
||||
audit.reference(),
|
||||
&audit_response.interface,
|
||||
audit_response.scope_ref.as_deref(),
|
||||
None,
|
||||
"status",
|
||||
BTreeMap::new(),
|
||||
@@ -276,7 +278,8 @@ pub fn run() -> Result<DemoReport, Box<dyn Error>> {
|
||||
let host_error = provider
|
||||
.call_operation(request(
|
||||
OBJECT_PATH,
|
||||
catalog.reference(),
|
||||
&catalog_response.interface,
|
||||
catalog_response.scope_ref.as_deref(),
|
||||
Some(CATALOG_VALIDATOR),
|
||||
"backend_status",
|
||||
BTreeMap::new(),
|
||||
@@ -288,7 +291,8 @@ pub fn run() -> Result<DemoReport, Box<dyn Error>> {
|
||||
let unknown_protocol_failure = provider
|
||||
.call_operation(request(
|
||||
OBJECT_PATH,
|
||||
catalog.reference(),
|
||||
&catalog_response.interface,
|
||||
catalog_response.scope_ref.as_deref(),
|
||||
Some(CATALOG_VALIDATOR),
|
||||
"uncertain_backend_status",
|
||||
BTreeMap::new(),
|
||||
@@ -299,7 +303,8 @@ pub fn run() -> Result<DemoReport, Box<dyn Error>> {
|
||||
let permission_failure = provider
|
||||
.call_operation(request(
|
||||
OBJECT_PATH,
|
||||
catalog.reference(),
|
||||
&catalog_response.interface,
|
||||
catalog_response.scope_ref.as_deref(),
|
||||
Some(CATALOG_VALIDATOR),
|
||||
"restricted_status",
|
||||
BTreeMap::new(),
|
||||
@@ -329,7 +334,8 @@ pub fn run() -> Result<DemoReport, Box<dyn Error>> {
|
||||
let interface_mismatch = provider
|
||||
.call_operation(request(
|
||||
OBJECT_PATH,
|
||||
diagnostics.reference(),
|
||||
&diagnostics.reference().protocol_reference(),
|
||||
catalog_response.scope_ref.as_deref(),
|
||||
None,
|
||||
"diagnostics",
|
||||
BTreeMap::new(),
|
||||
@@ -351,7 +357,8 @@ pub fn run() -> Result<DemoReport, Box<dyn Error>> {
|
||||
|
||||
let mut stale_scope = request(
|
||||
OBJECT_PATH,
|
||||
catalog.reference(),
|
||||
&catalog_response.interface,
|
||||
catalog_response.scope_ref.as_deref(),
|
||||
Some(CATALOG_VALIDATOR),
|
||||
"status",
|
||||
BTreeMap::new(),
|
||||
@@ -421,13 +428,14 @@ fn ensure_kind(error: &wip_host::CallError, expected: CallErrorKind) -> Result<(
|
||||
|
||||
fn dispatch_publish(
|
||||
provider: &ApplicationProvider,
|
||||
catalog: &InterfaceRef,
|
||||
catalog: &wip_protocol::FetchInterfaceResponse,
|
||||
request_value: PublishRequest,
|
||||
) -> Result<PublishOutcome, Box<dyn Error>> {
|
||||
let encoded_request = request_value.encode()?;
|
||||
let result = provider.call_operation(request(
|
||||
OBJECT_PATH,
|
||||
catalog,
|
||||
&catalog.interface,
|
||||
catalog.scope_ref.as_deref(),
|
||||
Some(CATALOG_VALIDATOR),
|
||||
"publish",
|
||||
BTreeMap::from([("request".into(), encoded_request)]),
|
||||
@@ -445,7 +453,8 @@ fn dispatch_string(
|
||||
|
||||
fn request(
|
||||
path: &str,
|
||||
interface: &InterfaceRef,
|
||||
interface: &InterfaceReference,
|
||||
scope_ref: Option<&str>,
|
||||
interface_validator: Option<&[u8]>,
|
||||
operation: &str,
|
||||
arguments: BTreeMap<String, Value>,
|
||||
@@ -456,8 +465,8 @@ fn request(
|
||||
validator: Some(OBJECT_VALIDATOR.to_vec()),
|
||||
},
|
||||
interface: InterfaceTarget {
|
||||
scope_ref: Some("article-service-instance".into()),
|
||||
reference: interface.protocol_reference(),
|
||||
scope_ref: scope_ref.map(str::to_owned),
|
||||
reference: interface.clone(),
|
||||
validator: interface_validator.map(<[u8]>::to_vec),
|
||||
},
|
||||
operation: operation.into(),
|
||||
|
||||
@@ -258,6 +258,40 @@ fn assert_schema_neutral(value: &Value) {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scoped_dtos_keep_identity_binding_and_validators_separate() {
|
||||
// Exact literals/destructuring guard the public Rust contract, including
|
||||
// omission of refs rather than requiring a fabricated identity.
|
||||
let reference = wip_protocol::InterfaceReference {
|
||||
scope: "/article-service".into(),
|
||||
name: "catalog::read#raw".into(),
|
||||
};
|
||||
let request = wip_protocol::FetchInterfaceRequest {
|
||||
interface: reference.clone(),
|
||||
};
|
||||
request.validate().unwrap();
|
||||
let wip_protocol::FetchInterfaceResponse {
|
||||
interface,
|
||||
scope_ref,
|
||||
descriptor,
|
||||
validator,
|
||||
} = wip_protocol::FetchInterfaceResponse {
|
||||
interface: reference.clone(),
|
||||
scope_ref: None,
|
||||
descriptor: CatalogDefinition::descriptor(),
|
||||
validator: None,
|
||||
};
|
||||
let target = wip_protocol::InterfaceTarget {
|
||||
reference: interface,
|
||||
scope_ref,
|
||||
validator,
|
||||
};
|
||||
assert_eq!(target.reference, reference);
|
||||
assert!(target.scope_ref.is_none());
|
||||
assert!(target.validator.is_none());
|
||||
assert_eq!(descriptor.format, INTERFACE_FORMAT_V1);
|
||||
}
|
||||
|
||||
fn root_reference(name: &str) -> wip_protocol::InterfaceReference {
|
||||
wip_protocol::InterfaceReference {
|
||||
scope: "/".into(),
|
||||
|
||||
@@ -0,0 +1,559 @@
|
||||
//! External-style consumer: a transactional provider, real TCP HTTP, and Client.
|
||||
use std::collections::BTreeMap;
|
||||
use std::io::{Read, Write};
|
||||
use std::net::{TcpListener, TcpStream};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
use wip_client::{
|
||||
CallOutcome, Client, ClientLimits, Completion, ObservationState, SecurityContext,
|
||||
};
|
||||
use wip_host::{
|
||||
ExecutionContext, InterfaceRegistry, ObjectProvider, ProviderResult, PublishedObject,
|
||||
RegisteredInterface, Scope, ScopeObservation, Validator, WipObject,
|
||||
};
|
||||
use wip_http::{http, *};
|
||||
use wip_interface as wip;
|
||||
use wip_protocol::{InterfaceReference, ObjectObservation, ProtocolErrorCode, ProtocolInteraction};
|
||||
|
||||
#[wip::interface]
|
||||
pub trait Status {
|
||||
fn status(&self) -> wip::OperationResult<String>;
|
||||
fn uncertain(&self) -> wip::OperationResult<String>;
|
||||
}
|
||||
|
||||
struct Service {
|
||||
name: String,
|
||||
reference: Option<String>,
|
||||
validator: Validator,
|
||||
calls: Arc<AtomicUsize>,
|
||||
}
|
||||
impl WipObject for Service {
|
||||
fn name(&self) -> &str {
|
||||
&self.name
|
||||
}
|
||||
fn reference(&self) -> Option<&str> {
|
||||
self.reference.as_deref()
|
||||
}
|
||||
fn validator(&self) -> Option<&Validator> {
|
||||
Some(&self.validator)
|
||||
}
|
||||
}
|
||||
impl Status for Service {
|
||||
fn status(&self) -> wip::OperationResult<String> {
|
||||
self.calls.fetch_add(1, Ordering::SeqCst);
|
||||
Ok(self.name.clone())
|
||||
}
|
||||
fn uncertain(&self) -> wip::OperationResult<String> {
|
||||
self.calls.fetch_add(1, Ordering::SeqCst);
|
||||
Err(wip::HostError::internal(
|
||||
"commit acknowledgement lost",
|
||||
wip::EffectGuarantee::Unknown,
|
||||
))
|
||||
}
|
||||
}
|
||||
struct Publication {
|
||||
scope: Scope,
|
||||
api: RegisteredInterface<StatusDefinition>,
|
||||
reference: Option<String>,
|
||||
}
|
||||
struct State {
|
||||
registry: InterfaceRegistry,
|
||||
root: Publication,
|
||||
service: Option<Publication>,
|
||||
calls: Arc<AtomicUsize>,
|
||||
}
|
||||
impl State {
|
||||
fn new(reference: Option<&str>) -> Self {
|
||||
let mut registry = InterfaceRegistry::new();
|
||||
let scope = Scope::new("/").unwrap();
|
||||
let root = Publication {
|
||||
api: registry
|
||||
.register(&scope, "status", StatusDefinition, None)
|
||||
.unwrap(),
|
||||
scope,
|
||||
reference: None,
|
||||
};
|
||||
let mut state = Self {
|
||||
registry,
|
||||
root,
|
||||
service: None,
|
||||
calls: Arc::default(),
|
||||
};
|
||||
state.republish(reference);
|
||||
state
|
||||
}
|
||||
fn remove(&mut self) {
|
||||
if let Some(old) = self.service.take() {
|
||||
self.registry.remove_scope(&old.scope);
|
||||
}
|
||||
}
|
||||
fn republish(&mut self, reference: Option<&str>) {
|
||||
self.remove();
|
||||
let scope = Scope::new("/service").unwrap();
|
||||
self.service = Some(Publication {
|
||||
api: self
|
||||
.registry
|
||||
.register(
|
||||
&scope,
|
||||
"status",
|
||||
StatusDefinition,
|
||||
Some(Validator::new(b"api-v1")),
|
||||
)
|
||||
.unwrap(),
|
||||
scope,
|
||||
reference: reference.map(str::to_owned),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// The HTTP adapter holds the State lock across all provider lookups, preconditions
|
||||
// and dispatch. No separate scope prelookup can race a replacement.
|
||||
struct Provider<'a> {
|
||||
state: &'a State,
|
||||
visible: bool,
|
||||
}
|
||||
impl ObjectProvider for Provider<'_> {
|
||||
type Object = Service;
|
||||
fn with_object<R>(
|
||||
&self,
|
||||
_: &ExecutionContext,
|
||||
path: &str,
|
||||
operation: impl FnOnce(&PublishedObject<Service>) -> R,
|
||||
) -> ProviderResult<R> {
|
||||
let publication = match path {
|
||||
"/" => &self.state.root,
|
||||
"/service" if self.visible => match &self.state.service {
|
||||
Some(value) => value,
|
||||
None => return ProviderResult::NotFound,
|
||||
},
|
||||
_ => return ProviderResult::NotFound,
|
||||
};
|
||||
let object = Service {
|
||||
name: if path == "/" { "" } else { "service" }.into(),
|
||||
reference: publication.reference.clone(),
|
||||
validator: Validator::new(b"object-v1"),
|
||||
calls: self.state.calls.clone(),
|
||||
};
|
||||
let current = ScopeObservation::new(&publication.scope, &object);
|
||||
let mut object = PublishedObject::new(object)
|
||||
.bind_scope(current)
|
||||
.expose(&publication.api)
|
||||
.unwrap();
|
||||
if path == "/service" {
|
||||
let root_object = Service {
|
||||
name: "".into(),
|
||||
reference: None,
|
||||
validator: Validator::new(b"object-v1"),
|
||||
calls: self.state.calls.clone(),
|
||||
};
|
||||
object = object
|
||||
.bind_scope(ScopeObservation::new(&self.state.root.scope, &root_object))
|
||||
.expose(&self.state.root.api)
|
||||
.unwrap();
|
||||
}
|
||||
ProviderResult::Found(operation(&object))
|
||||
}
|
||||
}
|
||||
fn limits() -> Limits {
|
||||
Limits::new(65536, 65536, 64).unwrap()
|
||||
}
|
||||
fn reference(scope: &str) -> InterfaceReference {
|
||||
InterfaceReference {
|
||||
scope: scope.into(),
|
||||
name: "status".into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn handle(state: &State, path: &str, body: &[u8], visible: bool) -> http::Response<Vec<u8>> {
|
||||
let provider = Provider { state, visible };
|
||||
match path {
|
||||
"/wip/v1/observe" => {
|
||||
let request = decode_observe_request(body, limits()).unwrap();
|
||||
match provider.observe_object(&request.path) {
|
||||
Ok(object) => encode_observe_response(
|
||||
&request,
|
||||
&ObjectObservation {
|
||||
object,
|
||||
children: None,
|
||||
},
|
||||
limits(),
|
||||
)
|
||||
.unwrap(),
|
||||
Err(error) => encode_protocol_error_response(
|
||||
ProtocolInteraction::Observe,
|
||||
&error.protocol_error(),
|
||||
limits(),
|
||||
)
|
||||
.unwrap(),
|
||||
}
|
||||
}
|
||||
"/wip/v1/fetch_interface" => {
|
||||
let request = decode_fetch_interface_request(body, limits()).unwrap();
|
||||
match provider.fetch_interface(&state.registry, &request.interface) {
|
||||
Ok(observation) => encode_fetch_interface_response(
|
||||
&request,
|
||||
&observation.protocol_response().unwrap(),
|
||||
limits(),
|
||||
)
|
||||
.unwrap(),
|
||||
Err(error) => encode_protocol_error_response(
|
||||
ProtocolInteraction::FetchInterface,
|
||||
&error,
|
||||
limits(),
|
||||
)
|
||||
.unwrap(),
|
||||
}
|
||||
}
|
||||
"/wip/v1/call_operation" => {
|
||||
let metadata = decode_call_operation_metadata(body, limits()).unwrap();
|
||||
let decoded = Mutex::new(None);
|
||||
let result = provider.call_operation_with_context_and_decoder(
|
||||
&ExecutionContext::default(),
|
||||
metadata.target.clone(),
|
||||
metadata.interface.clone(),
|
||||
metadata.operation.clone(),
|
||||
|descriptor| {
|
||||
let request = metadata.decode_request(body, descriptor, limits())?;
|
||||
let arguments = request.arguments.clone();
|
||||
*decoded.lock().unwrap() = Some((request, descriptor.clone()));
|
||||
Ok::<_, CodecError>(arguments)
|
||||
},
|
||||
);
|
||||
match result {
|
||||
Ok(result) => {
|
||||
let (request, descriptor) = decoded.into_inner().unwrap().unwrap();
|
||||
encode_call_operation_response(&request, &descriptor, &result, limits())
|
||||
.unwrap()
|
||||
}
|
||||
Err(error) => encode_protocol_error_response(
|
||||
ProtocolInteraction::CallOperation,
|
||||
&error.protocol_error(),
|
||||
limits(),
|
||||
)
|
||||
.unwrap(),
|
||||
}
|
||||
}
|
||||
_ => panic!("unexpected route {path}"),
|
||||
}
|
||||
}
|
||||
|
||||
// Small bounded test-only HTTP/1.1 listener. Production serving is deliberately
|
||||
// not an example dependency; both directions still cross an actual TCP socket.
|
||||
fn exchange(
|
||||
state: &Arc<Mutex<State>>,
|
||||
request: http::Request<Vec<u8>>,
|
||||
visible: bool,
|
||||
) -> http::Response<Vec<u8>> {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let address = listener.local_addr().unwrap();
|
||||
let state = state.clone();
|
||||
let server = thread::spawn(move || {
|
||||
let (mut stream, _) = listener.accept().unwrap();
|
||||
stream
|
||||
.set_read_timeout(Some(Duration::from_secs(5)))
|
||||
.unwrap();
|
||||
let mut raw = Vec::new();
|
||||
let (split, length) = loop {
|
||||
let mut chunk = [0; 1024];
|
||||
let count = stream.read(&mut chunk).unwrap();
|
||||
assert!(count > 0);
|
||||
raw.extend_from_slice(&chunk[..count]);
|
||||
assert!(raw.len() <= 70000);
|
||||
if let Some(split) = raw.windows(4).position(|part| part == b"\r\n\r\n") {
|
||||
let head = std::str::from_utf8(&raw[..split]).unwrap();
|
||||
let length = head
|
||||
.lines()
|
||||
.find_map(|line| line.strip_prefix("Content-Length: "))
|
||||
.unwrap()
|
||||
.parse::<usize>()
|
||||
.unwrap();
|
||||
if raw.len() >= split + 4 + length {
|
||||
break (split, length);
|
||||
}
|
||||
}
|
||||
};
|
||||
let path = std::str::from_utf8(&raw[..split])
|
||||
.unwrap()
|
||||
.lines()
|
||||
.next()
|
||||
.unwrap()
|
||||
.split_whitespace()
|
||||
.nth(1)
|
||||
.unwrap();
|
||||
let response = handle(
|
||||
&state.lock().unwrap(),
|
||||
path,
|
||||
&raw[split + 4..split + 4 + length],
|
||||
visible,
|
||||
);
|
||||
write!(stream, "HTTP/1.1 {} OK\r\nConnection: close\r\nContent-Length: {}\r\nContent-Type: application/json\r\n\r\n", response.status().as_u16(), response.body().len()).unwrap();
|
||||
stream.write_all(response.body()).unwrap();
|
||||
});
|
||||
let mut stream = TcpStream::connect(address).unwrap();
|
||||
stream
|
||||
.set_read_timeout(Some(Duration::from_secs(5)))
|
||||
.unwrap();
|
||||
write!(stream, "POST {} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n", request.uri().path(), request.body().len()).unwrap();
|
||||
stream.write_all(request.body()).unwrap();
|
||||
let mut raw = Vec::new();
|
||||
stream.read_to_end(&mut raw).unwrap();
|
||||
server.join().unwrap();
|
||||
let split = raw.windows(4).position(|part| part == b"\r\n\r\n").unwrap();
|
||||
let status = std::str::from_utf8(&raw[..split])
|
||||
.unwrap()
|
||||
.split_whitespace()
|
||||
.nth(1)
|
||||
.unwrap()
|
||||
.parse::<u16>()
|
||||
.unwrap();
|
||||
http::Response::builder()
|
||||
.status(status)
|
||||
.header("content-type", "application/json")
|
||||
.body(raw[split + 4..].to_vec())
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn setup() -> (Client, wip_client::SessionId) {
|
||||
let mut client = Client::new(ClientLimits::new(1, 16, 16, 16, 16).unwrap(), limits());
|
||||
let session = client
|
||||
.open_session("http://localhost/wip", SecurityContext::new("alice"))
|
||||
.unwrap();
|
||||
(client, session)
|
||||
}
|
||||
fn observe(
|
||||
client: &mut Client,
|
||||
session: &wip_client::SessionId,
|
||||
state: &Arc<Mutex<State>>,
|
||||
path: &str,
|
||||
) {
|
||||
let request = client.refresh_object(session, path).unwrap();
|
||||
let response = exchange(state, request.request, true);
|
||||
client.complete(request.id, response).unwrap();
|
||||
}
|
||||
fn fetch(
|
||||
client: &mut Client,
|
||||
session: &wip_client::SessionId,
|
||||
state: &Arc<Mutex<State>>,
|
||||
reference: &InterfaceReference,
|
||||
) {
|
||||
let request = client
|
||||
.prepare_interface(session, reference.clone())
|
||||
.unwrap();
|
||||
let body = std::str::from_utf8(request.request.body()).unwrap();
|
||||
assert!(body.contains("\"scope\""));
|
||||
assert!(body.contains("\"name\""));
|
||||
let response = exchange(state, request.request, true);
|
||||
assert!(matches!(
|
||||
client.complete(request.id, response).unwrap(),
|
||||
Completion::Interface(_)
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn real_http_root_self_same_name_and_optional_scope_refs() {
|
||||
for scope_ref in [Some("instance-a"), None] {
|
||||
let state = Arc::new(Mutex::new(State::new(scope_ref)));
|
||||
let (mut client, session) = setup();
|
||||
observe(&mut client, &session, &state, "/");
|
||||
let root = client.object(&session, "/").unwrap();
|
||||
assert_eq!(root.object.unwrap().name, "");
|
||||
observe(&mut client, &session, &state, "/service");
|
||||
for scope in ["/", "/service"] {
|
||||
let selected = reference(scope);
|
||||
fetch(&mut client, &session, &state, &selected);
|
||||
let observation = client.interface(&session, &selected).unwrap().clone();
|
||||
assert_eq!(
|
||||
observation.scope_ref.as_deref(),
|
||||
if scope == "/" { None } else { scope_ref }
|
||||
);
|
||||
let request = client
|
||||
.prepare_call(&session, "/service", &selected, "status", BTreeMap::new())
|
||||
.unwrap();
|
||||
let logical = decode_call_operation_request(
|
||||
request.request.body(),
|
||||
observation.descriptor.as_ref().unwrap(),
|
||||
limits(),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(logical.interface.reference, selected);
|
||||
assert_eq!(logical.interface.scope_ref, observation.scope_ref);
|
||||
let response = exchange(&state, request.request, true);
|
||||
assert!(matches!(
|
||||
client.complete(request.id, response).unwrap(),
|
||||
Completion::Call(record) if matches!(record.outcome, CallOutcome::Success(_))
|
||||
));
|
||||
}
|
||||
assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 2);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stale_scope_republication_and_ref_less_replacement_require_explicit_calls() {
|
||||
let selected = reference("/service");
|
||||
let state = Arc::new(Mutex::new(State::new(Some("instance-a"))));
|
||||
let (mut client, session) = setup();
|
||||
observe(&mut client, &session, &state, "/service");
|
||||
fetch(&mut client, &session, &state, &selected);
|
||||
let request = client
|
||||
.prepare_call(&session, "/service", &selected, "status", BTreeMap::new())
|
||||
.unwrap();
|
||||
state.lock().unwrap().republish(Some("instance-b"));
|
||||
let response = exchange(&state, request.request, true);
|
||||
assert_eq!(response.status(), http::StatusCode::CONFLICT);
|
||||
assert!(matches!(client.complete(request.id, response).unwrap(),
|
||||
Completion::Call(record) if matches!(record.outcome,
|
||||
CallOutcome::ProtocolFailure(ref error) if error.code == ProtocolErrorCode::InterfaceMismatch)));
|
||||
assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 0);
|
||||
assert_ne!(
|
||||
client.interface(&session, &selected).unwrap().state,
|
||||
ObservationState::Fresh
|
||||
);
|
||||
assert!(
|
||||
client
|
||||
.prepare_call(&session, "/service", &selected, "status", BTreeMap::new())
|
||||
.is_err()
|
||||
);
|
||||
observe(&mut client, &session, &state, "/service");
|
||||
fetch(&mut client, &session, &state, &selected);
|
||||
assert_eq!(
|
||||
client
|
||||
.interface(&session, &selected)
|
||||
.unwrap()
|
||||
.scope_ref
|
||||
.as_deref(),
|
||||
Some("instance-b")
|
||||
);
|
||||
assert_eq!(
|
||||
state.lock().unwrap().calls.load(Ordering::SeqCst),
|
||||
0,
|
||||
"re-observation is not execution"
|
||||
);
|
||||
|
||||
state.lock().unwrap().remove();
|
||||
observe(&mut client, &session, &state, "/service");
|
||||
assert!(client.interface(&session, &selected).is_none());
|
||||
state.lock().unwrap().republish(None);
|
||||
observe(&mut client, &session, &state, "/service");
|
||||
fetch(&mut client, &session, &state, &selected);
|
||||
let request = client
|
||||
.prepare_call(&session, "/service", &selected, "status", BTreeMap::new())
|
||||
.unwrap();
|
||||
// Ref-less replacement is intentionally not identity-trackable. Current
|
||||
// membership, live publication and both validators still apply at dispatch.
|
||||
state.lock().unwrap().republish(None);
|
||||
let response = exchange(&state, request.request, true);
|
||||
client.complete(request.id, response).unwrap();
|
||||
assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn real_http_unknown_outcome_is_never_automatically_retried() {
|
||||
let state = Arc::new(Mutex::new(State::new(Some("instance-a"))));
|
||||
let (mut client, session) = setup();
|
||||
observe(&mut client, &session, &state, "/service");
|
||||
let selected = reference("/service");
|
||||
fetch(&mut client, &session, &state, &selected);
|
||||
let request = client
|
||||
.prepare_call(
|
||||
&session,
|
||||
"/service",
|
||||
&selected,
|
||||
"uncertain",
|
||||
BTreeMap::new(),
|
||||
)
|
||||
.unwrap();
|
||||
let response = exchange(&state, request.request, true);
|
||||
let completion = client.complete(request.id, response).unwrap();
|
||||
assert!(
|
||||
matches!(completion, Completion::Call(record) if matches!(record.outcome, CallOutcome::Unknown { .. }))
|
||||
);
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 1);
|
||||
observe(&mut client, &session, &state, "/service");
|
||||
assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn principal_visibility_and_validator_precedence_cross_real_http() {
|
||||
let state = Arc::new(Mutex::new(State::new(Some("instance-a"))));
|
||||
let endpoint = Endpoint::parse("http://localhost/wip").unwrap();
|
||||
let selected = reference("/service");
|
||||
let fetch = wip_protocol::FetchInterfaceRequest {
|
||||
interface: selected.clone(),
|
||||
};
|
||||
let response = exchange(
|
||||
&state,
|
||||
encode_fetch_interface_request(&endpoint, &fetch, limits()).unwrap(),
|
||||
false,
|
||||
);
|
||||
assert!(
|
||||
matches!(decode_fetch_interface_response(&fetch, &response, limits()).unwrap(), DecodedResponse::ProtocolFailure { error, .. } if error.code == ProtocolErrorCode::InterfaceNotFound)
|
||||
);
|
||||
let descriptor = StatusDefinition::descriptor();
|
||||
let mut call = wip_protocol::CallOperationRequest {
|
||||
target: wip_protocol::Target {
|
||||
path: "/service".into(),
|
||||
validator: Some(b"stale-object".to_vec()),
|
||||
},
|
||||
interface: wip_protocol::InterfaceTarget {
|
||||
reference: selected,
|
||||
scope_ref: Some("stale-scope".into()),
|
||||
validator: Some(b"stale-api".to_vec()),
|
||||
},
|
||||
operation: "status".into(),
|
||||
arguments: BTreeMap::new(),
|
||||
};
|
||||
for (visible, object_validator, scope_ref, interface_validator, expected) in [
|
||||
(
|
||||
false,
|
||||
b"stale-object".as_slice(),
|
||||
Some("stale-scope"),
|
||||
Some(b"stale-api".as_slice()),
|
||||
ProtocolErrorCode::NotFound,
|
||||
),
|
||||
(
|
||||
true,
|
||||
b"stale-object".as_slice(),
|
||||
Some("stale-scope"),
|
||||
Some(b"stale-api".as_slice()),
|
||||
ProtocolErrorCode::ValidatorMismatch,
|
||||
),
|
||||
(
|
||||
true,
|
||||
b"object-v1".as_slice(),
|
||||
Some("stale-scope"),
|
||||
Some(b"stale-api".as_slice()),
|
||||
ProtocolErrorCode::InterfaceMismatch,
|
||||
),
|
||||
(
|
||||
true,
|
||||
b"object-v1".as_slice(),
|
||||
Some("instance-a"),
|
||||
None,
|
||||
ProtocolErrorCode::InterfaceValidatorRequired,
|
||||
),
|
||||
(
|
||||
true,
|
||||
b"object-v1".as_slice(),
|
||||
None,
|
||||
Some(b"stale-api".as_slice()),
|
||||
ProtocolErrorCode::InterfaceValidatorMismatch,
|
||||
),
|
||||
] {
|
||||
call.target.validator = Some(object_validator.to_vec());
|
||||
call.interface.scope_ref = scope_ref.map(str::to_owned);
|
||||
call.interface.validator = interface_validator.map(<[u8]>::to_vec);
|
||||
let response = exchange(
|
||||
&state,
|
||||
encode_call_operation_request(&endpoint, &call, &descriptor, limits()).unwrap(),
|
||||
visible,
|
||||
);
|
||||
assert!(
|
||||
matches!(decode_call_operation_response(&call, &descriptor, &response, limits()).unwrap(), DecodedResponse::ProtocolFailure { error, .. } if error.code == expected)
|
||||
);
|
||||
}
|
||||
assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
Reference in New Issue
Block a user