ryerraguntla commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3812641450
########## gateways/kafka/tests/server_e2e_tests.rs: ########## @@ -0,0 +1,466 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! End-to-end TCP tests through `KafkaServer` (full request/response cycle). + +#[path = "common/codec.rs"] +mod codec; +#[path = "common/fixtures.rs"] +mod fixtures; +#[path = "common/server.rs"] +mod server; +#[path = "common/tcp.rs"] +mod tcp; +#[path = "common/wire.rs"] +mod wire; + +use bytes::{BufMut, Bytes, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; + +use iggy_gateway_kafka::protocol::api::{ + API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_FETCH, API_KEY_LIST_OFFSETS, + API_KEY_METADATA, API_KEY_PRODUCE, ERROR_NOT_LEADER_OR_FOLLOWER, +}; + +use codec::Decoder; +use fixtures::load_fixture_body_or_skip; +use server::spawn_test_server; +use std::time::Duration; +use tcp::{ + ByteRead, build_list_offsets_v0_request_with_topic_t, build_metadata_legacy_request, + build_produce_v3_body, build_request_frame, parse_response_payload, read_byte_with_timeout, + read_response_frame, read_response_frame_with_timeout, round_trip, +}; +use wire::{ + OUT_OF_SCOPE_API_KEYS, build_create_topics_empty_request, build_fetch_empty_topics_request, + build_list_offsets_request, build_produce_flexible_empty_request, +}; + +#[tokio::test] +async fn e2e_apiversions_v1_preserves_correlation_id() { + let (addr, _shutdown) = spawn_test_server().await; + let (corr, body) = round_trip(addr, API_KEY_API_VERSIONS, 1, 42_001, &[]).await; + assert_eq!(corr, 42_001); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0); +} + +#[tokio::test] +async fn e2e_apiversions_v3_flexible_preserves_correlation_id() { + let (addr, _shutdown) = spawn_test_server().await; + let request = wire::build_api_versions_flexible_request("iggy-test", "0.1.0"); + let (corr, body) = round_trip(addr, API_KEY_API_VERSIONS, 3, 42_002, &request).await; + assert_eq!(corr, 42_002); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0); + let count = usize::try_from(d.read_varint().unwrap() - 1).expect("api count fits usize"); + assert_eq!(count, 6); +} + +#[tokio::test] +async fn e2e_metadata_v0_returns_stub_broker() { + let (addr, _shutdown) = spawn_test_server().await; + let mut req = BytesMut::new(); + req.put_i32(0); // empty topics + let (corr, body) = round_trip(addr, API_KEY_METADATA, 0, 77, &req).await; + assert_eq!(corr, 77); + let mut d = Decoder::new(body); + assert_eq!(d.read_i32().unwrap(), 1); + d.read_i32().unwrap(); + let host = d.read_nullable_string().unwrap().unwrap(); + assert_eq!(host, "127.0.0.1"); +} + +#[tokio::test] +async fn e2e_produce_v3_round_trip_with_fixture() { + let (addr, _shutdown) = spawn_test_server().await; + let Some(body) = load_fixture_body_or_skip(0, "Produce", 3) else { + return; + }; + let (corr, resp_body) = round_trip(addr, API_KEY_PRODUCE, 3, 88, &body).await; + assert_eq!(corr, 88); + assert!(!resp_body.is_empty()); +} + +#[tokio::test] +async fn e2e_unsupported_api_key_closes_connection() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + // Unknown api key (8, OffsetCommit) has no response schema this gateway can encode, so the + // server closes the connection without a (misparseable) response body. + let frame1 = build_request_frame(8, 2, 99, Some("e2e-test"), &[]); + stream.write_all(&frame1).await.unwrap(); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "unknown api key must close the connection without a response" + ); +} + +#[tokio::test] +async fn e2e_sequential_requests_on_one_connection() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + let requests = [(API_KEY_API_VERSIONS, 1i16), (API_KEY_METADATA, 0i16)]; + for (i, (key, ver)) in requests.iter().enumerate() { + let meta_body = { + let mut b = BytesMut::new(); + b.put_i32(0); + b + }; + let body: &[u8] = if *key == API_KEY_METADATA { + &meta_body + } else { + &[] + }; + let correlation_id = 1000 + i32::try_from(i).expect("test index fits i32"); + let frame = build_request_frame(*key, *ver, correlation_id, Some("seq-test"), body); + stream.write_all(&frame).await.unwrap(); + let payload = read_response_frame(&mut stream, 8 * 1024 * 1024).await; + let (corr, _) = parse_response_payload(*key, *ver, payload); + assert_eq!(corr, correlation_id); + } +} + +// Negative-frame-length-closes-connection coverage lives in listener_robustness_tests.rs +// (uses a timeout-guarded read helper, so a regression fails fast instead of hanging). + +#[tokio::test] +async fn e2e_oversized_frame_is_rejected() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + let mut frame = BytesMut::new(); + frame.put_i32(10_000_000); // exceeds default 8 MiB cap Review Comment: Test deleted (server_e2e_tests.rs); listener_robustness_tests.rs:118-144's timeout-guarded version is the sole survivor. Tradeoff closed: added default_max_frame_size_is_eight_mebibytes (server.rs) so the default 8 MiB value stays pinned without needing an 8 MiB e2e payload. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
