test(bdd): add stream CRUD coverage for Rust and Java

bdd/docker-compose.coverage.yml

@@ -34,6 +34,7 @@ services:
         && case "$$BDD_FEATURE" in
           basic_messaging)    cargo llvm-cov test -p bdd --features "$$BDD_RUST_FEATURES" --test basic_messaging --lcov --output-path /reports/rust-bdd-coverage.lcov ;;
           leader_redirection) cargo llvm-cov test -p bdd --features "$$BDD_RUST_FEATURES" --test leader_redirection --lcov --output-path /reports/rust-bdd-coverage.lcov ;;
+          stream_crud)        cargo llvm-cov test -p bdd --features "$$BDD_RUST_FEATURES" --test stream_crud --lcov --output-path /reports/rust-bdd-coverage.lcov ;;
           *)                  cargo llvm-cov test -p bdd --features "$$BDD_RUST_FEATURES" --lcov --output-path /reports/rust-bdd-coverage.lcov ;;
         esac
 
@@ -106,6 +107,12 @@ services:
     command:
       - sh
       - -c
-      - >-
+      - |
+        case "$$BDD_FEATURE" in
+          basic_messaging)    export CUCUMBER_FILTER_TAGS='@basic-messaging' ;;
+          leader_redirection) export CUCUMBER_FILTER_TAGS='@requires-leader-awareness' ;;
+          raw_command)        export CUCUMBER_FILTER_TAGS='@raw-command' ;;
+          stream_crud)        export CUCUMBER_FILTER_TAGS='@stream-crud' ;;
+        esac
         gradle --no-daemon test jacocoTestReport
-        && cp build/reports/jacoco/test/jacocoTestReport.xml /reports/java-bdd-coverage.xml
+        cp build/reports/jacoco/test/jacocoTestReport.xml /reports/java-bdd-coverage.xml

bdd/docker-compose.yml

@@ -32,6 +32,7 @@ services:
       - ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature
       - ./scenarios/leader_redirection.feature:/app/features/leader_redirection.feature
       - ./scenarios/raw_command.feature:/app/features/raw_command.feature
+      - ./scenarios/stream_crud.feature:/app/features/stream_crud.feature
     command:
       - sh
       - -c
@@ -40,6 +41,7 @@ services:
           basic_messaging)    cargo test -p bdd --features "$$BDD_RUST_FEATURES" --test basic_messaging ;;
           leader_redirection) cargo test -p bdd --features "$$BDD_RUST_FEATURES" --test leader_redirection ;;
           raw_command)        cargo test -p bdd --features "$$BDD_RUST_FEATURES" --test raw_command ;;
+          stream_crud)        cargo test -p bdd --features "$$BDD_RUST_FEATURES" --test stream_crud ;;
           *)                  cargo test -p bdd --features "$$BDD_RUST_FEATURES" ;;
         esac
     networks:
@@ -172,6 +174,7 @@ services:
       - ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature
       - ./scenarios/leader_redirection.feature:/app/features/leader_redirection.feature
       - ./scenarios/raw_command.feature:/app/features/raw_command.feature
+      - ./scenarios/stream_crud.feature:/app/features/stream_crud.feature
     command:
       - sh
       - -c
@@ -180,6 +183,7 @@ services:
           basic_messaging)    export CUCUMBER_FILTER_TAGS='@basic-messaging' ;;
           leader_redirection) export CUCUMBER_FILTER_TAGS='@requires-leader-awareness' ;;
           raw_command)        export CUCUMBER_FILTER_TAGS='@raw-command' ;;
+          stream_crud)        export CUCUMBER_FILTER_TAGS='@stream-crud' ;;
         esac
         gradle --no-daemon test
     networks:

bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java

@@ -98,6 +98,56 @@ public void streamHasName(String streamName) {
         assertEquals(streamName, stream.get().name(), "Stream should have expected name");
     }
 
+    @Given("a stream with name {string} exists")
+    public void streamExists(String streamName) {
+        createStream(streamName);
+    }
+
+    @When("I get the stream by its numeric ID")
+    public void getStreamByNumericId() {
+        Optional<StreamDetails> stream = getClient().streams().getStream(context.lastStreamId);
+        context.lastStreamName = stream.map(StreamDetails::name).orElse(null);
+    }
+
+    @Then("the returned stream should have name {string}")
+    public void returnedStreamHasName(String streamName) {
+        assertEquals(streamName, context.lastStreamName, "Stream should have expected name");
+    }
+
+    @When("I list all streams")
+    public void listAllStreams() {
+        List<StreamBase> streams = getClient().streams().getStreams();
+        context.lastStreamWasFound =
+                streams.stream().anyMatch(stream -> stream.id().equals(context.lastStreamId));
+    }
+
+    @Then("the stream list should contain the created stream")
+    public void streamListContainsCreatedStream() {
+        assertTrue(context.lastStreamWasFound, "Stream list should contain the created stream");
+    }
+
+    @When("I update the stream name to {string}")
+    public void updateStreamName(String streamName) {
+        getClient().streams().updateStream(context.lastStreamId, streamName);
+    }
+
+    @Then("getting the stream by its numeric ID should return name {string}")
+    public void getStreamReturnsName(String streamName) {
+        getStreamByNumericId();
+        returnedStreamHasName(streamName);
+    }
+
+    @When("I delete the stream by its numeric ID")
+    public void deleteStreamByNumericId() {
+        getClient().streams().deleteStream(context.lastStreamId);
+    }
+
+    @Then("getting the stream by its numeric ID should return no stream")
+    public void getStreamReturnsNoStream() {
+        Optional<StreamDetails> stream = getClient().streams().getStream(context.lastStreamId);
+        assertTrue(stream.isEmpty(), "Deleted stream should not be returned");
+    }
+
     @When("I create a topic with name {string} in stream {int} with {int} partitions")
     public void createTopic(String topicName, int streamId, int partitions) {
         TopicDetails topic = getClient()

bdd/java/src/test/java/org/apache/iggy/bdd/TestContext.java

@@ -27,6 +27,7 @@ class TestContext {
     String serverAddr;
     Long lastStreamId;
     String lastStreamName;
+    boolean lastStreamWasFound;
     Long lastTopicId;
     String lastTopicName;
     Long lastTopicPartitions;

bdd/rust/Cargo.toml

@@ -49,3 +49,8 @@ required-features = ["bdd"]
 name = "raw_command"
 harness = false
 required-features = ["bdd"]
+
+[[test]]
+name = "stream_crud"
+harness = false
+required-features = ["bdd"]

bdd/rust/tests/common/global_context.rs

@@ -26,6 +26,7 @@ pub struct GlobalContext {
     pub server_addr: Option<String>,
     pub last_stream_id: Option<u32>,
     pub last_stream_name: Option<String>,
+    pub last_stream_was_found: bool,
     pub last_topic_id: Option<u32>,
     pub last_topic_name: Option<String>,
     pub last_topic_partitions: Option<u32>,

bdd/rust/tests/steps/streams.rs

@@ -17,7 +17,7 @@
 
 use crate::common::global_context::GlobalContext;
 use cucumber::{given, then, when};
-use iggy::prelude::StreamClient;
+use iggy::prelude::{Identifier, StreamClient};
 
 #[given("I have no streams in the system")]
 pub async fn given_no_streams(world: &mut GlobalContext) {
@@ -32,7 +32,7 @@ pub async fn given_no_streams(world: &mut GlobalContext) {
     );
 }
 
-#[when(regex = r"^I create a stream with name (.+)$")]
+#[when(regex = r#"^I create a stream with name "(.+)"$"#)]
 pub async fn when_create_stream(world: &mut GlobalContext, stream_name: String) {
     let client = world.client.as_ref().expect("Client should be available");
     let stream = client
@@ -52,7 +52,7 @@ pub async fn then_stream_created_successfully(world: &mut GlobalContext) {
     );
 }
 
-#[then(regex = r"^the stream should have name (.+)$")]
+#[then(regex = r#"^the stream should have name "(.+)"$"#)]
 pub async fn then_stream_has_name(world: &mut GlobalContext, expected_name: String) {
     let stream_name = world
         .last_stream_name
@@ -63,3 +63,91 @@ pub async fn then_stream_has_name(world: &mut GlobalContext, expected_name: Stri
         "Stream should have expected name"
     );
 }
+
+#[given(regex = r#"^a stream with name "(.+)" exists$"#)]
+pub async fn given_stream_exists(world: &mut GlobalContext, stream_name: String) {
+    when_create_stream(world, stream_name).await;
+}
+
+#[when("I get the stream by its numeric ID")]
+pub async fn when_get_stream_by_numeric_id(world: &mut GlobalContext) {
+    let client = world.client.as_ref().expect("Client should be available");
+    let stream_id = world
+        .last_stream_id
+        .expect("Stream should have been created");
+    let stream = client
+        .get_stream(&Identifier::numeric(stream_id).expect("Stream ID should be valid"))
+        .await
+        .expect("Should be able to get stream");
+
+    world.last_stream_name = stream.map(|stream| stream.name);
+}
+
+#[then(regex = r#"^the returned stream should have name "(.+)"$"#)]
+pub async fn then_returned_stream_has_name(world: &mut GlobalContext, expected_name: String) {
+    then_stream_has_name(world, expected_name).await;
+}
+
+#[when("I list all streams")]
+pub async fn when_list_all_streams(world: &mut GlobalContext) {
+    let client = world.client.as_ref().expect("Client should be available");
+    let stream_id = world
+        .last_stream_id
+        .expect("Stream should have been created");
+    let streams = client
+        .get_streams()
+        .await
+        .expect("Should be able to get streams");
+
+    world.last_stream_was_found = streams.iter().any(|stream| stream.id == stream_id);
+}
+
+#[then("the stream list should contain the created stream")]
+pub async fn then_stream_list_contains_created_stream(world: &mut GlobalContext) {
+    assert!(
+        world.last_stream_was_found,
+        "Stream list should contain the created stream"
+    );
+}
+
+#[when(regex = r#"^I update the stream name to "(.+)"$"#)]
+pub async fn when_update_stream_name(world: &mut GlobalContext, stream_name: String) {
+    let client = world.client.as_ref().expect("Client should be available");
+    let stream_id = world
+        .last_stream_id
+        .expect("Stream should have been created");
+    client
+        .update_stream(
+            &Identifier::numeric(stream_id).expect("Stream ID should be valid"),
+            &stream_name,
+        )
+        .await
+        .expect("Should be able to update stream");
+}
+
+#[then(regex = r#"^getting the stream by its numeric ID should return name "(.+)"$"#)]
+pub async fn then_get_stream_returns_name(world: &mut GlobalContext, expected_name: String) {
+    when_get_stream_by_numeric_id(world).await;
+    then_returned_stream_has_name(world, expected_name).await;
+}
+
+#[when("I delete the stream by its numeric ID")]
+pub async fn when_delete_stream_by_numeric_id(world: &mut GlobalContext) {
+    let client = world.client.as_ref().expect("Client should be available");
+    let stream_id = world
+        .last_stream_id
+        .expect("Stream should have been created");
+    client
+        .delete_stream(&Identifier::numeric(stream_id).expect("Stream ID should be valid"))
+        .await
+        .expect("Should be able to delete stream");
+}
+
+#[then("getting the stream by its numeric ID should return no stream")]
+pub async fn then_get_stream_returns_no_stream(world: &mut GlobalContext) {
+    when_get_stream_by_numeric_id(world).await;
+    assert!(
+        world.last_stream_name.is_none(),
+        "Deleted stream should not be returned"
+    );
+}

bdd/rust/tests/stream_crud.rs

@@ -0,0 +1,31 @@
+// 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.
+
+pub(crate) mod common;
+pub(crate) mod helpers;
+pub(crate) mod steps;
+
+use crate::common::global_context::GlobalContext;
+use cucumber::World;
+
+#[tokio::main]
+async fn main() {
+    GlobalContext::cucumber()
+        .fail_on_skipped()
+        .run_and_exit("../../bdd/scenarios/stream_crud.feature")
+        .await;
+}

bdd/scenarios/stream_crud.feature

@@ -0,0 +1,52 @@
+# 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.
+
+@stream-crud
+Feature: Stream CRUD operations
+  As a developer using Apache Iggy
+  I want to manage streams
+  So that I can organize message data
+
+  Background:
+    Given I have a running Iggy server
+    And I am authenticated as the root user
+
+  Scenario: Create a stream
+    When I create a stream with name "stream-crud-create"
+    Then the stream should be created successfully
+    And the stream should have name "stream-crud-create"
+    And getting the stream by its numeric ID should return name "stream-crud-create"
+
+  Scenario: Get a stream by numeric ID
+    Given a stream with name "stream-crud-get" exists
+    When I get the stream by its numeric ID
+    Then the returned stream should have name "stream-crud-get"
+
+  Scenario: List streams
+    Given a stream with name "stream-crud-list" exists
+    When I list all streams
+    Then the stream list should contain the created stream
+
+  Scenario: Update a stream
+    Given a stream with name "stream-crud-update" exists
+    When I update the stream name to "stream-crud-updated"
+    Then getting the stream by its numeric ID should return name "stream-crud-updated"
+
+  Scenario: Delete a stream
+    Given a stream with name "stream-crud-delete" exists
+    When I delete the stream by its numeric ID
+    Then getting the stream by its numeric ID should return no stream

scripts/run-bdd-tests.sh

@@ -36,7 +36,7 @@ usage(){
   log "Usage: $0 [--coverage] <sdk> [feature]"
   log ""
   log "  sdk:     rust | python | php | go | go-race | node | csharp | java | cpp | all | clean (default: all)"
-  log "  feature: basic_messaging | leader_redirection | raw_command | all  (default: all)"
+  log "  feature: basic_messaging | leader_redirection | raw_command | stream_crud | all  (default: all)"
   log ""
   log "  Every suite runs against iggy-server, taken from IGGY_SERVER_PATH"
   log "  (default: target/debug/iggy-server) with an iggy CLI at IGGY_CLI_PATH."
@@ -49,7 +49,7 @@ usage(){
 }
 
 case "$FEATURE" in
-  basic_messaging|leader_redirection|raw_command|all) ;;
+  basic_messaging|leader_redirection|raw_command|stream_crud|all) ;;
   *)
     log "Unknown feature: ${FEATURE}"
     usage
@@ -69,7 +69,7 @@ ALL_COMPOSE_FILES=(
 
 COMPOSE_FILES=(-f docker-compose.yml)
 case "$FEATURE" in
-  basic_messaging|leader_redirection|raw_command|all)
+  basic_messaging|leader_redirection|raw_command|stream_crud|all)
     COMPOSE_FILES+=(-f docker-compose.server.yml) ;;
 esac
 case "$FEATURE" in
@@ -111,6 +111,21 @@ run_suite(){
     esac
   fi
 
+  if [ "$FEATURE" = "stream_crud" ]; then
+    case "$svc" in
+      rust-bdd|java-bdd) ;;
+      *)
+        if [ "$SDK" = "all" ]; then
+          log "⚠️ skipping ${svc%-bdd} (does not support ${FEATURE})"
+          return 0
+        else
+          log "❌ ${SDK} does not support feature '${FEATURE}'"
+          return 1
+        fi
+        ;;
+    esac
+  fi
+
   log "${emoji} ${label}..."
   local code=0
   docker compose "${COMPOSE_FILES[@]}" \