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 @@ -38,10 +38,8 @@ def test_buildInfo_same_result_any_database(collection):


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",
)
"""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")
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
"""Tests for listCommands command argument handling.

Validates that listCommands accepts any BSON type as its argument value
and rejects unrecognized fields.
"""

import pytest

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.property_checks import Eq

pytestmark = pytest.mark.admin


# listCommands accepts all BSON types as its argument value
LISTCOMMANDS_BSON_TYPE_PARAMS = [
BsonTypeTestCase(
id="argument",
msg="listCommands should accept any BSON type as argument",
keyword="listCommands",
valid_types=[
BsonType.DOUBLE,
BsonType.STRING,
BsonType.OBJECT,
BsonType.ARRAY,
BsonType.BIN_DATA,
BsonType.OBJECT_ID,
BsonType.BOOL,
BsonType.DATE,
BsonType.NULL,
BsonType.REGEX,
BsonType.JAVASCRIPT,
BsonType.INT,
BsonType.TIMESTAMP,
BsonType.LONG,
BsonType.DECIMAL,
BsonType.MIN_KEY,
BsonType.MAX_KEY,
],
),
]

ACCEPTANCE_TESTS = generate_bson_acceptance_test_cases(LISTCOMMANDS_BSON_TYPE_PARAMS)


@pytest.mark.parametrize("bson_type,sample_value,spec", ACCEPTANCE_TESTS)
def test_listCommands_argument_types(collection, bson_type, sample_value, spec):
"""Test that listCommands accepts any BSON type as argument value."""
result = execute_admin_command(collection, {"listCommands": sample_value})
assertProperties(result, {"ok": Eq(1.0)}, msg=f"{spec.msg} - {bson_type.value}", raw_res=True)


def test_listCommands_extra_fields_ignored(collection):
"""Test that listCommands ignores extra unrecognized fields."""
result = execute_admin_command(collection, {"listCommands": 1, "unknownField": 1})
assertProperties(
result, {"ok": Eq(1.0)}, msg="Should succeed even with extra fields", raw_res=True
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
"""Tests for listCommands command consistency and availability.

Validates that listCommands works across databases and returns consistent results.
"""

import pytest

from documentdb_tests.compatibility.tests.system.diagnostic.utils.diagnostic_test_case import (
DiagnosticTestCase,
)
from documentdb_tests.framework.assertions import (
assertProperties,
assertSuccess,
assertSuccessPartial,
)
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={"listCommands": 1},
use_admin=True,
checks={"ok": Eq(1.0)},
msg="Should succeed on admin database",
),
DiagnosticTestCase(
id="non_admin_database",
command={"listCommands": 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_listCommands_database_availability(collection, test):
"""Test listCommands works 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_listCommands_nonexistent_database(collection):
"""Test listCommands 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, {"listCommands": 1})
assertSuccessPartial(result, {"ok": 1.0}, msg="Should succeed on non-existent database")


def test_listCommands_idempotent(collection):
"""Test calling listCommands multiple times returns identical results."""
result1 = execute_admin_command(collection, {"listCommands": 1})
result2 = execute_admin_command(collection, {"listCommands": 1})
assertSuccess(result2, expected=result1, msg="Should return identical results", raw_res=True)


def test_listCommands_same_result_any_database(collection):
"""Test listCommands returns same result from admin and non-admin database."""
admin_result = execute_admin_command(collection, {"listCommands": 1})
db_result = execute_command(collection, {"listCommands": 1})
assertSuccess(
db_result,
expected=admin_result,
msg="Should return same result from any database",
raw_res=True,
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
"""Tests for listCommands command error conditions.

Validates that invalid usages of listCommands 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,
)
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="as_aggregation_stage",
command={"aggregate": "test", "pipeline": [{"$listCommands": {}}], "cursor": {}},
use_admin=False,
error_code=UNKNOWN_PIPELINE_STAGE_ERROR,
msg="$listCommands is not a valid aggregation stage",
),
DiagnosticTestCase(
id="case_sensitive",
command={"ListCommands": 1},
use_admin=True,
error_code=COMMAND_NOT_FOUND_ERROR,
msg="Case-mismatched command name should fail",
),
]


@pytest.mark.parametrize("test", pytest_params(ERROR_TESTS))
def test_listCommands_error_conditions(collection, test):
"""Verifies listCommands rejects invalid usages with appropriate error codes."""
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)
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
"""Tests for listCommands command response structure.

Validates the response format, field types, and presence of expected commands.
"""

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, Exists, IsType

pytestmark = pytest.mark.admin


RESPONSE_FIELD_TESTS: list[DiagnosticTestCase] = [
DiagnosticTestCase(
id="ok_is_1",
checks={"ok": Eq(1.0)},
msg="'ok' field should be 1.0",
),
DiagnosticTestCase(
id="commands_is_object",
checks={"commands": IsType("object")},
msg="'commands' field should be an object",
),
]


@pytest.mark.parametrize("test", pytest_params(RESPONSE_FIELD_TESTS))
def test_listCommands_response_fields(collection, test):
"""Verify listCommands response contains expected fields with correct types."""
result = execute_admin_command(collection, {"listCommands": 1})
assertProperties(result, test.checks, msg=test.msg, raw_res=True)


COMMAND_ENTRY_STRUCTURE_TESTS: list[DiagnosticTestCase] = [
DiagnosticTestCase(
id="find_has_help",
checks={"commands.find.help": IsType("string")},
msg="Command entry 'find' should have 'help' field of type string",
),
DiagnosticTestCase(
id="find_has_adminOnly",
checks={"commands.find.adminOnly": IsType("bool")},
msg="Command entry 'find' should have 'adminOnly' field of type boolean",
),
DiagnosticTestCase(
id="find_has_requiresAuth",
checks={"commands.find.requiresAuth": IsType("bool")},
msg="Command entry 'find' should have 'requiresAuth' field of type boolean",
),
DiagnosticTestCase(
id="find_has_secondaryOk",
checks={"commands.find.secondaryOk": IsType("bool")},
msg="Command entry 'find' should have 'secondaryOk' field of type boolean",
),
DiagnosticTestCase(
id="find_has_apiVersions",
checks={"commands.find.apiVersions": IsType("array")},
msg="Command entry 'find' should have 'apiVersions' field of type array",
),
DiagnosticTestCase(
id="find_has_deprecatedApiVersions",
checks={"commands.find.deprecatedApiVersions": IsType("array")},
msg="Command entry 'find' should have 'deprecatedApiVersions' field of type array",
),
]


@pytest.mark.parametrize("test", pytest_params(COMMAND_ENTRY_STRUCTURE_TESTS))
def test_listCommands_command_entry_structure(collection, test):
"""Verify each command entry contains expected fields with correct types."""
result = execute_admin_command(collection, {"listCommands": 1})
assertProperties(result, test.checks, msg=test.msg, raw_res=True)


KNOWN_COMMANDS_TESTS: list[DiagnosticTestCase] = [
DiagnosticTestCase(
id="find_present",
checks={"commands.find": Exists()},
msg="'find' should be listed",
),
DiagnosticTestCase(
id="insert_present",
checks={"commands.insert": Exists()},
msg="'insert' should be listed",
),
DiagnosticTestCase(
id="update_present",
checks={"commands.update": Exists()},
msg="'update' should be listed",
),
DiagnosticTestCase(
id="delete_present",
checks={"commands.delete": Exists()},
msg="'delete' should be listed",
),
DiagnosticTestCase(
id="aggregate_present",
checks={"commands.aggregate": Exists()},
msg="'aggregate' should be listed",
),
DiagnosticTestCase(
id="ping_present",
checks={"commands.ping": Exists()},
msg="'ping' should be listed",
),
]


@pytest.mark.parametrize("test", pytest_params(KNOWN_COMMANDS_TESTS))
def test_listCommands_known_commands_present(collection, test):
"""Verify well-known commands appear in the listCommands response."""
result = execute_admin_command(collection, {"listCommands": 1})
assertProperties(result, test.checks, msg=test.msg, raw_res=True)
Loading