Add setUserWriteBlockMode tests

documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_argument_validation.py

@@ -0,0 +1,150 @@
+"""Tests for setUserWriteBlockMode argument validation errors.
+
+Validates type rejection for the global and reason fields, missing required fields,
+invalid enum values, and unrecognized fields.
+"""
+
+from __future__ import annotations
+
+import pytest
+from bson import Decimal128, Int64
+
+from documentdb_tests.compatibility.tests.core.utils.command_test_case import (
+    CommandContext,
+)
+from documentdb_tests.compatibility.tests.system.administration.commands.setUserWriteBlockMode.utils.write_block_helpers import (  # noqa: E501
+    force_disable_write_block,
+)
+from documentdb_tests.compatibility.tests.system.administration.commands.utils.admin_test_case import (  # noqa: E501
+    AdminTestCase,
+)
+from documentdb_tests.framework.assertions import assertFailureCode
+from documentdb_tests.framework.error_codes import (
+    BAD_VALUE_ERROR,
+    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_constants import FLOAT_INFINITY, FLOAT_NAN
+
+pytestmark = [pytest.mark.admin, pytest.mark.no_parallel, pytest.mark.requires(cluster_admin=True)]
+
+
+@pytest.fixture(autouse=True)
+def _manage_write_block(collection):
+    """Ensure write block is disabled before and after each test."""
+    force_disable_write_block(collection)
+    yield
+    force_disable_write_block(collection)
+
+
+# Property [Global Field Type Rejection]: setUserWriteBlockMode rejects all non-boolean types
+# for the global field with no coercion.
+GLOBAL_TYPE_REJECTION_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        f"global_type_{tid}",
+        command=lambda ctx, v=val: {"setUserWriteBlockMode": 1, "global": v},
+        error_code=error,
+        msg=f"setUserWriteBlockMode should reject {tid} for global field",
+    )
+    for tid, val, error in [
+        ("int32_1", 1, TYPE_MISMATCH_ERROR),
+        ("int32_0", 0, TYPE_MISMATCH_ERROR),
+        ("double_1", 1.0, TYPE_MISMATCH_ERROR),
+        ("double_0", 0.0, TYPE_MISMATCH_ERROR),
+        ("int64", Int64(1), TYPE_MISMATCH_ERROR),
+        ("decimal128", Decimal128("1"), TYPE_MISMATCH_ERROR),
+        ("nan", FLOAT_NAN, TYPE_MISMATCH_ERROR),
+        ("infinity", FLOAT_INFINITY, TYPE_MISMATCH_ERROR),
+        ("negative_infinity", float("-inf"), TYPE_MISMATCH_ERROR),
+        ("negative_zero", -0.0, TYPE_MISMATCH_ERROR),
+        ("string", "true", TYPE_MISMATCH_ERROR),
+        ("array", [], TYPE_MISMATCH_ERROR),
+        ("object", {}, TYPE_MISMATCH_ERROR),
+    ]
+]
+
+# Property [Missing Global Field]: setUserWriteBlockMode requires the global field.
+# Null is treated as missing.
+MISSING_GLOBAL_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        "missing_global",
+        command=lambda ctx: {"setUserWriteBlockMode": 1},
+        error_code=MISSING_FIELD_ERROR,
+        msg="setUserWriteBlockMode should require the global field",
+    ),
+    AdminTestCase(
+        "global_null_treated_as_missing",
+        command=lambda ctx: {"setUserWriteBlockMode": 1, "global": None},
+        error_code=MISSING_FIELD_ERROR,
+        msg="setUserWriteBlockMode should treat null global as missing",
+    ),
+]
+
+# Property [Reason Field Type Rejection]: setUserWriteBlockMode rejects non-string types for
+# the reason field.
+REASON_TYPE_REJECTION_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        f"reason_type_{tid}",
+        command=lambda ctx, v=val: {
+            "setUserWriteBlockMode": 1,
+            "global": True,
+            "reason": v,
+        },
+        error_code=TYPE_MISMATCH_ERROR,
+        msg=f"setUserWriteBlockMode should reject {tid} for reason field",
+    )
+    for tid, val in [
+        ("int", 1),
+        ("bool", True),
+        ("array", []),
+        ("object", {}),
+    ]
+]
+
+# Property [Reason Field Invalid Enum]: setUserWriteBlockMode rejects unrecognized reason
+# strings.
+REASON_INVALID_ENUM_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        "reason_invalid_enum",
+        command=lambda ctx: {
+            "setUserWriteBlockMode": 1,
+            "global": True,
+            "reason": "InvalidReason",
+        },
+        error_code=BAD_VALUE_ERROR,
+        msg="setUserWriteBlockMode should reject unrecognized reason enum value",
+    ),
+]
+
+# Property [Unrecognized Fields]: setUserWriteBlockMode rejects unknown fields.
+UNRECOGNIZED_FIELD_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        "unrecognized_field",
+        command=lambda ctx: {
+            "setUserWriteBlockMode": 1,
+            "global": False,
+            "unknownField": 1,
+        },
+        error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR,
+        msg="setUserWriteBlockMode should reject unrecognized fields",
+    ),
+]
+
+ARGUMENT_ERROR_TESTS: list[AdminTestCase] = (
+    GLOBAL_TYPE_REJECTION_TESTS
+    + MISSING_GLOBAL_TESTS
+    + REASON_TYPE_REJECTION_TESTS
+    + REASON_INVALID_ENUM_TESTS
+    + UNRECOGNIZED_FIELD_TESTS
+)
+
+
+@pytest.mark.parametrize("test", pytest_params(ARGUMENT_ERROR_TESTS))
+def test_setUserWriteBlockMode_argument_error(collection, test):
+    """Test setUserWriteBlockMode rejects invalid arguments."""
+    ctx = CommandContext.from_collection(collection)
+    result = execute_admin_command(collection, test.build_command(ctx))
+    assertFailureCode(result, test.error_code, msg=test.msg)

documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_core_behavior.py

@@ -0,0 +1,118 @@
+"""Tests for setUserWriteBlockMode core behavior.
+
+Validates enable/disable semantics, idempotent behavior, and state restoration.
+"""
+
+import pytest
+
+from documentdb_tests.compatibility.tests.system.administration.commands.setUserWriteBlockMode.utils.write_block_helpers import (  # noqa: E501
+    force_disable_write_block,
+)
+from documentdb_tests.framework.assertions import assertSuccessPartial
+from documentdb_tests.framework.executor import execute_admin_command, execute_command
+
+pytestmark = [pytest.mark.admin, pytest.mark.no_parallel, pytest.mark.requires(cluster_admin=True)]
+
+
+@pytest.fixture(autouse=True)
+def _manage_write_block(collection):
+    """Ensure write block is disabled before and after each test."""
+    force_disable_write_block(collection)
+    yield
+    force_disable_write_block(collection)
+
+
+# Property [Idempotent Disable]: disabling write block when no block is active succeeds.
+def test_setUserWriteBlockMode_disable_when_no_block_active(collection):
+    """Test setUserWriteBlockMode global:false when no block is active succeeds."""
+    result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False})
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0},
+        msg="setUserWriteBlockMode should succeed when disabling with no active block",
+    )
+
+
+# Property [Write Restoration]: writes succeed after disabling a previously active block.
+def test_setUserWriteBlockMode_enable_disable_restores_writes(collection):
+    """Test setUserWriteBlockMode enable then disable allows writes again."""
+    execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True})
+    execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False})
+    result = execute_command(
+        collection, {"insert": collection.name, "documents": [{"_id": "restore_test"}]}
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0},
+        msg="setUserWriteBlockMode should allow writes after block is disabled",
+    )
+
+
+# Property [Repeated Toggle]: toggling write block multiple times does not produce errors.
+def test_setUserWriteBlockMode_toggle_multiple_times(collection):
+    """Test setUserWriteBlockMode toggling on and off multiple times succeeds."""
+    execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True})
+    execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False})
+    execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True})
+    execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False})
+    result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True})
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0},
+        msg="setUserWriteBlockMode should succeed after repeated toggling",
+    )
+
+
+# Property [Idempotent Enable]: re-enabling with same default reason is idempotent.
+def test_setUserWriteBlockMode_enable_idempotent_same_reason(collection):
+    """Test setUserWriteBlockMode re-enable with same reason is idempotent."""
+    execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True})
+    result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True})
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0},
+        msg="setUserWriteBlockMode should be idempotent when re-enabling with same reason",
+    )
+
+
+# Property [Same Explicit Reason Idempotent]: re-enabling with same explicit reason succeeds.
+def test_setUserWriteBlockMode_same_reason_unspecified_idempotent(collection):
+    """Test setUserWriteBlockMode re-enable with same reason Unspecified is idempotent."""
+    execute_admin_command(
+        collection,
+        {"setUserWriteBlockMode": 1, "global": True, "reason": "Unspecified"},
+    )
+    result = execute_admin_command(
+        collection,
+        {"setUserWriteBlockMode": 1, "global": True, "reason": "Unspecified"},
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0},
+        msg="setUserWriteBlockMode should be idempotent with same explicit reason",
+    )
+
+
+def test_setUserWriteBlockMode_same_reason_cluster_migration_idempotent(collection):
+    """Test setUserWriteBlockMode re-enable with same reason ClusterToClusterMigrationInProgress."""
+    execute_admin_command(
+        collection,
+        {
+            "setUserWriteBlockMode": 1,
+            "global": True,
+            "reason": "ClusterToClusterMigrationInProgress",
+        },
+    )
+    result = execute_admin_command(
+        collection,
+        {
+            "setUserWriteBlockMode": 1,
+            "global": True,
+            "reason": "ClusterToClusterMigrationInProgress",
+        },
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0},
+        msg="setUserWriteBlockMode should be idempotent with same explicit reason",
+    )

documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_errors.py

@@ -0,0 +1,70 @@
+"""Tests for setUserWriteBlockMode error cases.
+
+Validates mismatched reason errors when changing the reason on an active block.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.compatibility.tests.system.administration.commands.setUserWriteBlockMode.utils.write_block_helpers import (  # noqa: E501
+    force_disable_write_block,
+)
+from documentdb_tests.framework.assertions import assertFailureCode
+from documentdb_tests.framework.error_codes import ILLEGAL_OPERATION_ERROR
+from documentdb_tests.framework.executor import execute_admin_command
+
+pytestmark = [pytest.mark.admin, pytest.mark.no_parallel, pytest.mark.requires(cluster_admin=True)]
+
+
+@pytest.fixture(autouse=True)
+def _manage_write_block(collection):
+    """Ensure write block is disabled before and after each test."""
+    force_disable_write_block(collection)
+    yield
+    force_disable_write_block(collection)
+
+
+# Property [Mismatched Reason on Enable]: re-enabling with a different reason fails.
+def test_setUserWriteBlockMode_enable_mismatched_reason_fails(collection):
+    """Test setUserWriteBlockMode re-enable with different reason fails."""
+    execute_admin_command(
+        collection,
+        {"setUserWriteBlockMode": 1, "global": True, "reason": "Unspecified"},
+    )
+    result = execute_admin_command(
+        collection,
+        {
+            "setUserWriteBlockMode": 1,
+            "global": True,
+            "reason": "ClusterToClusterMigrationInProgress",
+        },
+    )
+    assertFailureCode(
+        result,
+        ILLEGAL_OPERATION_ERROR,
+        msg="setUserWriteBlockMode should reject mismatched reason on re-enable",
+    )
+
+
+# Property [Mismatched Reason on Disable]: disabling with a different reason than the active
+# block fails.
+def test_setUserWriteBlockMode_disable_mismatched_reason_fails(collection):
+    """Test setUserWriteBlockMode disable with different reason fails."""
+    execute_admin_command(
+        collection,
+        {"setUserWriteBlockMode": 1, "global": True, "reason": "Unspecified"},
+    )
+    result = execute_admin_command(
+        collection,
+        {
+            "setUserWriteBlockMode": 1,
+            "global": False,
+            "reason": "ClusterToClusterMigrationInProgress",
+        },
+    )
+    assertFailureCode(
+        result,
+        ILLEGAL_OPERATION_ERROR,
+        msg="setUserWriteBlockMode should reject mismatched reason on disable",
+    )