ryerraguntla commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3699061138
########## gateways/kafka/tests/server_e2e_tests.rs: ########## @@ -0,0 +1,480 @@ +// 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/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_NONE, ERROR_UNSUPPORTED_VERSION, +}; +use iggy_gateway_kafka::protocol::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 (corr, body) = round_trip(addr, API_KEY_API_VERSIONS, 3, 42_002, &[]).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_returns_error_then_closes() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.unwrap(); + + let frame1 = build_request_frame(8, 2, 99, Some("e2e-test"), &[]); + stream.write_all(&frame1).await.unwrap(); + let payload1 = read_response_frame(&mut stream, 8 * 1024 * 1024).await; + let (corr, body) = parse_response_payload(8, 2, payload1); + assert_eq!(corr, 99); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), ERROR_UNSUPPORTED_VERSION); + + // The unsupported-version error is terminal: the server closes the connection. + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "connection must close after the unsupported-version error 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); + } +} + +#[tokio::test] +async fn e2e_negative_frame_length_closes_connection() { Review Comment: fixed in [f016d4a](https://github.com/apache/iggy/pull/3519/commits/f016d4a46e2f749c58c01b9fcba7fe24ff6b8a53) -- 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]
