From 5fbb2965e8415193d0dc4a6d5a5e3a90524d5816 Mon Sep 17 00:00:00 2001 From: Ian Forster Date: Fri, 19 Jun 2026 10:27:20 -0700 Subject: [PATCH] Add dropConnections administration command tests Signed-off-by: Ian Forster --- .../commands/dropConnections/__init__.py | 0 .../test_dropConnections_argument_handling.py | 197 ++++++++++++++++++ .../test_dropConnections_comment.py | 43 ++++ .../test_dropConnections_core_behavior.py | 160 ++++++++++++++ 4 files changed, 400 insertions(+) create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/__init__.py create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_argument_handling.py create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_comment.py create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_core_behavior.py diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/__init__.py b/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_argument_handling.py b/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_argument_handling.py new file mode 100644 index 000000000..eb2941991 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_argument_handling.py @@ -0,0 +1,197 @@ +"""Tests for dropConnections command argument validation. + +Verifies type rejection for the hostAndPort field, missing field errors, +array element type handling, command field value variations, and +unrecognized field rejection. +""" + +from dataclasses import dataclass +from typing import Any + +import pytest + +from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial +from documentdb_tests.framework.bson_type_validator import ( + BsonTypeTestCase, + generate_bson_acceptance_test_cases, + generate_bson_rejection_test_cases, +) +from documentdb_tests.framework.error_codes import ( + MISSING_FIELD_ERROR, + TYPE_MISMATCH_ERROR, + UNRECOGNIZED_COMMAND_FIELD_ERROR, +) +from documentdb_tests.framework.executor import execute_admin_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.test_case import BaseTestCase +from documentdb_tests.framework.test_constants import BsonType + +pytestmark = [pytest.mark.admin, pytest.mark.no_parallel] + + +# --- hostAndPort field type rejection --- + +# Property [hostAndPort Type Rejection]: dropConnections rejects non-array +# BSON types for the hostAndPort field. +_HOST_AND_PORT_TYPE_SPEC = [ + BsonTypeTestCase( + id="hostAndPort", + msg="dropConnections should reject non-array types for hostAndPort", + keyword="hostAndPort", + valid_types=[BsonType.ARRAY], + default_error_code=TYPE_MISMATCH_ERROR, + skip_rejection_types=[BsonType.NULL], + ), +] + +_REJECTION_CASES = generate_bson_rejection_test_cases(_HOST_AND_PORT_TYPE_SPEC) + + +@pytest.mark.parametrize("bson_type,sample_value,spec", _REJECTION_CASES) +def test_dropConnections_hostAndPort_rejects_type(collection, bson_type, sample_value, spec): + """Test dropConnections rejects non-array types for hostAndPort.""" + result = execute_admin_command(collection, {"dropConnections": 1, "hostAndPort": sample_value}) + assertFailureCode( + result, + spec.expected_code(bson_type), + msg=f"dropConnections should reject {bson_type.value} for hostAndPort.", + ) + + +# --- hostAndPort missing field --- + + +def test_dropConnections_missing_hostAndPort(collection): + """Test dropConnections rejects command without hostAndPort field.""" + result = execute_admin_command(collection, {"dropConnections": 1}) + assertFailureCode( + result, + MISSING_FIELD_ERROR, + msg="dropConnections should require hostAndPort field.", + ) + + +def test_dropConnections_null_hostAndPort(collection): + """Test dropConnections rejects null for hostAndPort field.""" + result = execute_admin_command(collection, {"dropConnections": 1, "hostAndPort": None}) + assertFailureCode( + result, + MISSING_FIELD_ERROR, + msg="dropConnections should treat null hostAndPort as missing field.", + ) + + +# --- hostAndPort array element type validation --- + + +@dataclass(frozen=True) +class ElementTypeTestCase(BaseTestCase): + """Test case for hostAndPort array element type checking.""" + + element: Any = None + + +# Property [Array Element Types]: dropConnections rejects non-string array +# elements with TYPE_MISMATCH_ERROR, except null which is silently ignored. +_ELEMENT_REJECTION_TESTS: list[ElementTypeTestCase] = [ + ElementTypeTestCase( + "int_element", + element=1, + error_code=TYPE_MISMATCH_ERROR, + msg="dropConnections should reject int element in hostAndPort.", + ), + ElementTypeTestCase( + "bool_element", + element=True, + error_code=TYPE_MISMATCH_ERROR, + msg="dropConnections should reject bool element in hostAndPort.", + ), + ElementTypeTestCase( + "object_element", + element={"host": "x"}, + error_code=TYPE_MISMATCH_ERROR, + msg="dropConnections should reject object element in hostAndPort.", + ), + ElementTypeTestCase( + "nested_array_element", + element=[], + error_code=TYPE_MISMATCH_ERROR, + msg="dropConnections should reject nested array element in hostAndPort.", + ), + ElementTypeTestCase( + "double_element", + element=3.14, + error_code=TYPE_MISMATCH_ERROR, + msg="dropConnections should reject double element in hostAndPort.", + ), +] + +_ELEMENT_IGNORE_TESTS: list[ElementTypeTestCase] = [ + ElementTypeTestCase( + "null_element_ignored", + element=None, + expected={"ok": 1.0}, + msg="dropConnections should silently ignore null element in hostAndPort.", + ), +] + + +@pytest.mark.parametrize("test", pytest_params(_ELEMENT_REJECTION_TESTS)) +def test_dropConnections_hostAndPort_element_rejects_type(collection, test): + """Test dropConnections rejects non-string, non-null array elements.""" + result = execute_admin_command( + collection, {"dropConnections": 1, "hostAndPort": [test.element]} + ) + assertFailureCode(result, test.error_code, msg=test.msg) + + +@pytest.mark.parametrize("test", pytest_params(_ELEMENT_IGNORE_TESTS)) +def test_dropConnections_hostAndPort_element_ignored(collection, test): + """Test dropConnections silently ignores null array elements.""" + result = execute_admin_command( + collection, {"dropConnections": 1, "hostAndPort": [test.element]} + ) + assertSuccessPartial(result, test.expected, msg=test.msg) + + +# --- Command field value variations --- + +# Property [Command Field Acceptance]: dropConnections accepts various BSON +# types for the command field value (the value of the "dropConnections" key). +_COMMAND_FIELD_PARAMS = [ + BsonTypeTestCase( + id="dropConnections_value", + msg="dropConnections should accept all BSON types for command field value", + keyword="dropConnections", + valid_types=list(BsonType), + ), +] + +_COMMAND_FIELD_ACCEPTANCE = generate_bson_acceptance_test_cases(_COMMAND_FIELD_PARAMS) + + +@pytest.mark.parametrize("bson_type,sample_value,spec", _COMMAND_FIELD_ACCEPTANCE) +def test_dropConnections_command_field_accepts_type(collection, bson_type, sample_value, spec): + """Test dropConnections accepts any BSON type for command field value.""" + result = execute_admin_command(collection, {"dropConnections": sample_value, "hostAndPort": []}) + assertSuccessPartial( + result, + {"ok": 1.0}, + msg=f"dropConnections should accept {bson_type.value} for command field value.", + ) + + +# --- Unrecognized field rejection --- + + +def test_dropConnections_unrecognized_field(collection): + """Test dropConnections rejects unrecognized top-level fields.""" + result = execute_admin_command( + collection, + {"dropConnections": 1, "hostAndPort": [], "unknownField": 1}, + ) + assertFailureCode( + result, + UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="dropConnections should reject unrecognized fields.", + ) diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_comment.py b/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_comment.py new file mode 100644 index 000000000..7c9129a5d --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_comment.py @@ -0,0 +1,43 @@ +"""Tests for dropConnections command comment field acceptance. + +Verifies that the optional comment field accepts any valid BSON type. +""" + +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, pytest.mark.no_parallel] + +# Property [Comment Acceptance]: dropConnections accepts any BSON type for the +# optional comment field. +_COMMENT_PARAMS = [ + BsonTypeTestCase( + id="comment", + msg="dropConnections should accept all BSON types for comment field", + keyword="comment", + valid_types=list(BsonType), + ), +] + +_COMMENT_ACCEPTANCE_CASES = generate_bson_acceptance_test_cases(_COMMENT_PARAMS) + + +@pytest.mark.parametrize("bson_type,sample_value,spec", _COMMENT_ACCEPTANCE_CASES) +def test_dropConnections_comment_accepts_type(collection, bson_type, sample_value, spec): + """Test dropConnections accepts any BSON type for comment field.""" + result = execute_admin_command( + collection, + {"dropConnections": 1, "hostAndPort": [], "comment": sample_value}, + ) + assertSuccessPartial( + result, + {"ok": 1.0}, + msg=f"dropConnections should accept {bson_type.value} for comment field.", + ) diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_core_behavior.py b/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_core_behavior.py new file mode 100644 index 000000000..45a9be832 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/administration/commands/dropConnections/test_dropConnections_core_behavior.py @@ -0,0 +1,160 @@ +"""Tests for dropConnections command core behavior. + +Verifies that dropConnections succeeds with valid hostAndPort values, +silently ignores elements without a colon (no port separator), and +rejects entries with invalid host:port parsing. +""" + +from dataclasses import dataclass +from typing import Any + +import pytest + +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.error_codes import FAILED_TO_PARSE_ERROR +from documentdb_tests.framework.executor import execute_admin_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.test_case import BaseTestCase + +pytestmark = [pytest.mark.admin, pytest.mark.no_parallel] + + +@dataclass(frozen=True) +class DropConnectionsTestCase(BaseTestCase): + """Test case for dropConnections command.""" + + host_and_port: Any = None + + +# Property [Core Success]: dropConnections succeeds with valid hostAndPort arrays. +CORE_SUCCESS_TESTS: list[DropConnectionsTestCase] = [ + DropConnectionsTestCase( + "empty_array", + host_and_port=[], + expected={"ok": 1.0}, + msg="dropConnections should succeed with empty hostAndPort array (no-op).", + ), + DropConnectionsTestCase( + "single_valid_host_port", + host_and_port=["nonexistent.host.invalid:27017"], + expected={"ok": 1.0}, + msg="dropConnections should succeed with a valid host:port string.", + ), + DropConnectionsTestCase( + "multiple_valid_host_ports", + host_and_port=["host1.invalid:27017", "host2.invalid:27018"], + expected={"ok": 1.0}, + msg="dropConnections should succeed with multiple valid host:port strings.", + ), + DropConnectionsTestCase( + "nonexistent_host_no_connection", + host_and_port=["no.such.host.invalid:12345"], + expected={"ok": 1.0}, + msg="dropConnections should succeed silently for hosts with no active connection.", + ), + DropConnectionsTestCase( + "idempotent_repeated_call", + host_and_port=["nonexistent.host.invalid:27017"], + expected={"ok": 1.0}, + msg="dropConnections should be idempotent when called with same host.", + ), + DropConnectionsTestCase( + "ip_address_with_port", + host_and_port=["192.168.1.1:27017"], + expected={"ok": 1.0}, + msg="dropConnections should accept IP address with port.", + ), + DropConnectionsTestCase( + "duplicate_entries_in_array", + host_and_port=["host.invalid:27017", "host.invalid:27017"], + expected={"ok": 1.0}, + msg="dropConnections should handle duplicate entries idempotently.", + ), + DropConnectionsTestCase( + "ipv6_format", + host_and_port=["[::1]:27017"], + expected={"ok": 1.0}, + msg="dropConnections should accept IPv6 format address.", + ), +] + +# Property [Silent Ignore]: dropConnections silently ignores hostAndPort elements +# that contain no colon (no port separator). +SILENT_IGNORE_TESTS: list[DropConnectionsTestCase] = [ + DropConnectionsTestCase( + "hostname_only_no_colon", + host_and_port=["hostname_only"], + expected={"ok": 1.0}, + msg="dropConnections should silently ignore entry without a colon.", + ), +] + +# Property [Host Parse Rejection]: dropConnections rejects hostAndPort entries +# that contain a colon but fail host:port parsing (empty host, invalid port). +HOST_PARSE_ERROR_TESTS: list[DropConnectionsTestCase] = [ + DropConnectionsTestCase( + "missing_hostname_colon_port", + host_and_port=[":27017"], + error_code=FAILED_TO_PARSE_ERROR, + msg="dropConnections should reject entry with empty host component.", + ), + DropConnectionsTestCase( + "hostname_trailing_colon", + host_and_port=["host:"], + error_code=FAILED_TO_PARSE_ERROR, + msg="dropConnections should reject entry with empty port after colon.", + ), + DropConnectionsTestCase( + "empty_string", + host_and_port=[""], + error_code=FAILED_TO_PARSE_ERROR, + msg="dropConnections should reject empty string entry.", + ), + DropConnectionsTestCase( + "port_zero", + host_and_port=["host.invalid:0"], + error_code=FAILED_TO_PARSE_ERROR, + msg="dropConnections should reject port 0 (out of range).", + ), + DropConnectionsTestCase( + "high_port_number", + host_and_port=["host.invalid:99999"], + error_code=FAILED_TO_PARSE_ERROR, + msg="dropConnections should reject port exceeding valid range.", + ), + DropConnectionsTestCase( + "non_numeric_port", + host_and_port=["host.invalid:abc"], + error_code=FAILED_TO_PARSE_ERROR, + msg="dropConnections should reject non-numeric port.", + ), + DropConnectionsTestCase( + "multiple_colons_no_brackets", + host_and_port=["host:port:extra"], + error_code=FAILED_TO_PARSE_ERROR, + msg="dropConnections should reject entry with multiple colons without IPv6 brackets.", + ), + DropConnectionsTestCase( + "negative_port", + host_and_port=["host.invalid:-1"], + error_code=FAILED_TO_PARSE_ERROR, + msg="dropConnections should reject negative port number.", + ), +] + +ALL_TESTS = CORE_SUCCESS_TESTS + SILENT_IGNORE_TESTS + HOST_PARSE_ERROR_TESTS + + +@pytest.mark.parametrize("test", pytest_params(ALL_TESTS)) +def test_dropConnections_core_behavior(collection, test): + """Test dropConnections core behavior and host:port parsing.""" + result = execute_admin_command( + collection, {"dropConnections": 1, "hostAndPort": test.host_and_port} + ) + assertResult( + result, + expected=test.expected, + error_code=test.error_code, + msg=test.msg, + raw_res=True, + )