330 lines
12 KiB
Rust
330 lines
12 KiB
Rust
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
|
use std::thread;
|
|
use std::time::{Duration, Instant};
|
|
|
|
use tempfile::tempdir;
|
|
use wip_fs_server::{ServerConfig, WipFsServer};
|
|
use wip_http::encode_protocol_error_response;
|
|
use wip_http::http::{Request, Response};
|
|
use wip_protocol::{ProtocolError, ProtocolErrorCode, ProtocolInteraction};
|
|
use wip_tui::app::{ConnectionState, Focus, HistoryKind, InputKey, Modal, ViewState};
|
|
use wip_tui::config::Config;
|
|
use wip_tui::runtime::{
|
|
Runtime, Transport, TransportError, TransportErrorKind, default_wire_limits,
|
|
};
|
|
|
|
struct ScriptedTransport {
|
|
server: Arc<WipFsServer>,
|
|
sequence: Arc<AtomicUsize>,
|
|
delayed_after_startup: bool,
|
|
unknown_calls: bool,
|
|
fail_call_prepare: bool,
|
|
fail_tree_prepare: bool,
|
|
panic_calls: bool,
|
|
root_tree_protocol_failures: Arc<AtomicUsize>,
|
|
}
|
|
|
|
impl ScriptedTransport {
|
|
fn new(server: Arc<WipFsServer>) -> Self {
|
|
Self {
|
|
server,
|
|
sequence: Arc::new(AtomicUsize::new(0)),
|
|
delayed_after_startup: false,
|
|
unknown_calls: false,
|
|
fail_call_prepare: false,
|
|
fail_tree_prepare: false,
|
|
panic_calls: false,
|
|
root_tree_protocol_failures: Arc::new(AtomicUsize::new(0)),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Transport for ScriptedTransport {
|
|
type Prepared = Request<Vec<u8>>;
|
|
|
|
fn prepare(&self, request: Request<Vec<u8>>) -> Result<Self::Prepared, TransportError> {
|
|
if self.fail_call_prepare && request.uri().path().ends_with("/v1/call_operation") {
|
|
return Err(TransportError {
|
|
kind: TransportErrorKind::Disconnect,
|
|
detail: "scripted local conversion failure".into(),
|
|
});
|
|
}
|
|
if self.fail_tree_prepare && request.uri().path().ends_with("/v1/observe") {
|
|
return Err(TransportError {
|
|
kind: TransportErrorKind::Disconnect,
|
|
detail: "scripted tree conversion failure".into(),
|
|
});
|
|
}
|
|
Ok(request)
|
|
}
|
|
|
|
fn execute(&self, request: Self::Prepared) -> Result<Response<Vec<u8>>, TransportError> {
|
|
let sequence = self.sequence.fetch_add(1, Ordering::AcqRel);
|
|
if request.uri().path().ends_with("/v1/observe")
|
|
&& self
|
|
.root_tree_protocol_failures
|
|
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| {
|
|
remaining.checked_sub(1)
|
|
})
|
|
.is_ok()
|
|
{
|
|
return Ok(encode_protocol_error_response(
|
|
ProtocolInteraction::Observe,
|
|
&ProtocolError {
|
|
code: ProtocolErrorCode::Internal,
|
|
message: "scripted startup failure".into(),
|
|
},
|
|
default_wire_limits(),
|
|
)
|
|
.expect("encode tree failure"));
|
|
}
|
|
if self.delayed_after_startup && sequence > 0 {
|
|
// Delay the first refresh so the second wins and exercises exact
|
|
// RequestId ownership.
|
|
let delay = if sequence == 1 { 80 } else { 5 };
|
|
thread::sleep(Duration::from_millis(delay));
|
|
}
|
|
if self.panic_calls && request.uri().path().ends_with("/v1/call_operation") {
|
|
panic!("scripted transport worker panic");
|
|
}
|
|
if self.unknown_calls && request.uri().path().ends_with("/v1/call_operation") {
|
|
return Ok(encode_protocol_error_response(
|
|
ProtocolInteraction::CallOperation,
|
|
&ProtocolError {
|
|
code: ProtocolErrorCode::OperationOutcomeUnknown,
|
|
message: "scripted commit uncertainty".into(),
|
|
},
|
|
default_wire_limits(),
|
|
)
|
|
.expect("encode unknown outcome"));
|
|
}
|
|
Ok(self.server.handle_http(request))
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn out_of_order_refresh_and_cancel_keep_request_identity_safe() {
|
|
let temporary = tempdir().expect("temporary root");
|
|
std::fs::write(temporary.path().join("file"), b"contents").expect("file");
|
|
let server = Arc::new(
|
|
WipFsServer::new(ServerConfig::for_root(temporary.path())).expect("filesystem server"),
|
|
);
|
|
let mut transport = ScriptedTransport::new(server);
|
|
transport.delayed_after_startup = true;
|
|
let config = config();
|
|
let mut runtime = Runtime::new(&config, transport).expect("runtime");
|
|
runtime.connect();
|
|
settle(&mut runtime);
|
|
|
|
runtime.app_mut().focus = Focus::Tree;
|
|
runtime.handle_key(InputKey::Refresh);
|
|
runtime.handle_key(InputKey::Refresh);
|
|
settle(&mut runtime);
|
|
assert_eq!(runtime.app().entries["/"].state, ViewState::Fresh);
|
|
assert!(runtime.app().history.iter().any(|item| {
|
|
item.kind == HistoryKind::Information && item.title.contains("stale response")
|
|
}));
|
|
|
|
runtime.handle_key(InputKey::Refresh);
|
|
runtime.handle_key(InputKey::CancelRead);
|
|
runtime.handle_key(InputKey::CancelRead);
|
|
settle(&mut runtime);
|
|
assert!(runtime.app().in_flight.is_empty());
|
|
assert!(
|
|
runtime
|
|
.app()
|
|
.history
|
|
.iter()
|
|
.any(|item| item.title == "retrieval canceled")
|
|
);
|
|
|
|
// When a refreshed tree omits a formerly selected Entry, the client marks
|
|
// it stale and the presentation no longer exposes old metadata/operations
|
|
// as current.
|
|
std::fs::remove_file(temporary.path().join("file")).expect("remove file");
|
|
runtime.app_mut().selected_path = "/".into();
|
|
runtime.handle_key(InputKey::Refresh);
|
|
settle(&mut runtime);
|
|
assert_eq!(runtime.app().entries["/file"].state, ViewState::Stale);
|
|
runtime.app_mut().selected_path = "/file".into();
|
|
assert!(runtime.app().operation_rows().is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn startup_transport_preparation_failure_enters_failed_state() {
|
|
let temporary = tempdir().expect("temporary root");
|
|
let server = Arc::new(
|
|
WipFsServer::new(ServerConfig::for_root(temporary.path())).expect("filesystem server"),
|
|
);
|
|
let mut transport = ScriptedTransport::new(server);
|
|
transport.fail_tree_prepare = true;
|
|
let config = config();
|
|
let mut runtime = Runtime::new(&config, transport).expect("runtime");
|
|
|
|
runtime.connect();
|
|
settle(&mut runtime);
|
|
assert!(matches!(
|
|
runtime.app().connection,
|
|
ConnectionState::Failed(ref detail) if detail.contains("scripted tree conversion failure")
|
|
));
|
|
assert!(runtime.app().status.contains("press R to reconnect"));
|
|
assert!(runtime.app().in_flight.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn startup_protocol_failure_enters_failed_state_and_reconnects() {
|
|
let temporary = tempdir().expect("temporary root");
|
|
let server = Arc::new(
|
|
WipFsServer::new(ServerConfig::for_root(temporary.path())).expect("filesystem server"),
|
|
);
|
|
let transport = ScriptedTransport::new(server);
|
|
transport
|
|
.root_tree_protocol_failures
|
|
.store(1, Ordering::Release);
|
|
let config = config();
|
|
let mut runtime = Runtime::new(&config, transport).expect("runtime");
|
|
|
|
runtime.connect();
|
|
settle(&mut runtime);
|
|
assert!(matches!(
|
|
runtime.app().connection,
|
|
ConnectionState::Failed(ref detail) if detail.contains("scripted startup failure")
|
|
));
|
|
assert!(runtime.app().status.contains("press R to reconnect"));
|
|
assert!(runtime.app().history.iter().any(|item| {
|
|
item.kind == HistoryKind::DomainFailure && item.detail.contains("scripted startup failure")
|
|
}));
|
|
|
|
runtime.handle_key(InputKey::Reconnect);
|
|
settle(&mut runtime);
|
|
assert_eq!(runtime.app().connection, ConnectionState::Connected);
|
|
}
|
|
|
|
#[test]
|
|
fn local_transport_preparation_failure_is_definitely_not_dispatched() {
|
|
let temporary = tempdir().expect("temporary root");
|
|
std::fs::write(temporary.path().join("file"), b"contents").expect("file");
|
|
let server = Arc::new(
|
|
WipFsServer::new(ServerConfig::for_root(temporary.path())).expect("filesystem server"),
|
|
);
|
|
let mut transport = ScriptedTransport::new(server);
|
|
transport.fail_call_prepare = true;
|
|
let execute_calls = Arc::clone(&transport.sequence);
|
|
let config = config();
|
|
let mut runtime = Runtime::new(&config, transport).expect("runtime");
|
|
runtime.connect();
|
|
settle(&mut runtime);
|
|
|
|
open_read_editor(&mut runtime);
|
|
let calls_before_operation = execute_calls.load(Ordering::Acquire);
|
|
runtime.handle_key(InputKey::Enter);
|
|
runtime.handle_key(InputKey::Enter);
|
|
assert_eq!(
|
|
execute_calls.load(Ordering::Acquire),
|
|
calls_before_operation,
|
|
"preparation failure must not invoke transport I/O"
|
|
);
|
|
assert!(runtime.app().in_flight.is_empty());
|
|
assert_eq!(runtime.app().history[0].kind, HistoryKind::TransportFailure);
|
|
assert!(runtime.app().history[0].detail.contains("not dispatched"));
|
|
assert_eq!(
|
|
runtime.app().status,
|
|
"operation not dispatched; not executed"
|
|
);
|
|
assert_eq!(runtime.app().entries["/file"].state, ViewState::Fresh);
|
|
}
|
|
|
|
#[test]
|
|
fn worker_panic_settles_as_unknown_instead_of_stranding_the_main_loop() {
|
|
let temporary = tempdir().expect("temporary root");
|
|
std::fs::write(temporary.path().join("file"), b"contents").expect("file");
|
|
let server = Arc::new(
|
|
WipFsServer::new(ServerConfig::for_root(temporary.path())).expect("filesystem server"),
|
|
);
|
|
let mut transport = ScriptedTransport::new(server);
|
|
transport.panic_calls = true;
|
|
let config = config();
|
|
let mut runtime = Runtime::new(&config, transport).expect("runtime");
|
|
runtime.connect();
|
|
settle(&mut runtime);
|
|
|
|
open_read_editor(&mut runtime);
|
|
runtime.handle_key(InputKey::Enter);
|
|
runtime.handle_key(InputKey::Enter);
|
|
settle(&mut runtime);
|
|
assert!(runtime.app().in_flight.is_empty());
|
|
assert_eq!(runtime.app().history[0].kind, HistoryKind::OutcomeUnknown);
|
|
assert!(
|
|
runtime.app().history[0]
|
|
.detail
|
|
.contains("transport worker failed")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn post_dispatch_unknown_is_prominent_stale_and_never_retried() {
|
|
let temporary = tempdir().expect("temporary root");
|
|
std::fs::write(temporary.path().join("file"), b"contents").expect("file");
|
|
let server = Arc::new(
|
|
WipFsServer::new(ServerConfig::for_root(temporary.path())).expect("filesystem server"),
|
|
);
|
|
let mut transport = ScriptedTransport::new(server);
|
|
transport.unknown_calls = true;
|
|
let config = config();
|
|
let mut runtime = Runtime::new(&config, transport).expect("runtime");
|
|
runtime.connect();
|
|
settle(&mut runtime);
|
|
|
|
open_read_editor(&mut runtime);
|
|
runtime.handle_key(InputKey::Enter);
|
|
runtime.handle_key(InputKey::Enter);
|
|
settle(&mut runtime);
|
|
|
|
assert_eq!(runtime.app().history[0].kind, HistoryKind::OutcomeUnknown);
|
|
assert!(
|
|
runtime.app().history[0]
|
|
.detail
|
|
.contains("side effects are unknown")
|
|
);
|
|
assert_eq!(runtime.app().entries["/file"].state, ViewState::Stale);
|
|
assert!(runtime.app().in_flight.is_empty());
|
|
let history = runtime.app().history.len();
|
|
thread::sleep(Duration::from_millis(20));
|
|
runtime.poll();
|
|
assert_eq!(
|
|
runtime.app().history.len(),
|
|
history,
|
|
"no retry was manufactured"
|
|
);
|
|
}
|
|
|
|
fn config() -> Config {
|
|
Config::parse(
|
|
["https://example.test/wip"],
|
|
None,
|
|
Some("scripted-test".into()),
|
|
|_| None,
|
|
)
|
|
.expect("config")
|
|
}
|
|
|
|
fn open_read_editor(runtime: &mut Runtime<ScriptedTransport>) {
|
|
runtime.app_mut().selected_path = "/file".into();
|
|
runtime.app_mut().focus = Focus::Operations;
|
|
runtime.app_mut().operation_cursor = 0;
|
|
runtime.handle_key(InputKey::Enter);
|
|
settle(runtime);
|
|
runtime.app_mut().operation_cursor = 1;
|
|
runtime.handle_key(InputKey::Enter);
|
|
assert!(matches!(runtime.app().modal, Modal::Editor(_)));
|
|
}
|
|
|
|
fn settle(runtime: &mut Runtime<ScriptedTransport>) {
|
|
let deadline = Instant::now() + Duration::from_secs(5);
|
|
while !runtime.app().in_flight.is_empty() {
|
|
assert!(Instant::now() < deadline, "scripted runtime did not settle");
|
|
runtime.wait_and_poll(Duration::from_millis(10));
|
|
}
|
|
}
|