From a1c449843f7d66bbc0bbdb9154600ec65a57f751 Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 19:57:11 +0900 Subject: [PATCH] Use an in-memory FTP recovery transport --- Cargo.lock | 59 -------------- Cargo.toml | 2 +- src/session_tests.rs | 167 +++++++-------------------------------- src/test_host.rs | 181 +++++++++++++++++++++++++++++++++++-------- 4 files changed, 179 insertions(+), 230 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1ac10fb..f5c4452 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14,12 +14,6 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" -[[package]] -name = "bytes" -version = "1.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" - [[package]] name = "equivalent" version = "1.0.2" @@ -77,12 +71,6 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" -[[package]] -name = "libc" -version = "0.2.189" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" - [[package]] name = "log" version = "0.4.34" @@ -121,17 +109,6 @@ dependencies = [ "wit-bindgen", ] -[[package]] -name = "mio" -version = "1.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" -dependencies = [ - "libc", - "wasi", - "windows-sys", -] - [[package]] name = "pin-project-lite" version = "0.2.17" @@ -214,16 +191,6 @@ dependencies = [ "zmij", ] -[[package]] -name = "socket2" -version = "0.6.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" -dependencies = [ - "libc", - "windows-sys", -] - [[package]] name = "syn" version = "3.0.6" @@ -241,13 +208,8 @@ version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ - "bytes", - "libc", - "mio", "pin-project-lite", - "socket2", "tokio-macros", - "windows-sys", ] [[package]] @@ -267,12 +229,6 @@ version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" -[[package]] -name = "wasi" -version = "0.11.1+wasi-snapshot-preview1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" - [[package]] name = "wasm-encoder" version = "0.259.0" @@ -307,21 +263,6 @@ dependencies = [ "semver", ] -[[package]] -name = "windows-link" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" - -[[package]] -name = "windows-sys" -version = "0.61.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" -dependencies = [ - "windows-link", -] - [[package]] name = "wit-bindgen" version = "0.62.0" diff --git a/Cargo.toml b/Cargo.toml index f665f24..77a50dc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,7 @@ crate-type = ["cdylib", "rlib"] memona-plugin-sdk = { path = "sdk" } [dev-dependencies] -tokio = { version = "1", features = ["macros", "rt", "net", "io-util", "sync", "time"] } +tokio = { version = "1", features = ["macros", "rt"] } [lints] workspace = true diff --git a/src/session_tests.rs b/src/session_tests.rs index ddcb5e0..bef5fb2 100644 --- a/src/session_tests.rs +++ b/src/session_tests.rs @@ -1,25 +1,22 @@ -//! Real FTP commands against an async local server. The generated guest API is -//! synchronous, so each guest call is isolated with `spawn_blocking`; production -//! WASM host integration is exercised separately. +//! FTP command recovery against an in-memory protocol transport. Live QA +//! separately exercises the compiled WASM component over real host sockets. -use super::Ftp; +use super::{Ftp, test_host}; use memona_plugin_sdk::{ErrorCode, Guest, Locator, Setting}; -use std::time::Duration; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::oneshot; type TestResult = Result>; -type Channel = BufReader; fn guest(value: Result) -> TestResult { value.map_err(|error| format!("guest operation failed: {error:?}").into()) } -fn config(port: u16) -> Vec<(String, Setting)> { +fn config() -> Vec<(String, Setting)> { vec![ ("host".into(), Setting::Text("127.0.0.1".into())), - ("port".into(), Setting::Text(port.to_string())), + ( + "port".into(), + Setting::Text(test_host::CONTROL_PORT.to_string()), + ), ("username".into(), Setting::Text("fixture".into())), ( "password".into(), @@ -28,140 +25,36 @@ fn config(port: u16) -> Vec<(String, Setting)> { ] } -async fn reply(channel: &mut Channel, text: &str) -> TestResult { - channel.get_mut().write_all(text.as_bytes()).await?; - Ok(()) -} - -async fn expect(channel: &mut Channel, expected: &str) -> TestResult { - loop { - let mut line = String::new(); - channel.read_line(&mut line).await?; - if line == "NOOP\r\n" { - reply(channel, "200 Still connected\r\n").await?; - continue; - } - assert_eq!(line, format!("{expected}\r\n")); - return Ok(()); - } -} - -async fn login(listener: &TcpListener) -> TestResult { - let (stream, _) = listener.accept().await?; - let mut channel = BufReader::new(stream); - reply(&mut channel, "220 Fixture\r\n").await?; - expect(&mut channel, "USER fixture").await?; - reply(&mut channel, "331 Password required\r\n").await?; - expect(&mut channel, "PASS fixture secret ").await?; - reply(&mut channel, "230 Signed in\r\n").await?; - expect(&mut channel, "TYPE I").await?; - reply(&mut channel, "200 Binary\r\n").await?; - Ok(channel) -} - -async fn listing(channel: &mut Channel) -> TestResult { - expect(channel, "EPSV").await?; - let listener = TcpListener::bind("127.0.0.1:0").await?; - reply( - channel, - &format!("229 Passive (|||{}|)\r\n", listener.local_addr()?.port()), - ) - .await?; - expect(channel, "MLSD /").await?; - reply(channel, "150 Listing\r\n").await?; - let (mut data, _) = listener.accept().await?; - data.write_all(b"type=file;size=3; report.md \r\n").await?; - data.shutdown().await?; - drop(data); - reply(channel, "226 Complete\r\n").await -} - #[tokio::test] async fn a_listing_reconnects_after_the_server_expires_the_control_session() -> TestResult { - let listener = TcpListener::bind("127.0.0.1:0").await?; - let port = listener.local_addr()?.port(); - let (expired, wait_expired) = oneshot::channel(); - let server = tokio::spawn(async move { - tokio::time::timeout(Duration::from_secs(5), async { - let mut channel = login(&listener).await?; - listing(&mut channel).await?; - reply(&mut channel, "421 Control connection timed out\r\n").await?; - channel.get_mut().shutdown().await?; - drop(channel); - expired.send(()).map_err(|_| "guest disappeared")?; - let mut channel = login(&listener).await?; - listing(&mut channel).await?; - expect(&mut channel, "QUIT").await?; - reply(&mut channel, "221 Bye\r\n").await - }) - .await? - }); - // `Guest` is generated as a synchronous trait, so the repository rule for - // unavoidable blocking I/O requires isolating this native test call here. - let client = tokio::task::spawn_blocking(move || -> TestResult { - guest(Ftp::open("ftp".into(), config(port)))?; - let first = guest(Ftp::list(Locator::Path("/".into()), None, 10))?; - assert_eq!(first.entries[0].name, " report.md "); - wait_expired.blocking_recv()?; - let next = guest(Ftp::list(Locator::Path("/".into()), None, 10))?; - assert_eq!(next.entries[0].name, " report.md "); - Ftp::close(); - Ok(()) - }) - .await?; - if client.is_err() { - server.abort(); - } - client?; - server.await??; + let fixture = test_host::install(test_host::Scenario::ExpiredListing); + guest(Ftp::open("ftp".into(), config()))?; + let first = guest(Ftp::list(Locator::Path("/".into()), None, 10))?; + assert_eq!(first.entries[0].name, " report.md "); + let next = guest(Ftp::list(Locator::Path("/".into()), None, 10))?; + assert_eq!(next.entries[0].name, " report.md "); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 2); + assert_eq!(summary.listings, 2); + assert_eq!(summary.mkdir_commands, 0); Ok(()) } #[tokio::test] async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply() -> TestResult { - let listener = TcpListener::bind("127.0.0.1:0").await?; - let port = listener.local_addr()?.port(); - let (expired, wait_expired) = oneshot::channel(); - let server = tokio::spawn(async move { - tokio::time::timeout(Duration::from_secs(5), async { - let mut channel = login(&listener).await?; - channel.get_mut().shutdown().await?; - drop(channel); - expired.send(()).map_err(|_| "guest disappeared")?; - let mut channel = login(&listener).await?; - expect(&mut channel, "MKD /new").await?; - // The mutation may have succeeded; lose its reply deliberately. - channel.get_mut().shutdown().await?; - drop(channel); - Ok::<_, Box>(listener) - }) - .await? - }); - // `Guest` is generated as a synchronous trait, so the repository rule for - // unavoidable blocking I/O requires isolating this native test call here. - let client = tokio::task::spawn_blocking(move || -> TestResult { - guest(Ftp::open("ftp".into(), config(port)))?; - wait_expired.blocking_recv()?; - assert!(matches!( - Ftp::mkdir(Locator::Path("/new".into())), - Err(ErrorCode::Unavailable) - )); - Ftp::close(); - Ok(()) - }) - .await?; - if client.is_err() { - server.abort(); - } - client?; - let listener = server.await??; - assert!( - tokio::time::timeout(Duration::from_millis(50), listener.accept()) - .await - .is_err(), - "a mutation must not be replayed on a third control connection" - ); + let fixture = test_host::install(test_host::Scenario::LostMutationReply); + guest(Ftp::open("ftp".into(), config()))?; + assert!(matches!( + Ftp::mkdir(Locator::Path("/new".into())), + Err(ErrorCode::Unavailable) + )); + Ftp::close(); + let summary = fixture.summary(); + assert_eq!(summary.control_connections, 2); + assert_eq!(summary.listings, 0); + assert_eq!(summary.mkdir_commands, 1); Ok(()) } diff --git a/src/test_host.rs b/src/test_host.rs index 32f870b..9be6f4e 100644 --- a/src/test_host.rs +++ b/src/test_host.rs @@ -1,51 +1,166 @@ -//! Native test transport for the generated synchronous `Guest` API. It cannot -//! await a Tokio socket, so tests confine it to `spawn_blocking` while their FTP -//! server remains async. Production WASM always uses the host's async sockets. +//! In-memory FTP host for the generated synchronous `Guest` API. Production +//! WASM uses Memona's real async socket host; live QA covers that boundary. use std::cell::RefCell; -use std::io::{Read, Write}; -use std::net::{IpAddr, SocketAddr, TcpStream}; -use std::time::Duration; +use std::collections::VecDeque; +use std::rc::Rc; pub use memona_plugin_sdk::HostError; -pub struct Tcp(RefCell); +pub const CONTROL_PORT: u16 = 21_210; +const DATA_PORT: u16 = 31_310; +const LISTING: &[u8] = b"type=file;size=3; report.md \r\n"; + +#[derive(Clone, Copy)] +pub enum Scenario { + ExpiredListing, + LostMutationReply, +} + +struct State { + scenario: Scenario, + control_connections: usize, + listings: usize, + mkdir_commands: usize, +} + +thread_local! { + static ACTIVE: RefCell>>> = const { RefCell::new(None) }; +} + +pub struct Fixture(Rc>); + +pub struct Summary { + pub control_connections: usize, + pub listings: usize, + pub mkdir_commands: usize, +} + +pub fn install(scenario: Scenario) -> Fixture { + let state = Rc::new(RefCell::new(State { + scenario, + control_connections: 0, + listings: 0, + mkdir_commands: 0, + })); + ACTIVE.set(Some(state.clone())); + Fixture(state) +} + +impl Fixture { + pub fn summary(&self) -> Summary { + let state = self.0.borrow(); + Summary { + control_connections: state.control_connections, + listings: state.listings, + mkdir_commands: state.mkdir_commands, + } + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + ACTIVE.with_borrow_mut(|active| { + if active + .as_ref() + .is_some_and(|state| Rc::ptr_eq(state, &self.0)) + { + *active = None; + } + }); + } +} + +enum Kind { + Control(usize), + Data, +} + +pub struct Tcp { + state: Rc>, + kind: Kind, + input: RefCell>, +} pub fn connect(name: &str, port: u16, tls: bool) -> Result { - if tls { + if tls || name != "127.0.0.1" { return Err(HostError::InvalidInput); } - let ip: IpAddr = name.parse().map_err(|_| HostError::InvalidInput)?; - let timeout = Duration::from_secs(2); - let stream = TcpStream::connect_timeout(&SocketAddr::new(ip, port), timeout) - .map_err(|_| HostError::Unavailable)?; - stream - .set_read_timeout(Some(timeout)) - .map_err(|_| HostError::Unavailable)?; - stream - .set_write_timeout(Some(timeout)) - .map_err(|_| HostError::Unavailable)?; - Ok(Tcp(RefCell::new(stream))) + let state = ACTIVE + .with_borrow(Clone::clone) + .ok_or(HostError::Internal)?; + match port { + CONTROL_PORT => { + let index = { + let mut state = state.borrow_mut(); + let index = state.control_connections; + state.control_connections += 1; + index + }; + Ok(Tcp { + state, + kind: Kind::Control(index), + input: RefCell::new(VecDeque::from(b"220 Fixture\r\n".to_vec())), + }) + } + DATA_PORT => Ok(Tcp { + state, + kind: Kind::Data, + input: RefCell::new(VecDeque::from(LISTING.to_vec())), + }), + _ => Err(HostError::Denied), + } } impl Tcp { pub fn read(&self, limit: u32) -> Result, HostError> { - let mut bytes = vec![0; limit as usize]; - let count = self - .0 - .borrow_mut() - .read(&mut bytes) - .map_err(|_| HostError::Unavailable)?; - bytes.truncate(count); - Ok(bytes) + let mut input = self.input.borrow_mut(); + if input.is_empty() && matches!(self.kind, Kind::Control(_)) { + return Err(HostError::Unavailable); + } + let count = usize::min(limit as usize, input.len()); + Ok(input.drain(..count).collect()) } pub fn write(&self, bytes: &[u8]) -> Result { - self - .0 - .borrow_mut() - .write(bytes) - .map(|count| count as u32) - .map_err(|_| HostError::Unavailable) + let Kind::Control(index) = self.kind else { + return Err(HostError::InvalidInput); + }; + let command = std::str::from_utf8(bytes) + .map_err(|_| HostError::InvalidInput)? + .strip_suffix("\r\n") + .ok_or(HostError::InvalidInput)?; + let response = { + let mut state = self.state.borrow_mut(); + match command { + "USER fixture" => "331 Password required\r\n", + "PASS fixture secret " => "230 Signed in\r\n", + "TYPE I" => "200 Binary\r\n", + "QUIT" => "221 Bye\r\n", + "EPSV" => "229 Passive (|||31310|)\r\n", + "MLSD /" if matches!(state.scenario, Scenario::ExpiredListing) => { + state.listings += 1; + "150 Listing\r\n226 Complete\r\n" + } + "NOOP" + if index == 0 + && (matches!(state.scenario, Scenario::LostMutationReply) + || state.listings > 0) => + { + return Err(HostError::Unavailable); + } + "NOOP" => "200 Still connected\r\n", + "MKD /new" + if index == 1 + && matches!(state.scenario, Scenario::LostMutationReply) => + { + state.mkdir_commands += 1; + return Ok(bytes.len() as u32); + } + _ => return Err(HostError::InvalidInput), + } + }; + self.input.borrow_mut().extend(response.as_bytes()); + Ok(bytes.len() as u32) } }