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
Expand Up @@ -35,13 +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 regardless of database context."""
result = execute_admin_command(collection, {"buildInfo": 1})
assertSuccessPartial(
result,
{"ok": 1.0},
msg="Should succeed regardless of database existence",
)
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,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For consistency with buildInfo and listCommands tests, consider adding a test for ping on a nonexistent database (e.g., collection.database.client["nonexistent_db"]["coll"]).

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In investigating this, I have learned that the basic test would be functionally identical to this case, since collection generates a new database reference which acts identical to the ["nonexistent_db"] name until any data is inserted. As such I have rather than adding this case removed the duplicate buildInfo case for consistency.

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,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] = [
Comment thread
PatersonProjects marked this conversation as resolved.
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"
)
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