feat: support typed in-process WIP clients without HTTP
This commit is contained in:
@@ -0,0 +1,614 @@
|
||||
//! Public API tests: no HTTP types, codec, endpoint parser, or JSON dependency.
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use wip_client::direct::{Client, PreparedRequest, Request, Response};
|
||||
use wip_client::{
|
||||
CallOutcome, ClientErrorKind, ClientLimits, Completion, DispatchedTransportFailure,
|
||||
ObservationState, OutcomeUnknownReason, SecurityContext, SessionId,
|
||||
};
|
||||
use wip_protocol::{
|
||||
CallOperationResponse, FetchInterfaceResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor,
|
||||
InterfaceReference, Object, ObjectObservation, OperationDeclaration, ParameterDeclaration,
|
||||
ProtocolError, ProtocolErrorCode, ReturnDeclaration, TypeExpr, Value,
|
||||
};
|
||||
|
||||
fn setup() -> (Client, SessionId) {
|
||||
let mut client = Client::new(ClientLimits::new(4, 16, 8, 8, 4).unwrap());
|
||||
let session = client
|
||||
.open_session("in-process world", SecurityContext::new("alice"))
|
||||
.unwrap();
|
||||
(client, session)
|
||||
}
|
||||
|
||||
fn reference() -> InterfaceReference {
|
||||
InterfaceReference {
|
||||
scope: "/scope".into(),
|
||||
name: "math".into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn object(name: &str, identity: Option<&str>) -> ObjectObservation {
|
||||
ObjectObservation {
|
||||
object: Object {
|
||||
name: name.into(),
|
||||
description: None,
|
||||
interfaces: vec![reference()],
|
||||
r#ref: identity.map(str::to_owned),
|
||||
validator: None,
|
||||
},
|
||||
children: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn descriptor() -> InterfaceDescriptor {
|
||||
InterfaceDescriptor {
|
||||
format: INTERFACE_FORMAT_V1.into(),
|
||||
documentation: None,
|
||||
types: vec![],
|
||||
operations: vec![OperationDeclaration {
|
||||
name: "double".into(),
|
||||
documentation: None,
|
||||
parameters: vec![ParameterDeclaration {
|
||||
name: "n".into(),
|
||||
documentation: None,
|
||||
required: true,
|
||||
r#type: TypeExpr::Integer,
|
||||
}],
|
||||
returns: ReturnDeclaration {
|
||||
documentation: None,
|
||||
r#type: TypeExpr::Integer,
|
||||
},
|
||||
}],
|
||||
}
|
||||
}
|
||||
|
||||
fn interface(scope_ref: Option<&str>) -> FetchInterfaceResponse {
|
||||
FetchInterfaceResponse {
|
||||
interface: reference(),
|
||||
scope_ref: scope_ref.map(str::to_owned),
|
||||
validator: None,
|
||||
descriptor: descriptor(),
|
||||
}
|
||||
}
|
||||
|
||||
fn failure(code: ProtocolErrorCode) -> ProtocolError {
|
||||
ProtocolError {
|
||||
code,
|
||||
message: "test failure".into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn observe(client: &mut Client, session: &SessionId, path: &str, value: ObjectObservation) {
|
||||
let request = client.refresh_object(session, path).unwrap();
|
||||
assert!(
|
||||
matches!(&request.request, Request::Observe(value) if value.path == path && value.depth == 0)
|
||||
);
|
||||
client
|
||||
.complete(request.id, Response::Observe(Ok(value)))
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn ready(client: &mut Client, session: &SessionId) {
|
||||
observe(client, session, "/scope/item", object("item", None));
|
||||
let request = client.prepare_interface(session, reference()).unwrap();
|
||||
assert!(
|
||||
matches!(request.request, Request::FetchInterface(value) if value.interface == reference())
|
||||
);
|
||||
client
|
||||
.complete(request.id, Response::FetchInterface(Ok(interface(None))))
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn call(client: &mut Client, session: &SessionId) -> PreparedRequest {
|
||||
client
|
||||
.prepare_call(
|
||||
session,
|
||||
"/scope/item",
|
||||
&reference(),
|
||||
"double",
|
||||
BTreeMap::from([("n".into(), Value::Integer(3))]),
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn typed_interactions_succeed_with_optional_identity_and_validators_absent() {
|
||||
let (mut client, session) = setup();
|
||||
ready(&mut client, &session);
|
||||
assert!(
|
||||
client
|
||||
.ensure_observed(&session, "/scope/item", 0)
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
assert!(
|
||||
client
|
||||
.ensure_interface(&session, reference())
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
let request = call(&mut client, &session);
|
||||
let Request::CallOperation(logical) = request.request else {
|
||||
panic!("wrong request")
|
||||
};
|
||||
assert_eq!(logical.target.path, "/scope/item");
|
||||
assert_eq!(logical.target.validator, None);
|
||||
assert_eq!(logical.interface.scope_ref, None);
|
||||
assert_eq!(logical.interface.validator, None);
|
||||
assert_eq!(logical.arguments["n"], Value::Integer(3));
|
||||
client.mark_dispatched(request.id).unwrap();
|
||||
let result = CallOperationResponse {
|
||||
result: Value::Integer(6),
|
||||
validator: Some(vec![2]),
|
||||
};
|
||||
let completion = client
|
||||
.complete(request.id, Response::CallOperation(Ok(result.clone())))
|
||||
.unwrap();
|
||||
assert!(
|
||||
matches!(completion, Completion::Call(record) if record.outcome == CallOutcome::Success(result))
|
||||
);
|
||||
assert_eq!(
|
||||
client.object(&session, "/scope/item").unwrap().validator,
|
||||
Some(vec![2])
|
||||
);
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(
|
||||
request.id,
|
||||
Response::CallOperation(Ok(CallOperationResponse {
|
||||
result: Value::Integer(6),
|
||||
validator: None,
|
||||
}))
|
||||
)
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::InvalidRequest
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn worldspace_and_security_identities_partition_observations_without_urls() {
|
||||
let (mut client, session) = setup();
|
||||
ready(&mut client, &session);
|
||||
let other_world = client
|
||||
.open_session("another world", SecurityContext::new("alice"))
|
||||
.unwrap();
|
||||
let other_security = client
|
||||
.open_session("in-process world", SecurityContext::new("bob"))
|
||||
.unwrap();
|
||||
for other in [&other_world, &other_security] {
|
||||
assert!(client.object(other, "/scope/item").is_none());
|
||||
assert!(client.interface(other, &reference()).is_none());
|
||||
}
|
||||
assert_eq!(
|
||||
client
|
||||
.open_session("in-process world", SecurityContext::new("alice"))
|
||||
.unwrap(),
|
||||
session
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn typed_observations_validate_name_depth_duplicate_children_and_interface_scope() {
|
||||
let (mut client, session) = setup();
|
||||
let mut wrong_scope = object("item", None);
|
||||
wrong_scope.object.interfaces[0].scope = "/elsewhere".into();
|
||||
let mut unexpected_children = object("item", None);
|
||||
unexpected_children.children = Some(vec![]);
|
||||
let mut duplicate = object("item", None);
|
||||
duplicate.children = Some(vec![object("child", None), object("child", None)]);
|
||||
for (depth, response) in [
|
||||
(0, object("wrong-name", None)),
|
||||
(1, object("item", None)),
|
||||
(0, unexpected_children),
|
||||
(0, wrong_scope),
|
||||
(1, duplicate),
|
||||
] {
|
||||
let request = client
|
||||
.refresh_observed(&session, "/scope/item", depth)
|
||||
.unwrap();
|
||||
let error = client
|
||||
.complete(request.id, Response::Observe(Ok(response)))
|
||||
.unwrap_err();
|
||||
assert_eq!(error.kind, ClientErrorKind::InvalidResponse);
|
||||
assert!(matches!(
|
||||
client.object(&session, "/scope/item").unwrap().state,
|
||||
ObservationState::Error(_)
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn typed_descriptor_correspondence_structure_and_format_are_checked() {
|
||||
let (mut client, session) = setup();
|
||||
let mut mismatched = interface(None);
|
||||
mismatched.interface.name = "other".into();
|
||||
let mut malformed = interface(None);
|
||||
malformed
|
||||
.descriptor
|
||||
.operations
|
||||
.push(malformed.descriptor.operations[0].clone());
|
||||
let mut unsupported = interface(None);
|
||||
unsupported.descriptor.format = "future".into();
|
||||
for (value, kind) in [
|
||||
(mismatched, ClientErrorKind::InvalidResponse),
|
||||
(malformed, ClientErrorKind::InvalidResponse),
|
||||
(unsupported, ClientErrorKind::UnsupportedDescriptorFormat),
|
||||
] {
|
||||
let request = client.prepare_interface(&session, reference()).unwrap();
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(request.id, Response::FetchInterface(Ok(value)))
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
kind
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn typed_arguments_and_results_use_the_fixed_descriptor() {
|
||||
let (mut client, session) = setup();
|
||||
ready(&mut client, &session);
|
||||
let invalid = client
|
||||
.prepare_call(
|
||||
&session,
|
||||
"/scope/item",
|
||||
&reference(),
|
||||
"double",
|
||||
BTreeMap::from([("n".into(), Value::String("not integer".into()))]),
|
||||
)
|
||||
.unwrap_err();
|
||||
assert_eq!(invalid.kind, ClientErrorKind::InvalidRequest);
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
let request = call(&mut client, &session);
|
||||
client.mark_dispatched(request.id).unwrap();
|
||||
let completion = client
|
||||
.complete(
|
||||
request.id,
|
||||
Response::CallOperation(Ok(CallOperationResponse {
|
||||
result: Value::String("not integer".into()),
|
||||
validator: None,
|
||||
})),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(
|
||||
matches!(completion, Completion::Call(record) if matches!(record.outcome,
|
||||
CallOutcome::Unknown { reason: OutcomeUnknownReason::ResponseDecode, .. }))
|
||||
);
|
||||
assert_eq!(
|
||||
client.object(&session, "/scope/item").unwrap().state,
|
||||
ObservationState::Stale
|
||||
);
|
||||
assert!(
|
||||
client
|
||||
.ensure_observed(&session, "/scope/item", 0)
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn typed_protocol_failures_validate_interaction_and_apply_recovery() {
|
||||
let (mut client, session) = setup();
|
||||
let request = client.refresh_object(&session, "/scope/item").unwrap();
|
||||
assert!(matches!(
|
||||
client
|
||||
.complete(
|
||||
request.id,
|
||||
Response::Observe(Err(failure(ProtocolErrorCode::NotFound)))
|
||||
)
|
||||
.unwrap(),
|
||||
Completion::ProtocolFailure(_)
|
||||
));
|
||||
let request = client.prepare_interface(&session, reference()).unwrap();
|
||||
assert!(matches!(
|
||||
client
|
||||
.complete(
|
||||
request.id,
|
||||
Response::FetchInterface(Err(failure(ProtocolErrorCode::InterfaceNotFound)))
|
||||
)
|
||||
.unwrap(),
|
||||
Completion::ProtocolFailure(_)
|
||||
));
|
||||
let request = client.refresh_object(&session, "/scope/item").unwrap();
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(
|
||||
request.id,
|
||||
Response::Observe(Err(failure(ProtocolErrorCode::OperationOutcomeUnknown)))
|
||||
)
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::InvalidResponse
|
||||
);
|
||||
let request = client.prepare_interface(&session, reference()).unwrap();
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(
|
||||
request.id,
|
||||
Response::FetchInterface(Err(failure(ProtocolErrorCode::ValidatorMismatch)))
|
||||
)
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::InvalidResponse
|
||||
);
|
||||
ready(&mut client, &session);
|
||||
let request = call(&mut client, &session);
|
||||
let completion = client
|
||||
.complete(
|
||||
request.id,
|
||||
Response::CallOperation(Err(failure(ProtocolErrorCode::InterfaceMismatch))),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(
|
||||
matches!(completion, Completion::Call(record) if matches!(record.outcome, CallOutcome::ProtocolFailure(_)))
|
||||
);
|
||||
assert_eq!(
|
||||
client.object(&session, "/scope/item").unwrap().state,
|
||||
ObservationState::Stale
|
||||
);
|
||||
assert_eq!(
|
||||
client.interface(&session, &reference()).unwrap().state,
|
||||
ObservationState::Stale
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mismatched_interactions_are_not_trusted_and_calls_remain_unknown() {
|
||||
let (mut client, session) = setup();
|
||||
let request = client.refresh_object(&session, "/scope/item").unwrap();
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(request.id, Response::FetchInterface(Ok(interface(None))))
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::InvalidResponse
|
||||
);
|
||||
let request = client.prepare_interface(&session, reference()).unwrap();
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(request.id, Response::Observe(Ok(object("item", None))))
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::InvalidResponse
|
||||
);
|
||||
ready(&mut client, &session);
|
||||
let request = call(&mut client, &session);
|
||||
let completion = client
|
||||
.complete(
|
||||
request.id,
|
||||
Response::Observe(Err(failure(ProtocolErrorCode::NotFound))),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(
|
||||
matches!(completion, Completion::Call(record) if matches!(record.outcome, CallOutcome::Unknown { .. }))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn typed_stale_responses_cannot_replace_new_observations() {
|
||||
let (mut client, session) = setup();
|
||||
let old = client.refresh_object(&session, "/scope/item").unwrap();
|
||||
assert!(
|
||||
client
|
||||
.ensure_observed(&session, "/scope/item", 0)
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
let new = client.refresh_object(&session, "/scope/item").unwrap();
|
||||
client
|
||||
.complete(new.id, Response::Observe(Ok(object("item", Some("new")))))
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(old.id, Response::Observe(Ok(object("item", Some("old")))))
|
||||
.unwrap(),
|
||||
Completion::StaleResponseRejected(old.id)
|
||||
);
|
||||
assert_eq!(
|
||||
client
|
||||
.object(&session, "/scope/item")
|
||||
.unwrap()
|
||||
.object
|
||||
.unwrap()
|
||||
.r#ref
|
||||
.as_deref(),
|
||||
Some("new")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scope_replacement_retires_descriptors_and_pending_fetch_but_not_call_snapshot() {
|
||||
let (mut client, session) = setup();
|
||||
observe(
|
||||
&mut client,
|
||||
&session,
|
||||
"/scope",
|
||||
object("scope", Some("old")),
|
||||
);
|
||||
ready(&mut client, &session);
|
||||
let call = call(&mut client, &session);
|
||||
let fetch = client.prepare_interface(&session, reference()).unwrap();
|
||||
observe(
|
||||
&mut client,
|
||||
&session,
|
||||
"/scope",
|
||||
object("scope", Some("new")),
|
||||
);
|
||||
assert!(
|
||||
client
|
||||
.interface(&session, &reference())
|
||||
.unwrap()
|
||||
.descriptor
|
||||
.is_none()
|
||||
);
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(
|
||||
fetch.id,
|
||||
Response::FetchInterface(Ok(interface(Some("old"))))
|
||||
)
|
||||
.unwrap(),
|
||||
Completion::StaleResponseRejected(fetch.id)
|
||||
);
|
||||
assert_eq!(
|
||||
client.pending_call_context(call.id).unwrap().descriptor(),
|
||||
&descriptor()
|
||||
);
|
||||
assert_eq!(
|
||||
client
|
||||
.pending_call_context(call.id)
|
||||
.unwrap()
|
||||
.request()
|
||||
.interface
|
||||
.scope_ref,
|
||||
None
|
||||
);
|
||||
// A newer descriptor has a different return type; the prepared call still uses Integer.
|
||||
let fetch = client.prepare_interface(&session, reference()).unwrap();
|
||||
let mut next = interface(Some("new"));
|
||||
next.descriptor.operations[0].returns.r#type = TypeExpr::String;
|
||||
client
|
||||
.complete(fetch.id, Response::FetchInterface(Ok(next)))
|
||||
.unwrap();
|
||||
let completion = client
|
||||
.complete(
|
||||
call.id,
|
||||
Response::CallOperation(Ok(CallOperationResponse {
|
||||
result: Value::Integer(6),
|
||||
validator: None,
|
||||
})),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(
|
||||
matches!(completion, Completion::Call(record) if matches!(record.outcome, CallOutcome::Success(_)))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scope_not_found_retires_interfaces_and_late_fetch_cannot_restore_them() {
|
||||
let (mut client, session) = setup();
|
||||
ready(&mut client, &session);
|
||||
let old = client.prepare_interface(&session, reference()).unwrap();
|
||||
let missing = client.refresh_object(&session, "/scope").unwrap();
|
||||
client
|
||||
.complete(
|
||||
missing.id,
|
||||
Response::Observe(Err(failure(ProtocolErrorCode::NotFound))),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(client.interface(&session, &reference()).is_none());
|
||||
assert_eq!(
|
||||
client
|
||||
.complete(old.id, Response::FetchInterface(Ok(interface(None))))
|
||||
.unwrap(),
|
||||
Completion::StaleResponseRejected(old.id)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn failures_before_and_after_dispatch_never_retry_operations() {
|
||||
let (mut client, session) = setup();
|
||||
ready(&mut client, &session);
|
||||
let before = call(&mut client, &session);
|
||||
assert!(
|
||||
matches!(client.fail_transport(before.id, "not handed to Host").unwrap(),
|
||||
Completion::Call(record) if matches!(record.outcome, CallOutcome::NotDispatched(_)))
|
||||
);
|
||||
assert_eq!(
|
||||
client.object(&session, "/scope/item").unwrap().state,
|
||||
ObservationState::Fresh
|
||||
);
|
||||
let after = call(&mut client, &session);
|
||||
client.mark_dispatched(after.id).unwrap();
|
||||
assert!(
|
||||
matches!(client.fail_dispatched_call(after.id, DispatchedTransportFailure::Timeout, "deadline").unwrap(),
|
||||
Completion::Call(record) if matches!(record.outcome, CallOutcome::Unknown { reason: OutcomeUnknownReason::Timeout, .. }))
|
||||
);
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
assert!(
|
||||
client
|
||||
.ensure_observed(&session, "/scope/item", 0)
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
ready(&mut client, &session);
|
||||
let unknown = call(&mut client, &session);
|
||||
assert!(
|
||||
matches!(client.complete(unknown.id, Response::CallOperation(Err(failure(ProtocolErrorCode::OperationOutcomeUnknown)))).unwrap(),
|
||||
Completion::Call(record) if matches!(record.outcome, CallOutcome::Unknown { reason: OutcomeUnknownReason::Protocol, .. }))
|
||||
);
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
assert_eq!(client.call_history(&session).unwrap().len(), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn direct_limits_bound_sessions_inflight_cache_and_history() {
|
||||
let mut client = Client::new(ClientLimits::new(1, 1, 1, 1, 1).unwrap());
|
||||
let session = client
|
||||
.open_session("world", SecurityContext::new("alice"))
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
client
|
||||
.open_session("other", SecurityContext::new("alice"))
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::Capacity
|
||||
);
|
||||
let first = client.refresh_object(&session, "/scope/item").unwrap();
|
||||
assert!(
|
||||
client
|
||||
.ensure_observed(&session, "/scope/other", 0)
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
assert_eq!(
|
||||
client
|
||||
.refresh_object(&session, "/scope/other")
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::Capacity
|
||||
);
|
||||
client
|
||||
.complete(first.id, Response::Observe(Ok(object("item", None))))
|
||||
.unwrap();
|
||||
observe(&mut client, &session, "/scope/other", object("other", None));
|
||||
assert_eq!(client.known_space(&session).unwrap().len(), 1);
|
||||
for _ in 0..3 {
|
||||
let request = client.refresh_object(&session, "/scope/other").unwrap();
|
||||
client
|
||||
.complete(request.id, Response::Observe(Ok(object("wrong", None))))
|
||||
.unwrap_err();
|
||||
}
|
||||
assert_eq!(client.diagnostics(&session).unwrap().len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invalid_direct_requests_do_not_acquire_pending_state() {
|
||||
let (mut client, session) = setup();
|
||||
assert_eq!(
|
||||
client
|
||||
.refresh_object(&session, "not/absolute")
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::InvalidRequest
|
||||
);
|
||||
assert_eq!(
|
||||
client
|
||||
.prepare_interface(
|
||||
&session,
|
||||
InterfaceReference {
|
||||
scope: "/".into(),
|
||||
name: "".into()
|
||||
}
|
||||
)
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::InvalidRequest
|
||||
);
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
assert!(client.known_space(&session).unwrap().is_empty());
|
||||
}
|
||||
@@ -0,0 +1,289 @@
|
||||
#![cfg(feature = "http")]
|
||||
//! The same stateful scenario runs through direct DTOs and the HTTP facade.
|
||||
use std::collections::BTreeMap;
|
||||
use wip_client::direct::{Client as DirectClient, Request, Response};
|
||||
use wip_client::{
|
||||
CallOutcome, Client, ClientLimits, Completion, InterfaceObservation, ObjectObservation,
|
||||
RequestId, SecurityContext, SessionId,
|
||||
};
|
||||
use wip_http::{
|
||||
Limits, decode_call_operation_request, decode_fetch_interface_request, decode_observe_request,
|
||||
encode_call_operation_response, encode_fetch_interface_response, encode_observe_response,
|
||||
encode_protocol_error_response,
|
||||
};
|
||||
use wip_protocol::{
|
||||
CallOperationResponse, FetchInterfaceResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor,
|
||||
InterfaceReference, Object, ObjectObservation as ProtocolObservation, OperationDeclaration,
|
||||
ProtocolError, ProtocolErrorCode, ProtocolInteraction, ReturnDeclaration, TypeExpr, Value,
|
||||
};
|
||||
|
||||
fn limits() -> Limits {
|
||||
Limits::new(64 * 1024, 64 * 1024, 64).unwrap()
|
||||
}
|
||||
fn reference() -> InterfaceReference {
|
||||
InterfaceReference {
|
||||
scope: "/scope".into(),
|
||||
name: "math".into(),
|
||||
}
|
||||
}
|
||||
fn descriptor() -> InterfaceDescriptor {
|
||||
InterfaceDescriptor {
|
||||
format: INTERFACE_FORMAT_V1.into(),
|
||||
documentation: None,
|
||||
types: vec![],
|
||||
operations: vec![OperationDeclaration {
|
||||
name: "run".into(),
|
||||
documentation: None,
|
||||
parameters: vec![],
|
||||
returns: ReturnDeclaration {
|
||||
documentation: None,
|
||||
r#type: TypeExpr::Integer,
|
||||
},
|
||||
}],
|
||||
}
|
||||
}
|
||||
fn observation(name: &str, identity: &str) -> ProtocolObservation {
|
||||
ProtocolObservation {
|
||||
object: Object {
|
||||
name: name.into(),
|
||||
description: None,
|
||||
interfaces: vec![reference()],
|
||||
r#ref: Some(identity.into()),
|
||||
validator: None,
|
||||
},
|
||||
children: None,
|
||||
}
|
||||
}
|
||||
fn interface(identity: &str) -> FetchInterfaceResponse {
|
||||
FetchInterfaceResponse {
|
||||
interface: reference(),
|
||||
scope_ref: Some(identity.into()),
|
||||
validator: None,
|
||||
descriptor: descriptor(),
|
||||
}
|
||||
}
|
||||
|
||||
enum Channel {
|
||||
Direct(DirectClient),
|
||||
Http(Client),
|
||||
}
|
||||
impl Channel {
|
||||
fn core(&self) -> &DirectClient {
|
||||
match self {
|
||||
Self::Direct(client) => client,
|
||||
Self::Http(client) => client,
|
||||
}
|
||||
}
|
||||
fn observe(&mut self, session: &SessionId, path: &str) -> (RequestId, Request) {
|
||||
match self {
|
||||
Self::Direct(client) => {
|
||||
let p = client.refresh_object(session, path).unwrap();
|
||||
(p.id, p.request)
|
||||
}
|
||||
Self::Http(client) => {
|
||||
let p = client.refresh_object(session, path).unwrap();
|
||||
(
|
||||
p.id,
|
||||
Request::Observe(decode_observe_request(p.request.body(), limits()).unwrap()),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
fn fetch(&mut self, session: &SessionId) -> (RequestId, Request) {
|
||||
match self {
|
||||
Self::Direct(client) => {
|
||||
let p = client.prepare_interface(session, reference()).unwrap();
|
||||
(p.id, p.request)
|
||||
}
|
||||
Self::Http(client) => {
|
||||
let p = client.prepare_interface(session, reference()).unwrap();
|
||||
(
|
||||
p.id,
|
||||
Request::FetchInterface(
|
||||
decode_fetch_interface_request(p.request.body(), limits()).unwrap(),
|
||||
),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
fn call(&mut self, session: &SessionId) -> (RequestId, Request) {
|
||||
match self {
|
||||
Self::Direct(client) => {
|
||||
let p = client
|
||||
.prepare_call(session, "/scope/item", &reference(), "run", BTreeMap::new())
|
||||
.unwrap();
|
||||
client.mark_dispatched(p.id).unwrap();
|
||||
(p.id, p.request)
|
||||
}
|
||||
Self::Http(client) => {
|
||||
let p = client
|
||||
.prepare_call(session, "/scope/item", &reference(), "run", BTreeMap::new())
|
||||
.unwrap();
|
||||
client.mark_dispatched(p.id).unwrap();
|
||||
(
|
||||
p.id,
|
||||
Request::CallOperation(
|
||||
decode_call_operation_request(p.request.body(), &descriptor(), limits())
|
||||
.unwrap(),
|
||||
),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
fn complete(&mut self, pending: (RequestId, Request), result: Response) -> Completion {
|
||||
let (id, request) = pending;
|
||||
match self {
|
||||
Self::Direct(client) => client.complete(id, result).unwrap(),
|
||||
Self::Http(client) => {
|
||||
let wire = match (request, result) {
|
||||
(Request::Observe(request), Response::Observe(Ok(value))) => {
|
||||
encode_observe_response(&request, &value, limits())
|
||||
}
|
||||
(Request::FetchInterface(request), Response::FetchInterface(Ok(value))) => {
|
||||
encode_fetch_interface_response(&request, &value, limits())
|
||||
}
|
||||
(Request::CallOperation(request), Response::CallOperation(Ok(value))) => {
|
||||
encode_call_operation_response(&request, &descriptor(), &value, limits())
|
||||
}
|
||||
(_, Response::Observe(Err(error))) => encode_protocol_error_response(
|
||||
ProtocolInteraction::Observe,
|
||||
&error,
|
||||
limits(),
|
||||
),
|
||||
(_, Response::FetchInterface(Err(error))) => encode_protocol_error_response(
|
||||
ProtocolInteraction::FetchInterface,
|
||||
&error,
|
||||
limits(),
|
||||
),
|
||||
(_, Response::CallOperation(Err(error))) => encode_protocol_error_response(
|
||||
ProtocolInteraction::CallOperation,
|
||||
&error,
|
||||
limits(),
|
||||
),
|
||||
_ => panic!("test interaction mismatch"),
|
||||
}
|
||||
.unwrap();
|
||||
client.complete(id, wire).unwrap()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
struct Snapshot {
|
||||
objects: Vec<ObjectObservation>,
|
||||
interface: Option<InterfaceObservation>,
|
||||
outcomes: Vec<CallOutcome>,
|
||||
requests: Vec<Request>,
|
||||
}
|
||||
fn scenario(http: bool) -> Snapshot {
|
||||
let state_limits = ClientLimits::new(4, 16, 8, 8, 8).unwrap();
|
||||
let (mut channel, session) = if http {
|
||||
let mut client = Client::new(state_limits, limits());
|
||||
let session = client
|
||||
.open_session("https://example.test/wip", SecurityContext::new("alice"))
|
||||
.unwrap();
|
||||
(Channel::Http(client), session)
|
||||
} else {
|
||||
let mut client = DirectClient::new(state_limits);
|
||||
let session = client
|
||||
.open_session("local world", SecurityContext::new("alice"))
|
||||
.unwrap();
|
||||
(Channel::Direct(client), session)
|
||||
};
|
||||
let mut requests = vec![];
|
||||
for (path, name, identity) in [
|
||||
("/scope", "scope", "scope-v1"),
|
||||
("/scope/item", "item", "item-v1"),
|
||||
] {
|
||||
let p = channel.observe(&session, path);
|
||||
requests.push(p.1.clone());
|
||||
channel.complete(p, Response::Observe(Ok(observation(name, identity))));
|
||||
}
|
||||
let p = channel.fetch(&session);
|
||||
requests.push(p.1.clone());
|
||||
channel.complete(p, Response::FetchInterface(Ok(interface("scope-v1"))));
|
||||
let p = channel.call(&session);
|
||||
requests.push(p.1.clone());
|
||||
channel.complete(
|
||||
p,
|
||||
Response::CallOperation(Ok(CallOperationResponse {
|
||||
result: Value::Integer(42),
|
||||
validator: Some(vec![2]),
|
||||
})),
|
||||
);
|
||||
let p = channel.call(&session);
|
||||
requests.push(p.1.clone());
|
||||
channel.complete(
|
||||
p,
|
||||
Response::CallOperation(Err(ProtocolError {
|
||||
code: ProtocolErrorCode::InterfaceMismatch,
|
||||
message: "changed".into(),
|
||||
})),
|
||||
);
|
||||
let p = channel.observe(&session, "/scope/item");
|
||||
channel.complete(p, Response::Observe(Ok(observation("item", "item-v1"))));
|
||||
let p = channel.fetch(&session);
|
||||
channel.complete(p, Response::FetchInterface(Ok(interface("scope-v1"))));
|
||||
let old = channel.fetch(&session);
|
||||
let p = channel.observe(&session, "/scope");
|
||||
channel.complete(p, Response::Observe(Ok(observation("scope", "scope-v2"))));
|
||||
let old_id = old.0;
|
||||
assert_eq!(
|
||||
channel.complete(old, Response::FetchInterface(Ok(interface("scope-v1")))),
|
||||
Completion::StaleResponseRejected(old_id)
|
||||
);
|
||||
let p = channel.fetch(&session);
|
||||
channel.complete(p, Response::FetchInterface(Ok(interface("scope-v2"))));
|
||||
let p = channel.call(&session);
|
||||
requests.push(p.1.clone());
|
||||
channel.complete(
|
||||
p,
|
||||
Response::CallOperation(Err(ProtocolError {
|
||||
code: ProtocolErrorCode::OperationOutcomeUnknown,
|
||||
message: "unknown".into(),
|
||||
})),
|
||||
);
|
||||
assert_eq!(channel.core().in_flight_len(), 0);
|
||||
Snapshot {
|
||||
objects: channel.core().known_space(&session).unwrap(),
|
||||
interface: channel.core().interface(&session, &reference()).cloned(),
|
||||
outcomes: channel
|
||||
.core()
|
||||
.call_history(&session)
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|record| record.outcome.clone())
|
||||
.collect(),
|
||||
requests,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn direct_and_http_agree_on_requests_state_calls_scope_invalidation_and_uncertainty() {
|
||||
assert_eq!(scenario(false), scenario(true));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn http_endpoint_and_identical_direct_identity_do_not_share_cache() {
|
||||
let mut client = Client::new(ClientLimits::new(4, 16, 8, 8, 8).unwrap(), limits());
|
||||
let http = client
|
||||
.open_session("https://example.test/wip", SecurityContext::new("alice"))
|
||||
.unwrap();
|
||||
let direct = DirectClient::open_session(
|
||||
&mut client,
|
||||
"https://example.test/wip/",
|
||||
SecurityContext::new("alice"),
|
||||
)
|
||||
.unwrap();
|
||||
assert_ne!(http, direct);
|
||||
let request = DirectClient::refresh_object(&mut client, &direct, "/scope").unwrap();
|
||||
DirectClient::complete(
|
||||
&mut client,
|
||||
request.id,
|
||||
Response::Observe(Ok(observation("scope", "local"))),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(client.object(&http, "/scope").is_none());
|
||||
assert!(client.refresh_object(&direct, "/scope").is_err());
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
#![cfg(feature = "http")]
|
||||
//! Wire budgets are adapter-only and failed encoding must not acquire state.
|
||||
use std::collections::BTreeMap;
|
||||
use wip_client::{Client, ClientErrorKind, ClientLimits, ObservationState, SecurityContext};
|
||||
use wip_http::{Limits, encode_fetch_interface_response, encode_observe_response};
|
||||
use wip_protocol::{
|
||||
FetchInterfaceRequest, FetchInterfaceResponse, INTERFACE_FORMAT_V1, InterfaceDescriptor,
|
||||
InterfaceReference, Object, ObjectObservation, ObserveRequest, OperationDeclaration,
|
||||
ReturnDeclaration, TypeExpr,
|
||||
};
|
||||
|
||||
fn state_limits() -> ClientLimits {
|
||||
ClientLimits::new(2, 8, 8, 8, 8).unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn request_encoding_failure_leaves_retrieval_cache_and_inflight_unchanged() {
|
||||
let mut client = Client::new(state_limits(), Limits::new(1, 4096, 32).unwrap());
|
||||
let session = client
|
||||
.open_session("https://example.test/wip", SecurityContext::new("alice"))
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
client.refresh_object(&session, "/item").unwrap_err().kind,
|
||||
ClientErrorKind::InvalidRequest
|
||||
);
|
||||
let reference = InterfaceReference {
|
||||
scope: "/".into(),
|
||||
name: "test".into(),
|
||||
};
|
||||
assert_eq!(
|
||||
client
|
||||
.prepare_interface(&session, reference.clone())
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::InvalidRequest
|
||||
);
|
||||
assert!(client.object(&session, "/item").is_none());
|
||||
assert!(client.interface(&session, &reference).is_none());
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn call_encoding_failure_keeps_fresh_observations_and_no_pending_call() {
|
||||
let limits = Limits::new(80, 4096, 32).unwrap();
|
||||
let mut client = Client::new(state_limits(), limits);
|
||||
let session = client
|
||||
.open_session("https://example.test/wip", SecurityContext::new("alice"))
|
||||
.unwrap();
|
||||
let reference = InterfaceReference {
|
||||
scope: "/".into(),
|
||||
name: "test".into(),
|
||||
};
|
||||
let observe = client.refresh_object(&session, "/item").unwrap();
|
||||
client
|
||||
.complete(
|
||||
observe.id,
|
||||
encode_observe_response(
|
||||
&ObserveRequest {
|
||||
path: "/item".into(),
|
||||
depth: 0,
|
||||
},
|
||||
&ObjectObservation {
|
||||
object: Object {
|
||||
name: "item".into(),
|
||||
description: None,
|
||||
interfaces: vec![reference.clone()],
|
||||
r#ref: None,
|
||||
validator: None,
|
||||
},
|
||||
children: None,
|
||||
},
|
||||
limits,
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
let fetch = client
|
||||
.prepare_interface(&session, reference.clone())
|
||||
.unwrap();
|
||||
client
|
||||
.complete(
|
||||
fetch.id,
|
||||
encode_fetch_interface_response(
|
||||
&FetchInterfaceRequest {
|
||||
interface: reference.clone(),
|
||||
},
|
||||
&FetchInterfaceResponse {
|
||||
interface: reference.clone(),
|
||||
scope_ref: None,
|
||||
validator: None,
|
||||
descriptor: InterfaceDescriptor {
|
||||
format: INTERFACE_FORMAT_V1.into(),
|
||||
documentation: None,
|
||||
types: vec![],
|
||||
operations: vec![OperationDeclaration {
|
||||
name: "run".into(),
|
||||
documentation: None,
|
||||
parameters: vec![],
|
||||
returns: ReturnDeclaration {
|
||||
documentation: None,
|
||||
r#type: TypeExpr::Integer,
|
||||
},
|
||||
}],
|
||||
},
|
||||
},
|
||||
limits,
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
client
|
||||
.prepare_call(&session, "/item", &reference, "run", BTreeMap::new())
|
||||
.unwrap_err()
|
||||
.kind,
|
||||
ClientErrorKind::InvalidRequest
|
||||
);
|
||||
assert_eq!(
|
||||
client.object(&session, "/item").unwrap().state,
|
||||
ObservationState::Fresh
|
||||
);
|
||||
assert_eq!(
|
||||
client.interface(&session, &reference).unwrap().state,
|
||||
ObservationState::Fresh
|
||||
);
|
||||
assert!(client.call_history(&session).unwrap().is_empty());
|
||||
assert_eq!(client.in_flight_len(), 0);
|
||||
}
|
||||
Reference in New Issue
Block a user