Add $connectionStatus tests

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)

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)

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,
+    )

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)

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}