diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/__init__.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/__init__.py new file mode 100644 index 000000000..49a4f47a0 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/__init__.py @@ -0,0 +1 @@ +"""Tests for the lockInfo diagnostic command.""" diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_argument_validation.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_argument_validation.py new file mode 100644 index 000000000..d4f9504bd --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_argument_validation.py @@ -0,0 +1,63 @@ +"""Tests for lockInfo command argument validation. + +Verifies that lockInfo accepts any BSON type for the command field value. +""" + +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 + +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) + + +INT_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", + ), +] + + +@pytest.mark.parametrize("test", pytest_params(INT_EDGE_TESTS)) +def test_lockInfo_accepts_int_edge_values(collection, test): + """Verify lockInfo accepts zero and negative int values for the command field.""" + result = execute_admin_command(collection, test.command) + assertProperties(result, test.checks, msg=test.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_core_behavior.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_core_behavior.py new file mode 100644 index 000000000..1cb8e68db --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_core_behavior.py @@ -0,0 +1,50 @@ +"""Tests for lockInfo command core behavior. + +Verifies that lockInfo succeeds on admin database, returns a lockInfo array, +and behaves correctly under successive calls. +""" + +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 ( # IsType used in CORE_BEHAVIOR_TESTS + Eq, + IsType, +) + +pytestmark = pytest.mark.admin + + +CORE_BEHAVIOR_TESTS: list[DiagnosticTestCase] = [ + DiagnosticTestCase( + id="returns_ok", + command={"lockInfo": 1}, + checks={"ok": Eq(1.0)}, + msg="lockInfo should return ok:1", + ), + DiagnosticTestCase( + id="returns_lockInfo_array", + command={"lockInfo": 1}, + checks={"lockInfo": IsType("array")}, + msg="lockInfo field should be array", + ), +] + + +@pytest.mark.parametrize("test", pytest_params(CORE_BEHAVIOR_TESTS)) +def test_lockInfo_core_behavior(collection, test): + """Verifies lockInfo returns expected fields on the admin database.""" + 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) diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_errors.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_errors.py new file mode 100644 index 000000000..bcbdc1efe --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_errors.py @@ -0,0 +1,48 @@ +"""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 ( + 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", + ), +] + + +@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) diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_response_structure.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/lockInfo/test_lockInfo_response_structure.py new file mode 100644 index 000000000..42e27eabf --- /dev/null +++ b/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, + )