diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/__init__.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_argument_validation.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_argument_validation.py new file mode 100644 index 00000000..8dcb35b9 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_argument_validation.py @@ -0,0 +1,56 @@ +"""Tests for connectionStatus command argument validation. + +Verifies that connectionStatus accepts all BSON types as the command field +value and returns ok: 1 for each. +""" + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any + +import pytest +from bson import Binary, Code, Decimal128, Int64, MaxKey, MinKey, ObjectId, Regex, Timestamp + +from documentdb_tests.framework.assertions import assertSuccessPartial +from documentdb_tests.framework.executor import execute_admin_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.test_case import BaseTestCase + +pytestmark = pytest.mark.admin + + +@dataclass(frozen=True) +class ConnStatusArgTest(BaseTestCase): + value: Any = None + + +ARG_TYPE_TESTS: list[ConnStatusArgTest] = [ + ConnStatusArgTest("int_1", value=1, msg="int should succeed"), + ConnStatusArgTest("int_0", value=0, msg="int 0 should succeed"), + ConnStatusArgTest("double", value=1.5, msg="double should succeed"), + ConnStatusArgTest("long", value=Int64(1), msg="long should succeed"), + ConnStatusArgTest("decimal128", value=Decimal128("1"), msg="decimal128 should succeed"), + ConnStatusArgTest("string", value="test", msg="string should succeed"), + ConnStatusArgTest("bool_true", value=True, msg="bool true should succeed"), + ConnStatusArgTest("bool_false", value=False, msg="bool false should succeed"), + ConnStatusArgTest( + "date", value=datetime(2024, 1, 1, tzinfo=timezone.utc), msg="date should succeed" + ), + ConnStatusArgTest("null", value=None, msg="null should succeed"), + ConnStatusArgTest("object", value={}, msg="object should succeed"), + ConnStatusArgTest("array", value=[], msg="array should succeed"), + ConnStatusArgTest("binData", value=Binary(b""), msg="binData should succeed"), + ConnStatusArgTest("objectId", value=ObjectId(), msg="objectId should succeed"), + ConnStatusArgTest("regex", value=Regex("test"), msg="regex should succeed"), + ConnStatusArgTest("timestamp", value=Timestamp(0, 0), msg="timestamp should succeed"), + ConnStatusArgTest("minKey", value=MinKey(), msg="minKey should succeed"), + ConnStatusArgTest("maxKey", value=MaxKey(), msg="maxKey should succeed"), + ConnStatusArgTest("code", value=Code("function(){}"), msg="code should succeed"), +] + + +@pytest.mark.parametrize("test", pytest_params(ARG_TYPE_TESTS)) +def test_connectionStatus_accepts_any_type(collection, test): + """Verify connectionStatus succeeds when the command field value is a given BSON type.""" + result = execute_admin_command(collection, {"connectionStatus": test.value}) + assertSuccessPartial(result, {"ok": 1.0}, msg=test.msg) diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_errors.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_errors.py new file mode 100644 index 00000000..2e2af06e --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_errors.py @@ -0,0 +1,126 @@ +"""Tests for connectionStatus command error conditions. + +Verifies that connectionStatus rejects invalid inputs with the correct error +codes. Covers unrecognized fields (UNRECOGNIZED_COMMAND_FIELD_ERROR) and +showPrivileges type-mismatch errors (TYPE_MISMATCH_ERROR) for all non-boolean, +non-numeric BSON types. +""" + +from datetime import datetime, timezone + +import pytest +from bson import Binary, Code, MaxKey, MinKey, ObjectId, Regex, Timestamp + +from documentdb_tests.compatibility.tests.system.diagnostic.utils.diagnostic_test_case import ( + DiagnosticErrorTest, +) +from documentdb_tests.framework.assertions import assertFailureCode +from documentdb_tests.framework.error_codes import ( + TYPE_MISMATCH_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[DiagnosticErrorTest] = [ + DiagnosticErrorTest( + id="unrecognized_field", + command={"connectionStatus": 1, "foo": 1}, + error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="Unrecognized field should error", + ), + DiagnosticErrorTest( + id="showPrivileges_string_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": "yes"}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges string should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_empty_string_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": ""}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges empty string should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_string_false_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": "false"}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges string 'false' should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_array_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": [1]}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges array should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_object_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": {"a": 1}}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges object should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_date_type_mismatch", + command={ + "connectionStatus": 1, + "showPrivileges": datetime(2024, 1, 1, tzinfo=timezone.utc), + }, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges date should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_objectId_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": ObjectId()}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges objectId should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_minKey_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": MinKey()}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges minKey should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_maxKey_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": MaxKey()}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges maxKey should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_binData_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": Binary(b"")}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges binData should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_regex_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": Regex("test")}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges regex should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_timestamp_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": Timestamp(0, 0)}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges timestamp should be rejected as wrong type", + ), + DiagnosticErrorTest( + id="showPrivileges_code_type_mismatch", + command={"connectionStatus": 1, "showPrivileges": Code("function(){}")}, + error_code=TYPE_MISMATCH_ERROR, + msg="showPrivileges code should be rejected as wrong type", + ), +] + + +@pytest.mark.parametrize("test", pytest_params(ERROR_TESTS)) +def test_connectionStatus_error_conditions(collection, test): + """Verify connectionStatus returns the expected error code for an invalid input.""" + 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/connectionStatus/test_connectionStatus_response_validation.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_response_validation.py new file mode 100644 index 00000000..cb9c2e5e --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_response_validation.py @@ -0,0 +1,91 @@ +"""Tests for connectionStatus response structure validation. + +Validates the shape of the connectionStatus response on an unauthenticated +connection: authInfo exists, authenticatedUsers and authenticatedUserRoles +are arrays (and empty without auth), and the uuid field is binData. + +Also verifies cross-database behavior: the command succeeds when run against +a non-existent database and returns identical authInfo regardless of which +database it is executed on. + +Note: Sub-document field validation (user, db, role entries) requires +creating users with specific roles and is not covered here. +""" + +import pytest + +from documentdb_tests.compatibility.tests.system.diagnostic.utils.diagnostic_test_case import ( + DiagnosticPropertyTest, +) +from documentdb_tests.framework.assertions import assertProperties, assertResult +from documentdb_tests.framework.executor import execute_admin_command, execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Eq, Exists, IsType, Len + +pytestmark = pytest.mark.admin + + +RESPONSE_PROPERTY_TESTS: list[DiagnosticPropertyTest] = [ + DiagnosticPropertyTest( + id="authenticatedUsers_is_array", + checks={"authInfo": {"authenticatedUsers": IsType("array")}}, + msg="authenticatedUsers should be array", + ), + DiagnosticPropertyTest( + id="authenticatedUserRoles_is_array", + checks={"authInfo": {"authenticatedUserRoles": IsType("array")}}, + msg="authenticatedUserRoles should be array", + ), + DiagnosticPropertyTest( + id="authInfo_exists", + checks={"authInfo": Exists()}, + msg="authInfo should always exist", + ), + DiagnosticPropertyTest( + id="unauthenticated_users_empty", + checks={"authInfo": {"authenticatedUsers": Len(0)}}, + msg="without auth, authenticatedUsers should be empty", + ), + DiagnosticPropertyTest( + id="unauthenticated_roles_empty", + checks={"authInfo": {"authenticatedUserRoles": Len(0)}}, + msg="without auth, authenticatedUserRoles should be empty", + ), + DiagnosticPropertyTest( + id="uuid_is_binData", + checks={"uuid": IsType("binData")}, + msg="uuid should be a binData (UUID) field", + ), +] + + +@pytest.mark.parametrize("test", pytest_params(RESPONSE_PROPERTY_TESTS)) +def test_connectionStatus_response_properties(collection, test): + """Verify a response field exists and has the expected type.""" + result = execute_admin_command(collection, {"connectionStatus": 1}) + assertProperties(result, test.checks, msg=test.msg, raw_res=True) + + +def test_connectionStatus_succeeds_on_nonexistent_database(collection): + """Verify connectionStatus returns ok and authInfo when run on a non-existent database.""" + other_col = collection.database.client["nonexistent_db_for_connstatus_test"]["dummy"] + result = execute_command(other_col, {"connectionStatus": 1}) + assertProperties( + result, + {"ok": Exists(), "authInfo": Exists()}, + msg="connectionStatus should succeed on non-existent database", + raw_res=True, + ) + + +def test_connectionStatus_same_result_across_databases(collection): + """Verify authInfo is identical whether run on admin or a different database.""" + admin_result = execute_admin_command(collection, {"connectionStatus": 1}) + other_col = collection.database.client["nonexistent_db_for_connstatus_test"]["dummy"] + other_result = execute_command(other_col, {"connectionStatus": 1}) + assertResult( + other_result, + expected={"authInfo": Eq(admin_result["authInfo"])}, + msg="authInfo should be identical across databases", + raw_res=True, + ) diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_show_privileges.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_show_privileges.py new file mode 100644 index 00000000..6440ac7a --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_connectionStatus_show_privileges.py @@ -0,0 +1,93 @@ +"""Tests for connectionStatus showPrivileges parameter. + +Verifies the truthy/falsy/omit behavior of the showPrivileges field. +Truthy values (true, int 1, double 1.0, long 1, decimal128 1) should cause +authenticatedUserPrivileges to appear as an array. Falsy values (false, +int 0, double 0.0, long 0, decimal128 0, null) and omitting the field +entirely should exclude authenticatedUserPrivileges from the response. +""" + +from dataclasses import dataclass +from typing import Any + +import pytest +from bson import Decimal128, Int64 + +from documentdb_tests.compatibility.tests.system.diagnostic.utils.diagnostic_test_case import ( + DiagnosticPropertyTest, +) +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, NotExists +from documentdb_tests.framework.test_case import BaseTestCase + +pytestmark = pytest.mark.admin + + +@dataclass(frozen=True) +class ShowPrivTest(BaseTestCase): + show_priv: Any = None + + +TRUTHY_TESTS: list[ShowPrivTest] = [ + ShowPrivTest("true", show_priv=True, msg="true should show privileges"), + ShowPrivTest("int_1", show_priv=1, msg="int 1 (truthy) should show privileges"), + ShowPrivTest("double_1", show_priv=1.0, msg="double 1.0 (truthy) should show"), + ShowPrivTest("long_1", show_priv=Int64(1), msg="long 1 (truthy) should show"), + ShowPrivTest("decimal128_1", show_priv=Decimal128("1"), msg="decimal128 1 should show"), + ShowPrivTest("int_neg1", show_priv=-1, msg="int -1 (truthy) should show privileges"), + ShowPrivTest("double_neg1", show_priv=-1.0, msg="double -1.0 (truthy) should show"), +] + +FALSY_TESTS: list[ShowPrivTest] = [ + ShowPrivTest("false", show_priv=False, msg="false should hide privileges"), + ShowPrivTest("int_0", show_priv=0, msg="int 0 (falsy) should hide privileges"), + ShowPrivTest("double_0", show_priv=0.0, msg="double 0.0 (falsy) should hide"), + ShowPrivTest("long_0", show_priv=Int64(0), msg="long 0 (falsy) should hide"), + ShowPrivTest("decimal128_0", show_priv=Decimal128("0"), msg="decimal128 0 should hide"), + ShowPrivTest("null", show_priv=None, msg="null (falsy) should hide privileges"), +] + +OMIT_TESTS: list[DiagnosticPropertyTest] = [ + DiagnosticPropertyTest( + id="omit_showPrivileges", + checks={"authInfo": {"authenticatedUserPrivileges": NotExists()}}, + msg="Omitting showPrivileges should not return privileges", + ), +] + + +@pytest.mark.parametrize("test", pytest_params(TRUTHY_TESTS)) +def test_connectionStatus_show_privileges_truthy(collection, test): + """Verify a truthy showPrivileges value includes authenticatedUserPrivileges as an array.""" + result = execute_admin_command( + collection, {"connectionStatus": 1, "showPrivileges": test.show_priv} + ) + assertProperties( + result, + {"authInfo": {"authenticatedUserPrivileges": IsType("array")}}, + msg=test.msg, + raw_res=True, + ) + + +@pytest.mark.parametrize("test", pytest_params(FALSY_TESTS)) +def test_connectionStatus_show_privileges_falsy(collection, test): + """Verify a falsy showPrivileges value excludes authenticatedUserPrivileges.""" + result = execute_admin_command( + collection, {"connectionStatus": 1, "showPrivileges": test.show_priv} + ) + assertProperties( + result, + {"authInfo": {"authenticatedUserPrivileges": NotExists()}}, + msg=test.msg, + raw_res=True, + ) + + +@pytest.mark.parametrize("test", pytest_params(OMIT_TESTS)) +def test_connectionStatus_omit_showPrivileges(collection, test): + """Verify omitting showPrivileges excludes authenticatedUserPrivileges (same as false).""" + result = execute_admin_command(collection, {"connectionStatus": 1}) + assertProperties(result, test.checks, msg=test.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_smoke_connectionStatus.py b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_smoke_connectionStatus.py index 612a9388..2cd7fc2a 100644 --- a/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_smoke_connectionStatus.py +++ b/documentdb_tests/compatibility/tests/system/diagnostic/commands/connectionStatus/test_smoke_connectionStatus.py @@ -1,7 +1,7 @@ """ Smoke test for connectionStatus command. -Tests basic connectionStatus command functionality. +Verifies the command executes successfully and returns ok: 1. """ import pytest @@ -13,7 +13,7 @@ def test_smoke_connectionStatus(collection): - """Test basic connectionStatus command behavior.""" + """Verify connectionStatus executes successfully and returns ok: 1.""" result = execute_admin_command(collection, {"connectionStatus": 1}) expected = {"ok": 1.0} diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/utils/__init__.py b/documentdb_tests/compatibility/tests/system/diagnostic/utils/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/documentdb_tests/compatibility/tests/system/diagnostic/utils/diagnostic_test_case.py b/documentdb_tests/compatibility/tests/system/diagnostic/utils/diagnostic_test_case.py new file mode 100644 index 00000000..10b620e8 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/diagnostic/utils/diagnostic_test_case.py @@ -0,0 +1,30 @@ +"""Shared test cases for diagnostic command tests.""" + +from dataclasses import dataclass, field +from typing import Any, Dict, Optional + +from documentdb_tests.framework.test_case import BaseTestCase + + +@dataclass(frozen=True) +class DiagnosticErrorTest(BaseTestCase): + """Test case for diagnostic command error conditions. + + Attributes: + command: The command document to execute. + use_admin: If True, execute against the admin database. + """ + + command: Optional[Dict[str, Any]] = None + use_admin: bool = True + + +@dataclass(frozen=True) +class DiagnosticPropertyTest(BaseTestCase): + """Test case for diagnostic command response property checks. + + Attributes: + checks: Mapping of dotted field paths to property check objects. + """ + + checks: Dict[str, Any] = field(default_factory=dict)