Add $connPoolStats tests (documentdb#242)

Signed-off-by: PatersonProjects <keldonhoff@gmail.com>

Author: PatersonProjects

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

@@ -0,0 +1,50 @@
+"""Tests for connPoolStats command argument validation.
+
+Verifies that connPoolStats accepts all BSON types for the command field
+value and ignores unrecognized fields.
+"""
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertSuccessPartial
+from documentdb_tests.framework.bson_type_validator import (
+    BsonTypeTestCase,
+    generate_bson_acceptance_test_cases,
+)
+from documentdb_tests.framework.executor import execute_admin_command
+from documentdb_tests.framework.test_constants import BsonType
+
+pytestmark = pytest.mark.admin
+
+# connPoolStats ignores the command field value — all types should succeed
+CONNPOOLSTATS_PARAMS = [
+    BsonTypeTestCase(
+        id="connPoolStats_value",
+        msg="connPoolStats should accept all BSON types",
+        keyword="connPoolStats",
+        valid_types=list(BsonType),
+    ),
+]
+
+ACCEPTANCE_CASES = generate_bson_acceptance_test_cases(CONNPOOLSTATS_PARAMS)
+
+
+@pytest.mark.parametrize("bson_type,sample_value,spec", ACCEPTANCE_CASES)
+def test_connPoolStats_accepts_any_type(collection, bson_type, sample_value, spec):
+    """Test connPoolStats accepts all BSON types for command field value."""
+    result = execute_admin_command(collection, {"connPoolStats": sample_value})
+    assertSuccessPartial(result, {"ok": 1.0}, msg=f"connPoolStats should accept {bson_type.value}")
+
+
+def test_connPoolStats_unrecognized_field(collection):
+    """Test connPoolStats with unrecognized extra field succeeds."""
+    result = execute_admin_command(collection, {"connPoolStats": 1, "foo": 1})
+    assertSuccessPartial(result, {"ok": 1.0}, msg="Unrecognized field should be ignored")
+
+
+def test_connPoolStats_multiple_unrecognized_fields(collection):
+    """Test connPoolStats with multiple unrecognized fields succeeds."""
+    result = execute_admin_command(
+        collection, {"connPoolStats": 1, "foo": 1, "bar": "baz", "qux": []}
+    )
+    assertSuccessPartial(result, {"ok": 1.0}, msg="Multiple unrecognized fields should be ignored")

documentdb_tests/compatibility/tests/system/diagnostic/commands/connPoolStats/test_connPoolStats_argument_validation.py

@@ -0,0 +1,60 @@
+"""Tests for connPoolStats consistency and database context.
+
+Verifies connection count invariants (totalCreated is non-decreasing,
+totalCreated >= totalInUse + totalAvailable), repeated call stability,
+and that the command succeeds regardless of database context.
+"""
+
+import pytest
+
+from documentdb_tests.framework.assertions import (
+    assertResult,
+    assertSuccessPartial,
+)
+from documentdb_tests.framework.executor import execute_admin_command, execute_command
+from documentdb_tests.framework.property_checks import Gte
+from documentdb_tests.framework.test_constants import FLOAT_INFINITY
+
+pytestmark = pytest.mark.admin
+
+
+def test_connPoolStats_repeated_calls(collection):
+    """Verify connPoolStats can be called repeatedly without error."""
+    for _ in range(5):
+        result = execute_admin_command(collection, {"connPoolStats": 1})
+        assertSuccessPartial(result, {"ok": 1.0}, msg="Repeated call should succeed")
+
+
+def test_connPoolStats_totalCreated_non_decreasing(collection):
+    """Verify totalCreated is monotonically non-decreasing across calls."""
+    r1 = execute_admin_command(collection, {"connPoolStats": 1})
+    created1 = r1.get("totalCreated", FLOAT_INFINITY)
+    r2 = execute_admin_command(collection, {"connPoolStats": 1})
+    assertResult(
+        r2,
+        expected={"totalCreated": Gte(created1)},
+        raw_res=True,
+        msg="totalCreated should not decrease",
+    )
+
+
+def test_connPoolStats_totalCreated_gte_inUse_plus_available(collection):
+    """Verify totalCreated >= totalInUse + totalAvailable."""
+    result = execute_admin_command(collection, {"connPoolStats": 1})
+    in_use = result.get("totalInUse", FLOAT_INFINITY)
+    available = result.get("totalAvailable", FLOAT_INFINITY)
+    minimum = in_use + available
+    assertResult(
+        result,
+        expected={"totalCreated": Gte(minimum)},
+        raw_res=True,
+        msg="totalCreated should be >= totalInUse + totalAvailable",
+    )
+
+
+def test_connPoolStats_succeeds_on_nonexistent_database(collection):
+    """Verify connPoolStats succeeds when run on a non-existent database."""
+    other_db = f"{collection.name}_nonexistent_db"
+    other_col = collection.database.client[other_db][collection.name]
+    result = execute_command(other_col, {"connPoolStats": 1})
+    assertSuccessPartial(result, {"ok": 1.0}, msg="Should succeed on non-existent database")

documentdb_tests/compatibility/tests/system/diagnostic/commands/connPoolStats/test_connPoolStats_consistency.py

@@ -0,0 +1,67 @@
+"""Tests for connPoolStats command core behavior.
+
+Verifies that each response field exists and has the expected type or 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.executor import execute_admin_command
+from documentdb_tests.framework.parametrize import pytest_params
+from documentdb_tests.framework.property_checks import Gte, IsType
+
+pytestmark = pytest.mark.admin
+
+
+RESPONSE_PROPERTY_TESTS: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        id="totalInUse_gte_0",
+        checks={"totalInUse": Gte(0)},
+        msg="totalInUse should be >= 0",
+    ),
+    DiagnosticTestCase(
+        id="totalAvailable_gte_0",
+        checks={"totalAvailable": Gte(0)},
+        msg="totalAvailable should be >= 0",
+    ),
+    DiagnosticTestCase(
+        id="totalCreated_gte_0",
+        checks={"totalCreated": Gte(0)},
+        msg="totalCreated should be >= 0",
+    ),
+    DiagnosticTestCase(
+        id="totalRefreshing_gte_0",
+        checks={"totalRefreshing": Gte(0)},
+        msg="totalRefreshing should be >= 0",
+    ),
+    DiagnosticTestCase(
+        id="pools_is_object",
+        checks={"pools": IsType("object")},
+        msg="pools should be a document",
+    ),
+    DiagnosticTestCase(
+        id="hosts_is_object",
+        checks={"hosts": IsType("object")},
+        msg="hosts should be a document",
+    ),
+    DiagnosticTestCase(
+        id="numClientConnections_gte_0",
+        checks={"numClientConnections": Gte(0)},
+        msg="numClientConnections should be >= 0",
+    ),
+    DiagnosticTestCase(
+        id="numAScopedConnections_gte_0",
+        checks={"numAScopedConnections": Gte(0)},
+        msg="numAScopedConnections should be >= 0",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(RESPONSE_PROPERTY_TESTS))
+def test_connPoolStats_response_properties(collection, test):
+    """Verify a connPoolStats response field exists and has the expected type or value."""
+    result = execute_admin_command(collection, {"connPoolStats": 1})
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)