Add lockInfo tests

documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/__init__.py

@@ -0,0 +1 @@
+"""Tests for the lockInfo diagnostic command."""

documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_argument_validation.py

@@ -0,0 +1,78 @@
+"""Tests for lockInfo command argument validation.
+
+Verifies that lockInfo accepts any BSON type for the command field value and
+that successive calls succeed (point-in-time snapshot behaviour).
+"""
+
+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.bson_type_validator import (
+    BsonType,
+    BsonTypeTestCase,
+    generate_bson_acceptance_test_cases,
+)
+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
+from documentdb_tests.framework.test_constants import FLOAT_INFINITY
+
+pytestmark = pytest.mark.admin
+
+
+LOCKINFO_BSON_TYPE_SPECS = [
+    BsonTypeTestCase(
+        id="command_field",
+        msg="lockInfo accepts any BSON type for the command field",
+        keyword="lockInfo",
+        valid_types=list(BsonType),
+    )
+]
+
+ACCEPTANCE_TESTS = generate_bson_acceptance_test_cases(LOCKINFO_BSON_TYPE_SPECS)
+
+
+@pytest.mark.parametrize("bson_type,sample_value,spec", ACCEPTANCE_TESTS)
+def test_lockInfo_accepts_any_type(collection, bson_type, sample_value, spec):
+    """Verify lockInfo succeeds when the command field value is a given BSON type."""
+    result = execute_admin_command(collection, {"lockInfo": sample_value})
+    assertProperties(result, {"ok": Eq(1.0)}, msg=spec.msg, raw_res=True)
+
+
+VALUE_EDGE_TESTS: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        id="int_zero",
+        command={"lockInfo": 0},
+        checks={"ok": Eq(1.0)},
+        msg="lockInfo should accept int 0",
+    ),
+    DiagnosticTestCase(
+        id="int_neg1",
+        command={"lockInfo": -1},
+        checks={"ok": Eq(1.0)},
+        msg="lockInfo should accept int -1",
+    ),
+    DiagnosticTestCase(
+        id="float_infinity",
+        command={"lockInfo": FLOAT_INFINITY},
+        checks={"ok": Eq(1.0)},
+        msg="lockInfo should accept float infinity as the command field value",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(VALUE_EDGE_TESTS))
+def test_lockInfo_accepts_value_edge_cases(collection, test):
+    """Verify lockInfo accepts int and float edge values for the command field."""
+    result = execute_admin_command(collection, test.command)
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)
+
+
+def test_lockInfo_point_in_time_snapshot(collection):
+    """Test lockInfo result is a point-in-time snapshot (successive calls succeed)."""
+    execute_admin_command(collection, {"lockInfo": 1})
+    result = execute_admin_command(collection, {"lockInfo": 1})
+    assertProperties(result, {"ok": Eq(1.0)}, msg="Successive calls should succeed", raw_res=True)

documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_errors.py

@@ -0,0 +1,56 @@
+"""Tests for lockInfo command error cases.
+
+Verifies that lockInfo returns correct error codes when run on non-admin
+database or with unrecognized fields.
+"""
+
+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 (
+    COMMAND_NOT_FOUND_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(
+        id="non_admin_database",
+        command={"lockInfo": 1},
+        use_admin=False,
+        error_code=UNAUTHORIZED_ERROR,
+        msg="lockInfo on non-admin db should be unauthorized",
+    ),
+    DiagnosticTestCase(
+        id="unrecognized_field",
+        command={"lockInfo": 1, "foo": 1},
+        use_admin=True,
+        error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR,
+        msg="Unrecognized field should error",
+    ),
+    DiagnosticTestCase(
+        id="wrong_case",
+        command={"LockInfo": 1},
+        use_admin=True,
+        error_code=COMMAND_NOT_FOUND_ERROR,
+        msg="Command name is case-sensitive; 'LockInfo' should not be found",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(ERROR_TESTS))
+def test_lockInfo_error_conditions(collection, test):
+    """Verifies lockInfo returns appropriate error codes for invalid usages."""
+    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/lockInfo/test_lockInfo_response_structure.py

@@ -0,0 +1,64 @@
+"""Tests for lockInfo command response structure.
+
+Verifies the top-level structure of the lockInfo response: ok field and
+lockInfo array field. Also verifies the structure of individual lock entries
+using an fsync lock to guarantee a non-empty snapshot.
+"""
+
+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 Eq, IsType
+
+pytestmark = [pytest.mark.admin, pytest.mark.no_parallel]
+
+
+RESPONSE_STRUCTURE_TESTS: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        "ok_field",
+        checks={"ok": Eq(1.0)},
+        msg="Response should contain ok: 1.0",
+    ),
+    DiagnosticTestCase(
+        "lockInfo_is_array",
+        checks={"lockInfo": IsType("array")},
+        msg="lockInfo field should be an array",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(RESPONSE_STRUCTURE_TESTS))
+def test_lockInfo_response_structure(collection, test):
+    """Verify lockInfo response contains expected fields with correct types."""
+    result = execute_admin_command(collection, {"lockInfo": 1})
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)
+
+
+def test_lockInfo_entry_structure(collection):
+    """Test a lock entry exposes resourceId (string), granted (array), and pending (array).
+
+    Acquires a global fsync lock so the lock manager reports at least one entry,
+    guaranteeing the assertion runs against a real lock entry rather than an
+    empty snapshot. Must not run in parallel — fsyncLock is a global server lock.
+    """
+    execute_admin_command(collection, {"fsync": 1, "lock": True})
+    try:
+        result = execute_admin_command(collection, {"lockInfo": 1})
+        entry = result["lockInfo"][0]
+    finally:
+        execute_admin_command(collection, {"fsyncUnlock": 1})
+    assertProperties(
+        entry,
+        {
+            "resourceId": IsType("string"),
+            "granted": IsType("array"),
+            "pending": IsType("array"),
+        },
+        msg="lock entry should expose resourceId (string), granted (array), pending (array)",
+        raw_res=True,
+    )