Add $ping tests

documentdb_tests/compatibility/tests/system/diagnostic/commands/buildInfo/test_buildInfo_consistency.py

@@ -35,11 +35,3 @@ def test_buildInfo_same_result_any_database(collection):
         msg="Should return same result from any database",
         raw_res=True,
     )
-
-
-def test_buildInfo_nonexistent_database(collection):
-    """Test buildInfo 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, {"buildInfo": 1})
-    assertSuccessPartial(result, {"ok": 1.0}, msg="Should succeed on non-existent database")

documentdb_tests/compatibility/tests/system/diagnostic/commands/ping/test_ping_argument_handling.py

@@ -0,0 +1,71 @@
+"""Tests for ping command argument handling.
+
+Validates that ping accepts any BSON type as its command value, as well as
+numeric edge cases (negative int, zero, infinity). The command value does
+not affect behavior — all inputs should return ok: 1.
+"""
+
+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
+
+
+PING_BSON_TYPE_SPECS = [
+    BsonTypeTestCase(
+        id="ping_value",
+        msg="ping command should accept any BSON type as value",
+        valid_types=list(BsonType),
+    ),
+]
+
+ACCEPTANCE_CASES = generate_bson_acceptance_test_cases(PING_BSON_TYPE_SPECS)
+
+
+@pytest.mark.parametrize("bson_type,sample_value,spec", ACCEPTANCE_CASES)
+def test_ping_argument_types(collection, bson_type, sample_value, spec):
+    """Test that ping accepts various BSON types as command value."""
+    result = execute_admin_command(collection, {"ping": sample_value})
+    assertProperties(result, {"ok": Eq(1.0)}, msg=spec.msg, raw_res=True)
+
+
+NUMERIC_EDGE_CASES: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        id="negative_int",
+        command={"ping": -1},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept negative int",
+    ),
+    DiagnosticTestCase(
+        id="int_0",
+        command={"ping": 0},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept int 0",
+    ),
+    DiagnosticTestCase(
+        id="infinity",
+        command={"ping": FLOAT_INFINITY},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept infinity",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(NUMERIC_EDGE_CASES))
+def test_ping_argument_numeric_edge_cases(collection, test):
+    """Test that ping accepts edge-case numeric values as command value."""
+    result = execute_admin_command(collection, test.command)
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)

documentdb_tests/compatibility/tests/system/diagnostic/commands/ping/test_ping_core_behavior.py

@@ -0,0 +1,64 @@
+"""Tests for ping command core behavior.
+
+Validates that ping succeeds on both admin and non-admin databases,
+can be executed repeatedly in succession, and works after other commands
+or write activity.
+"""
+
+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, execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+from documentdb_tests.framework.property_checks import Eq
+
+pytestmark = pytest.mark.admin
+
+
+DATABASE_TESTS: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        id="admin_database",
+        command={"ping": 1},
+        use_admin=True,
+        checks={"ok": Eq(1.0)},
+        msg="Should succeed on admin database",
+    ),
+    DiagnosticTestCase(
+        id="non_admin_database",
+        command={"ping": 1},
+        use_admin=False,
+        checks={"ok": Eq(1.0)},
+        msg="Should succeed on non-admin database",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(DATABASE_TESTS))
+def test_ping_on_database(collection, test):
+    """Test ping returns ok:1 on both admin and non-admin databases."""
+    if test.use_admin:
+        result = execute_admin_command(collection, test.command)
+    else:
+        result = execute_command(collection, test.command)
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)
+
+
+def test_ping_multiple_times_in_succession(collection):
+    """Test ping can be executed multiple times in succession."""
+    for _ in range(3):
+        result = execute_admin_command(collection, {"ping": 1})
+        assertProperties(
+            result, {"ok": Eq(1.0)}, msg="Should succeed on repeated execution", raw_res=True
+        )
+
+
+def test_ping_after_write_activity(collection):
+    """Test ping returns ok:1 immediately after write activity."""
+    collection.insert_many([{"_id": i, "data": "x" * 100} for i in range(100)])
+    result = execute_admin_command(collection, {"ping": 1})
+    assertProperties(
+        result, {"ok": Eq(1.0)}, msg="Should succeed after write activity", raw_res=True
+    )

documentdb_tests/compatibility/tests/system/diagnostic/commands/ping/test_ping_error_conditions.py

@@ -0,0 +1,54 @@
+"""Tests for ping command error conditions.
+
+Validates that invalid usages of ping produce appropriate errors.
+"""
+
+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,
+    UNKNOWN_PIPELINE_STAGE_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="unrecognized_field",
+        command={"ping": 1, "unknownField": "test"},
+        error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR,
+        msg="Should reject unrecognized fields",
+    ),
+    DiagnosticTestCase(
+        id="case_sensitive",
+        command={"Ping": 1},
+        error_code=COMMAND_NOT_FOUND_ERROR,
+        msg="Case-mismatched command name should fail",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(ERROR_TESTS))
+def test_ping_error_conditions(collection, test):
+    """Verifies ping rejects invalid usages with appropriate error codes."""
+    result = execute_admin_command(collection, test.command)
+    assertFailureCode(result, test.error_code, msg=test.msg)
+
+
+def test_ping_as_pipeline_stage(collection):
+    """Test that $ping is not a valid aggregation pipeline stage."""
+    result = execute_command(
+        collection,
+        {"aggregate": collection.name, "pipeline": [{"$ping": 1}], "cursor": {}},
+    )
+    assertFailureCode(
+        result, UNKNOWN_PIPELINE_STAGE_ERROR, msg="ping should not be a valid pipeline stage"
+    )

documentdb_tests/compatibility/tests/system/diagnostic/commands/ping/test_ping_response_structure.py

@@ -0,0 +1,37 @@
+"""Tests for ping command response structure.
+
+Validates the response fields and their types.
+"""
+
+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
+
+
+RESPONSE_TESTS: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        id="ok_field_value",
+        checks={"ok": Eq(1.0)},
+        msg="'ok' field should be 1.0",
+    ),
+    DiagnosticTestCase(
+        id="ok_field_type",
+        checks={"ok": IsType("double")},
+        msg="'ok' field should be a double",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(RESPONSE_TESTS))
+def test_ping_response_properties(collection, test):
+    """Verifies ping response fields have expected types and values."""
+    result = execute_admin_command(collection, {"ping": 1})
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)