diff --git a/Cargo.toml b/Cargo.toml index 0c4066048..d4d02b6c1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -217,6 +217,30 @@ name = "bdd_pool" path = "tests/bdd_pool/mod.rs" required-features = ["advanced-tests", "pool"] +[[test]] +name = "connection" +path = "tests/connection/mod.rs" + +[[test]] +name = "formal_tooling" +path = "tests/formal_tooling/mod.rs" + +[[test]] +name = "fragment_transport" +path = "tests/fragment_transport/mod.rs" + +[[test]] +name = "preamble" +path = "tests/preamble/mod.rs" + +[[test]] +name = "prepared_app" +path = "tests/prepared_app/mod.rs" + +[[test]] +name = "response" +path = "tests/response/mod.rs" + [[bench]] name = "codec_performance" harness = false diff --git a/docs/adr-002-streaming-requests-and-shared-message-assembly.md b/docs/adr-002-streaming-requests-and-shared-message-assembly.md index 1bd6f7503..1c801ddc0 100644 --- a/docs/adr-002-streaming-requests-and-shared-message-assembly.md +++ b/docs/adr-002-streaming-requests-and-shared-message-assembly.md @@ -137,9 +137,9 @@ sequenceDiagram #### Implementation decisions (2026-02-12) -- The inbound integration seam for this ADR is `src/app/inbound_handler.rs` and - `src/app/frame_handling/`, not `src/connection/`, because inbound decode and - dispatch already occur in the app-layer connection loop. +- The inbound integration seam for this ADR is `src/app/inbound_handler/mod.rs` + and `src/app/frame_handling/`, not `src/connection/`, because inbound decode + and dispatch already occur in the app-layer connection loop. - Message-assembly parsing, length, and continuity violations are treated as inbound deserialization failures and therefore use the same threshold-based invalid-data policy as other malformed inbound input. @@ -319,8 +319,8 @@ Precedence is: #### Implementation decisions (2026-02-23) - Roadmap item `8.3.3` now implements soft-limit back-pressure in the inbound - read loop (`src/app/inbound_handler.rs`) by pausing briefly before polling - the next frame when buffered assembly bytes are under pressure. + read loop (`src/app/inbound_handler/mod.rs`) by pausing briefly before + polling the next frame when buffered assembly bytes are under pressure. - Soft pressure is computed from assembly-state buffered bytes versus the smaller aggregate cap (`min(bytes_per_connection, bytes_in_flight)`), with a threshold of 80% to engage pacing before hard-cap rejection. @@ -342,7 +342,7 @@ Precedence is: `evaluate_memory_pressure()` function in `src/app/frame_handling/backpressure.rs`, returning a `MemoryPressureAction` enum (`Continue`, `Pause`, or `Abort`). This keeps the inbound loop - (`src/app/inbound_handler.rs`) within the 400-line file limit. + (`src/app/inbound_handler/mod.rs`) within the 400-line file limit. - The hard cap uses `>` (strictly exceeds) for the comparison, matching the `check_aggregate_budgets` convention in `budget.rs`. The limit value itself is permitted. diff --git a/docs/adr-004-pluggable-protocol-codecs.md b/docs/adr-004-pluggable-protocol-codecs.md index 0eafeb7ec..8fefe072c 100644 --- a/docs/adr-004-pluggable-protocol-codecs.md +++ b/docs/adr-004-pluggable-protocol-codecs.md @@ -13,7 +13,7 @@ length-delimited implementation. Wireframe currently hardcodes `tokio_util::codec::LengthDelimitedCodec` with a 4-byte big-endian length prefix in the connection pipeline (notably in -`src/app/inbound_handler.rs` and `src/app/frame_handling.rs`). This makes +`src/app/inbound_handler/mod.rs` and `src/app/frame_handling/`). This makes framing inflexible and prevents protocols with alternative framing rules (such as Hotline, MySQL, and Redis Serialization Protocol (RESP)) from reusing Wireframe's routing, middleware, and serialization infrastructure. @@ -156,7 +156,7 @@ pub trait FrameCodec: Send + Sync + Clone + 'static { ### Phase 1: Trait definition (non-breaking) -- Create `src/codec.rs` with the `FrameCodec` trait and +- Create `src/codec/mod.rs` with the `FrameCodec` trait and `LengthDelimitedFrameCodec` default. - Add unit tests covering the default codec behaviour. @@ -301,7 +301,7 @@ let extracted = MyCodec::frame_payload_bytes(&frame); assert_eq!(frame.payload.as_ptr(), extracted.as_ptr()); ``` -See `src/codec/tests.rs` for zero-copy regression tests covering +See `src/codec/tests/mod.rs` for zero-copy regression tests covering `LengthDelimitedFrameCodec`, `HotlineFrameCodec`, and `MysqlFrameCodec`. ### Property-based codec round-trip hardening (resolved 2026-02-19) diff --git a/docs/adr-011-runtime-ownership-and-task-lifetime-boundaries.md b/docs/adr-011-runtime-ownership-and-task-lifetime-boundaries.md index 5492f8e6a..17391520e 100644 --- a/docs/adr-011-runtime-ownership-and-task-lifetime-boundaries.md +++ b/docs/adr-011-runtime-ownership-and-task-lifetime-boundaries.md @@ -85,7 +85,7 @@ and especially the following runtime surfaces: - `src/push/queues/handle.rs`; - `src/middleware.rs`; - `src/hooks.rs`; -- `src/server/runtime.rs` and `src/server/connection_spawner.rs`; +- `src/server/runtime/mod.rs` and `src/server/connection_spawner.rs`; - `src/client/pool/*`. It complements rather than supersedes: @@ -237,7 +237,7 @@ Examples governed by this rule include: `WireframeServer::run_with_shutdown` is a supervisor for independently scheduled accept-loop tasks. It owns their `CancellationToken` and -`TaskTracker`, and `src/server/runtime.rs` holds a named `drop_guard_ref()` +`TaskTracker`, and `src/server/runtime/mod.rs` holds a named `drop_guard_ref()` borrow in the supervisor frame. If the supervisor future is dropped, including through `JoinHandle::abort()`, the guard cancels the accept loops so they eventually release their listener references. The guarantee is eventual: the diff --git a/docs/adr-012-prepared-application-and-connection-runtime.md b/docs/adr-012-prepared-application-and-connection-runtime.md index c0d116952..266ab8cdd 100644 --- a/docs/adr-012-prepared-application-and-connection-runtime.md +++ b/docs/adr-012-prepared-application-and-connection-runtime.md @@ -73,9 +73,9 @@ Primary code surfaces: - `src/app/builder/routing.rs`; - `src/app/builder/lifecycle.rs`; - `src/app/builder/protocol.rs`; -- `src/app/inbound_handler.rs`; +- `src/app/inbound_handler/mod.rs`; - `src/server/mod.rs`; -- `src/server/runtime.rs`; +- `src/server/runtime/mod.rs`; - `src/server/runtime/accept.rs`; - `src/server/connection_spawner.rs`; - `examples/ping_pong.rs` and `examples/packet_enum.rs`. diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 8e462e911..95a83f65f 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -392,9 +392,9 @@ the rustdoc for `wireframe::NoProtocolError` for the public API contract. ## Message sequence validation architecture -Message continuation ordering lives in `src/message_assembler/series.rs`. The -public-facing control point is `validate_and_advance_sequence()`, which keeps -the series state machine readable by delegating to two private helpers: +Message continuation ordering lives in `src/message_assembler/series/mod.rs`. +The public-facing control point is `validate_and_advance_sequence()`, which +keeps the series state machine readable by delegating to two private helpers: - `start_sequence_tracking()` handles the first numbered continuation after an untracked start. It switches the series into tracked mode and then delegates @@ -1033,13 +1033,14 @@ evaluations and four middleware transforms for the former path, versus one of each for the prepared root. The counters are collected for each Criterion sample so the counts remain visible alongside the timing results. -Fragment transport integration tests import `tests/common/fragment_helpers.rs` -as a facade. Keep the public re-export surface stable there, and place helper -implementation details in responsibility-focused modules under -`tests/common/fragment_helpers/`: app construction and spawning, assertions, -fragmentation configuration, envelope building, error types, and framed -transport. New fragment helpers should be added to the smallest matching module -and re-exported only when more than one test binary needs the helper. +Fragment transport integration tests import +`tests/common/fragment_helpers/mod.rs` as a facade. Keep the public re-export +surface stable there, and place helper implementation details in +responsibility-focused modules under `tests/common/fragment_helpers/`: app +construction and spawning, assertions, fragmentation configuration, envelope +building, error types, and framed transport. New fragment helpers should be +added to the smallest matching module and re-exported only when more than one +test binary needs the helper. ## Formal verification tooling @@ -1192,6 +1193,16 @@ because a model that cannot finish hangs rather than failing. ## Test infrastructure and framework +### Integration-test target layout + +Integration-test roots may be relocated from `tests/.rs` to +`tests//mod.rs` so that supporting modules sit beside the root. Cargo +auto-discovers `tests/*.rs` and `tests//main.rs`, but not +`tests//mod.rs`, so a relocated root must be declared as an explicit +`[[test]]` target in `Cargo.toml` whose `path` names that `mod.rs`. Without the +declaration the root is not built, so its tests stop running with no compile +error to flag the loss. + ### rstest and rstest-bdd The test suite uses [`rstest`](https://crates.io/crates/rstest) for @@ -1295,7 +1306,7 @@ propagates both the `JoinError` and the inner error with `handle.await??`. interrupted before calling `finish_server`, so a server-side failure is never silently discarded on the happy path. See `ClientLifecycleWorld::finish_server` in `tests/fixtures/client_lifecycle.rs` and the precedent in -`tests/fixtures/client_preamble.rs`. +`tests/fixtures/client_preamble/mod.rs`. **The server-task cleanup contract**, as implemented by `run_hook_test_with_server` in `src/client/tests/request_hooks_support.rs`: on diff --git a/docs/execplans/8-5-3-deterministic-reassembly-outcome-assertion-helpers.md b/docs/execplans/8-5-3-deterministic-reassembly-outcome-assertion-helpers.md index fa421f9be..08f5234af 100644 --- a/docs/execplans/8-5-3-deterministic-reassembly-outcome-assertion-helpers.md +++ b/docs/execplans/8-5-3-deterministic-reassembly-outcome-assertion-helpers.md @@ -263,7 +263,7 @@ Testkit/public API touch points: - [`wireframe_testing/src/lib.rs`](../../wireframe_testing/src/lib.rs) is the public re-export surface. -- [`wireframe_testing/src/helpers.rs`](../../wireframe_testing/src/helpers.rs) +- [`wireframe_testing/src/helpers/mod.rs`](../../wireframe_testing/src/helpers/mod.rs) currently re-exports the in-memory driver helpers added in `8.5.1` and `8.5.2`. - [`docs/wireframe-testing-crate.md`](../wireframe-testing-crate.md) is the diff --git a/docs/formal-verification-methods-in-wireframe.md b/docs/formal-verification-methods-in-wireframe.md index 4368652e9..39456a604 100644 --- a/docs/formal-verification-methods-in-wireframe.md +++ b/docs/formal-verification-methods-in-wireframe.md @@ -1188,7 +1188,7 @@ organizational friction. [^22]: Wireframe [`src/message_assembler/header.rs`](../src/message_assembler/header.rs) [^23]: Wireframe - [`src/message_assembler/series.rs`](../src/message_assembler/series.rs) + [`src/message_assembler/series/mod.rs`](../src/message_assembler/series/mod.rs) [^24]: Wireframe [`src/message_assembler/state.rs`](../src/message_assembler/state.rs) [^25]: Wireframe [`src/fairness.rs`](../src/fairness.rs) diff --git a/docs/frame-vec-u8-inventory.md b/docs/frame-vec-u8-inventory.md index a8d84d3ac..4504ad209 100644 --- a/docs/frame-vec-u8-inventory.md +++ b/docs/frame-vec-u8-inventory.md @@ -118,11 +118,11 @@ PR #283 moved more of the codec path toward `Bytes`.[^issue-286] The middleware layer currently sits entirely inside an owned-`Vec` segment of the request/response pipeline: -1. `src/app/inbound_handler.rs` builds or reuses the middleware chain through - `WireframeApp::build_chains()`, wrapping each route handler inside +1. `src/app/inbound_handler/mod.rs` builds or reuses the middleware chain + through `WireframeApp::build_chains()`, wrapping each route handler inside `HandlerService`. -2. `src/app/frame_handling/decode.rs` and `src/app/inbound_handler.rs` decode - an inbound codec frame into an `Envelope`. +2. `src/app/frame_handling/decode.rs` and `src/app/inbound_handler/mod.rs` + decode an inbound codec frame into an `Envelope`. 3. `src/app/frame_handling/response.rs` crosses the middleware boundary by calling `ServiceRequest::new(env.payload, env.correlation_id)`. 4. `src/middleware.rs` exposes that payload to middleware as @@ -152,8 +152,9 @@ The practical consequence is that middleware currently forms a hard - `src/app/frame_handling/response.rs` is the request/response bridge into and out of middleware. Any type change has to preserve correlation handling and packet reconstruction here. -- `src/app/inbound_handler.rs` owns middleware-chain construction and therefore - any staging or compatibility story for mixed old/new middleware types. +- `src/app/inbound_handler/mod.rs` owns middleware-chain construction and + therefore any staging or compatibility story for mixed old/new middleware + types. - `tests/middleware.rs` and `tests/middleware_order.rs` demonstrate the current mutation semantics directly through `push`, `clear`, and `extend_from_slice` on `frame_mut()`. @@ -257,7 +258,7 @@ themselves `Frame = Vec` bindings: `F: FrameLike + CorrelatableFrame + Packet` for the `ConnectionActor`. - `src/app/builder/protocol.rs` stores protocols as `WireframeProtocol`. -- `src/codec.rs` defines the default `LengthDelimitedFrameCodec` with +- `src/codec/mod.rs` defines the default `LengthDelimitedFrameCodec` with `Bytes` frames, not `Vec`. Taken together, these show that the main frame-level tension is between generic @@ -405,7 +406,7 @@ ADRs: [comment](https://github.com/leynos/wireframe/pull/283#issuecomment-3167997856). [^codec]: See [Pluggable protocol codecs](execplans/pluggable-protocol-codecs.md) - and `src/codec.rs`. + and `src/codec/mod.rs`. [^hooks]: See `src/hooks.rs` and the hook overview in [users-guide.md](users-guide.md). [^unified-path]: See diff --git a/docs/multi-packet-and-streaming-responses-design.md b/docs/multi-packet-and-streaming-responses-design.md index 49bbcd036..ea866d4ba 100644 --- a/docs/multi-packet-and-streaming-responses-design.md +++ b/docs/multi-packet-and-streaming-responses-design.md @@ -515,15 +515,15 @@ staying true to the architecture already in production use. mirrors the message attribution strategy outlined in the capability roadmap. - **Correlation provenance today.** The current implementation expects correlation to live inside the serialized `Envelope` payload. See the decode - path in `src/app/inbound_handler.rs` and the payload-carried correlation + path in `src/app/inbound_handler/mod.rs` and the payload-carried correlation fields in `src/app/envelope.rs`. - **Header correlation (if that is not the intent).** The boundary for injecting header correlation is the frame-to-`Envelope` step in - `WireframeApp::decode_envelope` (`src/app/inbound_handler.rs`). That is where - the decoded frame and the deserialized `Envelope` are both available, so the - runtime can merge header correlation into `Envelope::correlation_id` (for - example, “if payload correlation is `None`, inherit from the header; if both - exist and differ, log and pick a precedence rule”). On the outbound path, + `WireframeApp::decode_envelope` (`src/app/inbound_handler/mod.rs`). That is + where the decoded frame and the deserialized `Envelope` are both available, + so the runtime can merge header correlation into `Envelope::correlation_id` + (for example, “if payload correlation is `None`, inherit from the header; if + both exist and differ, log and pick a precedence rule”). On the outbound path, `FrameCodec::wrap_payload` receives the codec instance and raw `Bytes`, but it still cannot see correlation unless it is embedded in the payload. Supporting header correlation therefore requires a codec API that can wrap diff --git a/docs/repository-layout.md b/docs/repository-layout.md index 399549b5f..7e9fa697c 100644 --- a/docs/repository-layout.md +++ b/docs/repository-layout.md @@ -6,7 +6,11 @@ and shared test support. Its main areas are: - `src/` contains the published `wireframe` library. - `crates/wireframe-verification/` contains verification support. - `wireframe_testing/` contains shared test fixtures and helpers. -- `tests/` contains integration and behaviour-driven tests. +- `tests/` contains integration and behaviour-driven tests. An + integration-test root laid out as `tests//mod.rs` needs an explicit + `[[test]]` entry in `Cargo.toml`, because Cargo auto-discovers `tests/*.rs` + and `tests//main.rs` but not `tests//mod.rs`. Without the entry + the root is silently not built or run. - `tools/dev-fast/config.toml` is the explicitly selected development Cargo configuration. The [development builds](developers-guide.md#development-builds) section diff --git a/docs/roadmap.md b/docs/roadmap.md index 54a619250..1fd42adc7 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -396,7 +396,8 @@ integration boundaries. [outbound messaging design](asynchronous-outbound-messaging-design.md)). - [x] Route app-level request and response handling through the `FramePipeline` so fragmentation and metrics apply consistently. - - [x] Remove duplicate codec construction in `src/app/inbound_handler.rs`; the + - [x] Remove duplicate codec construction in + `src/app/inbound_handler/mod.rs`; the `FramePipeline` owns outbound fragmentation. - [x] Add integration tests covering the unified pipeline (round-trip, fragmentation, sequential requests, disabled fragmentation). diff --git a/docs/users-guide.md b/docs/users-guide.md index 8084b4fc5..2b61677c1 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -2655,8 +2655,8 @@ directories (one per shard on full dispatch runs) for offline triage. [^3]: Implemented in `src/app/builder.rs` (lines 100-121, 347-360). [^4]: Implemented in `src/app/builder.rs` (lines 326-344). [^5]: Implemented in `src/error.rs` (lines 1-68). -[^6]: Implemented in `src/app/inbound_handler.rs` (lines 41-205). -[^7]: Implemented in `src/app/inbound_handler.rs` (lines 207-289). +[^6]: Implemented in `src/app/inbound_handler/mod.rs` (lines 41-205). +[^7]: Implemented in `src/app/inbound_handler/mod.rs` (lines 207-289). [^8]: Implemented in `src/app/envelope.rs` (lines 11-172). [^9]: Implemented in `src/frame/metadata.rs` (lines 1-40). [^10]: Implemented in `src/message.rs` (lines 4-41). @@ -2665,10 +2665,10 @@ directories (one per shard on full dispatch runs) for offline triage. [^13]: Implemented in `src/extractor.rs` (lines 1-236). [^14]: Implemented in `src/hooks.rs` (lines 18-175). [^15]: Implemented in `src/server/config/mod.rs` (lines 47-152). -[^16]: Implemented in `src/server/runtime.rs` (lines 46-83). +[^16]: Implemented in `src/server/runtime/mod.rs` (lines 46-83). [^17]: Implemented in `src/server/config/binding.rs` (lines 68-214). -[^18]: Implemented in `src/server/runtime.rs` (lines 90-233). -[^19]: Implemented in `src/server/runtime.rs` (lines 240-333). +[^18]: Implemented in `src/server/runtime/mod.rs` (lines 90-233). +[^19]: Implemented in `src/server/runtime/mod.rs` (lines 240-333). [^20]: Implemented in `src/server/config/preamble.rs` (lines 14-135) and `src/server/connection_spawner.rs` (lines 1-222). [^21]: Implemented in `src/server/error.rs` (lines 7-18). diff --git a/src/app/frame_handling/assembly_tests.rs b/src/app/frame_handling/assembly_tests/mod.rs similarity index 100% rename from src/app/frame_handling/assembly_tests.rs rename to src/app/frame_handling/assembly_tests/mod.rs diff --git a/src/app/frame_handling/backpressure_tests.rs b/src/app/frame_handling/backpressure_tests/mod.rs similarity index 100% rename from src/app/frame_handling/backpressure_tests.rs rename to src/app/frame_handling/backpressure_tests/mod.rs diff --git a/src/app/inbound_handler.rs b/src/app/inbound_handler/mod.rs similarity index 100% rename from src/app/inbound_handler.rs rename to src/app/inbound_handler/mod.rs diff --git a/src/codec.rs b/src/codec/mod.rs similarity index 100% rename from src/codec.rs rename to src/codec/mod.rs diff --git a/src/codec/tests.rs b/src/codec/tests/mod.rs similarity index 100% rename from src/codec/tests.rs rename to src/codec/tests/mod.rs diff --git a/src/fragment/tests.rs b/src/fragment/tests/mod.rs similarity index 100% rename from src/fragment/tests.rs rename to src/fragment/tests/mod.rs diff --git a/src/message_assembler/series.rs b/src/message_assembler/series/mod.rs similarity index 100% rename from src/message_assembler/series.rs rename to src/message_assembler/series/mod.rs diff --git a/src/server/runtime.rs b/src/server/runtime/mod.rs similarity index 100% rename from src/server/runtime.rs rename to src/server/runtime/mod.rs diff --git a/src/test_helpers.rs b/src/test_helpers/mod.rs similarity index 100% rename from src/test_helpers.rs rename to src/test_helpers/mod.rs diff --git a/tests/common/fragment_helpers.rs b/tests/common/fragment_helpers/mod.rs similarity index 70% rename from tests/common/fragment_helpers.rs rename to tests/common/fragment_helpers/mod.rs index d1730ea62..f82fedec2 100644 --- a/tests/common/fragment_helpers.rs +++ b/tests/common/fragment_helpers/mod.rs @@ -4,17 +4,17 @@ //! fragment envelope construction, framed transport, test app construction, //! and assertions. -#[path = "fragment_helpers/app.rs"] +#[path = "app.rs"] mod app; -#[path = "fragment_helpers/assertions.rs"] +#[path = "assertions.rs"] mod assertions; -#[path = "fragment_helpers/config.rs"] +#[path = "config.rs"] mod config; -#[path = "fragment_helpers/envelopes.rs"] +#[path = "envelopes.rs"] mod envelopes; -#[path = "fragment_helpers/errors.rs"] +#[path = "errors.rs"] mod errors; -#[path = "fragment_helpers/transport.rs"] +#[path = "transport.rs"] mod transport; pub use app::{make_app, make_handler, spawn_app}; @@ -22,7 +22,13 @@ pub use assertions::assert_handler_observed; pub use config::{fragmentation_config, fragmentation_config_with_timeout}; pub use envelopes::{build_envelopes, fragment_envelope}; pub use errors::{TestError, TestResult}; -pub use transport::{read_reassembled_response, read_response_payload, send_envelopes}; +pub use transport::{ + FramedResponse, + read_framed_response, + read_reassembled_response, + read_response_payload, + send_envelopes, +}; /// Default route ID used in fragmentation tests. pub const ROUTE_ID: u32 = 42; diff --git a/tests/common/fragment_helpers/transport.rs b/tests/common/fragment_helpers/transport.rs index 7c5a6e3a1..e2884a2ac 100644 --- a/tests/common/fragment_helpers/transport.rs +++ b/tests/common/fragment_helpers/transport.rs @@ -28,30 +28,55 @@ pub async fn send_envelopes( Ok(()) } -/// Read and reassemble a fragmented response from a client connection. +/// A reassembled response payload plus the number of frames that carried it. +/// +/// The frame count is kept separately so a test can distinguish a genuinely +/// unfragmented response from a fragmented one, which +/// [`read_reassembled_response`] alone cannot show. +pub struct FramedResponse { + /// Reassembled response payload. + pub payload: Vec, + /// Number of transport frames observed for this response. + pub frame_count: usize, +} + +/// Read a response, preserving the frame boundaries that carried it. +/// +/// Returns once the payload reassembles, together with the number of frames +/// consumed. A single frame means the response was not fragmented. /// /// # Errors /// /// Returns an error if reading, decoding the envelope, or reassembly fails, /// or if the stream ends before reassembly completes. -pub async fn read_reassembled_response( +pub async fn read_framed_response( client: &mut Framed, cfg: &FragmentationConfig, -) -> TestResult> { +) -> TestResult { let serializer = BincodeSerializer; let mut reassembler = Reassembler::new(cfg.max_message_size, cfg.reassembly_timeout); + let mut frame_count = 0usize; while let Some(frame) = client.next().await { let bytes = frame?; + frame_count += 1; let (env, _) = serializer.deserialize::(&bytes)?; let payload = env.into_parts().into_payload(); match decode_fragment_payload(&payload)? { Some((header, fragment)) => { if let Some(message) = reassembler.push(header, fragment)? { - return Ok(message.into_payload()); + return Ok(FramedResponse { + payload: message.into_payload(), + frame_count, + }); } } - None => return Ok(payload), + None => { + return Ok(FramedResponse { + payload, + frame_count, + }); + } } } @@ -60,7 +85,20 @@ pub async fn read_reassembled_response( )) } -/// Read and return the response payload with a 1-second timeout. +/// Read and reassemble a fragmented response from a client connection. +/// +/// # Errors +/// +/// Returns an error if reading, decoding the envelope, or reassembly fails, +/// or if the stream ends before reassembly completes. +pub async fn read_reassembled_response( + client: &mut Framed, + cfg: &FragmentationConfig, +) -> TestResult> { + Ok(read_framed_response(client, cfg).await?.payload) +} + +/// Read the response with a 1-second timeout, preserving frame boundaries. /// /// # Errors /// @@ -68,10 +106,10 @@ pub async fn read_reassembled_response( pub async fn read_response_payload( client: &mut Framed, config: &FragmentationConfig, -) -> TestResult> { +) -> TestResult { let response = timeout( TokioDuration::from_secs(1), - read_reassembled_response(client, config), + read_framed_response(client, config), ) .await??; Ok(response) diff --git a/tests/connection.rs b/tests/connection/mod.rs similarity index 78% rename from tests/connection.rs rename to tests/connection/mod.rs index 996bcf32e..e2430de72 100644 --- a/tests/connection.rs +++ b/tests/connection/mod.rs @@ -7,11 +7,11 @@ //! - Queue polling and lifecycle state (`state.rs`) #![cfg(not(loom))] -#[path = "connection/logs.rs"] +#[path = "logs.rs"] mod logs; -#[path = "connection/multi_packet.rs"] +#[path = "multi_packet.rs"] mod multi_packet; -#[path = "connection/state.rs"] +#[path = "state.rs"] mod state; -#[path = "connection/support.rs"] +#[path = "support.rs"] mod support; diff --git a/tests/fixtures/client_preamble.rs b/tests/fixtures/client_preamble/mod.rs similarity index 100% rename from tests/fixtures/client_preamble.rs rename to tests/fixtures/client_preamble/mod.rs diff --git a/tests/fixtures/client_streaming.rs b/tests/fixtures/client_streaming/mod.rs similarity index 100% rename from tests/fixtures/client_streaming.rs rename to tests/fixtures/client_streaming/mod.rs diff --git a/tests/fixtures/derived_memory_budgets.rs b/tests/fixtures/derived_memory_budgets/mod.rs similarity index 100% rename from tests/fixtures/derived_memory_budgets.rs rename to tests/fixtures/derived_memory_budgets/mod.rs diff --git a/tests/fixtures/message_assembler.rs b/tests/fixtures/message_assembler/mod.rs similarity index 100% rename from tests/fixtures/message_assembler.rs rename to tests/fixtures/message_assembler/mod.rs diff --git a/tests/formal_tooling.rs b/tests/formal_tooling/mod.rs similarity index 53% rename from tests/formal_tooling.rs rename to tests/formal_tooling/mod.rs index 4a426899a..468160a26 100644 --- a/tests/formal_tooling.rs +++ b/tests/formal_tooling/mod.rs @@ -18,10 +18,10 @@ use cap_std::{ fs_utf8::Dir, }; -#[path = "common/formal_tooling_support.rs"] +#[path = "../common/formal_tooling_support.rs"] mod formal_tooling_support; -#[path = "common/repo_access.rs"] +#[path = "../common/repo_access.rs"] mod repo_access; use formal_tooling_support::{ @@ -292,252 +292,4 @@ fn make_targets_dry_run_to_expected_prover_tools_command( ) } -#[rstest] -#[case::install_kani("install-kani")] -#[case::check_kani_version("check-kani-version")] -#[case::install_verus("install-verus")] -#[case::run_verus("run-verus")] -fn makefile_targets_do_not_embed_bespoke_installer_logic(#[case] target: &str) -> TestResult { - let makefile_str = makefile()?; - let makefile = MakefileContent(&makefile_str); - let recipe = makefile - .target_recipe(target) - .ok_or_else(|| format!("expected `{target}` target in Makefile"))?; - - for forbidden in [ - "cargo install", - "cargo kani setup", - "curl", - "unzip", - "sha256sum", - "shasum", - "rustup toolchain install", - ] { - ensure( - !recipe.contains(forbidden), - format!("`{target}` should not contain bespoke installer command `{forbidden}`"), - )?; - } - Ok(()) -} - -#[rstest] -#[case::test_verification("test-verification")] -#[case::kani("kani")] -#[case::kani_full("kani-full")] -#[case::verus("verus")] -#[case::formal_pr("formal-pr")] -#[case::formal_nightly("formal-nightly")] -fn formal_execution_targets_are_declared(#[case] target: &str) -> TestResult { - let makefile_str = makefile()?; - let makefile = MakefileContent(&makefile_str); - - ensure( - makefile.has_phony_target(target), - format!("`{target}` should be declared as a phony Make target"), - )?; - ensure( - makefile.target_prerequisites(target).is_some(), - format!("`{target}` should have a Make rule"), - ) -} - -#[rstest] -#[case::test_verification("test-verification", "test -p $(VERIFICATION_CRATE)")] -#[case::kani("kani", "$(FORMAL_STUB) kani")] -#[case::kani_full("kani-full", "$(FORMAL_STUB) kani-full")] -#[case::verus("verus", "$(FORMAL_STUB) verus")] -fn direct_recipe_targets_have_expected_content( - #[case] target: &str, - #[case] expected_content: &str, -) -> TestResult { - let makefile_str = makefile()?; - let makefile = MakefileContent(&makefile_str); - let recipe = makefile - .target_recipe(target) - .ok_or_else(|| format!("expected `{target}` target in Makefile"))?; - - ensure( - recipe.contains(expected_content), - format!("`{target}` should contain `{expected_content}`"), - ) -} - -#[rstest] -fn test_verification_executes_the_configured_cargo_command() -> TestResult { - let cargo_wrapper = TemporaryCargoWrapper::new()?; - let output = Command::new("make") - .arg("test-verification") - .arg(format!("CARGO={}", cargo_wrapper.path.as_str())) - .current_dir(repo_access::repo_root()?) - .output()?; - let stdout = String::from_utf8(output.stdout)?; - let wrapper_output = stdout - .lines() - .filter(|line| line.starts_with("WRAPPER_")) - .collect::>(); - - ensure( - output.status.success(), - "`make test-verification` should execute the configured Cargo command", - )?; - ensure( - wrapper_output - == [ - "WRAPPER_RUSTFLAGS=-D warnings", - "WRAPPER_ARG=test", - "WRAPPER_ARG=-p", - "WRAPPER_ARG=wireframe-verification", - ], - "`make test-verification` should invoke Cargo with `RUSTFLAGS=-D warnings` and `test -p \ - wireframe-verification`", - ) -} - -#[rstest] -#[case::formal_pr("formal-pr", &["test-verification", "kani", "verus"])] -#[case::formal_nightly( - "formal-nightly", - &["test-verification", "kani-full", "verus"] -)] -fn aggregate_targets_declare_expected_prerequisites( - #[case] target: &str, - #[case] expected_prerequisites: &[&str], -) -> TestResult { - let makefile_str = makefile()?; - let makefile = MakefileContent(&makefile_str); - let prerequisites = makefile - .target_prerequisites(target) - .ok_or_else(|| format!("expected `{target}` target in Makefile"))?; - let expected_prerequisites = expected_prerequisites - .iter() - .map(|prerequisite| (*prerequisite).to_owned()) - .collect::>(); - - ensure( - prerequisites == expected_prerequisites, - format!("`{target}` should have the expected prerequisites"), - ) -} - -#[rstest] -#[case::kani("kani", "roadmap 15.3.1 adds src/frame Kani smoke harnesses")] -#[case::kani_full("kani-full", "roadmap 15.3.x adds the full Kani harness set")] -#[case::verus("verus", "roadmap 15.5.2 adds verus/wireframe_proofs.rs")] -fn stub_targets_skip_and_exit_zero( - #[case] target: &str, - #[case] roadmap_message: &str, -) -> TestResult { - let (status, _stdout, stderr) = run_make(target, false)?; - - ensure( - status.success(), - format!("`make {target}` should exit zero"), - )?; - ensure( - stderr == format!("FORMAL-SKIP: {target} not yet implemented — {roadmap_message}\n"), - format!("`make {target}` should report its complete formal skip"), - ) -} - -#[rstest] -#[case::kani("kani", "roadmap 15.3.1 adds src/frame Kani smoke harnesses")] -#[case::kani_full("kani-full", "roadmap 15.3.x adds the full Kani harness set")] -#[case::verus("verus", "roadmap 15.5.2 adds verus/wireframe_proofs.rs")] -fn stub_targets_fail_under_formal_strict( - #[case] target: &str, - #[case] roadmap_message: &str, -) -> TestResult { - let (status, _stdout, stderr) = run_make(target, true)?; - let expected_skip = format!("FORMAL-SKIP: {target} not yet implemented — {roadmap_message}"); - - ensure( - !status.success(), - format!("`FORMAL_STRICT=1 make {target}` should fail"), - )?; - // GNU Make appends its failure diagnostic after the stub's fixed line. - ensure( - stderr.lines().next() == Some(&expected_skip), - format!("`FORMAL_STRICT=1 make {target}` should first report its complete formal skip"), - ) -} - -#[rstest] -#[case::test_verification("test-verification", "wireframe-verification")] -#[case::formal_pr("formal-pr", "formal-stub.sh")] -#[case::formal_nightly("formal-nightly", "formal-stub.sh")] -fn aggregate_targets_dry_run_zero( - #[case] target: &str, - #[case] expected_content: &str, -) -> TestResult { - let output = run_make_dry_run(target)?; - - ensure( - output.contains(expected_content), - format!("`make --dry-run {target}` should emit `{expected_content}`"), - ) -} - -#[rstest] -fn formal_alias_delegates_to_formal_pr() -> TestResult { - let makefile_str = makefile()?; - let makefile = MakefileContent(&makefile_str); - let prerequisites = makefile - .target_prerequisites("formal") - .ok_or_else(|| "expected `formal` target in Makefile".to_owned())?; - - ensure( - makefile.has_phony_target("formal"), - "`formal` should be declared as a phony Make target", - )?; - ensure( - prerequisites == ["formal-pr"], - "`formal` should delegate to `formal-pr`", - ) -} - -proptest! { - #[test] - fn three_part_numeric_versions_accept_exactly_three_numeric_parts( - major in "[0-9]+", - minor in "[0-9]+", - patch in "[0-9]+", - ) { - let version = format!("{major}.{minor}.{patch}"); - - prop_assert!(is_three_part_numeric_version(version)); - } - - #[test] - fn three_part_numeric_versions_reject_non_matching_strings( - candidate in "\\PC*", - ) { - let expected = candidate - .split('.') - .collect::>() - .as_slice() - .iter() - .copied() - .all(|part| !part.is_empty() && part.chars().all(|character| character.is_ascii_digit())) - && candidate.split('.').count() == 3; - - prop_assert_eq!(is_three_part_numeric_version(&candidate), expected); - } - - #[test] - fn sha256_hex_accepts_sixty_four_ascii_hex_digits( - digest in "[0-9A-Fa-f]{64}", - ) { - prop_assert!(is_sha256_hex(&digest)); - } - - #[test] - fn sha256_hex_rejects_non_matching_strings( - candidate in "\\PC*", - ) { - let expected = candidate.len() == 64 - && candidate.chars().all(|character| character.is_ascii_hexdigit()); - - prop_assert_eq!(is_sha256_hex(&candidate), expected); - } -} +mod targets; diff --git a/tests/formal_tooling/targets.rs b/tests/formal_tooling/targets.rs new file mode 100644 index 000000000..2715833af --- /dev/null +++ b/tests/formal_tooling/targets.rs @@ -0,0 +1,351 @@ +//! Regression cases for formal-verification Make targets and version syntax. + +use super::*; + +/// A `make` target named by the formal-verification contract. +/// +/// The case tables below name targets and the text their recipes owe. Holding +/// those in domain types keeps a string from meaning "target" in one case and +/// "expected recipe text" in the next. +#[derive(Debug, Clone, Copy)] +struct MakeTarget { + /// Target name exactly as the Makefile declares it. + name: &'static str, +} + +impl MakeTarget { + /// Name the target under test. + const fn new(name: &'static str) -> Self { Self { name } } +} + +impl AsRef for MakeTarget { + fn as_ref(&self) -> &str { self.name } +} + +impl core::fmt::Display for MakeTarget { + fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + formatter.write_str(self.name) + } +} + +/// Text a Make recipe is contractually expected to contain. +#[derive(Debug, Clone, Copy)] +struct RecipeText { + /// Fragment the recipe must contain. + text: &'static str, +} + +impl RecipeText { + /// Record the expected recipe fragment. + const fn new(text: &'static str) -> Self { Self { text } } +} + +impl AsRef for RecipeText { + fn as_ref(&self) -> &str { self.text } +} + +/// Roadmap note quoted when a formal stub target declines to run. +#[derive(Debug, Clone, Copy)] +struct RoadmapNote { + /// Roadmap reference reported by the stub. + note: &'static str, +} + +impl RoadmapNote { + /// Record the roadmap note. + const fn new(note: &'static str) -> Self { Self { note } } +} + +impl AsRef for RoadmapNote { + fn as_ref(&self) -> &str { self.note } +} + +#[rstest] +#[case::install_kani(MakeTarget::new("install-kani"))] +#[case::check_kani_version(MakeTarget::new("check-kani-version"))] +#[case::install_verus(MakeTarget::new("install-verus"))] +#[case::run_verus(MakeTarget::new("run-verus"))] +fn makefile_targets_do_not_embed_bespoke_installer_logic(#[case] target: MakeTarget) -> TestResult { + let makefile_str = makefile()?; + let makefile = MakefileContent(&makefile_str); + let recipe = makefile + .target_recipe(target) + .ok_or_else(|| format!("expected `{target}` target in Makefile"))?; + + for forbidden in [ + "cargo install", + "cargo kani setup", + "curl", + "unzip", + "sha256sum", + "shasum", + "rustup toolchain install", + ] { + ensure( + !recipe.contains(forbidden), + format!("`{target}` should not contain bespoke installer command `{forbidden}`"), + )?; + } + Ok(()) +} + +#[rstest] +#[case::test_verification(MakeTarget::new("test-verification"))] +#[case::kani(MakeTarget::new("kani"))] +#[case::kani_full(MakeTarget::new("kani-full"))] +#[case::verus(MakeTarget::new("verus"))] +#[case::formal_pr(MakeTarget::new("formal-pr"))] +#[case::formal_nightly(MakeTarget::new("formal-nightly"))] +fn formal_execution_targets_are_declared(#[case] target: MakeTarget) -> TestResult { + let makefile_str = makefile()?; + let makefile = MakefileContent(&makefile_str); + + ensure( + makefile.has_phony_target(target), + format!("`{target}` should be declared as a phony Make target"), + )?; + ensure( + makefile.target_prerequisites(target).is_some(), + format!("`{target}` should have a Make rule"), + ) +} + +#[rstest] +#[case::test_verification( + MakeTarget::new("test-verification"), + RecipeText::new("test -p $(VERIFICATION_CRATE)") +)] +#[case::kani(MakeTarget::new("kani"), RecipeText::new("$(FORMAL_STUB) kani"))] +#[case::kani_full( + MakeTarget::new("kani-full"), + RecipeText::new("$(FORMAL_STUB) kani-full") +)] +#[case::verus(MakeTarget::new("verus"), RecipeText::new("$(FORMAL_STUB) verus"))] +fn direct_recipe_targets_have_expected_content( + #[case] target: MakeTarget, + #[case] expected_content: RecipeText, +) -> TestResult { + let makefile_str = makefile()?; + let makefile = MakefileContent(&makefile_str); + let recipe = makefile + .target_recipe(target) + .ok_or_else(|| format!("expected `{target}` target in Makefile"))?; + let expected_content = expected_content.as_ref(); + + ensure( + recipe.contains(expected_content), + format!("`{target}` should contain `{expected_content}`"), + ) +} + +#[rstest] +fn test_verification_executes_the_configured_cargo_command() -> TestResult { + let cargo_wrapper = TemporaryCargoWrapper::new()?; + let output = Command::new("make") + .arg("test-verification") + .arg(format!("CARGO={}", cargo_wrapper.path.as_str())) + .current_dir(repo_access::repo_root()?) + .output()?; + let stdout = String::from_utf8(output.stdout)?; + let wrapper_output = stdout + .lines() + .filter(|line| line.starts_with("WRAPPER_")) + .collect::>(); + + ensure( + output.status.success(), + "`make test-verification` should execute the configured Cargo command", + )?; + ensure( + wrapper_output + == [ + "WRAPPER_RUSTFLAGS=-D warnings", + "WRAPPER_ARG=test", + "WRAPPER_ARG=-p", + "WRAPPER_ARG=wireframe-verification", + ], + "`make test-verification` should invoke Cargo with `RUSTFLAGS=-D warnings` and `test -p \ + wireframe-verification`", + ) +} + +#[rstest] +#[case::formal_pr( + MakeTarget::new("formal-pr"), + &[MakeTarget::new("test-verification"), MakeTarget::new("kani"), MakeTarget::new("verus")] +)] +#[case::formal_nightly( + MakeTarget::new("formal-nightly"), + &[ + MakeTarget::new("test-verification"), + MakeTarget::new("kani-full"), + MakeTarget::new("verus") + ] +)] +fn aggregate_targets_declare_expected_prerequisites( + #[case] target: MakeTarget, + #[case] expected_prerequisites: &[MakeTarget], +) -> TestResult { + let makefile_str = makefile()?; + let makefile = MakefileContent(&makefile_str); + let prerequisites = makefile + .target_prerequisites(target) + .ok_or_else(|| format!("expected `{target}` target in Makefile"))?; + let expected_prerequisites = expected_prerequisites + .iter() + .map(|prerequisite| prerequisite.as_ref().to_owned()) + .collect::>(); + + ensure( + prerequisites == expected_prerequisites, + format!("`{target}` should have the expected prerequisites"), + ) +} + +#[rstest] +#[case::kani( + MakeTarget::new("kani"), + RoadmapNote::new("roadmap 15.3.1 adds src/frame Kani smoke harnesses") +)] +#[case::kani_full( + MakeTarget::new("kani-full"), + RoadmapNote::new("roadmap 15.3.x adds the full Kani harness set") +)] +#[case::verus( + MakeTarget::new("verus"), + RoadmapNote::new("roadmap 15.5.2 adds verus/wireframe_proofs.rs") +)] +fn stub_targets_skip_and_exit_zero( + #[case] target: MakeTarget, + #[case] roadmap_message: RoadmapNote, +) -> TestResult { + let (status, _stdout, stderr) = run_make(target, false)?; + let roadmap_message = roadmap_message.as_ref(); + + ensure( + status.success(), + format!("`make {target}` should exit zero"), + )?; + ensure( + stderr == format!("FORMAL-SKIP: {target} not yet implemented — {roadmap_message}\n"), + format!("`make {target}` should report its complete formal skip"), + ) +} + +#[rstest] +#[case::kani( + MakeTarget::new("kani"), + RoadmapNote::new("roadmap 15.3.1 adds src/frame Kani smoke harnesses") +)] +#[case::kani_full( + MakeTarget::new("kani-full"), + RoadmapNote::new("roadmap 15.3.x adds the full Kani harness set") +)] +#[case::verus( + MakeTarget::new("verus"), + RoadmapNote::new("roadmap 15.5.2 adds verus/wireframe_proofs.rs") +)] +fn stub_targets_fail_under_formal_strict( + #[case] target: MakeTarget, + #[case] roadmap_message: RoadmapNote, +) -> TestResult { + let (status, _stdout, stderr) = run_make(target, true)?; + let expected_skip = format!( + "FORMAL-SKIP: {target} not yet implemented — {}", + roadmap_message.as_ref() + ); + + ensure( + !status.success(), + format!("`FORMAL_STRICT=1 make {target}` should fail"), + )?; + // GNU Make appends its failure diagnostic after the stub's fixed line. + ensure( + stderr.lines().next() == Some(&expected_skip), + format!("`FORMAL_STRICT=1 make {target}` should first report its complete formal skip"), + ) +} + +#[rstest] +#[case::test_verification( + MakeTarget::new("test-verification"), + RecipeText::new("wireframe-verification") +)] +#[case::formal_pr(MakeTarget::new("formal-pr"), RecipeText::new("formal-stub.sh"))] +#[case::formal_nightly(MakeTarget::new("formal-nightly"), RecipeText::new("formal-stub.sh"))] +fn aggregate_targets_dry_run_zero( + #[case] target: MakeTarget, + #[case] expected_content: RecipeText, +) -> TestResult { + let output = run_make_dry_run(target)?; + let expected_content = expected_content.as_ref(); + + ensure( + output.contains(expected_content), + format!("`make --dry-run {target}` should emit `{expected_content}`"), + ) +} + +#[rstest] +fn formal_alias_delegates_to_formal_pr() -> TestResult { + let makefile_str = makefile()?; + let makefile = MakefileContent(&makefile_str); + let prerequisites = makefile + .target_prerequisites(MakeTarget::new("formal")) + .ok_or_else(|| "expected `formal` target in Makefile".to_owned())?; + + ensure( + makefile.has_phony_target(MakeTarget::new("formal")), + "`formal` should be declared as a phony Make target", + )?; + ensure( + prerequisites == ["formal-pr"], + "`formal` should delegate to `formal-pr`", + ) +} + +proptest! { + #[test] + fn three_part_numeric_versions_accept_exactly_three_numeric_parts( + major in "[0-9]+", + minor in "[0-9]+", + patch in "[0-9]+", + ) { + let version = format!("{major}.{minor}.{patch}"); + + prop_assert!(is_three_part_numeric_version(version)); + } + + #[test] + fn three_part_numeric_versions_reject_non_matching_strings( + candidate in "\\PC*", + ) { + let expected = candidate + .split('.') + .collect::>() + .as_slice() + .iter() + .copied() + .all(|part| !part.is_empty() && part.chars().all(|character| character.is_ascii_digit())) + && candidate.split('.').count() == 3; + + prop_assert_eq!(is_three_part_numeric_version(&candidate), expected); + } + + #[test] + fn sha256_hex_accepts_sixty_four_ascii_hex_digits( + digest in "[0-9A-Fa-f]{64}", + ) { + prop_assert!(is_sha256_hex(&digest)); + } + + #[test] + fn sha256_hex_rejects_non_matching_strings( + candidate in "\\PC*", + ) { + let expected = candidate.len() == 64 + && candidate.chars().all(|character| character.is_ascii_hexdigit()); + + prop_assert_eq!(is_sha256_hex(&candidate), expected); + } +} diff --git a/tests/fragment_transport.rs b/tests/fragment_transport.rs deleted file mode 100644 index d9775b7d4..000000000 --- a/tests/fragment_transport.rs +++ /dev/null @@ -1,294 +0,0 @@ -//! Integration tests for transport-level fragmentation and reassembly. -//! -//! Tests are organized into submodules by concern: -//! - Round-trip tests (this file) -//! - Rejection tests (`fragment_transport/rejection.rs`) -//! - Eviction tests (`fragment_transport/eviction.rs`) -#![cfg(not(loom))] - -use std::time::Duration; - -use futures::{SinkExt, StreamExt}; -use tokio::{io::AsyncWriteExt, sync::mpsc, time::timeout}; -use wireframe::{ - app::{Envelope, Packet, WireframeApp}, - fragment::{Fragmenter, decode_fragment_payload}, - serializer::{BincodeSerializer, Serializer}, -}; - -#[path = "fragment_transport/mod.rs"] -mod fragment_transport; - -#[path = "common/fragment_helpers.rs"] -mod fragment_helpers; - -use crate::fragment_helpers::{ - CORRELATION, - ROUTE_ID, - TestError, - TestResult, - assert_handler_observed, - build_envelopes, - fragment_envelope, - fragmentation_config, - make_app, - make_handler, - read_reassembled_response, - read_response_payload, - send_envelopes, - spawn_app, -}; - -/// Common helper for round-trip fragmentation tests. -/// Returns the response payload for additional test-specific assertions. -async fn run_round_trip_test( - buffer_capacity: usize, - payload: Vec, - should_fragment: bool, -) -> TestResult> { - let config = fragmentation_config(buffer_capacity)?; - let (tx, mut rx) = mpsc::unbounded_channel(); - let app = make_app(buffer_capacity, config, &tx)?; - let (mut client, server) = spawn_app(app); - - let request = Envelope::new(ROUTE_ID, CORRELATION, payload.clone()); - - let envelopes = build_envelopes(request, &config, should_fragment)?; - - send_envelopes(&mut client, &envelopes).await?; - client.flush().await?; - - assert_handler_observed(&mut rx, &payload).await?; - client.get_mut().shutdown().await?; - let response = read_response_payload(&mut client, &config).await?; - if response != payload { - return Err(TestError::Assertion(format!( - "response payload mismatch: expected {payload:?}, got {response:?}" - ))); - } - - server.await??; - - Ok(response) -} - -#[tokio::test] -async fn fragmented_request_and_response_round_trip() -> TestResult { - let buffer_capacity = 512; - let payload = vec![b'Z'; 1_200]; - run_round_trip_test(buffer_capacity, payload, true).await?; - Ok(()) -} - -#[tokio::test] -async fn unfragmented_request_and_response_round_trip() -> TestResult { - let buffer_capacity = 512; - let config = fragmentation_config(buffer_capacity)?; - let cap = config.fragment_payload_cap.get(); - let payload_len = cap.saturating_sub(8).max(1); - let payload = vec![b's'; payload_len]; - - let response = run_round_trip_test(buffer_capacity, payload, false).await?; - if decode_fragment_payload(&response)?.is_some() { - return Err(TestError::Assertion( - "small payload should pass through unfragmented".to_string(), - )); - } - - Ok(()) -} - -#[tokio::test] -async fn fragmentation_can_be_disabled_via_public_api() -> TestResult { - let capacity = 1024; - let (tx, mut rx) = mpsc::unbounded_channel(); - let config = fragmentation_config(capacity)?; - - let handler = make_handler(&tx); - - let app: WireframeApp = WireframeApp::new()? - .buffer_capacity(capacity) - .fragmentation(None) - .route(ROUTE_ID, handler)?; - - let (mut client, server) = spawn_app(app); - - let half_capacity = capacity - .checked_div(2) - .ok_or(TestError::Setup("capacity must be at least two"))?; - let payload = vec![b'X'; half_capacity]; - let request = Envelope::new(ROUTE_ID, CORRELATION, payload.clone()); - let serializer = BincodeSerializer; - let bytes = serializer.serialize(&request)?; - client.send(bytes.into()).await?; - - let observed = timeout(Duration::from_secs(1), rx.recv()) - .await? - .ok_or(TestError::Setup("handler payload missing"))?; - if observed != payload { - return Err(TestError::Assertion(format!( - "observed payload mismatch: expected {payload:?}, got {observed:?}" - ))); - } - - client.get_mut().shutdown().await?; - let response = timeout( - Duration::from_secs(1), - read_reassembled_response(&mut client, &config), - ) - .await??; - if decode_fragment_payload(&response)?.is_some() { - return Err(TestError::Assertion( - "expected no fragmentation when fragmentation is disabled".to_string(), - )); - } - - server.await??; - - Ok(()) -} - -#[tokio::test] -async fn fragmentation_is_opt_in_by_default() -> TestResult { - let capacity = 512; - let payload = vec![b'd'; 128]; - let (tx, mut rx) = mpsc::unbounded_channel(); - let handler = make_handler(&tx); - - let app: WireframeApp = WireframeApp::new()? - .buffer_capacity(capacity) - .route(ROUTE_ID, handler)?; - let (mut client, server) = spawn_app(app); - - let request = Envelope::new(ROUTE_ID, CORRELATION, payload.clone()); - let serializer = BincodeSerializer; - let bytes = serializer.serialize(&request)?; - client.send(bytes.into()).await?; - client.get_mut().shutdown().await?; - - assert_handler_observed(&mut rx, &payload).await?; - let response_bytes = timeout(Duration::from_secs(1), client.next()) - .await? - .ok_or(TestError::Setup("response frame missing"))??; - let (response_env, _) = serializer.deserialize::(&response_bytes)?; - let response_payload = response_env.into_parts().into_payload(); - if decode_fragment_payload(&response_payload)?.is_some() { - return Err(TestError::Assertion( - "default app should keep fragmentation disabled".to_string(), - )); - } - - server.await??; - - Ok(()) -} - -#[tokio::test] -async fn duplicate_fragment_is_suppressed_and_reassembles() -> TestResult { - let buffer_capacity = 512; - let config = fragmentation_config(buffer_capacity)?; - let (tx, mut rx) = mpsc::unbounded_channel(); - let app = make_app(buffer_capacity, config, &tx)?; - let (mut client, server) = spawn_app(app); - let fragmenter = Fragmenter::new(config.fragment_payload_cap); - - let payload = vec![b'D'; 800]; - let request = Envelope::new(ROUTE_ID, CORRELATION, payload.clone()); - let mut fragments = fragment_envelope(&request, &fragmenter)?; - let duplicate = fragments - .first() - .cloned() - .ok_or(TestError::Setup("fragmenter produced no fragments"))?; - fragments.insert(1, duplicate); - - send_envelopes(&mut client, &fragments).await?; - client.flush().await?; - client.get_mut().shutdown().await?; - - assert_handler_observed(&mut rx, &payload).await?; - if let Ok(Some(_)) = timeout(Duration::from_millis(200), rx.recv()).await { - return Err(TestError::Assertion( - "duplicate suppression should prevent duplicate handler delivery".to_string(), - )); - } - - let response = read_reassembled_response(&mut client, &config).await?; - if response != payload { - return Err(TestError::Assertion( - "reassembled payload mismatch after duplicate suppression".to_string(), - )); - } - - server.await??; - Ok(()) -} - -#[tokio::test] -async fn interleaved_fragment_streams_reassemble_independently() -> TestResult { - let buffer_capacity = 512; - let config = fragmentation_config(buffer_capacity)?; - let (tx, mut rx) = mpsc::unbounded_channel(); - let app = make_app(buffer_capacity, config, &tx)?; - let (mut client, server) = spawn_app(app); - let fragmenter = Fragmenter::new(config.fragment_payload_cap); - - let payload_a = vec![b'A'; 800]; - let payload_b = vec![b'B'; 900]; - let request_a = Envelope::new(ROUTE_ID, Some(101), payload_a.clone()); - let request_b = Envelope::new(ROUTE_ID, Some(202), payload_b.clone()); - let fragments_a = fragment_envelope(&request_a, &fragmenter)?; - let fragments_b = fragment_envelope(&request_b, &fragmenter)?; - - let mut interleaved = Vec::with_capacity(fragments_a.len() + fragments_b.len()); - let mut idx = 0usize; - while idx < fragments_a.len() || idx < fragments_b.len() { - if let Some(fragment) = fragments_a.get(idx) { - interleaved.push(fragment.clone()); - } - if let Some(fragment) = fragments_b.get(idx) { - interleaved.push(fragment.clone()); - } - idx += 1; - } - - send_envelopes(&mut client, &interleaved).await?; - client.flush().await?; - client.get_mut().shutdown().await?; - - let response_a = timeout( - Duration::from_secs(1), - read_reassembled_response(&mut client, &config), - ) - .await??; - let response_b = timeout( - Duration::from_secs(1), - read_reassembled_response(&mut client, &config), - ) - .await??; - let mut observed_responses = vec![response_a, response_b]; - observed_responses.sort(); - let mut expected_payloads = vec![payload_a.clone(), payload_b.clone()]; - expected_payloads.sort(); - if observed_responses != expected_payloads { - return Err(TestError::Assertion( - "interleaved responses were not reassembled correctly".to_string(), - )); - } - - let first = timeout(Duration::from_secs(1), rx.recv()) - .await? - .ok_or(TestError::Setup("first handler payload missing"))?; - let second = timeout(Duration::from_secs(1), rx.recv()) - .await? - .ok_or(TestError::Setup("second handler payload missing"))?; - let mut observed_requests = vec![first, second]; - observed_requests.sort(); - if observed_requests != expected_payloads { - return Err(TestError::Assertion( - "interleaved reassembly delivered unexpected handler payloads".to_string(), - )); - } - - server.await??; - Ok(()) -} diff --git a/tests/fragment_transport/mod.rs b/tests/fragment_transport/mod.rs index 8c88c0d37..44371aee5 100644 --- a/tests/fragment_transport/mod.rs +++ b/tests/fragment_transport/mod.rs @@ -1,8 +1,368 @@ -//! Module organization for fragment transport integration tests. +//! Integration tests for transport-level fragmentation and reassembly. //! -//! Splits tests by concern: -//! - `rejection`: Tests for malformed and out-of-order fragments -//! - `eviction`: Tests for reassembly timeout and eviction behaviour +//! Tests are organized into submodules by concern: +//! - Round-trip tests (this file) +//! - Rejection tests (`fragment_transport/rejection.rs`) +//! - Eviction tests (`fragment_transport/eviction.rs`) +#![cfg(not(loom))] -pub mod eviction; -pub mod rejection; +use std::{io, time::Duration}; + +use futures::{SinkExt, StreamExt}; +use rstest::{fixture, rstest}; +use tokio::{io::AsyncWriteExt, sync::mpsc, time::timeout}; +use tokio_util::codec::{Framed, LengthDelimitedCodec}; +use wireframe::{ + app::{Envelope, Packet, WireframeApp}, + fragment::{FragmentationConfig, Fragmenter, decode_fragment_payload}, + serializer::{BincodeSerializer, Serializer}, +}; + +#[path = "subtests.rs"] +mod fragment_transport; + +#[path = "../common/fragment_helpers/mod.rs"] +mod fragment_helpers; + +use crate::fragment_helpers::{ + CORRELATION, + FramedResponse, + ROUTE_ID, + TestError, + TestResult, + assert_handler_observed, + build_envelopes, + fragment_envelope, + fragmentation_config, + make_app, + make_handler, + read_framed_response, + read_reassembled_response, + read_response_payload, + send_envelopes, + spawn_app, +}; + +/// Buffer capacity shared by the tests that drive [`Fragmenter`] directly. +const BUFFER_CAPACITY: usize = 512; + +/// Scaffold for tests that build their own fragment stream. +/// +/// Grouping these five values keeps the fallible setup in one place; a test +/// that needs different fragmentation settings should build its own app +/// instead of widening this fixture. +struct FragmentHarness { + /// Fragmentation settings matching [`BUFFER_CAPACITY`]. + config: FragmentationConfig, + /// Client end of the duplex connection. + client: Framed, + /// Server task driving the application. + server: tokio::task::JoinHandle>, + /// Payloads observed by the handler. + rx: mpsc::UnboundedReceiver>, +} + +/// Build a fragmentation-enabled harness at [`BUFFER_CAPACITY`]. +#[fixture] +fn fragment_harness() -> TestResult { + let config = fragmentation_config(BUFFER_CAPACITY)?; + let (tx, rx) = mpsc::unbounded_channel(); + let app = make_app(BUFFER_CAPACITY, config, &tx)?; + let (client, server) = spawn_app(app); + Ok(FragmentHarness { + config, + client, + server, + rx, + }) +} + +/// Common helper for round-trip fragmentation tests. +/// Returns the framed response for additional test-specific assertions. +async fn run_round_trip_test( + buffer_capacity: usize, + payload: Vec, + should_fragment: bool, +) -> TestResult { + let config = fragmentation_config(buffer_capacity)?; + let (tx, mut rx) = mpsc::unbounded_channel(); + let app = make_app(buffer_capacity, config, &tx)?; + let (mut client, server) = spawn_app(app); + + let request = Envelope::new(ROUTE_ID, CORRELATION, payload.clone()); + + let envelopes = build_envelopes(request, &config, should_fragment)?; + + send_envelopes(&mut client, &envelopes).await?; + client.flush().await?; + + assert_handler_observed(&mut rx, &payload).await?; + client.get_mut().shutdown().await?; + let response = read_response_payload(&mut client, &config).await?; + if response.payload != payload { + return Err(TestError::Assertion(format!( + "response payload mismatch: expected {payload:?}, got {:?}", + response.payload + ))); + } + + server.await??; + + Ok(response) +} + +#[tokio::test] +async fn fragmented_request_and_response_round_trip() -> TestResult { + let buffer_capacity = 512; + let payload = vec![b'Z'; 1_200]; + + // `read_framed_response` accepts an unfragmented response, so without this + // assertion a regression that stopped fragmenting the response would still + // satisfy the reassembly and payload checks inside `run_round_trip_test`. + let response = run_round_trip_test(buffer_capacity, payload, true).await?; + if response.frame_count <= 1 { + return Err(TestError::Assertion(format!( + "a large response must cross multiple frames, saw {}", + response.frame_count + ))); + } + + Ok(()) +} + +#[tokio::test] +async fn unfragmented_request_and_response_round_trip() -> TestResult { + let buffer_capacity = 512; + let config = fragmentation_config(buffer_capacity)?; + let cap = config.fragment_payload_cap.get(); + let payload_len = cap.saturating_sub(8).max(1); + let payload = vec![b's'; payload_len]; + + let response = run_round_trip_test(buffer_capacity, payload, false).await?; + if decode_fragment_payload(&response.payload)?.is_some() { + return Err(TestError::Assertion( + "small payload should pass through unfragmented".to_string(), + )); + } + if response.frame_count != 1 { + return Err(TestError::Assertion(format!( + "a payload within the fragment cap should cross exactly one frame, saw {}", + response.frame_count + ))); + } + + Ok(()) +} + +#[tokio::test] +async fn fragmentation_can_be_disabled_via_public_api() -> TestResult { + let capacity = 1024; + let (tx, mut rx) = mpsc::unbounded_channel(); + let config = fragmentation_config(capacity)?; + + let handler = make_handler(&tx); + + let app: WireframeApp = WireframeApp::new()? + .buffer_capacity(capacity) + .fragmentation(None) + .route(ROUTE_ID, handler)?; + + let (mut client, server) = spawn_app(app); + + // One byte past the cap proves the threshold is inert while fragmentation + // is disabled, rather than merely confirming a half-sized payload passes. + let payload_len = config + .fragment_payload_cap + .get() + .checked_add(1) + .ok_or(TestError::Setup("fragment payload cap overflow"))?; + let payload = vec![b'X'; payload_len]; + let request = Envelope::new(ROUTE_ID, CORRELATION, payload.clone()); + let serializer = BincodeSerializer; + let bytes = serializer.serialize(&request)?; + client.send(bytes.into()).await?; + + let observed = timeout(Duration::from_secs(1), rx.recv()) + .await? + .ok_or(TestError::Setup("handler payload missing"))?; + if observed != payload { + return Err(TestError::Assertion(format!( + "observed payload mismatch: expected {payload:?}, got {observed:?}" + ))); + } + + client.get_mut().shutdown().await?; + let response = timeout( + Duration::from_secs(1), + read_framed_response(&mut client, &config), + ) + .await??; + if decode_fragment_payload(&response.payload)?.is_some() { + return Err(TestError::Assertion( + "expected no fragmentation when fragmentation is disabled".to_string(), + )); + } + if response.frame_count != 1 { + return Err(TestError::Assertion(format!( + "a payload past the cap must still cross one frame when fragmentation is off, saw {}", + response.frame_count + ))); + } + + server.await??; + + Ok(()) +} + +#[tokio::test] +async fn fragmentation_is_opt_in_by_default() -> TestResult { + let capacity = 512; + let payload = vec![b'd'; 128]; + let (tx, mut rx) = mpsc::unbounded_channel(); + let handler = make_handler(&tx); + + let app: WireframeApp = WireframeApp::new()? + .buffer_capacity(capacity) + .route(ROUTE_ID, handler)?; + let (mut client, server) = spawn_app(app); + + let request = Envelope::new(ROUTE_ID, CORRELATION, payload.clone()); + let serializer = BincodeSerializer; + let bytes = serializer.serialize(&request)?; + client.send(bytes.into()).await?; + client.get_mut().shutdown().await?; + + assert_handler_observed(&mut rx, &payload).await?; + let response_bytes = timeout(Duration::from_secs(1), client.next()) + .await? + .ok_or(TestError::Setup("response frame missing"))??; + let (response_env, _) = serializer.deserialize::(&response_bytes)?; + let response_payload = response_env.into_parts().into_payload(); + if decode_fragment_payload(&response_payload)?.is_some() { + return Err(TestError::Assertion( + "default app should keep fragmentation disabled".to_string(), + )); + } + + server.await??; + + Ok(()) +} + +#[rstest] +#[tokio::test] +async fn duplicate_fragment_is_suppressed_and_reassembles( + fragment_harness: TestResult, +) -> TestResult { + let FragmentHarness { + config, + mut client, + server, + mut rx, + } = fragment_harness?; + let fragmenter = Fragmenter::new(config.fragment_payload_cap); + + let payload = vec![b'D'; 800]; + let request = Envelope::new(ROUTE_ID, CORRELATION, payload.clone()); + let mut fragments = fragment_envelope(&request, &fragmenter)?; + let duplicate = fragments + .first() + .cloned() + .ok_or(TestError::Setup("fragmenter produced no fragments"))?; + fragments.insert(1, duplicate); + + send_envelopes(&mut client, &fragments).await?; + client.flush().await?; + client.get_mut().shutdown().await?; + + assert_handler_observed(&mut rx, &payload).await?; + if let Ok(Some(_)) = timeout(Duration::from_millis(200), rx.recv()).await { + return Err(TestError::Assertion( + "duplicate suppression should prevent duplicate handler delivery".to_string(), + )); + } + + let response = read_reassembled_response(&mut client, &config).await?; + if response != payload { + return Err(TestError::Assertion( + "reassembled payload mismatch after duplicate suppression".to_string(), + )); + } + + server.await??; + Ok(()) +} + +#[rstest] +#[tokio::test] +async fn interleaved_fragment_streams_reassemble_independently( + fragment_harness: TestResult, +) -> TestResult { + let FragmentHarness { + config, + mut client, + server, + mut rx, + } = fragment_harness?; + let fragmenter = Fragmenter::new(config.fragment_payload_cap); + + let payload_a = vec![b'A'; 800]; + let payload_b = vec![b'B'; 900]; + let request_a = Envelope::new(ROUTE_ID, Some(101), payload_a.clone()); + let request_b = Envelope::new(ROUTE_ID, Some(202), payload_b.clone()); + let fragments_a = fragment_envelope(&request_a, &fragmenter)?; + let fragments_b = fragment_envelope(&request_b, &fragmenter)?; + + let mut interleaved = Vec::with_capacity(fragments_a.len() + fragments_b.len()); + let mut idx = 0usize; + while idx < fragments_a.len() || idx < fragments_b.len() { + if let Some(fragment) = fragments_a.get(idx) { + interleaved.push(fragment.clone()); + } + if let Some(fragment) = fragments_b.get(idx) { + interleaved.push(fragment.clone()); + } + idx += 1; + } + + send_envelopes(&mut client, &interleaved).await?; + client.flush().await?; + client.get_mut().shutdown().await?; + + let response_a = timeout( + Duration::from_secs(1), + read_reassembled_response(&mut client, &config), + ) + .await??; + let response_b = timeout( + Duration::from_secs(1), + read_reassembled_response(&mut client, &config), + ) + .await??; + let mut observed_responses = vec![response_a, response_b]; + observed_responses.sort(); + let mut expected_payloads = vec![payload_a.clone(), payload_b.clone()]; + expected_payloads.sort(); + if observed_responses != expected_payloads { + return Err(TestError::Assertion( + "interleaved responses were not reassembled correctly".to_string(), + )); + } + + let first = timeout(Duration::from_secs(1), rx.recv()) + .await? + .ok_or(TestError::Setup("first handler payload missing"))?; + let second = timeout(Duration::from_secs(1), rx.recv()) + .await? + .ok_or(TestError::Setup("second handler payload missing"))?; + let mut observed_requests = vec![first, second]; + observed_requests.sort(); + if observed_requests != expected_payloads { + return Err(TestError::Assertion( + "interleaved reassembly delivered unexpected handler payloads".to_string(), + )); + } + + server.await??; + Ok(()) +} diff --git a/tests/fragment_transport/subtests.rs b/tests/fragment_transport/subtests.rs new file mode 100644 index 000000000..5396eccb5 --- /dev/null +++ b/tests/fragment_transport/subtests.rs @@ -0,0 +1,10 @@ +//! Module organization for fragment transport integration tests. +//! +//! Splits tests by concern: +//! - `rejection`: Tests for malformed and out-of-order fragments +//! - `eviction`: Tests for reassembly timeout and eviction behaviour + +#[path = "eviction.rs"] +pub mod eviction; +#[path = "rejection.rs"] +pub mod rejection; diff --git a/tests/preamble.rs b/tests/preamble.rs deleted file mode 100644 index a05756ff6..000000000 --- a/tests/preamble.rs +++ /dev/null @@ -1,13 +0,0 @@ -//! Tests for connection preamble reading. -#![cfg(not(loom))] - -#[path = "preamble/basic.rs"] -mod basic; -#[path = "preamble/callbacks.rs"] -mod callbacks; -#[path = "preamble/responses.rs"] -mod responses; -#[path = "preamble/support.rs"] -mod support; -#[path = "preamble/timeouts.rs"] -mod timeouts; diff --git a/tests/preamble/mod.rs b/tests/preamble/mod.rs new file mode 100644 index 000000000..f9da56372 --- /dev/null +++ b/tests/preamble/mod.rs @@ -0,0 +1,13 @@ +//! Tests for connection preamble reading. +#![cfg(not(loom))] + +#[path = "basic.rs"] +mod basic; +#[path = "callbacks.rs"] +mod callbacks; +#[path = "responses.rs"] +mod responses; +#[path = "support.rs"] +mod support; +#[path = "timeouts.rs"] +mod timeouts; diff --git a/tests/prepared_app.rs b/tests/prepared_app/mod.rs similarity index 98% rename from tests/prepared_app.rs rename to tests/prepared_app/mod.rs index 18a728d2d..257f3c186 100644 --- a/tests/prepared_app.rs +++ b/tests/prepared_app/mod.rs @@ -3,9 +3,10 @@ //! Real-socket scenarios live in `prepared_app/tcp.rs`; the scenarios here //! cover startup instrumentation, teardown, and shared-route reuse. +#[path = "../common/mod.rs"] mod common; -#[path = "prepared_app/tcp.rs"] +#[path = "tcp.rs"] mod tcp; use std::{ @@ -16,7 +17,7 @@ use std::{ time::Duration, }; -#[path = "common/fallible_assertions/check_equal.rs"] +#[path = "../common/fallible_assertions/check_equal.rs"] mod fallible_check_equal; use common::prepared_app::{ diff --git a/tests/response.rs b/tests/response/mod.rs similarity index 99% rename from tests/response.rs rename to tests/response/mod.rs index ffd2425e4..aff8dabe2 100644 --- a/tests/response.rs +++ b/tests/response/mod.rs @@ -26,7 +26,7 @@ use wireframe_testing::{ run_app, }; -#[path = "response/response_errors.rs"] +#[path = "response_errors.rs"] mod response_errors; // Larger cap used for oversized frame tests. diff --git a/tests/unified_codec.rs b/tests/unified_codec.rs index 4f8e1183b..870e68aec 100644 --- a/tests/unified_codec.rs +++ b/tests/unified_codec.rs @@ -17,7 +17,7 @@ use wireframe::{ fragment::{FragmentationConfig, decode_fragment_payload}, }; -#[path = "common/fragment_helpers.rs"] +#[path = "common/fragment_helpers/mod.rs"] mod fragment_helpers; #[path = "common/unified_codec_transport.rs"] mod unified_codec_transport; @@ -75,7 +75,9 @@ fn keep_fragment_helper_facade_reexports_linked() { let _ = fragment_helpers::assert_handler_observed; let _ = fragment_helpers::fragmentation_config_with_timeout; let _ = fragment_helpers::fragment_envelope; + let _ = fragment_helpers::read_framed_response; let _ = fragment_helpers::read_reassembled_response; + let _ = core::mem::size_of::(); let _ = fragment_helpers::TestError::Assertion(String::new()); } @@ -158,9 +160,14 @@ async fn fragmented_response_passes_through_pipeline( // Read fragmented response and reassemble let response = read_response_payload(&mut client, &config).await?; assert_eq!( - response, payload, + response.payload, payload, "reassembled response should match original payload" ); + assert!( + response.frame_count > 1, + "a payload past the fragment cap must cross several frames, saw {}", + response.frame_count + ); server.await??; Ok(()) diff --git a/wireframe_testing/src/helpers.rs b/wireframe_testing/src/helpers/mod.rs similarity index 100% rename from wireframe_testing/src/helpers.rs rename to wireframe_testing/src/helpers/mod.rs