diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/__init__.py b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_argument_validation.py b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_argument_validation.py
new file mode 100644
index 000000000..b29c2c161
--- /dev/null
+++ b/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)
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_core_behavior.py b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_core_behavior.py
new file mode 100644
index 000000000..d3048f64b
--- /dev/null
+++ b/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",
+    )
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_errors.py b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_errors.py
new file mode 100644
index 000000000..3f0056fdf
--- /dev/null
+++ b/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",
+    )
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_success.py b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_success.py
new file mode 100644
index 000000000..52756d162
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_success.py
@@ -0,0 +1,193 @@
+"""Tests for setUserWriteBlockMode success cases.
+
+Validates argument acceptance, read operations not blocked while active,
+and write operations succeeding when block is disabled.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+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 assertSuccessPartial
+from documentdb_tests.framework.executor import execute_admin_command, execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+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 Boolean Acceptance]: setUserWriteBlockMode accepts only boolean values
+# for the global field.
+GLOBAL_VALID_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        "global_true",
+        command=lambda ctx: {"setUserWriteBlockMode": 1, "global": True},
+        expected={"ok": 1.0},
+        msg="setUserWriteBlockMode should accept global:true",
+    ),
+    AdminTestCase(
+        "global_false",
+        command=lambda ctx: {"setUserWriteBlockMode": 1, "global": False},
+        expected={"ok": 1.0},
+        msg="setUserWriteBlockMode should accept global:false",
+    ),
+]
+
+# Property [Reason Field Valid Values]: setUserWriteBlockMode accepts valid reason enum strings.
+REASON_VALID_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        f"reason_{tid}",
+        command=lambda ctx, r=reason: {
+            "setUserWriteBlockMode": 1,
+            "global": True,
+            "reason": r,
+        },
+        expected={"ok": 1.0},
+        msg=f"setUserWriteBlockMode should accept reason:{reason}",
+    )
+    for tid, reason in [
+        ("unspecified", "Unspecified"),
+        ("cluster_migration", "ClusterToClusterMigrationInProgress"),
+        ("disk_threshold", "DiskUseThresholdExceeded"),
+    ]
+]
+
+# Property [Reason Field Optional]: the reason field can be null (treated as omitted).
+REASON_OPTIONAL_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        "reason_null",
+        command=lambda ctx: {"setUserWriteBlockMode": 1, "global": True, "reason": None},
+        expected={"ok": 1.0},
+        msg="setUserWriteBlockMode should treat null reason as omitted",
+    ),
+]
+
+# Property [Read Operations Not Blocked]: read operations succeed while the block is active.
+READ_NOT_BLOCKED_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        "read_find",
+        use_admin=False,
+        partial_success=True,
+        docs=[{"_id": "read_doc", "x": 1}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {"find": ctx.collection, "filter": {"_id": "read_doc"}},
+        expected={"ok": 1.0, "cursor": {"firstBatch": [{"_id": "read_doc", "x": 1}]}},
+        msg="setUserWriteBlockMode should not block find while active",
+    ),
+    AdminTestCase(
+        "read_aggregate",
+        use_admin=False,
+        partial_success=True,
+        docs=[{"_id": "agg_doc", "x": 5}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {
+            "aggregate": ctx.collection,
+            "pipeline": [{"$match": {"_id": "agg_doc"}}],
+            "cursor": {},
+        },
+        expected={"ok": 1.0, "cursor": {"firstBatch": [{"_id": "agg_doc", "x": 5}]}},
+        msg="setUserWriteBlockMode should not block aggregate while active",
+    ),
+    AdminTestCase(
+        "read_count",
+        use_admin=False,
+        partial_success=True,
+        docs=[{"_id": "c1"}, {"_id": "c2"}, {"_id": "c3"}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {"count": ctx.collection},
+        expected={"ok": 1.0, "n": 3},
+        msg="setUserWriteBlockMode should not block count while active",
+    ),
+    AdminTestCase(
+        "read_distinct",
+        use_admin=False,
+        partial_success=True,
+        docs=[{"_id": "d1", "x": 1}, {"_id": "d2", "x": 2}, {"_id": "d3", "x": 1}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {"distinct": ctx.collection, "key": "x"},
+        expected={"ok": 1.0, "values": [1, 2]},
+        msg="setUserWriteBlockMode should not block distinct while active",
+    ),
+]
+
+# Property [Writes Succeed When Disabled]: write operations succeed when no block is active.
+WRITE_SUCCEEDS_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        "write_insert_no_block",
+        use_admin=False,
+        partial_success=True,
+        docs=[],
+        command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": "new_doc", "v": 42}]},
+        expected={"ok": 1.0, "n": 1},
+        msg="setUserWriteBlockMode should allow insert when block is not active",
+    ),
+    AdminTestCase(
+        "write_update_no_block",
+        use_admin=False,
+        partial_success=True,
+        docs=[{"_id": "target", "x": 1}],
+        command=lambda ctx: {
+            "update": ctx.collection,
+            "updates": [{"q": {"_id": "target"}, "u": {"$set": {"x": 99}}}],
+        },
+        expected={"ok": 1.0, "n": 1, "nModified": 1},
+        msg="setUserWriteBlockMode should allow update when block is not active",
+    ),
+    AdminTestCase(
+        "write_delete_no_block",
+        use_admin=False,
+        partial_success=True,
+        docs=[{"_id": "target"}],
+        command=lambda ctx: {
+            "delete": ctx.collection,
+            "deletes": [{"q": {"_id": "target"}, "limit": 1}],
+        },
+        expected={"ok": 1.0, "n": 1},
+        msg="setUserWriteBlockMode should allow delete when block is not active",
+    ),
+]
+
+SUCCESS_TESTS: list[AdminTestCase] = (
+    GLOBAL_VALID_TESTS
+    + REASON_VALID_TESTS
+    + REASON_OPTIONAL_TESTS
+    + READ_NOT_BLOCKED_TESTS
+    + WRITE_SUCCEEDS_TESTS
+)
+
+
+@pytest.mark.parametrize("test", pytest_params(SUCCESS_TESTS))
+def test_setUserWriteBlockMode_success(database_client, collection, test):
+    """Test setUserWriteBlockMode success cases."""
+    collection = test.prepare(database_client, collection)
+    test.run_pre_command(collection)
+    ctx = CommandContext.from_collection(collection)
+    if test.use_admin:
+        result = execute_admin_command(collection, test.build_command(ctx))
+    else:
+        result = execute_command(collection, test.build_command(ctx))
+    assertSuccessPartial(result, test.expected, msg=test.msg)
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_write_block_enforcement.py b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_write_block_enforcement.py
new file mode 100644
index 000000000..c6fce6074
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_write_block_enforcement.py
@@ -0,0 +1,194 @@
+"""Tests for setUserWriteBlockMode write block enforcement errors.
+
+Validates that write operations are rejected while the block is active.
+"""
+
+from __future__ import annotations
+
+import pytest
+from pymongo import IndexModel
+
+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 USER_WRITES_BLOCKED_ERROR
+from documentdb_tests.framework.executor import execute_admin_command, execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+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 [Write Operations Blocked]: all write operations are rejected while the block is
+# active.
+WRITE_BLOCKED_TESTS: list[AdminTestCase] = [
+    AdminTestCase(
+        "blocked_insert",
+        use_admin=False,
+        docs=[{"_id": "seed", "a": 1}],
+        indexes=[IndexModel([("a", 1)], name="a_1")],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": "blocked"}]},
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block insert while active",
+    ),
+    AdminTestCase(
+        "blocked_update",
+        use_admin=False,
+        docs=[{"_id": "seed", "a": 1}],
+        indexes=[IndexModel([("a", 1)], name="a_1")],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {
+            "update": ctx.collection,
+            "updates": [{"q": {}, "u": {"$set": {"x": 2}}}],
+        },
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block update while active",
+    ),
+    AdminTestCase(
+        "blocked_delete",
+        use_admin=False,
+        docs=[{"_id": "seed", "a": 1}],
+        indexes=[IndexModel([("a", 1)], name="a_1")],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {
+            "delete": ctx.collection,
+            "deletes": [{"q": {}, "limit": 1}],
+        },
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block delete while active",
+    ),
+    AdminTestCase(
+        "blocked_findAndModify_update",
+        use_admin=False,
+        docs=[{"_id": "seed", "a": 1}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {
+            "findAndModify": ctx.collection,
+            "query": {},
+            "update": {"$set": {"x": 2}},
+        },
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block findAndModify update while active",
+    ),
+    AdminTestCase(
+        "blocked_findAndModify_remove",
+        use_admin=False,
+        docs=[{"_id": "seed", "a": 1}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {
+            "findAndModify": ctx.collection,
+            "query": {},
+            "remove": True,
+        },
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block findAndModify remove while active",
+    ),
+    AdminTestCase(
+        "blocked_createIndexes",
+        use_admin=False,
+        docs=[{"_id": "seed", "a": 1}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {
+            "createIndexes": ctx.collection,
+            "indexes": [{"key": {"blocked_field": 1}, "name": "blocked_field_1"}],
+        },
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block createIndexes while active",
+    ),
+    AdminTestCase(
+        "blocked_dropIndexes",
+        use_admin=False,
+        docs=[{"_id": "seed", "a": 1}],
+        indexes=[IndexModel([("a", 1)], name="a_1")],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {"dropIndexes": ctx.collection, "index": "a_1"},
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block dropIndexes while active",
+    ),
+    AdminTestCase(
+        "blocked_drop_collection",
+        use_admin=False,
+        docs=[{"_id": "seed"}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {"drop": ctx.collection},
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block drop collection while active",
+    ),
+    AdminTestCase(
+        "blocked_create_collection",
+        use_admin=False,
+        docs=[{"_id": "seed"}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {"create": f"{ctx.collection}_blocked_new"},
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block create collection while active",
+    ),
+    AdminTestCase(
+        "blocked_dropDatabase",
+        use_admin=False,
+        docs=[{"_id": "seed"}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {"dropDatabase": 1},
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block dropDatabase while active",
+    ),
+    AdminTestCase(
+        "blocked_batch_insert",
+        use_admin=False,
+        docs=[{"_id": "seed"}],
+        pre_command=lambda c: execute_admin_command(
+            c, {"setUserWriteBlockMode": 1, "global": True}
+        ),
+        command=lambda ctx: {
+            "insert": ctx.collection,
+            "documents": [{"_id": "bulk1"}, {"_id": "bulk2"}],
+        },
+        error_code=USER_WRITES_BLOCKED_ERROR,
+        msg="setUserWriteBlockMode should block multi-document insert while active",
+    ),
+]
+
+
+@pytest.mark.parametrize("test", pytest_params(WRITE_BLOCKED_TESTS))
+def test_setUserWriteBlockMode_blocked(database_client, collection, test):
+    """Test setUserWriteBlockMode blocks write operations while active."""
+    collection = test.prepare(database_client, collection)
+    test.run_pre_command(collection)
+    ctx = CommandContext.from_collection(collection)
+    result = execute_command(collection, test.build_command(ctx))
+    assertFailureCode(result, test.error_code, msg=test.msg)
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/utils/__init__.py b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/utils/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/utils/write_block_helpers.py b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/utils/write_block_helpers.py
new file mode 100644
index 000000000..3f05a357c
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/utils/write_block_helpers.py
@@ -0,0 +1,21 @@
+"""Shared utilities for setUserWriteBlockMode tests."""
+
+
+def force_disable_write_block(collection):
+    """Force-disable write block regardless of current reason."""
+    admin = collection.database.client.admin
+    try:
+        admin.command({"setUserWriteBlockMode": 1, "global": False})
+        return
+    except Exception:
+        pass
+    for reason in [
+        "Unspecified",
+        "ClusterToClusterMigrationInProgress",
+        "DiskUseThresholdExceeded",
+    ]:
+        try:
+            admin.command({"setUserWriteBlockMode": 1, "global": False, "reason": reason})
+            return
+        except Exception:
+            continue
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/utils/__init__.py b/documentdb_tests/compatibility/tests/system/administration/commands/utils/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/documentdb_tests/compatibility/tests/system/administration/commands/utils/admin_test_case.py b/documentdb_tests/compatibility/tests/system/administration/commands/utils/admin_test_case.py
new file mode 100644
index 000000000..d35be87bb
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/system/administration/commands/utils/admin_test_case.py
@@ -0,0 +1,49 @@
+"""Shared test case for administration command tests."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from pymongo.collection import Collection
+
+from documentdb_tests.compatibility.tests.core.utils.command_test_case import (
+    CommandContext,
+    CommandTestCase,
+)
+
+
+@dataclass(frozen=True)
+class AdminTestCase(CommandTestCase):
+    """Test case for administration command tests.
+
+    Extends CommandTestCase with fields for admin-specific execution:
+
+    Attributes:
+        use_admin: If True (the default), execute the command against
+            the admin database via ``execute_admin_command``.  If False,
+            execute against the test database via ``execute_command``.
+        pre_command: Optional callable ``(collection) -> None`` invoked
+            after ``prepare`` completes (docs inserted, indexes created)
+            but before the test command executes. Use this for stateful
+            setup like enabling a write block.
+        partial_success: If True, success assertions use partial matching
+            (only checks that expected keys are present in the result).
+            Useful for commands that return extra metadata fields.
+    """
+
+    use_admin: bool = True
+    pre_command: Callable[[Collection], Any] | None = None
+    partial_success: bool = False
+
+    def run_pre_command(self, collection: Collection) -> None:
+        """Execute the pre_command callable if defined."""
+        if self.pre_command is not None:
+            self.pre_command(collection)
+
+    def build_expected(self, ctx: CommandContext) -> dict[str, Any] | list[dict[str, Any]] | None:
+        """Resolve expected from a callable or plain value."""
+        if self.expected is None or isinstance(self.expected, (dict, list)):
+            return self.expected
+        return self.expected(ctx)
diff --git a/documentdb_tests/framework/error_codes.py b/documentdb_tests/framework/error_codes.py
index 423b3fe65..efa0b35cb 100644
--- a/documentdb_tests/framework/error_codes.py
+++ b/documentdb_tests/framework/error_codes.py
@@ -61,6 +61,7 @@
 API_VERSION_ERROR = 322
 API_STRICT_ERROR = 323
 COLLECTION_UUID_MISMATCH_ERROR = 361
+USER_WRITES_BLOCKED_ERROR = 371
 QUERYSETTINGS_QUERY_REJECTED_ERROR = 411
 EXPRESSION_NOT_OBJECT_ERROR = 10065
 BSON_OBJECT_TOO_LARGE_ERROR = 10334