Add getLog tests

documentdb_tests/compatibility/tests/system/diagnostic/commands/getLog/test_getLog_argument_validation.py

@@ -0,0 +1,41 @@
+"""Tests for getLog command argument validation.
+
+Covers BSON type handling for the ``getLog`` field value. Only a string is
+accepted; every non-string type is rejected with TypeMismatch.
+
+Invalid string values (e.g. unknown components, the deprecated "rs") and
+unrecognized command fields are covered in test_getLog_errors.py.
+"""
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertFailureCode
+from documentdb_tests.framework.bson_type_validator import (
+    BsonTypeTestCase,
+    generate_bson_rejection_test_cases,
+)
+from documentdb_tests.framework.error_codes import MISSING_FIELD_ERROR, TYPE_MISMATCH_ERROR
+from documentdb_tests.framework.executor import execute_admin_command
+from documentdb_tests.framework.test_constants import BsonType
+
+pytestmark = pytest.mark.admin
+
+BSON_TYPE_PARAMS = [
+    BsonTypeTestCase(
+        id="getLog_value",
+        msg="getLog should reject non-string value types",
+        keyword="getLog",
+        valid_types=[BsonType.STRING],
+        default_error_code=TYPE_MISMATCH_ERROR,
+        error_code_overrides={BsonType.NULL: MISSING_FIELD_ERROR},
+    ),
+]
+
+REJECTION_CASES = generate_bson_rejection_test_cases(BSON_TYPE_PARAMS)
+
+
+@pytest.mark.parametrize("bson_type,sample_value,spec", REJECTION_CASES)
+def test_getLog_rejects_non_string_value(collection, bson_type, sample_value, spec):
+    """Test getLog rejects each non-string BSON type for its value."""
+    result = execute_admin_command(collection, {"getLog": sample_value})
+    assertFailureCode(result, spec.expected_code(bson_type), msg=spec.msg)

documentdb_tests/compatibility/tests/system/diagnostic/commands/getLog/test_getLog_errors.py

@@ -0,0 +1,70 @@
+"""Tests for getLog command error conditions.
+
+Covers invalid log component names (unknown component, the deprecated "rs"
+value, empty string), unrecognized command fields, and the admin-database
+requirement.
+
+BSON type rejection/acceptance for the value is covered in
+test_getLog_argument_validation.py.
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.system.diagnostic.utils.diagnostic_test_case import (
+    DiagnosticTestCase,
+)
+from documentdb_tests.framework.assertions import assertFailureCode
+from documentdb_tests.framework.error_codes import (
+    OPERATION_FAILED_ERROR,
+    UNAUTHORIZED_ERROR,
+    UNRECOGNIZED_COMMAND_FIELD_ERROR,
+)
+from documentdb_tests.framework.executor import execute_admin_command, execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+pytestmark = pytest.mark.admin
+
+
+ERROR_TESTS: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        "unknown_component",
+        command={"getLog": "invalid"},
+        error_code=OPERATION_FAILED_ERROR,
+        msg="Unknown log component name should error",
+    ),
+    DiagnosticTestCase(
+        "deprecated_rs",
+        command={"getLog": "rs"},
+        error_code=OPERATION_FAILED_ERROR,
+        msg="Deprecated 'rs' value should error",
+    ),
+    DiagnosticTestCase(
+        "empty_string",
+        command={"getLog": ""},
+        error_code=OPERATION_FAILED_ERROR,
+        msg="Empty string component should error",
+    ),
+    DiagnosticTestCase(
+        "unrecognized_field",
+        command={"getLog": "global", "unknownField": 1},
+        error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR,
+        msg="Unrecognized command field should error",
+    ),
+    DiagnosticTestCase(
+        "non_admin_database",
+        command={"getLog": "global"},
+        use_admin=False,
+        error_code=UNAUTHORIZED_ERROR,
+        msg="getLog should only run on the admin database",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(ERROR_TESTS))
+def test_getLog_error(collection, test):
+    """Test getLog returns the expected error code for invalid arguments."""
+    if test.use_admin:
+        result = execute_admin_command(collection, test.command)
+    else:
+        result = execute_command(collection, test.command)
+    assertFailureCode(result, test.error_code, msg=test.msg)

documentdb_tests/compatibility/tests/system/diagnostic/commands/getLog/test_getLog_response_structure.py

@@ -0,0 +1,104 @@
+"""Tests for getLog command response structure.
+
+Covers response fields for the "global" filter (totalLinesWritten, log array
+capped at 1024 entries, string log entries, ok), the "startupWarnings" filter
+(totalLinesWritten, log array, ok), and the "*" filter (names array, ok).
+Each test asserts a single response property.
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.system.diagnostic.utils.diagnostic_test_case import (
+    DiagnosticTestCase,
+)
+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 ContainsElement, Eq, Gte, IsType, LenLte
+
+pytestmark = pytest.mark.admin
+
+MAX_LOG_EVENTS = 1024
+
+
+RESPONSE_TESTS: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        "global_totalLinesWritten_number",
+        command={"getLog": "global"},
+        checks={"totalLinesWritten": Gte(0)},
+        msg="global should return a non-negative totalLinesWritten",
+    ),
+    DiagnosticTestCase(
+        "global_log_is_array",
+        command={"getLog": "global"},
+        checks={"log": IsType("array")},
+        msg="global should return a log array",
+    ),
+    DiagnosticTestCase(
+        "global_log_capped_at_1024",
+        command={"getLog": "global"},
+        checks={"log": LenLte(MAX_LOG_EVENTS)},
+        msg="global log array should contain at most 1024 entries",
+    ),
+    DiagnosticTestCase(
+        "global_log_entry_is_string",
+        command={"getLog": "global"},
+        checks={"log.0": IsType("string")},
+        msg="global log entries should be JSON-formatted strings",
+    ),
+    DiagnosticTestCase(
+        "global_ok",
+        command={"getLog": "global"},
+        checks={"ok": Eq(1.0)},
+        msg="global should return ok:1",
+    ),
+    DiagnosticTestCase(
+        "startupWarnings_log_is_array",
+        command={"getLog": "startupWarnings"},
+        checks={"log": IsType("array")},
+        msg="startupWarnings should return a log array",
+    ),
+    DiagnosticTestCase(
+        "startupWarnings_ok",
+        command={"getLog": "startupWarnings"},
+        checks={"ok": Eq(1.0)},
+        msg="startupWarnings should return ok:1",
+    ),
+    DiagnosticTestCase(
+        "startupWarnings_totalLinesWritten_number",
+        command={"getLog": "startupWarnings"},
+        checks={"totalLinesWritten": Gte(0)},
+        msg="startupWarnings should return a non-negative totalLinesWritten",
+    ),
+    DiagnosticTestCase(
+        "wildcard_names_is_array",
+        command={"getLog": "*"},
+        checks={"names": IsType("array")},
+        msg="'*' should return a names array",
+    ),
+    DiagnosticTestCase(
+        "wildcard_names_contains_global",
+        command={"getLog": "*"},
+        checks={"names": ContainsElement("global")},
+        msg="'*' names should include 'global'",
+    ),
+    DiagnosticTestCase(
+        "wildcard_names_contains_startupWarnings",
+        command={"getLog": "*"},
+        checks={"names": ContainsElement("startupWarnings")},
+        msg="'*' names should include 'startupWarnings'",
+    ),
+    DiagnosticTestCase(
+        "wildcard_ok",
+        command={"getLog": "*"},
+        checks={"ok": Eq(1.0)},
+        msg="'*' should return ok:1",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(RESPONSE_TESTS))
+def test_getLog_response_properties(collection, test):
+    """Verify a getLog response field exists and has the expected type or value."""
+    result = execute_admin_command(collection, test.command)
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)

documentdb_tests/framework/property_checks.py

@@ -165,6 +165,25 @@ def __repr__(self) -> str:
         return f"{type(self).__name__}({self.expected!r})"
 
 
+class LenLte(Check):
+    """Assert that the field is a list whose length is at most ``maximum``."""
+
+    def __init__(self, maximum: int) -> None:
+        self.maximum = maximum
+
+    def check(self, value: Any, path: str) -> str | None:
+        if value is _FIELD_ABSENT:
+            return f"expected '{path}' to have length <= {self.maximum}, 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.maximum:
+            return f"expected '{path}' length <= {self.maximum}, got {len(value)}"
+        return None
+
+    def __repr__(self) -> str:
+        return f"{type(self).__name__}({self.maximum!r})"
+
+
 class Contains(Check):
     """Assert that a list contains a dict where ``key`` equals ``value``."""