diff --git a/src/channels/adapter.rs b/src/channels/adapter.rs index 0d18e5b..cf1d1fd 100644 --- a/src/channels/adapter.rs +++ b/src/channels/adapter.rs @@ -258,6 +258,8 @@ mod tests { use super::*; use crate::channels::manager::ChannelManager; use crate::channels::mux::MuxRunner; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; use tokio::io::AsyncWriteExt; #[test] @@ -265,6 +267,149 @@ mod tests { assert_eq!(CHANNELS_ALPN, b"alknet/channels"); } + #[test] + fn channels_adapter_alpn_returns_alknet_channels() { + let installed = Arc::new(AtomicBool::new(false)); + let install: InstallChannelZero = Arc::new(move |_m, _c, _a| { + installed.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + let policy = crate::channels::policy::default_policy(); + let adapter = ChannelsAdapter::new(install, policy); + assert_eq!(adapter.alpn(), b"alknet/channels"); + } + + #[test] + fn channels_adapter_with_limits_constructs() { + let installed = Arc::new(AtomicBool::new(false)); + let install: InstallChannelZero = Arc::new(move |_m, _c, _a| { + installed.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + let policy = crate::channels::policy::default_policy(); + let adapter = ChannelsAdapter::with_limits(install, 128, 32, policy); + assert_eq!(adapter.alpn(), b"alknet/channels"); + } + + #[tokio::test] + async fn handle_installs_channel_zero_and_runs_demux_loop() { + let (client, server) = tokio::io::duplex(64 * 1024); + let conn = Connection::from_bidi(server, b"alknet/channels".to_vec(), None); + + let installed = Arc::new(AtomicBool::new(false)); + let installed_clone = Arc::clone(&installed); + + let install_channel_zero: InstallChannelZero = Arc::new(move |_manager, _conn, _auth| { + installed_clone.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + + let policy = crate::channels::policy::default_policy(); + let adapter = ChannelsAdapter::new(install_channel_zero, policy); + + let auth = AuthContext::anonymous(b"alknet/channels"); + + let handle_task = tokio::spawn(async move { adapter.handle(conn, &auth).await }); + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + assert!( + installed.load(Ordering::SeqCst), + "install_channel_zero was called" + ); + + drop(client); + + let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle_task) + .await + .expect("handle task timed out") + .expect("handle task panicked"); + + assert!(result.is_ok(), "handle() should return Ok"); + } + + #[tokio::test] + async fn handle_demux_loop_processes_frames() { + let (client, server) = tokio::io::duplex(64 * 1024); + let conn = Connection::from_bidi(server, b"alknet/channels".to_vec(), None); + + let installed = Arc::new(AtomicBool::new(false)); + let installed_clone = Arc::clone(&installed); + + let install_channel_zero: InstallChannelZero = Arc::new(move |_manager, _conn, _auth| { + installed_clone.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + + let policy = crate::channels::policy::default_policy(); + let adapter = ChannelsAdapter::new(install_channel_zero, policy); + + let auth = AuthContext::anonymous(b"alknet/channels"); + + let handle_task = tokio::spawn(async move { adapter.handle(conn, &auth).await }); + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + assert!(installed.load(Ordering::SeqCst)); + + let mut header = [0u8; 8]; + super::super::wire::write_header(0, 5, &mut header).expect("write header"); + let mut client_write = client; + client_write.write_all(&header).await.expect("write header"); + client_write + .write_all(b"hello") + .await + .expect("write payload"); + drop(client_write); + + let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle_task) + .await + .expect("handle task timed out") + .expect("handle task panicked"); + + assert!( + result.is_ok(), + "handle() should return Ok after processing frames" + ); + } + + #[tokio::test] + async fn handle_connection_closed_on_accept_bi_returns_error() { + use crate::core::types::{BiStream, BidiStreamSource, StreamError}; + use async_trait::async_trait; + use std::net::SocketAddr; + + struct ClosedSource; + #[async_trait] + impl BidiStreamSource for ClosedSource { + async fn accept_bi(&self) -> Result { + Err(StreamError::ConnectionClosed) + } + async fn open_bi(&self) -> Result { + Err(StreamError::StreamClosed) + } + fn remote_addr(&self) -> Option { + None + } + fn close(&self, _code: u32, _reason: &str) {} + } + + let conn = Connection::from_source(ClosedSource, b"alknet/channels".to_vec()); + + let installed = Arc::new(AtomicBool::new(false)); + let install: InstallChannelZero = Arc::new(move |_m, _c, _a| { + installed.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + let policy = crate::channels::policy::default_policy(); + let adapter = ChannelsAdapter::new(install, policy); + let auth = AuthContext::anonymous(b"alknet/channels"); + + let result = adapter.handle(conn, &auth).await; + match result { + Err(HandlerError::ConnectionClosed) => {} + other => panic!("expected ConnectionClosed, got {other:?}"), + } + } + /// C-25 #2 — demux resync on `TooLarge`. Send an oversized chunk /// (length > MAX_CHUNK_LEN) followed by a valid chunk. The demux /// must skip the oversized payload bytes and correctly parse the diff --git a/src/channels/operations.rs b/src/channels/operations.rs index cfdc493..1c739db 100644 --- a/src/channels/operations.rs +++ b/src/channels/operations.rs @@ -701,11 +701,13 @@ fn make_open_handler_sink( mod tests { use super::*; use crate::channels::mux::MuxRunner; - use crate::core::auth::Identity; + use crate::core::auth::{AuthContext, Identity}; use crate::registry::context::{AbortPolicy, ScopedPeerEnv}; use crate::registry::env::OperationEnv; + use crate::registry::spec::ChannelOpenSpec; use futures::stream::StreamExt; use std::collections::HashMap; + use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use tokio::io::duplex; @@ -883,4 +885,403 @@ mod tests { Ok(_) => panic!("missing channel_id should error"), } } + + #[test] + fn channel_control_spec_is_mutation_external() { + let spec = channel_control_spec(); + assert_eq!(spec.name, OP_CHANNEL_CONTROL); + assert_eq!(spec.op_type, OperationType::Mutation); + assert_eq!(spec.visibility, Visibility::External); + } + + #[tokio::test] + async fn channel_operations_new_with_explicit_policy() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let ops = ChannelOperations::new(manager, policy); + let mut registry = OperationRegistry::new(); + ops.register_on(&mut registry).expect("register"); + assert!(registry.registration(OP_CHANNEL_CLOSE).is_some()); + } + + #[tokio::test] + async fn close_handler_success_path_closes_open_channel() { + let manager = make_manager().await; + let (id, _send, _recv) = manager + .open_channel("alknet/tty", "alice", None) + .await + .expect("open"); + assert!(manager.has_channel(id)); + let policy = super::super::policy::default_policy(); + let handler = make_close_handler(manager.clone(), policy); + let env = handler(json!({ "channel_id": id }), test_context("close-ok")).await; + match env.result { + Ok(v) => assert_eq!(v["closed"], true), + Err(e) => panic!("close should succeed, got: {e:?}"), + } + assert!(!manager.has_channel(id)); + } + + #[tokio::test] + async fn close_handler_missing_channel_id_returns_invalid_input() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let handler = make_close_handler(manager, policy); + let env = handler(json!({}), test_context("close-missing")).await; + match env.result { + Err(e) => assert_eq!(e.code, "INVALID_INPUT"), + Ok(_) => panic!("missing channel_id should error"), + } + } + + #[tokio::test] + async fn control_handler_success_path_returns_not_implemented() { + let manager = make_manager().await; + let (id, _send, _recv) = manager + .open_channel("alknet/tty", "alice", None) + .await + .expect("open"); + let handler = make_control_handler(manager); + let env = handler( + json!({ "channel_id": id, "message": {} }), + test_context("ctrl-ok"), + ) + .await; + match env.result { + Err(e) => assert_eq!(e.code, "channel:control_not_implemented"), + Ok(_) => panic!("control should return not-implemented"), + } + } + + #[test] + fn channel_core_new_and_accessors() { + let (_client, server) = duplex(1024); + let (_reader, writer) = tokio::io::split(server); + let (handle, _runner) = MuxRunner::new(Box::new(writer)); + let manager = ChannelManager::with_defaults(handle, None); + let policy = super::super::policy::default_policy(); + let core = ChannelCore::new(manager.clone(), policy.clone()); + assert!(!core.manager().has_channel(0)); + assert!(Arc::ptr_eq(core.policy(), &policy)); + } + + #[test] + fn channel_core_check_open_delegates_to_policy() { + let (_client, server) = duplex(1024); + let (_reader, writer) = tokio::io::split(server); + let (handle, _runner) = MuxRunner::new(Box::new(writer)); + let manager = ChannelManager::with_defaults(handle, None); + let policy = super::super::policy::default_policy(); + let core = ChannelCore::new(manager, policy); + let id = Identity { + id: "test-peer".to_string(), + scopes: vec![], + resources: HashMap::new(), + }; + assert!(core.check_open(&id).is_ok()); + } + + #[test] + fn channel_core_on_close_delegates_to_policy() { + let (_client, server) = duplex(1024); + let (_reader, writer) = tokio::io::split(server); + let (handle, _runner) = MuxRunner::new(Box::new(writer)); + let manager = ChannelManager::with_defaults(handle, None); + let policy = super::super::policy::default_policy(); + let core = ChannelCore::new(manager, policy); + let id = Identity { + id: "test-peer".to_string(), + scopes: vec![], + resources: HashMap::new(), + }; + core.on_close(&id); + } + + #[test] + fn opener_identity_from_context_returns_identity_when_present() { + let ctx = test_context("opener-1"); + let id = opener_identity_from_context(&ctx); + assert_eq!(id.id, "alice"); + } + + #[test] + fn opener_identity_from_context_falls_back_to_anonymous() { + let mut ctx = test_context("opener-2"); + ctx.identity = None; + let id = opener_identity_from_context(&ctx); + assert_eq!(id.id, "anonymous"); + } + + #[test] + fn map_channel_error_too_many_channels_to_call_error() { + let err = ChannelError::TooManyChannels { + identity: "alice".to_string(), + count: 256, + cap: 256, + }; + let call_err = map_channel_error_to_call_error(&err); + assert_eq!(call_err.code, "channel:too_many_channels"); + assert!(call_err.message.contains("alice")); + assert!(call_err.message.contains("256")); + } + + #[tokio::test] + async fn make_open_handler_once_success_path() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let spawned = Arc::new(AtomicBool::new(false)); + let spawned_clone = Arc::clone(&spawned); + let open_handler: OpenHandler = Arc::new(move |_input, _conn, _auth| { + spawned_clone.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + let auth = AuthContext::anonymous(b"alknet/call"); + let handler = make_open_handler_once( + manager.clone(), + policy, + open_handler, + auth, + "alknet/tty".to_string(), + ); + let env = handler(json!({}), test_context("open-once-1")).await; + match env.result { + Ok(v) => { + let channel_id = v["channel_id"].as_u64().expect("channel_id"); + assert!(channel_id > 0); + assert!(manager.has_channel(channel_id as u32)); + } + Err(e) => panic!("open should succeed, got: {e:?}"), + } + assert!(spawned.load(Ordering::SeqCst)); + } + + #[tokio::test] + async fn make_open_handler_stream_success_path() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let spawned = Arc::new(AtomicBool::new(false)); + let spawned_clone = Arc::clone(&spawned); + let open_handler: OpenHandler = Arc::new(move |_input, _conn, _auth| { + spawned_clone.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + let auth = AuthContext::anonymous(b"alknet/call"); + let handler = make_open_handler_stream( + manager.clone(), + policy, + open_handler, + auth, + "alknet/tty".to_string(), + ); + let mut stream = handler(json!({}), test_context("open-stream-1")); + let env = stream.next().await.expect("one envelope"); + match env.result { + Ok(v) => { + let channel_id = v["channel_id"].as_u64().expect("channel_id"); + assert!(channel_id > 0); + assert!(manager.has_channel(channel_id as u32)); + } + Err(e) => panic!("open should succeed, got: {e:?}"), + } + assert!(spawned.load(Ordering::SeqCst)); + assert!( + stream.next().await.is_none(), + "stream completes after one item" + ); + } + + #[tokio::test] + async fn make_open_handler_sink_returns_not_implemented() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let open_handler: OpenHandler = Arc::new(|_input, _conn, _auth| tokio::spawn(async {})); + let auth = AuthContext::anonymous(b"alknet/call"); + let handler = make_open_handler_sink( + manager, + policy, + open_handler, + auth, + "alknet/tty".to_string(), + ); + let publish_stream: crate::registry::registration::PublishStream = + Box::pin(futures::stream::empty()); + let env = handler(json!({}), test_context("open-sink-1"), publish_stream).await; + match env.result { + Err(e) => assert_eq!(e.code, "INTERNAL"), + Ok(_) => panic!("sink open should return not-implemented"), + } + } + + #[tokio::test] + async fn register_openable_query_registers_and_invokes() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let core = ChannelCore::new(manager.clone(), policy); + let spec = OperationSpec::new( + "channels/tty/sub", + OperationType::Query, + Visibility::External, + json!({}), + json!({ "type": "object", "properties": { "channel_id": { "type": "integer" } } }), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new("alknet/tty")); + let spawned = Arc::new(AtomicBool::new(false)); + let spawned_clone = Arc::clone(&spawned); + let open_handler: OpenHandler = Arc::new(move |_input, _conn, _auth| { + spawned_clone.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + let mut registry = OperationRegistry::new(); + core.register_openable( + spec, + open_handler, + &mut registry, + AuthContext::anonymous(b"alknet/call"), + ) + .expect("register"); + assert!(registry.registration("channels/tty/sub").is_some()); + let env = registry + .invoke("channels/tty/sub", json!({}), test_context("reg-open-1")) + .await; + match env.result { + Ok(v) => { + assert!(v["channel_id"].as_u64().is_some()); + } + Err(e) => panic!("invoke should succeed, got: {e:?}"), + } + assert!(spawned.load(Ordering::SeqCst)); + } + + #[tokio::test] + async fn register_openable_sub_registers_and_invokes() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let core = ChannelCore::new(manager.clone(), policy); + let spec = OperationSpec::new( + "channels/tty/sub", + OperationType::Sub, + Visibility::External, + json!({}), + json!({ "type": "object", "properties": { "channel_id": { "type": "integer" } } }), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new("alknet/tty")); + let spawned = Arc::new(AtomicBool::new(false)); + let spawned_clone = Arc::clone(&spawned); + let open_handler: OpenHandler = Arc::new(move |_input, _conn, _auth| { + spawned_clone.store(true, Ordering::SeqCst); + tokio::spawn(async {}) + }); + let mut registry = OperationRegistry::new(); + core.register_openable( + spec, + open_handler, + &mut registry, + AuthContext::anonymous(b"alknet/call"), + ) + .expect("register"); + let mut stream = + registry.invoke_streaming("channels/tty/sub", json!({}), test_context("reg-open-2")); + let env = stream.next().await.expect("one envelope"); + match env.result { + Ok(v) => { + assert!(v["channel_id"].as_u64().is_some()); + } + Err(e) => panic!("invoke_streaming should succeed, got: {e:?}"), + } + assert!(spawned.load(Ordering::SeqCst)); + } + + #[tokio::test] + async fn register_openable_pub_registers_stub() { + let manager = make_manager().await; + let policy = super::super::policy::default_policy(); + let core = ChannelCore::new(manager.clone(), policy); + let spec = OperationSpec::new( + "channels/tty/pub", + OperationType::Pub, + Visibility::External, + json!({}), + json!({ "type": "object", "properties": { "channel_id": { "type": "integer" } } }), + vec![], + AccessControl::default(), + None, + ) + .with_channel_open(ChannelOpenSpec::new("alknet/tty")); + let open_handler: OpenHandler = Arc::new(|_input, _conn, _auth| tokio::spawn(async {})); + let mut registry = OperationRegistry::new(); + core.register_openable( + spec, + open_handler, + &mut registry, + AuthContext::anonymous(b"alknet/call"), + ) + .expect("register"); + assert!(registry.registration("channels/tty/pub").is_some()); + } + + #[test] + fn register_openable_rejects_spec_without_channel_open_marker() { + let (_client, server) = duplex(1024); + let (_reader, writer) = tokio::io::split(server); + let (handle, _runner) = MuxRunner::new(Box::new(writer)); + let manager = ChannelManager::with_defaults(handle, None); + let policy = super::super::policy::default_policy(); + let core = ChannelCore::new(manager, policy); + let spec = OperationSpec::new( + "channels/tty/sub", + OperationType::Query, + Visibility::External, + json!({}), + json!({}), + vec![], + AccessControl::default(), + None, + ); + let open_handler: OpenHandler = Arc::new(|_input, _conn, _auth| tokio::spawn(async {})); + let mut registry = OperationRegistry::new(); + let result = core.register_openable( + spec, + open_handler, + &mut registry, + AuthContext::anonymous(b"alknet/call"), + ); + assert!(result.is_err()); + assert!(result.unwrap_err().contains("channel_open marker")); + } + + #[tokio::test] + async fn run_open_wrapper_denies_when_policy_check_open_fails() { + let manager = make_manager().await; + let policy: Arc = + Arc::new(super::super::policy::PerIdentityChannelPolicy::new(0)); + let open_handler: OpenHandler = Arc::new(|_input, _conn, _auth| tokio::spawn(async {})); + let auth = AuthContext::anonymous(b"alknet/call"); + let opener_id = Identity { + id: "alice".to_string(), + scopes: vec![], + resources: HashMap::new(), + }; + let env = run_open_wrapper( + &manager, + &policy, + &open_handler, + &auth, + "alknet/tty", + json!({}), + opener_id.id.clone(), + opener_id, + "req-cap-deny".to_string(), + ) + .await; + match env.result { + Err(e) => assert_eq!(e.code, "channel:too_many_channels"), + Ok(_) => panic!("cap 0 should deny open"), + } + } }