test(csharp): add leader_redirection scenario to BDD tests

core/integration/tests/data_integrity/mod.rs

@@ -18,4 +18,5 @@
 
 mod verify_after_server_restart;
 mod verify_consumer_group_partition_assignment;
+mod verify_no_plaintext_credentials_on_disk;
 mod verify_user_login_after_restart;

core/integration/tests/data_integrity/verify_no_plaintext_credentials_on_disk.rs

@@ -0,0 +1,105 @@
+/* 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.
+ */
+
+use iggy::prelude::*;
+use integration::iggy_harness;
+use std::fs;
+use std::path::{Path, PathBuf};
+
+const USERNAME: &str = "plaintext-regression-user";
+const PLAINTEXT_PASSWORD: &str = "plaintext-password-regression-2943";
+const PAT_NAME: &str = "plaintext-regression-pat";
+
+#[iggy_harness(test_client_transport = [Tcp, Http, Quic, WebSocket])]
+async fn should_not_persist_plaintext_password_or_pat_to_disk(harness: &mut TestHarness) {
+    let root_client = harness.root_client().await.unwrap();
+    root_client
+        .create_user(USERNAME, PLAINTEXT_PASSWORD, UserStatus::Active, None)
+        .await
+        .unwrap();
+
+    let raw_pat = root_client
+        .create_personal_access_token(PAT_NAME, IggyExpiry::NeverExpire)
+        .await
+        .unwrap();
+
+    assert!(!raw_pat.token.is_empty(), "Expected non-empty PAT value");
+
+    let data_path = harness.server().data_path();
+
+    drop(root_client);
+    harness.stop().await.unwrap();
+
+    assert_secret_not_persisted(&data_path, PLAINTEXT_PASSWORD, "plaintext password");
+    assert_secret_not_persisted(&data_path, &raw_pat.token, "raw PAT");
+}
+
+fn assert_secret_not_persisted(root: &Path, secret: &str, secret_name: &str) {
+    let secret = secret.as_bytes();
+    for path in collect_files(root) {
+        let contents = fs::read(&path).unwrap_or_else(|e| {
+            panic!("Failed to read persisted file {}: {e}", path.display());
+        });
+        assert!(
+            !contains_subslice(&contents, secret),
+            "Found {secret_name} persisted in file {}",
+            path.display()
+        );
+    }
+}
+
+fn collect_files(root: &Path) -> Vec<PathBuf> {
+    let mut files = Vec::new();
+    collect_files_recursive(root, &mut files);
+    files
+}
+
+fn collect_files_recursive(path: &Path, files: &mut Vec<PathBuf>) {
+    let entries = fs::read_dir(path).unwrap_or_else(|e| {
+        panic!("Failed to read persisted directory {}: {e}", path.display());
+    });
+
+    for entry in entries {
+        let entry = entry.unwrap_or_else(|e| {
+            panic!(
+                "Failed to read entry in persisted directory {}: {e}",
+                path.display()
+            );
+        });
+        let entry_path = entry.path();
+        let file_type = entry.file_type().unwrap_or_else(|e| {
+            panic!("Failed to get file type for {}: {e}", entry_path.display());
+        });
+
+        if file_type.is_dir() {
+            collect_files_recursive(&entry_path, files);
+        } else if file_type.is_file() {
+            files.push(entry_path);
+        }
+    }
+}
+
+fn contains_subslice(haystack: &[u8], needle: &[u8]) -> bool {
+    if needle.is_empty() {
+        return true;
+    }
+
+    haystack
+        .windows(needle.len())
+        .any(|window| window == needle)
+}

foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestContext.cs

@@ -25,8 +25,12 @@ public class TestContext
 {
     public IIggyClient IggyClient { get; set; } = null!;
     public string TcpUrl { get; set; } = string.Empty;
+    public string LeaderTcpUrl { get; set; } = string.Empty;
+    public string FollowerTcpUrl { get; set; } = string.Empty;
+    public Dictionary<string, IIggyClient> Clients { get; set; } = [];
     public StreamResponse? CreatedStream { get; set; }
     public TopicResponse? CreatedTopic { get; set; }
     public List<MessageResponse> PolledMessages { get; set; } = new();
     public Message? LastSendMessage { get; set; }
+    public string? InitialAddress { get; set; }
 }

foreign/csharp/Iggy_SDK.Tests.BDD/Context/TestHooks.cs

@@ -33,12 +33,29 @@ public TestHooks(TestContext context)
     public void BeforeScenario()
     {
         _context.TcpUrl = Environment.GetEnvironmentVariable("IGGY_TCP_ADDRESS") ?? "127.0.0.1:8090";
+        _context.LeaderTcpUrl = Environment.GetEnvironmentVariable("IGGY_TCP_ADDRESS_LEADER") ?? "127.0.0.1:8091";
+        _context.FollowerTcpUrl = Environment.GetEnvironmentVariable("IGGY_TCP_ADDRESS_FOLLOWER") ?? "127.0.0.1:8092";
+        _context.Clients.Clear();
+        _context.InitialAddress = null;
+        _context.CreatedStream = null;
     }
 
     [AfterScenario]
     public void AfterScenario()
     {
-        //
+        var clients = _context.Clients.Values
+            .Distinct()
+            .ToList();
+
+        if (_context.IggyClient is not null && !clients.Contains(_context.IggyClient))
+        {
+            clients.Add(_context.IggyClient);
+        }
+
+        foreach (var client in clients)
+        {
+            client.Dispose();
+        }
     }
 
     [BeforeFeature]

foreign/csharp/Iggy_SDK.Tests.BDD/Iggy_SDK.Tests.BDD.csproj

@@ -37,6 +37,10 @@
             <Link>Features\basic_messaging.feature</Link>
             <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
         </Content>
+        <Content Include="..\..\..\bdd\scenarios\leader_redirection.feature">
+            <Link>Features\leader_redirection.feature</Link>
+            <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+        </Content>
     </ItemGroup>
 
     <ItemGroup>

foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/LeaderRedirectionSteps.cs

@@ -0,0 +1,231 @@
+// 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.
+
+using Apache.Iggy.Configuration;
+using Apache.Iggy.Contracts;
+using Apache.Iggy.Enums;
+using Apache.Iggy.Factory;
+using Apache.Iggy.IggyClient;
+using Reqnroll;
+using Shouldly;
+using TestContext = Apache.Iggy.Tests.BDD.Context.TestContext;
+
+namespace Apache.Iggy.Tests.BDD.StepDefinitions;
+
+[Binding]
+public class LeaderRedirectionSteps
+{
+    private readonly TestContext _context;
+
+    public LeaderRedirectionSteps(TestContext context)
+    {
+        _context = context;
+    }
+
+    [Given(@"I have cluster configuration enabled with (\d+) nodes")]
+    public void GivenIHaveClusterConfigurationEnabledWithNodes(int nodeCount)
+    {
+        nodeCount.ShouldBe(2);
+    }
+
+    [Given(@"node (\d+) is configured on port (\d+)")]
+    public void GivenNodeIsConfiguredOnPort(int nodeId, int port)
+    {
+        _ = nodeId;
+        ResolveAddressForPort(port).ShouldNotBeNullOrEmpty();
+    }
+
+    [Given(@"I start server (\d+) on port (\d+) as (leader|follower)")]
+    public void GivenIStartServerOnPortAs(int nodeId, int port, string role)
+    {
+        _ = nodeId;
+        var address = ResolveAddressForRole(role);
+        address.ShouldEndWith($":{port}");
+    }
+
+    [Given(@"I start a single server on port (\d+) without clustering enabled")]
+    public void GivenIStartASingleServerOnPortWithoutClusteringEnabled(int port)
+    {
+        _context.TcpUrl.ShouldEndWith($":{port}");
+    }
+
+    [When(@"I create a client connecting to (follower|leader) on port (\d+)")]
+    public async Task WhenICreateAClientConnectingToOnPort(string role, int port)
+    {
+        var address = ResolveAddressForRole(role);
+        address.ShouldEndWith($":{port}");
+
+        await CreateAndConnectClient("main", address);
+        _context.InitialAddress = address;
+    }
+
+    [When(@"I create a client connecting directly to leader on port (\d+)")]
+    public async Task WhenICreateAClientConnectingDirectlyToLeaderOnPort(int port)
+    {
+        var address = _context.LeaderTcpUrl;
+        address.ShouldEndWith($":{port}");
+
+        await CreateAndConnectClient("main", address);
+        _context.InitialAddress = address;
+    }
+
+    [When(@"I create a client connecting to port (\d+)")]
+    public async Task WhenICreateAClientConnectingToPort(int port)
+    {
+        var address = ResolveAddressForPort(port);
+        await CreateAndConnectClient("main", address);
+        _context.InitialAddress = address;
+    }
+
+    [When(@"I create client ([A-Z]) connecting to port (\d+)")]
+    public async Task WhenICreateClientConnectingToPort(string clientName, int port)
+    {
+        var address = ResolveAddressForPort(port);
+        await CreateAndConnectClient(clientName, address);
+    }
+
+    [When(@"I authenticate as root user")]
+    public async Task WhenIAuthenticateAsRootUser()
+    {
+        var client = GetClient("main");
+        var loginResult = await client.LoginUser("iggy", "iggy");
+
+        loginResult.ShouldNotBeNull();
+        loginResult.UserId.ShouldBe(0);
+    }
+
+    [When(@"both clients authenticate as root user")]
+    public async Task WhenBothClientsAuthenticateAsRootUser()
+    {
+        foreach (var clientName in _context.Clients.Keys.OrderBy(name => name).ToList())
+        {
+            var loginResult = await _context.Clients[clientName].LoginUser("iggy", "iggy");
+            loginResult.ShouldNotBeNull();
+            loginResult.UserId.ShouldBe(0);
+        }
+    }
+
+    [When(@"I create a stream named ""([^""]+)""")]
+    public async Task WhenICreateAStreamNamed(string streamName)
+    {
+        _context.CreatedStream = await GetClient("main").CreateStreamAsync(streamName);
+    }
+
+    [Then(@"the client should automatically redirect to leader on port (\d+)")]
+    public async Task ThenTheClientShouldAutomaticallyRedirectToLeaderOnPort(int port)
+    {
+        await AssertClientAddress("main", port);
+        GetClient("main").GetCurrentAddress().ShouldNotBe(_context.InitialAddress);
+    }
+
+    [Then(@"the stream should be created successfully on the leader")]
+    public void ThenTheStreamShouldBeCreatedSuccessfullyOnTheLeader()
+    {
+        _context.CreatedStream.ShouldNotBeNull();
+        _context.CreatedStream.Name.ShouldNotBeNullOrEmpty();
+    }
+
+    [Then(@"the client should not perform any redirection")]
+    public void ThenTheClientShouldNotPerformAnyRedirection()
+    {
+        GetClient("main").GetCurrentAddress().ShouldBe(_context.InitialAddress);
+    }
+
+    [Then(@"the connection should remain on port (\d+)")]
+    public async Task ThenTheConnectionShouldRemainOnPort(int port)
+    {
+        await AssertClientAddress("main", port);
+    }
+
+    [Then(@"the client should connect successfully without redirection")]
+    public void ThenTheClientShouldConnectSuccessfullyWithoutRedirection()
+    {
+        GetClient("main").ShouldNotBeNull();
+        GetClient("main").GetCurrentAddress().ShouldBe(_context.InitialAddress);
+    }
+
+    [Then(@"client ([A-Z]) should stay connected to port (\d+)")]
+    public async Task ThenClientShouldStayConnectedToPort(string clientName, int port)
+    {
+        await AssertClientAddress(clientName, port);
+    }
+
+    [Then(@"client ([A-Z]) should redirect to port (\d+)")]
+    public async Task ThenClientShouldRedirectToPort(string clientName, int port)
+    {
+        await AssertClientAddress(clientName, port);
+    }
+
+    [Then(@"both clients should be using the same server")]
+    public void ThenBothClientsShouldBeUsingTheSameServer()
+    {
+        GetClient("A").GetCurrentAddress().ShouldBe(GetClient("B").GetCurrentAddress());
+    }
+
+    private async Task CreateAndConnectClient(string name, string address)
+    {
+        var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
+        {
+            BaseAddress = address,
+            Protocol = Protocol.Tcp,
+            ReconnectionSettings = new ReconnectionSettings { Enabled = true },
+            AutoLoginSettings = new AutoLoginSettings { Enabled = false }
+        });
+
+        await client.ConnectAsync();
+        await client.PingAsync();
+
+        _context.Clients[name] = client;
+        if (name == "main")
+        {
+            _context.IggyClient = client;
+        }
+    }
+
+    private IIggyClient GetClient(string name)
+    {
+        return _context.Clients[name];
+    }
+
+    private async Task AssertClientAddress(string clientName, int expectedPort)
+    {
+        var client = GetClient(clientName);
+        await client.PingAsync();
+        client.GetCurrentAddress().ShouldEndWith($":{expectedPort}");
+    }
+
+    private string ResolveAddressForRole(string role)
+    {
+        return role switch
+        {
+            "leader" => _context.LeaderTcpUrl,
+            "follower" => _context.FollowerTcpUrl,
+            _ => throw new ArgumentOutOfRangeException(nameof(role), role, "Unsupported server role")
+        };
+    }
+
+    private string ResolveAddressForPort(int port)
+    {
+        return port switch
+        {
+            8090 => _context.TcpUrl,
+            8091 => _context.LeaderTcpUrl,
+            8092 => _context.FollowerTcpUrl,
+            _ => throw new ArgumentOutOfRangeException(nameof(port), port, "Unsupported test port")
+        };
+    }
+}