config/tests: add tests for invalid yaml values

Test read_yaml behavior with validator errors, type mismatches, and
throwing validators.

Author: nvartolomei

src/v/config/tests/config_store_test.cc

@@ -13,21 +13,33 @@
 #include "json/writer.h"
 
 #include <seastar/core/sstring.hh>
-#include <seastar/core/thread.hh>
 #include <seastar/testing/thread_test_case.hh>
 #include <seastar/util/log.hh>
 
 #include <cstdint>
 #include <iostream>
 #include <iterator>
-#include <random>
 #include <string>
 #include <utility>
 
 namespace {
 
 ss::logger lg("config_test"); // NOLINT
 
+std::optional<ss::sstring> validate_magic_prefix(const ss::sstring& value) {
+    if (!value.starts_with("magic_")) {
+        return fmt::format("value must start with 'magic_', got: {}", value);
+    }
+    return std::nullopt;
+}
+
+std::optional<ss::sstring> validate_throwing(const ss::sstring& value) {
+    if (value == "throw") {
+        throw std::runtime_error("validator threw an exception");
+    }
+    return std::nullopt;
+}
+
 struct test_config : public config::config_store {
     config::property<int> optional_int;
     config::property<ss::sstring> required_string;
@@ -44,6 +56,8 @@ struct test_config : public config::config_store {
     config::property<ss::sstring> default_secret_string;
     config::property<ss::sstring> secret_string;
     config::property<bool> aliased_bool;
+    config::property<ss::sstring> validated_string;
+    config::property<ss::sstring> throwing_validator_string;
 
     test_config()
       : optional_int(
@@ -113,11 +127,38 @@ struct test_config : public config::config_store {
           "aliased_bool",
           "Property with a compat alias",
           {.aliases = {"aliased_bool_legacy"}},
-          true) {}
+          true)
+      , validated_string(
+          *this,
+          "validated_string",
+          "String that must start with magic_",
+          {},
+          "magic_foo",
+          &validate_magic_prefix)
+      , throwing_validator_string(
+          *this,
+          "throwing_validator_string",
+          "String with a validator that throws",
+          {},
+          "safe",
+          &validate_throwing) {}
 };
 
 struct noop_config : public config::config_store {};
 
+struct required_validated_config : public config::config_store {
+    config::property<ss::sstring> required_validated_string;
+
+    required_validated_config()
+      : required_validated_string(
+          *this,
+          "required_validated_string",
+          "Required string that must start with magic_",
+          {.required = config::required::yes},
+          "magic_bar",
+          &validate_magic_prefix) {}
+};
+
 YAML::Node minimal_valid_configuration() {
     return YAML::Load(
       "required_string: test_value_1\n"
@@ -257,10 +298,59 @@ SEASTAR_THREAD_TEST_CASE(validate_valid_configuration) {
     BOOST_TEST(errors.size() == 0);
 }
 
-SEASTAR_THREAD_TEST_CASE(validate_invalid_configuration) {
+SEASTAR_THREAD_TEST_CASE(validate_with_validator_error) {
     auto cfg = test_config();
-    auto errors = cfg.read_yaml(valid_configuration());
-    BOOST_TEST(errors.size() == 0);
+
+    auto invalid_yaml = YAML::Load("validated_string: invalid_value\n");
+
+    auto errors = cfg.read_yaml(invalid_yaml);
+
+    // Should have error from validator
+    BOOST_TEST(errors.size() > 0);
+
+    // Property should retain default value
+    // Surprising but the current behavior is that invalid values actually
+    // update the property.
+    BOOST_TEST(cfg.validated_string() == "invalid_value");
+}
+
+SEASTAR_THREAD_TEST_CASE(validate_with_type_mismatch) {
+    auto cfg = test_config();
+
+    // Provide string where int is expected
+    auto invalid_yaml = YAML::Load("optional_int: not_an_int\n");
+
+    auto errors = cfg.read_yaml(invalid_yaml);
+
+    BOOST_REQUIRE(!errors.empty());
+    auto& [key, msg] = *errors.begin();
+    BOOST_TEST(key == "optional_int");
+    BOOST_TEST_INFO(msg);
+    BOOST_TEST(msg.contains("bad conversion"));
+
+    BOOST_TEST(cfg.optional_int() == 100); // expect default retained
+}
+
+SEASTAR_THREAD_TEST_CASE(validate_with_throwing_validator) {
+    auto cfg = test_config();
+
+    auto yaml = YAML::Load("throwing_validator_string: throw\n");
+
+    BOOST_CHECK_THROW(cfg.read_yaml(yaml), std::runtime_error);
+}
+
+SEASTAR_THREAD_TEST_CASE(validate_required_with_validator_error) {
+    auto cfg = required_validated_config();
+
+    auto invalid_yaml = YAML::Load(
+      "required_validated_string: invalid_value\n");
+
+    // Required property with validation error throws std::invalid_argument
+    // BOOST_CHECK_THROW(cfg.read_yaml(invalid_yaml), std::invalid_argument);
+
+    auto errors = cfg.read_yaml(invalid_yaml);
+    BOOST_TEST(errors.size() > 0);
+    BOOST_TEST(cfg.required_validated_string() == "invalid_value");
 }
 
 SEASTAR_THREAD_TEST_CASE(config_json_serialization) {

tests/rptest/tests/cluster_config_test.py

@@ -205,6 +205,38 @@ def test_unknown_redpanda_yaml(self):
         )
 
 
+class ClusterConfigBoundedPropertyTest(RedpandaTest):
+    """
+    Test that bounded properties clamp out-of-range values from bootstrap config.
+    """
+
+    def __init__(self, *args, **kwargs):
+        # Set log_segment_size to 100 bytes (below minimum of 1MB)
+        super().__init__(
+            *args,
+            extra_rp_conf={"log_segment_size": 100},
+            **kwargs,
+        )
+
+    @cluster(num_nodes=1)
+    def test_bounded_property_clamped_to_minimum(self):
+        """
+        Verify that bounded properties are clamped to their minimum bound when
+        an out-of-range value is provided in bootstrap config. Setting
+        log_segment_size to 100 should result in the value being clamped to
+        1MB (the minimum bound).
+        """
+        admin = Admin(self.redpanda)
+        config = admin.get_cluster_config()
+
+        # The value should be clamped to 1MB (1048576 bytes)
+        min_segment_size = 1024 * 1024  # 1 MiB
+        assert config["log_segment_size"] == min_segment_size, (
+            f"Expected log_segment_size to be clamped to {min_segment_size} (1MB), "
+            f"but got {config['log_segment_size']}"
+        )
+
+
 class HasRedpandaAndAdmin(Protocol):
     redpanda: RedpandaService
     admin: Admin