Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -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)
Original file line number Diff line number Diff line change
@@ -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
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
"""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,
UNRECOGNIZED_COMMAND_FIELD_ERROR,
)
from documentdb_tests.framework.executor import execute_admin_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)
Original file line number Diff line number Diff line change
@@ -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)
Loading