Add getClusterParameter tests

documentdb_tests/compatibility/tests/system/administration/commands/getClusterParameter/__init__.py

@@ -0,0 +1 @@
+"""getClusterParameter command tests."""

documentdb_tests/compatibility/tests/system/administration/commands/getClusterParameter/test_getClusterParameter_argument_handling.py

@@ -0,0 +1,92 @@
+"""Tests for getClusterParameter argument handling.
+
+Covers accepted argument forms (single string, wildcard, array of
+strings, duplicate names, unrecognized extra field) and BSON type
+rejection for the command argument.
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.system.administration.utils.administration_test_case import (  # noqa: E501
+    AdministrationTestCase,
+)
+from documentdb_tests.framework.assertions import assertFailureCode, assertProperties
+from documentdb_tests.framework.bson_type_validator import (
+    BsonTypeTestCase,
+    generate_bson_rejection_test_cases,
+)
+from documentdb_tests.framework.error_codes import TYPE_MISMATCH_ERROR
+from documentdb_tests.framework.executor import execute_admin_command
+from documentdb_tests.framework.parametrize import pytest_params
+from documentdb_tests.framework.property_checks import Eq, Len
+from documentdb_tests.framework.test_constants import BsonType
+
+pytestmark = pytest.mark.admin
+
+_VALID_PARAM = "changeStreamOptions"
+_VALID_PARAM_2 = "defaultMaxTimeMS"
+
+_ARGUMENT_TYPE_SPEC = [
+    BsonTypeTestCase(
+        id="getClusterParameter_argument",
+        msg="getClusterParameter should reject non-string/non-array argument types",
+        keyword="getClusterParameter",
+        valid_types=[BsonType.STRING, BsonType.ARRAY],
+        default_error_code=TYPE_MISMATCH_ERROR,
+        skip_rejection_types=[BsonType.NULL],
+    ),
+]
+
+_ARGUMENT_REJECTION_CASES = generate_bson_rejection_test_cases(_ARGUMENT_TYPE_SPEC)
+
+
+@pytest.mark.parametrize("bson_type,sample_value,spec", _ARGUMENT_REJECTION_CASES)
+def test_getClusterParameter_argument_rejects_type(collection, bson_type, sample_value, spec):
+    """Test getClusterParameter rejects non-string/non-array argument types."""
+    result = execute_admin_command(collection, {"getClusterParameter": sample_value})
+    assertFailureCode(
+        result,
+        spec.expected_code(bson_type),
+        msg=f"getClusterParameter should reject {bson_type.value} argument.",
+    )
+
+
+ARGUMENT_FORM_TESTS: list[AdministrationTestCase] = [
+    AdministrationTestCase(
+        id="wildcard_returns_all",
+        command={"getClusterParameter": "*"},
+        checks={"ok": Eq(1.0)},
+        msg="Wildcard '*' should return ok:1",
+    ),
+    AdministrationTestCase(
+        id="single_name_returns_one",
+        command={"getClusterParameter": _VALID_PARAM},
+        checks={"ok": Eq(1.0), "clusterParameters": Len(1)},
+        msg="Single name should return ok:1 with one parameter",
+    ),
+    AdministrationTestCase(
+        id="array_two_names_returns_two",
+        command={"getClusterParameter": [_VALID_PARAM, _VALID_PARAM_2]},
+        checks={"ok": Eq(1.0), "clusterParameters": Len(2)},
+        msg="Array of two names should return two parameters",
+    ),
+    AdministrationTestCase(
+        id="array_duplicate_names",
+        command={"getClusterParameter": [_VALID_PARAM, _VALID_PARAM]},
+        checks={"ok": Eq(1.0)},
+        msg="Duplicate names in array should succeed",
+    ),
+    AdministrationTestCase(
+        id="unrecognized_field_accepted",
+        command={"getClusterParameter": "*", "unknownField": "test"},
+        checks={"ok": Eq(1.0)},
+        msg="Unrecognized extra field should be accepted",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(ARGUMENT_FORM_TESTS))
+def test_getClusterParameter_argument_forms(collection, test):
+    """Test accepted argument forms each return ok:1 with expected clusterParameters length."""
+    result = execute_admin_command(collection, test.command)
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)

documentdb_tests/compatibility/tests/system/administration/commands/getClusterParameter/test_getClusterParameter_core_behavior.py

@@ -0,0 +1,41 @@
+"""Tests for getClusterParameter core retrieval behavior.
+
+Verifies that the wildcard returns more than one cluster parameter and
+that the result includes known parameters by name.
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.system.administration.utils.administration_test_case import (  # noqa: E501
+    AdministrationTestCase,
+)
+from documentdb_tests.framework.assertions import assertProperties
+from documentdb_tests.framework.executor import execute_admin_command
+from documentdb_tests.framework.parametrize import pytest_params
+from documentdb_tests.framework.property_checks import Contains, LenGt
+
+pytestmark = pytest.mark.admin
+
+_VALID_PARAM = "changeStreamOptions"
+
+CORE_BEHAVIOR_TESTS: list[AdministrationTestCase] = [
+    AdministrationTestCase(
+        id="wildcard_returns_multiple_params",
+        command={"getClusterParameter": "*"},
+        checks={"clusterParameters": LenGt(1)},
+        msg="Wildcard should return more than one cluster parameter",
+    ),
+    AdministrationTestCase(
+        id="wildcard_includes_known_param",
+        command={"getClusterParameter": "*"},
+        checks={"clusterParameters": Contains("_id", _VALID_PARAM)},
+        msg=f"Wildcard result should include '{_VALID_PARAM}'",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(CORE_BEHAVIOR_TESTS))
+def test_getClusterParameter_core_behavior(collection, test):
+    """Test core retrieval behavior: wildcard parameter count and inclusion."""
+    result = execute_admin_command(collection, test.command)
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)

documentdb_tests/compatibility/tests/system/administration/commands/getClusterParameter/test_getClusterParameter_errors.py

@@ -0,0 +1,136 @@
+"""Tests for getClusterParameter error cases.
+
+Covers error-producing inputs: unknown parameter names, empty and
+null arguments, array element type errors, command-key case
+sensitivity, and key ordering enforcement. Also verifies that the
+command is rejected on non-admin databases.
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.system.administration.utils.administration_test_case import (  # noqa: E501
+    AdministrationTestCase,
+)
+from documentdb_tests.framework.assertions import assertFailureCode
+from documentdb_tests.framework.error_codes import (
+    BAD_VALUE_ERROR,
+    COMMAND_NOT_FOUND_ERROR,
+    NO_SUCH_KEY_ERROR,
+    TYPE_MISMATCH_ERROR,
+    UNAUTHORIZED_ERROR,
+)
+from documentdb_tests.framework.executor import execute_admin_command, execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+pytestmark = pytest.mark.admin
+
+_VALID_PARAM = "changeStreamOptions"
+
+
+_NO_SUCH_KEY_CASES: list[AdministrationTestCase] = [
+    AdministrationTestCase(
+        id="unknown_single_string_errors",
+        command={"getClusterParameter": "unknownParam"},
+        error_code=NO_SUCH_KEY_ERROR,
+        msg="Single unknown name should fail with no-such-parameter error",
+    ),
+    AdministrationTestCase(
+        id="empty_string_argument",
+        command={"getClusterParameter": ""},
+        error_code=NO_SUCH_KEY_ERROR,
+        msg="Empty-string argument should be treated as an unknown parameter name",
+    ),
+    AdministrationTestCase(
+        id="case_altered",
+        command={"getClusterParameter": "ChangeStreamOptions"},
+        error_code=NO_SUCH_KEY_ERROR,
+        msg="Altered-case known name should not match (case-sensitive)",
+    ),
+    AdministrationTestCase(
+        id="star_in_array_is_literal_name",
+        command={"getClusterParameter": ["*"]},
+        error_code=NO_SUCH_KEY_ERROR,
+        msg="'*' inside an array is a literal name, not a wildcard",
+    ),
+    AdministrationTestCase(
+        id="array_mixed_valid_unknown_errors",
+        command={"getClusterParameter": [_VALID_PARAM, "unknownParam"]},
+        error_code=NO_SUCH_KEY_ERROR,
+        msg="Unknown entry in mixed array should fail",
+    ),
+]
+
+_TYPE_ERROR_CASES: list[AdministrationTestCase] = [
+    AdministrationTestCase(
+        id="empty_array_errors",
+        command={"getClusterParameter": []},
+        error_code=BAD_VALUE_ERROR,
+        msg="Empty array must supply at least one name",
+    ),
+    AdministrationTestCase(
+        id="array_nonstring_element_rejects",
+        command={"getClusterParameter": [_VALID_PARAM, 123]},
+        error_code=TYPE_MISMATCH_ERROR,
+        msg="Non-string array element should be rejected",
+    ),
+    AdministrationTestCase(
+        id="null_argument",
+        command={"getClusterParameter": None},
+        error_code=TYPE_MISMATCH_ERROR,
+        msg="Null argument should be rejected as a type mismatch",
+    ),
+    AdministrationTestCase(
+        id="array_null_element_rejects",
+        command={"getClusterParameter": [None]},
+        error_code=TYPE_MISMATCH_ERROR,
+        msg="Null element in array should be rejected",
+    ),
+    AdministrationTestCase(
+        id="array_doc_element_rejects",
+        command={"getClusterParameter": [{"a": 1}]},
+        error_code=TYPE_MISMATCH_ERROR,
+        msg="Document element in array should be rejected",
+    ),
+    AdministrationTestCase(
+        id="array_nested_array_rejects",
+        command={"getClusterParameter": [[_VALID_PARAM]]},
+        error_code=TYPE_MISMATCH_ERROR,
+        msg="Nested array element should be rejected",
+    ),
+]
+
+_COMMAND_ROUTING_CASES: list[AdministrationTestCase] = [
+    AdministrationTestCase(
+        id="wrong_case_command_key_rejected",
+        command={"getclusterparameter": "*"},
+        error_code=COMMAND_NOT_FOUND_ERROR,
+        msg="Wrong-case command key should be rejected",
+    ),
+    AdministrationTestCase(
+        id="command_key_not_first_fails",
+        command={"comment": "test", "getClusterParameter": "*"},
+        error_code=COMMAND_NOT_FOUND_ERROR,
+        msg="getClusterParameter must be the first key in the command document",
+    ),
+]
+
+_ERROR_CASES: list[AdministrationTestCase] = (
+    _NO_SUCH_KEY_CASES + _TYPE_ERROR_CASES + _COMMAND_ROUTING_CASES
+)
+
+
+@pytest.mark.parametrize("test", pytest_params(_ERROR_CASES))
+def test_getClusterParameter_errors(collection, test):
+    """Test all error-producing inputs: unknown names, type mismatches, and command routing."""
+    result = execute_admin_command(collection, test.command)
+    assertFailureCode(result, test.error_code, msg=test.msg)
+
+
+def test_getClusterParameter_rejected_on_non_admin_database(collection):
+    """Test getClusterParameter is rejected against a non-admin database."""
+    result = execute_command(collection, {"getClusterParameter": "*"})
+    assertFailureCode(
+        result,
+        UNAUTHORIZED_ERROR,
+        msg="getClusterParameter should be rejected on a non-admin database.",
+    )

documentdb_tests/compatibility/tests/system/administration/commands/getClusterParameter/test_getClusterParameter_response_structure.py

@@ -0,0 +1,54 @@
+"""Tests for getClusterParameter response structure.
+
+Verifies the top-level shape of the success response: ok is 1,
+clusterParameters is an array, and a single-name request returns
+exactly one element.
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.system.administration.utils.administration_test_case import (  # noqa: E501
+    AdministrationTestCase,
+)
+from documentdb_tests.framework.assertions import assertProperties
+from documentdb_tests.framework.executor import execute_admin_command
+from documentdb_tests.framework.parametrize import pytest_params
+from documentdb_tests.framework.property_checks import Eq, IsType, Len
+
+pytestmark = pytest.mark.admin
+
+_VALID_PARAM = "changeStreamOptions"
+
+PROPERTY_TESTS: list[AdministrationTestCase] = [
+    AdministrationTestCase(
+        id="ok_is_1",
+        command={"getClusterParameter": "*"},
+        checks={"ok": Eq(1.0)},
+        msg="Response ok should be 1.0",
+    ),
+    AdministrationTestCase(
+        id="clusterParameters_is_array",
+        command={"getClusterParameter": "*"},
+        checks={"clusterParameters": IsType("array")},
+        msg="clusterParameters should be an array",
+    ),
+    AdministrationTestCase(
+        id="single_name_length_is_one",
+        command={"getClusterParameter": _VALID_PARAM},
+        checks={"clusterParameters": Len(1)},
+        msg="Single-name request should return exactly one element",
+    ),
+    AdministrationTestCase(
+        id="element_id_matches_request",
+        command={"getClusterParameter": _VALID_PARAM},
+        checks={"clusterParameters.0._id": Eq(_VALID_PARAM)},
+        msg=f"Single-name request should return element with _id equal to '{_VALID_PARAM}'",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(PROPERTY_TESTS))
+def test_getClusterParameter_response_properties(collection, test):
+    """Verifies getClusterParameter response fields have expected types and values."""
+    result = execute_admin_command(collection, test.command)
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)

documentdb_tests/compatibility/tests/system/administration/utils/administration_test_case.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional
+
+from documentdb_tests.framework.test_case import BaseTestCase
+
+
+@dataclass(frozen=True)
+class AdministrationTestCase(BaseTestCase):
+    """Test case for administration command tests.
+
+    Attributes:
+        command: The command dict to execute.
+        checks: Mapping of dotted field paths to property check objects.
+    """
+
+    command: Optional[Dict[str, Any]] = None
+    checks: Dict[str, Any] = field(default_factory=dict)

documentdb_tests/framework/property_checks.py

@@ -165,6 +165,25 @@ def __repr__(self) -> str:
         return f"{type(self).__name__}({self.expected!r})"
 
 
+class LenGt(Check):
+    """Assert that the field is a list with length strictly greater than a minimum."""
+
+    def __init__(self, minimum: int) -> None:
+        self.minimum = minimum
+
+    def check(self, value: Any, path: str) -> str | None:
+        if value is _FIELD_ABSENT:
+            return f"expected '{path}' length > {self.minimum}, but field is missing"
+        if not isinstance(value, list):
+            return f"expected '{path}' to be a list, got {type(value).__name__}"
+        if len(value) <= self.minimum:
+            return f"expected '{path}' length > {self.minimum}, got {len(value)}"
+        return None
+
+    def __repr__(self) -> str:
+        return f"{type(self).__name__}({self.minimum!r})"
+
+
 class Contains(Check):
     """Assert that a list contains a dict where ``key`` equals ``value``."""