From ea141ee71f559bc72f23d6acb8596d6c5f1a1d7f Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 12:00:38 -0700 Subject: [PATCH 01/10] initial generated setUserWriteBlockMode tests Signed-off-by: Alina (Xi) Li --- .../setUserWriteBlockMode/__init__.py | 0 ...tUserWriteBlockMode_argument_validation.py | 289 ++++++++++++++++++ ...est_setUserWriteBlockMode_core_behavior.py | 95 ++++++ .../test_setUserWriteBlockMode_errors.py | 128 ++++++++ ...rWriteBlockMode_write_block_enforcement.py | 232 ++++++++++++++ documentdb_tests/framework/error_codes.py | 1 + 6 files changed, 745 insertions(+) create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/__init__.py create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_argument_validation.py create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_core_behavior.py create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_errors.py create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_write_block_enforcement.py 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..637e62d69 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_argument_validation.py @@ -0,0 +1,289 @@ +"""Tests for setUserWriteBlockMode argument validation. + +Validates type checking for the global and reason fields, unrecognized fields, +and command value handling. +""" + +import pytest +from bson import Decimal128, Int64 + +from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial +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.test_constants import FLOAT_INFINITY, FLOAT_NAN + +pytestmark = [pytest.mark.admin, pytest.mark.no_parallel, pytest.mark.requires(cluster_admin=True)] + + +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 + + +@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) + + +# --- global field: valid boolean values --- + + +def test_setUserWriteBlockMode_global_true_succeeds(collection): + """Test global:true (boolean) succeeds.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) + assertSuccessPartial(result, {"ok": 1.0}, msg="global:true should succeed") + + +def test_setUserWriteBlockMode_global_false_succeeds(collection): + """Test global:false (boolean) succeeds.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False}) + assertSuccessPartial(result, {"ok": 1.0}, msg="global:false should succeed") + + +# --- global field: numeric types rejected (no coercion) --- + + +def test_setUserWriteBlockMode_global_int_1_rejected(collection): + """Test global:1 (int32) is rejected — no coercion to bool.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": 1}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:int should be rejected") + + +def test_setUserWriteBlockMode_global_int_0_rejected(collection): + """Test global:0 (int32) is rejected — no coercion to bool.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": 0}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:int(0) should be rejected") + + +def test_setUserWriteBlockMode_global_double_1_rejected(collection): + """Test global:1.0 (double) is rejected — no coercion to bool.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": 1.0}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:double should be rejected") + + +def test_setUserWriteBlockMode_global_double_0_rejected(collection): + """Test global:0.0 (double) is rejected — no coercion to bool.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": 0.0}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:double(0.0) should be rejected") + + +def test_setUserWriteBlockMode_global_int64_rejected(collection): + """Test global:NumberLong(1) (int64) is rejected — no coercion to bool.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": Int64(1)}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:int64 should be rejected") + + +def test_setUserWriteBlockMode_global_decimal128_rejected(collection): + """Test global:NumberDecimal('1') (decimal128) is rejected — no coercion to bool.""" + result = execute_admin_command( + collection, {"setUserWriteBlockMode": 1, "global": Decimal128("1")} + ) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:decimal128 should be rejected") + + +def test_setUserWriteBlockMode_global_nan_rejected(collection): + """Test global:NaN (double) is rejected — no coercion to bool.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": FLOAT_NAN}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:NaN should be rejected") + + +def test_setUserWriteBlockMode_global_infinity_rejected(collection): + """Test global:Infinity (double) is rejected — no coercion to bool.""" + result = execute_admin_command( + collection, {"setUserWriteBlockMode": 1, "global": FLOAT_INFINITY} + ) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:Infinity should be rejected") + + +def test_setUserWriteBlockMode_global_negative_infinity_rejected(collection): + """Test global:-Infinity (double) is rejected — no coercion to bool.""" + result = execute_admin_command( + collection, {"setUserWriteBlockMode": 1, "global": float("-inf")} + ) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:-Infinity should be rejected") + + +def test_setUserWriteBlockMode_global_negative_zero_rejected(collection): + """Test global:-0.0 (negative zero double) is rejected — no coercion to bool.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": -0.0}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:-0.0 should be rejected") + + +# --- global field: non-numeric invalid types --- + + +def test_setUserWriteBlockMode_global_null_fails(collection): + """Test global:null is treated as missing required field.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": None}) + assertFailureCode(result, MISSING_FIELD_ERROR, msg="global:null should fail as missing") + + +def test_setUserWriteBlockMode_global_string_fails(collection): + """Test global:'true' (string) fails with type error.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": "true"}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:string should fail") + + +def test_setUserWriteBlockMode_global_array_fails(collection): + """Test global:[] (array) fails with type error.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": []}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:array should fail") + + +def test_setUserWriteBlockMode_global_object_fails(collection): + """Test global:{} (object) fails with type error.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": {}}) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:object should fail") + + +# --- missing global field --- + + +def test_setUserWriteBlockMode_missing_global_fails(collection): + """Test setUserWriteBlockMode without global field fails with missing field error.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1}) + assertFailureCode(result, MISSING_FIELD_ERROR, msg="Missing global should fail") + + +# --- reason field: valid values --- + + +def test_setUserWriteBlockMode_reason_unspecified_succeeds(collection): + """Test reason:'Unspecified' succeeds.""" + result = execute_admin_command( + collection, + {"setUserWriteBlockMode": 1, "global": True, "reason": "Unspecified"}, + ) + assertSuccessPartial(result, {"ok": 1.0}, msg="reason:Unspecified should succeed") + + +def test_setUserWriteBlockMode_reason_cluster_migration_succeeds(collection): + """Test reason:'ClusterToClusterMigrationInProgress' succeeds.""" + result = execute_admin_command( + collection, + { + "setUserWriteBlockMode": 1, + "global": True, + "reason": "ClusterToClusterMigrationInProgress", + }, + ) + assertSuccessPartial( + result, {"ok": 1.0}, msg="reason:ClusterToClusterMigrationInProgress should succeed" + ) + + +def test_setUserWriteBlockMode_reason_disk_threshold_succeeds(collection): + """Test reason:'DiskUseThresholdExceeded' succeeds.""" + result = execute_admin_command( + collection, + { + "setUserWriteBlockMode": 1, + "global": True, + "reason": "DiskUseThresholdExceeded", + }, + ) + assertSuccessPartial(result, {"ok": 1.0}, msg="reason:DiskUseThresholdExceeded should succeed") + + +def test_setUserWriteBlockMode_no_reason_field_succeeds(collection): + """Test omitting reason field succeeds (defaults to Unspecified).""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) + assertSuccessPartial(result, {"ok": 1.0}, msg="Omitting reason should succeed") + + +def test_setUserWriteBlockMode_reason_null_succeeds(collection): + """Test reason:null is treated as omitted and succeeds.""" + result = execute_admin_command( + collection, {"setUserWriteBlockMode": 1, "global": True, "reason": None} + ) + assertSuccessPartial(result, {"ok": 1.0}, msg="reason:null should succeed as omitted") + + +# --- reason field: invalid types --- + + +def test_setUserWriteBlockMode_reason_int_fails(collection): + """Test reason:1 (int) fails with type error.""" + result = execute_admin_command( + collection, {"setUserWriteBlockMode": 1, "global": True, "reason": 1} + ) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="reason:int should fail") + + +def test_setUserWriteBlockMode_reason_bool_fails(collection): + """Test reason:true (bool) fails with type error.""" + result = execute_admin_command( + collection, {"setUserWriteBlockMode": 1, "global": True, "reason": True} + ) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="reason:bool should fail") + + +def test_setUserWriteBlockMode_reason_array_fails(collection): + """Test reason:[] (array) fails with type error.""" + result = execute_admin_command( + collection, {"setUserWriteBlockMode": 1, "global": True, "reason": []} + ) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="reason:array should fail") + + +def test_setUserWriteBlockMode_reason_object_fails(collection): + """Test reason:{} (object) fails with type error.""" + result = execute_admin_command( + collection, {"setUserWriteBlockMode": 1, "global": True, "reason": {}} + ) + assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="reason:object should fail") + + +def test_setUserWriteBlockMode_reason_invalid_enum_fails(collection): + """Test reason with unrecognized string value fails with BadValue.""" + result = execute_admin_command( + collection, + {"setUserWriteBlockMode": 1, "global": True, "reason": "InvalidReason"}, + ) + assertFailureCode(result, BAD_VALUE_ERROR, msg="Invalid reason enum should fail") + + +# --- unrecognized fields --- + + +def test_setUserWriteBlockMode_unrecognized_field_fails(collection): + """Test setUserWriteBlockMode with extra unrecognized field fails.""" + result = execute_admin_command( + collection, + {"setUserWriteBlockMode": 1, "global": False, "unknownField": 1}, + ) + assertFailureCode( + result, UNRECOGNIZED_COMMAND_FIELD_ERROR, msg="Unrecognized field should fail" + ) + + +# --- command field value --- + + +def test_setUserWriteBlockMode_command_value_1_succeeds(collection): + """Test setUserWriteBlockMode:1 is the correct command value.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False}) + assertSuccessPartial(result, {"ok": 1.0}, msg="Command value 1 should succeed") 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..1fa47b1ea --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_core_behavior.py @@ -0,0 +1,95 @@ +"""Tests for setUserWriteBlockMode core behavior. + +Validates enable/disable semantics, response structure, idempotent behavior, +and state restoration. +""" + +import pytest + +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)] + + +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 + + +@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) + + +def test_setUserWriteBlockMode_enable_returns_ok(collection): + """Test setUserWriteBlockMode with global:true returns ok:1.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) + assertSuccessPartial(result, {"ok": 1.0}, msg="Should return ok:1 on enable") + + +def test_setUserWriteBlockMode_disable_returns_ok(collection): + """Test setUserWriteBlockMode with global:false returns ok:1.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False}) + assertSuccessPartial(result, {"ok": 1.0}, msg="Should return ok:1 on disable") + + +def test_setUserWriteBlockMode_disable_when_no_block_active_idempotent(collection): + """Test global:false when no block is active succeeds (idempotent).""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False}) + assertSuccessPartial( + result, {"ok": 1.0}, msg="Should succeed when disabling with no active block" + ) + + +def test_setUserWriteBlockMode_enable_disable_cycle_restores_writes(collection): + """Test enable block then disable, verify writes succeed 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="Write should succeed after block disabled") + + +def test_setUserWriteBlockMode_toggle_multiple_times_no_error(collection): + """Test toggling write block off and on multiple times without errors.""" + 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="Should succeed after repeated toggling") + + +def test_setUserWriteBlockMode_enable_idempotent_same_reason(collection): + """Test enabling write block again with same default 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="Should succeed when re-enabling with same reason" + ) + + +def test_setUserWriteBlockMode_response_contains_ok_field(collection): + """Test successful command returns document with ok:1.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) + assertSuccessPartial(result, {"ok": 1.0}, msg="Response should contain ok:1") 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..13e819e78 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_errors.py @@ -0,0 +1,128 @@ +"""Tests for setUserWriteBlockMode error cases. + +Validates mismatched reason errors and idempotent behavior with same reason. +""" + +import pytest + +from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial +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)] + + +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 + + +@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) + + +# --- Mismatched reason errors --- + + +def test_setUserWriteBlockMode_enable_mismatched_reason_fails(collection): + """Test re-enabling with different reason fails with IllegalOperation.""" + 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="Mismatched reason on enable should fail with IllegalOperation", + ) + + +def test_setUserWriteBlockMode_disable_mismatched_reason_fails(collection): + """Test disabling with different reason than enabled fails with IllegalOperation.""" + 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="Mismatched reason on disable should fail with IllegalOperation", + ) + + +# --- Idempotent with same reason --- + + +def test_setUserWriteBlockMode_same_reason_unspecified_idempotent(collection): + """Test re-enabling with same reason 'Unspecified' succeeds (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="Same reason Unspecified re-enable should be idempotent" + ) + + +def test_setUserWriteBlockMode_same_reason_migration_idempotent(collection): + """Test re-enabling with same reason 'ClusterToClusterMigrationInProgress' succeeds.""" + 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="Same reason ClusterToClusterMigrationInProgress re-enable should be idempotent", + ) 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..ef2494c0b --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_write_block_enforcement.py @@ -0,0 +1,232 @@ +"""Tests for setUserWriteBlockMode write block enforcement. + +Validates that write operations are blocked while the block is active, +read operations are not affected, and operations succeed when block is disabled. +""" + +import pytest + +from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial +from documentdb_tests.framework.error_codes import USER_WRITES_BLOCKED_ERROR +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)] + + +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 + + +@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) + + +def _enable_write_block(collection): + """Enable write block.""" + execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) + + +# --- Writes blocked while block is active --- + + +def test_setUserWriteBlockMode_insert_blocked(collection): + """Test insert is rejected with UserWritesBlocked while write block is active.""" + _enable_write_block(collection) + result = execute_command( + collection, {"insert": collection.name, "documents": [{"_id": "blocked"}]} + ) + assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Insert should be blocked") + + +def test_setUserWriteBlockMode_update_blocked(collection): + """Test update is rejected with UserWritesBlocked while write block is active.""" + collection.insert_one({"_id": "upd_target", "x": 1}) + _enable_write_block(collection) + result = execute_command( + collection, + { + "update": collection.name, + "updates": [{"q": {"_id": "upd_target"}, "u": {"$set": {"x": 2}}}], + }, + ) + assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Update should be blocked") + + +def test_setUserWriteBlockMode_delete_blocked(collection): + """Test delete is rejected with UserWritesBlocked while write block is active.""" + collection.insert_one({"_id": "del_target"}) + _enable_write_block(collection) + result = execute_command( + collection, + {"delete": collection.name, "deletes": [{"q": {"_id": "del_target"}, "limit": 1}]}, + ) + assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Delete should be blocked") + + +def test_setUserWriteBlockMode_findAndModify_update_blocked(collection): + """Test findAndModify with update is blocked while write block is active.""" + collection.insert_one({"_id": "fam_target", "x": 1}) + _enable_write_block(collection) + result = execute_command( + collection, + { + "findAndModify": collection.name, + "query": {"_id": "fam_target"}, + "update": {"$set": {"x": 2}}, + }, + ) + assertFailureCode( + result, USER_WRITES_BLOCKED_ERROR, msg="findAndModify update should be blocked" + ) + + +def test_setUserWriteBlockMode_findAndModify_delete_blocked(collection): + """Test findAndModify with remove is blocked while write block is active.""" + collection.insert_one({"_id": "fam_del"}) + _enable_write_block(collection) + result = execute_command( + collection, + {"findAndModify": collection.name, "query": {"_id": "fam_del"}, "remove": True}, + ) + assertFailureCode( + result, USER_WRITES_BLOCKED_ERROR, msg="findAndModify remove should be blocked" + ) + + +def test_setUserWriteBlockMode_createIndexes_blocked(collection): + """Test createIndexes is blocked while write block is active.""" + collection.insert_one({"_id": "idx_doc", "a": 1}) + _enable_write_block(collection) + result = execute_command( + collection, + { + "createIndexes": collection.name, + "indexes": [{"key": {"a": 1}, "name": "a_1"}], + }, + ) + assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="createIndexes should be blocked") + + +def test_setUserWriteBlockMode_dropIndexes_blocked(collection): + """Test dropIndexes is blocked while write block is active.""" + collection.insert_one({"_id": "didx", "a": 1}) + collection.create_index([("a", 1)], name="a_1_to_drop") + _enable_write_block(collection) + result = execute_command(collection, {"dropIndexes": collection.name, "index": "a_1_to_drop"}) + assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="dropIndexes should be blocked") + + +def test_setUserWriteBlockMode_drop_collection_blocked(collection): + """Test drop collection is blocked while write block is active.""" + collection.insert_one({"_id": "drop_coll"}) + _enable_write_block(collection) + result = execute_command(collection, {"drop": collection.name}) + assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Drop collection should be blocked") + + +def test_setUserWriteBlockMode_create_collection_blocked(collection): + """Test creating a collection is blocked while write block is active.""" + _enable_write_block(collection) + result = execute_command(collection, {"create": "blocked_new_coll"}) + assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Create collection should be blocked") + + +def test_setUserWriteBlockMode_dropDatabase_blocked(database_client): + """Test dropDatabase is blocked while write block is active.""" + coll = database_client.create_collection("db_to_drop") + coll.insert_one({"_id": "seed"}) + execute_admin_command(coll, {"setUserWriteBlockMode": 1, "global": True}) + result = execute_command(coll, {"dropDatabase": 1}) + assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="dropDatabase should be blocked") + execute_admin_command(coll, {"setUserWriteBlockMode": 1, "global": False}) + + +# --- Reads not blocked while block is active --- + + +def test_setUserWriteBlockMode_find_not_blocked(collection): + """Test find succeeds while write block is active.""" + collection.insert_one({"_id": "read_doc", "val": 42}) + _enable_write_block(collection) + result = execute_command(collection, {"find": collection.name, "filter": {"_id": "read_doc"}}) + assertSuccessPartial(result, {"ok": 1.0}, msg="Find should succeed while block active") + + +def test_setUserWriteBlockMode_aggregate_read_not_blocked(collection): + """Test aggregate with no write stages succeeds while write block is active.""" + collection.insert_one({"_id": "agg_doc", "val": 1}) + _enable_write_block(collection) + result = execute_command( + collection, + {"aggregate": collection.name, "pipeline": [{"$match": {"val": 1}}], "cursor": {}}, + ) + assertSuccessPartial( + result, {"ok": 1.0}, msg="Aggregate read should succeed while block active" + ) + + +def test_setUserWriteBlockMode_count_not_blocked(collection): + """Test count succeeds while write block is active.""" + collection.insert_one({"_id": "count_doc"}) + _enable_write_block(collection) + result = execute_command(collection, {"count": collection.name}) + assertSuccessPartial(result, {"ok": 1.0}, msg="Count should succeed while block active") + + +def test_setUserWriteBlockMode_distinct_not_blocked(collection): + """Test distinct succeeds while write block is active.""" + collection.insert_one({"_id": "dist_doc", "x": 1}) + _enable_write_block(collection) + result = execute_command(collection, {"distinct": collection.name, "key": "x"}) + assertSuccessPartial(result, {"ok": 1.0}, msg="Distinct should succeed while block active") + + +# --- Writes succeed when block is disabled --- + + +def test_setUserWriteBlockMode_insert_succeeds_when_disabled(collection): + """Test insert succeeds when write block is not active.""" + result = execute_command( + collection, {"insert": collection.name, "documents": [{"_id": "not_blocked"}]} + ) + assertSuccessPartial(result, {"ok": 1.0}, msg="Insert should succeed with no block") + + +def test_setUserWriteBlockMode_update_succeeds_when_disabled(collection): + """Test update succeeds when write block is not active.""" + collection.insert_one({"_id": "upd_ok", "x": 1}) + result = execute_command( + collection, + {"update": collection.name, "updates": [{"q": {"_id": "upd_ok"}, "u": {"$set": {"x": 2}}}]}, + ) + assertSuccessPartial(result, {"ok": 1.0}, msg="Update should succeed with no block") + + +def test_setUserWriteBlockMode_delete_succeeds_when_disabled(collection): + """Test delete succeeds when write block is not active.""" + collection.insert_one({"_id": "del_ok"}) + result = execute_command( + collection, + {"delete": collection.name, "deletes": [{"q": {"_id": "del_ok"}, "limit": 1}]}, + ) + assertSuccessPartial(result, {"ok": 1.0}, msg="Delete should succeed with no block") 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 From e7cd17b798165e1dc61fcf5f7ead65c35ebed42a Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 12:17:35 -0700 Subject: [PATCH 02/10] change according to review Signed-off-by: Alina (Xi) Li --- ...tUserWriteBlockMode_argument_validation.py | 290 ++++++------------ ...est_setUserWriteBlockMode_core_behavior.py | 59 ++-- .../test_setUserWriteBlockMode_errors.py | 60 ++-- ...rWriteBlockMode_write_block_enforcement.py | 288 +++++++---------- 4 files changed, 263 insertions(+), 434 deletions(-) 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 index 637e62d69..1c7e7ef15 100644 --- 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 @@ -48,242 +48,148 @@ def _manage_write_block(collection): _force_disable_write_block(collection) -# --- global field: valid boolean values --- - - -def test_setUserWriteBlockMode_global_true_succeeds(collection): - """Test global:true (boolean) succeeds.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) - assertSuccessPartial(result, {"ok": 1.0}, msg="global:true should succeed") - - -def test_setUserWriteBlockMode_global_false_succeeds(collection): - """Test global:false (boolean) succeeds.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False}) - assertSuccessPartial(result, {"ok": 1.0}, msg="global:false should succeed") - - -# --- global field: numeric types rejected (no coercion) --- - - -def test_setUserWriteBlockMode_global_int_1_rejected(collection): - """Test global:1 (int32) is rejected — no coercion to bool.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": 1}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:int should be rejected") - - -def test_setUserWriteBlockMode_global_int_0_rejected(collection): - """Test global:0 (int32) is rejected — no coercion to bool.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": 0}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:int(0) should be rejected") - - -def test_setUserWriteBlockMode_global_double_1_rejected(collection): - """Test global:1.0 (double) is rejected — no coercion to bool.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": 1.0}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:double should be rejected") - - -def test_setUserWriteBlockMode_global_double_0_rejected(collection): - """Test global:0.0 (double) is rejected — no coercion to bool.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": 0.0}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:double(0.0) should be rejected") - - -def test_setUserWriteBlockMode_global_int64_rejected(collection): - """Test global:NumberLong(1) (int64) is rejected — no coercion to bool.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": Int64(1)}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:int64 should be rejected") - - -def test_setUserWriteBlockMode_global_decimal128_rejected(collection): - """Test global:NumberDecimal('1') (decimal128) is rejected — no coercion to bool.""" - result = execute_admin_command( - collection, {"setUserWriteBlockMode": 1, "global": Decimal128("1")} - ) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:decimal128 should be rejected") - - -def test_setUserWriteBlockMode_global_nan_rejected(collection): - """Test global:NaN (double) is rejected — no coercion to bool.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": FLOAT_NAN}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:NaN should be rejected") - - -def test_setUserWriteBlockMode_global_infinity_rejected(collection): - """Test global:Infinity (double) is rejected — no coercion to bool.""" - result = execute_admin_command( - collection, {"setUserWriteBlockMode": 1, "global": FLOAT_INFINITY} +# Property [Global Field Boolean Acceptance]: setUserWriteBlockMode accepts only boolean values +# for the global field. +@pytest.mark.parametrize( + "value", + [pytest.param(True, id="global_true"), pytest.param(False, id="global_false")], +) +def test_setUserWriteBlockMode_global_valid(collection, value): + """Test setUserWriteBlockMode accepts boolean global field.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": value}) + assertSuccessPartial( + result, {"ok": 1.0}, msg="setUserWriteBlockMode should accept boolean global field" ) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:Infinity should be rejected") -def test_setUserWriteBlockMode_global_negative_infinity_rejected(collection): - """Test global:-Infinity (double) is rejected — no coercion to bool.""" - result = execute_admin_command( - collection, {"setUserWriteBlockMode": 1, "global": float("-inf")} +# Property [Global Field Type Rejection]: setUserWriteBlockMode rejects all non-boolean types +# for the global field with no coercion. +@pytest.mark.parametrize( + "value,error_code", + [ + pytest.param(1, TYPE_MISMATCH_ERROR, id="int32_1"), + pytest.param(0, TYPE_MISMATCH_ERROR, id="int32_0"), + pytest.param(1.0, TYPE_MISMATCH_ERROR, id="double_1"), + pytest.param(0.0, TYPE_MISMATCH_ERROR, id="double_0"), + pytest.param(Int64(1), TYPE_MISMATCH_ERROR, id="int64"), + pytest.param(Decimal128("1"), TYPE_MISMATCH_ERROR, id="decimal128"), + pytest.param(FLOAT_NAN, TYPE_MISMATCH_ERROR, id="nan"), + pytest.param(FLOAT_INFINITY, TYPE_MISMATCH_ERROR, id="infinity"), + pytest.param(float("-inf"), TYPE_MISMATCH_ERROR, id="negative_infinity"), + pytest.param(-0.0, TYPE_MISMATCH_ERROR, id="negative_zero"), + pytest.param("true", TYPE_MISMATCH_ERROR, id="string"), + pytest.param([], TYPE_MISMATCH_ERROR, id="array"), + pytest.param({}, TYPE_MISMATCH_ERROR, id="object"), + pytest.param(None, MISSING_FIELD_ERROR, id="null"), + ], +) +def test_setUserWriteBlockMode_global_type_rejected(collection, value, error_code): + """Test setUserWriteBlockMode rejects non-boolean global field.""" + result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": value}) + assertFailureCode( + result, error_code, msg="setUserWriteBlockMode should reject non-boolean global field" ) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:-Infinity should be rejected") - - -def test_setUserWriteBlockMode_global_negative_zero_rejected(collection): - """Test global:-0.0 (negative zero double) is rejected — no coercion to bool.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": -0.0}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:-0.0 should be rejected") - - -# --- global field: non-numeric invalid types --- - - -def test_setUserWriteBlockMode_global_null_fails(collection): - """Test global:null is treated as missing required field.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": None}) - assertFailureCode(result, MISSING_FIELD_ERROR, msg="global:null should fail as missing") - - -def test_setUserWriteBlockMode_global_string_fails(collection): - """Test global:'true' (string) fails with type error.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": "true"}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:string should fail") - - -def test_setUserWriteBlockMode_global_array_fails(collection): - """Test global:[] (array) fails with type error.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": []}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:array should fail") - - -def test_setUserWriteBlockMode_global_object_fails(collection): - """Test global:{} (object) fails with type error.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": {}}) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="global:object should fail") - - -# --- missing global field --- +# Property [Missing Global Field]: setUserWriteBlockMode requires the global field. def test_setUserWriteBlockMode_missing_global_fails(collection): - """Test setUserWriteBlockMode without global field fails with missing field error.""" + """Test setUserWriteBlockMode without global field fails.""" result = execute_admin_command(collection, {"setUserWriteBlockMode": 1}) - assertFailureCode(result, MISSING_FIELD_ERROR, msg="Missing global should fail") - - -# --- reason field: valid values --- - - -def test_setUserWriteBlockMode_reason_unspecified_succeeds(collection): - """Test reason:'Unspecified' succeeds.""" - result = execute_admin_command( - collection, - {"setUserWriteBlockMode": 1, "global": True, "reason": "Unspecified"}, + assertFailureCode( + result, + MISSING_FIELD_ERROR, + msg="setUserWriteBlockMode should require the global field", ) - assertSuccessPartial(result, {"ok": 1.0}, msg="reason:Unspecified should succeed") -def test_setUserWriteBlockMode_reason_cluster_migration_succeeds(collection): - """Test reason:'ClusterToClusterMigrationInProgress' succeeds.""" +# Property [Reason Field Valid Values]: setUserWriteBlockMode accepts valid reason enum strings. +@pytest.mark.parametrize( + "reason", + [ + pytest.param("Unspecified", id="unspecified"), + pytest.param("ClusterToClusterMigrationInProgress", id="cluster_migration"), + pytest.param("DiskUseThresholdExceeded", id="disk_threshold"), + ], +) +def test_setUserWriteBlockMode_reason_valid(collection, reason): + """Test setUserWriteBlockMode accepts valid reason values.""" result = execute_admin_command( collection, - { - "setUserWriteBlockMode": 1, - "global": True, - "reason": "ClusterToClusterMigrationInProgress", - }, + {"setUserWriteBlockMode": 1, "global": True, "reason": reason}, ) assertSuccessPartial( - result, {"ok": 1.0}, msg="reason:ClusterToClusterMigrationInProgress should succeed" + result, {"ok": 1.0}, msg="setUserWriteBlockMode should accept valid reason string" ) -def test_setUserWriteBlockMode_reason_disk_threshold_succeeds(collection): - """Test reason:'DiskUseThresholdExceeded' succeeds.""" - result = execute_admin_command( - collection, - { - "setUserWriteBlockMode": 1, - "global": True, - "reason": "DiskUseThresholdExceeded", - }, - ) - assertSuccessPartial(result, {"ok": 1.0}, msg="reason:DiskUseThresholdExceeded should succeed") - - -def test_setUserWriteBlockMode_no_reason_field_succeeds(collection): - """Test omitting reason field succeeds (defaults to Unspecified).""" +# Property [Reason Field Optional]: the reason field can be omitted or null. +def test_setUserWriteBlockMode_reason_omitted_succeeds(collection): + """Test setUserWriteBlockMode succeeds when reason field is omitted.""" result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) - assertSuccessPartial(result, {"ok": 1.0}, msg="Omitting reason should succeed") + assertSuccessPartial( + result, {"ok": 1.0}, msg="setUserWriteBlockMode should succeed without reason field" + ) def test_setUserWriteBlockMode_reason_null_succeeds(collection): - """Test reason:null is treated as omitted and succeeds.""" + """Test setUserWriteBlockMode succeeds when reason is null.""" result = execute_admin_command( collection, {"setUserWriteBlockMode": 1, "global": True, "reason": None} ) - assertSuccessPartial(result, {"ok": 1.0}, msg="reason:null should succeed as omitted") - - -# --- reason field: invalid types --- - - -def test_setUserWriteBlockMode_reason_int_fails(collection): - """Test reason:1 (int) fails with type error.""" - result = execute_admin_command( - collection, {"setUserWriteBlockMode": 1, "global": True, "reason": 1} - ) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="reason:int should fail") - - -def test_setUserWriteBlockMode_reason_bool_fails(collection): - """Test reason:true (bool) fails with type error.""" - result = execute_admin_command( - collection, {"setUserWriteBlockMode": 1, "global": True, "reason": True} + assertSuccessPartial( + result, + {"ok": 1.0}, + msg="setUserWriteBlockMode should treat null reason as omitted", ) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="reason:bool should fail") -def test_setUserWriteBlockMode_reason_array_fails(collection): - """Test reason:[] (array) fails with type error.""" +# Property [Reason Field Type Rejection]: setUserWriteBlockMode rejects non-string types for +# the reason field. +@pytest.mark.parametrize( + "value,error_code", + [ + pytest.param(1, TYPE_MISMATCH_ERROR, id="int"), + pytest.param(True, TYPE_MISMATCH_ERROR, id="bool"), + pytest.param([], TYPE_MISMATCH_ERROR, id="array"), + pytest.param({}, TYPE_MISMATCH_ERROR, id="object"), + ], +) +def test_setUserWriteBlockMode_reason_type_rejected(collection, value, error_code): + """Test setUserWriteBlockMode rejects non-string reason field.""" result = execute_admin_command( - collection, {"setUserWriteBlockMode": 1, "global": True, "reason": []} + collection, {"setUserWriteBlockMode": 1, "global": True, "reason": value} ) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="reason:array should fail") - - -def test_setUserWriteBlockMode_reason_object_fails(collection): - """Test reason:{} (object) fails with type error.""" - result = execute_admin_command( - collection, {"setUserWriteBlockMode": 1, "global": True, "reason": {}} + assertFailureCode( + result, + error_code, + msg="setUserWriteBlockMode should reject non-string reason field", ) - assertFailureCode(result, TYPE_MISMATCH_ERROR, msg="reason:object should fail") +# Property [Reason Field Invalid Enum]: setUserWriteBlockMode rejects unrecognized reason +# strings. def test_setUserWriteBlockMode_reason_invalid_enum_fails(collection): - """Test reason with unrecognized string value fails with BadValue.""" + """Test setUserWriteBlockMode rejects unrecognized reason string.""" result = execute_admin_command( collection, {"setUserWriteBlockMode": 1, "global": True, "reason": "InvalidReason"}, ) - assertFailureCode(result, BAD_VALUE_ERROR, msg="Invalid reason enum should fail") - - -# --- unrecognized fields --- + assertFailureCode( + result, + BAD_VALUE_ERROR, + msg="setUserWriteBlockMode should reject unrecognized reason enum value", + ) +# Property [Unrecognized Fields]: setUserWriteBlockMode rejects unknown fields. def test_setUserWriteBlockMode_unrecognized_field_fails(collection): - """Test setUserWriteBlockMode with extra unrecognized field fails.""" + """Test setUserWriteBlockMode rejects extra unrecognized fields.""" result = execute_admin_command( collection, {"setUserWriteBlockMode": 1, "global": False, "unknownField": 1}, ) assertFailureCode( - result, UNRECOGNIZED_COMMAND_FIELD_ERROR, msg="Unrecognized field should fail" + result, + UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="setUserWriteBlockMode should reject unrecognized fields", ) - - -# --- command field value --- - - -def test_setUserWriteBlockMode_command_value_1_succeeds(collection): - """Test setUserWriteBlockMode:1 is the correct command value.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False}) - assertSuccessPartial(result, {"ok": 1.0}, msg="Command value 1 should succeed") 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 index 1fa47b1ea..43db8bf2f 100644 --- 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 @@ -1,7 +1,6 @@ """Tests for setUserWriteBlockMode core behavior. -Validates enable/disable semantics, response structure, idempotent behavior, -and state restoration. +Validates enable/disable semantics, idempotent behavior, and state restoration. """ import pytest @@ -40,56 +39,54 @@ def _manage_write_block(collection): _force_disable_write_block(collection) -def test_setUserWriteBlockMode_enable_returns_ok(collection): - """Test setUserWriteBlockMode with global:true returns ok:1.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) - assertSuccessPartial(result, {"ok": 1.0}, msg="Should return ok:1 on enable") - - -def test_setUserWriteBlockMode_disable_returns_ok(collection): - """Test setUserWriteBlockMode with global:false returns ok:1.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": False}) - assertSuccessPartial(result, {"ok": 1.0}, msg="Should return ok:1 on disable") - - -def test_setUserWriteBlockMode_disable_when_no_block_active_idempotent(collection): - """Test global:false when no block is active succeeds (idempotent).""" +# 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="Should succeed when disabling with no active block" + result, + {"ok": 1.0}, + msg="setUserWriteBlockMode should succeed when disabling with no active block", ) -def test_setUserWriteBlockMode_enable_disable_cycle_restores_writes(collection): - """Test enable block then disable, verify writes succeed again.""" +# 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="Write should succeed after block disabled") + assertSuccessPartial( + result, + {"ok": 1.0}, + msg="setUserWriteBlockMode should allow writes after block is disabled", + ) -def test_setUserWriteBlockMode_toggle_multiple_times_no_error(collection): - """Test toggling write block off and on multiple times without errors.""" +# 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="Should succeed after repeated toggling") + 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 enabling write block again with same default reason is idempotent.""" + """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="Should succeed when re-enabling with same reason" + result, + {"ok": 1.0}, + msg="setUserWriteBlockMode should be idempotent when re-enabling with same reason", ) - - -def test_setUserWriteBlockMode_response_contains_ok_field(collection): - """Test successful command returns document with ok:1.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) - assertSuccessPartial(result, {"ok": 1.0}, msg="Response should contain ok:1") 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 index 13e819e78..72ef5907a 100644 --- 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 @@ -1,6 +1,6 @@ """Tests for setUserWriteBlockMode error cases. -Validates mismatched reason errors and idempotent behavior with same reason. +Validates mismatched reason errors when changing the reason on an active block. """ import pytest @@ -40,11 +40,9 @@ def _manage_write_block(collection): _force_disable_write_block(collection) -# --- Mismatched reason errors --- - - +# Property [Mismatched Reason on Enable]: re-enabling with a different reason fails. def test_setUserWriteBlockMode_enable_mismatched_reason_fails(collection): - """Test re-enabling with different reason fails with IllegalOperation.""" + """Test setUserWriteBlockMode re-enable with different reason fails.""" execute_admin_command( collection, {"setUserWriteBlockMode": 1, "global": True, "reason": "Unspecified"}, @@ -60,12 +58,14 @@ def test_setUserWriteBlockMode_enable_mismatched_reason_fails(collection): assertFailureCode( result, ILLEGAL_OPERATION_ERROR, - msg="Mismatched reason on enable should fail with IllegalOperation", + 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 disabling with different reason than enabled fails with IllegalOperation.""" + """Test setUserWriteBlockMode disable with different reason fails.""" execute_admin_command( collection, {"setUserWriteBlockMode": 1, "global": True, "reason": "Unspecified"}, @@ -81,48 +81,30 @@ def test_setUserWriteBlockMode_disable_mismatched_reason_fails(collection): assertFailureCode( result, ILLEGAL_OPERATION_ERROR, - msg="Mismatched reason on disable should fail with IllegalOperation", - ) - - -# --- Idempotent with same reason --- - - -def test_setUserWriteBlockMode_same_reason_unspecified_idempotent(collection): - """Test re-enabling with same reason 'Unspecified' succeeds (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="Same reason Unspecified re-enable should be idempotent" + msg="setUserWriteBlockMode should reject mismatched reason on disable", ) -def test_setUserWriteBlockMode_same_reason_migration_idempotent(collection): - """Test re-enabling with same reason 'ClusterToClusterMigrationInProgress' succeeds.""" +# Property [Same Reason Idempotent]: re-enabling with same explicit reason succeeds. +@pytest.mark.parametrize( + "reason", + [ + pytest.param("Unspecified", id="unspecified"), + pytest.param("ClusterToClusterMigrationInProgress", id="cluster_migration"), + ], +) +def test_setUserWriteBlockMode_same_reason_idempotent(collection, reason): + """Test setUserWriteBlockMode re-enable with same reason is idempotent.""" execute_admin_command( collection, - { - "setUserWriteBlockMode": 1, - "global": True, - "reason": "ClusterToClusterMigrationInProgress", - }, + {"setUserWriteBlockMode": 1, "global": True, "reason": reason}, ) result = execute_admin_command( collection, - { - "setUserWriteBlockMode": 1, - "global": True, - "reason": "ClusterToClusterMigrationInProgress", - }, + {"setUserWriteBlockMode": 1, "global": True, "reason": reason}, ) assertSuccessPartial( result, {"ok": 1.0}, - msg="Same reason ClusterToClusterMigrationInProgress re-enable should be idempotent", + msg="setUserWriteBlockMode should be idempotent with same explicit reason", ) 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 index ef2494c0b..18a7fd96b 100644 --- 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 @@ -46,187 +46,131 @@ def _enable_write_block(collection): execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) -# --- Writes blocked while block is active --- - - -def test_setUserWriteBlockMode_insert_blocked(collection): - """Test insert is rejected with UserWritesBlocked while write block is active.""" - _enable_write_block(collection) - result = execute_command( - collection, {"insert": collection.name, "documents": [{"_id": "blocked"}]} - ) - assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Insert should be blocked") - - -def test_setUserWriteBlockMode_update_blocked(collection): - """Test update is rejected with UserWritesBlocked while write block is active.""" - collection.insert_one({"_id": "upd_target", "x": 1}) - _enable_write_block(collection) - result = execute_command( - collection, - { - "update": collection.name, - "updates": [{"q": {"_id": "upd_target"}, "u": {"$set": {"x": 2}}}], - }, - ) - assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Update should be blocked") - - -def test_setUserWriteBlockMode_delete_blocked(collection): - """Test delete is rejected with UserWritesBlocked while write block is active.""" - collection.insert_one({"_id": "del_target"}) - _enable_write_block(collection) - result = execute_command( - collection, - {"delete": collection.name, "deletes": [{"q": {"_id": "del_target"}, "limit": 1}]}, - ) - assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Delete should be blocked") - - -def test_setUserWriteBlockMode_findAndModify_update_blocked(collection): - """Test findAndModify with update is blocked while write block is active.""" - collection.insert_one({"_id": "fam_target", "x": 1}) - _enable_write_block(collection) - result = execute_command( - collection, - { - "findAndModify": collection.name, - "query": {"_id": "fam_target"}, - "update": {"$set": {"x": 2}}, +# Property [Write Operations Blocked]: all write operations are rejected while the block is +# active. +_WRITE_BLOCKED_PARAMS = [ + pytest.param( + lambda name: {"insert": name, "documents": [{"_id": "blocked"}]}, + id="insert", + ), + pytest.param( + lambda name: {"update": name, "updates": [{"q": {}, "u": {"$set": {"x": 2}}}]}, + id="update", + ), + pytest.param( + lambda name: {"delete": name, "deletes": [{"q": {}, "limit": 1}]}, + id="delete", + ), + pytest.param( + lambda name: {"findAndModify": name, "query": {}, "update": {"$set": {"x": 2}}}, + id="findAndModify_update", + ), + pytest.param( + lambda name: {"findAndModify": name, "query": {}, "remove": True}, + id="findAndModify_remove", + ), + pytest.param( + lambda name: { + "createIndexes": name, + "indexes": [{"key": {"blocked_field": 1}, "name": "blocked_field_1"}], }, - ) - assertFailureCode( - result, USER_WRITES_BLOCKED_ERROR, msg="findAndModify update should be blocked" - ) - - -def test_setUserWriteBlockMode_findAndModify_delete_blocked(collection): - """Test findAndModify with remove is blocked while write block is active.""" - collection.insert_one({"_id": "fam_del"}) + id="createIndexes", + ), + pytest.param( + lambda name: {"dropIndexes": name, "index": "a_1"}, + id="dropIndexes", + ), + pytest.param( + lambda name: {"drop": name}, + id="drop_collection", + ), + pytest.param( + lambda name: {"create": f"{name}_blocked_new"}, + id="create_collection", + ), + pytest.param( + lambda name: {"dropDatabase": 1}, + id="dropDatabase", + ), +] + + +@pytest.mark.parametrize("build_command", _WRITE_BLOCKED_PARAMS) +def test_setUserWriteBlockMode_write_operation_blocked(collection, build_command): + """Test write operation is rejected while write block is active.""" + collection.insert_one({"_id": "seed", "a": 1}) + collection.create_index([("a", 1)], name="a_1") _enable_write_block(collection) - result = execute_command( - collection, - {"findAndModify": collection.name, "query": {"_id": "fam_del"}, "remove": True}, - ) + command = build_command(collection.name) + result = execute_command(collection, command) assertFailureCode( - result, USER_WRITES_BLOCKED_ERROR, msg="findAndModify remove should be blocked" + result, + USER_WRITES_BLOCKED_ERROR, + msg="setUserWriteBlockMode should block write operations while active", ) -def test_setUserWriteBlockMode_createIndexes_blocked(collection): - """Test createIndexes is blocked while write block is active.""" - collection.insert_one({"_id": "idx_doc", "a": 1}) - _enable_write_block(collection) - result = execute_command( - collection, - { - "createIndexes": collection.name, - "indexes": [{"key": {"a": 1}, "name": "a_1"}], - }, - ) - assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="createIndexes should be blocked") - - -def test_setUserWriteBlockMode_dropIndexes_blocked(collection): - """Test dropIndexes is blocked while write block is active.""" - collection.insert_one({"_id": "didx", "a": 1}) - collection.create_index([("a", 1)], name="a_1_to_drop") - _enable_write_block(collection) - result = execute_command(collection, {"dropIndexes": collection.name, "index": "a_1_to_drop"}) - assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="dropIndexes should be blocked") - - -def test_setUserWriteBlockMode_drop_collection_blocked(collection): - """Test drop collection is blocked while write block is active.""" - collection.insert_one({"_id": "drop_coll"}) +# Property [Read Operations Not Blocked]: read operations succeed while the block is active. +_READ_NOT_BLOCKED_PARAMS = [ + pytest.param( + lambda name: {"find": name, "filter": {}}, + id="find", + ), + pytest.param( + lambda name: {"aggregate": name, "pipeline": [{"$match": {}}], "cursor": {}}, + id="aggregate", + ), + pytest.param( + lambda name: {"count": name}, + id="count", + ), + pytest.param( + lambda name: {"distinct": name, "key": "x"}, + id="distinct", + ), +] + + +@pytest.mark.parametrize("build_command", _READ_NOT_BLOCKED_PARAMS) +def test_setUserWriteBlockMode_read_operation_not_blocked(collection, build_command): + """Test read operation succeeds while write block is active.""" + collection.insert_one({"_id": "read_doc", "x": 1}) _enable_write_block(collection) - result = execute_command(collection, {"drop": collection.name}) - assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Drop collection should be blocked") - - -def test_setUserWriteBlockMode_create_collection_blocked(collection): - """Test creating a collection is blocked while write block is active.""" - _enable_write_block(collection) - result = execute_command(collection, {"create": "blocked_new_coll"}) - assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="Create collection should be blocked") - - -def test_setUserWriteBlockMode_dropDatabase_blocked(database_client): - """Test dropDatabase is blocked while write block is active.""" - coll = database_client.create_collection("db_to_drop") - coll.insert_one({"_id": "seed"}) - execute_admin_command(coll, {"setUserWriteBlockMode": 1, "global": True}) - result = execute_command(coll, {"dropDatabase": 1}) - assertFailureCode(result, USER_WRITES_BLOCKED_ERROR, msg="dropDatabase should be blocked") - execute_admin_command(coll, {"setUserWriteBlockMode": 1, "global": False}) - - -# --- Reads not blocked while block is active --- - - -def test_setUserWriteBlockMode_find_not_blocked(collection): - """Test find succeeds while write block is active.""" - collection.insert_one({"_id": "read_doc", "val": 42}) - _enable_write_block(collection) - result = execute_command(collection, {"find": collection.name, "filter": {"_id": "read_doc"}}) - assertSuccessPartial(result, {"ok": 1.0}, msg="Find should succeed while block active") - - -def test_setUserWriteBlockMode_aggregate_read_not_blocked(collection): - """Test aggregate with no write stages succeeds while write block is active.""" - collection.insert_one({"_id": "agg_doc", "val": 1}) - _enable_write_block(collection) - result = execute_command( - collection, - {"aggregate": collection.name, "pipeline": [{"$match": {"val": 1}}], "cursor": {}}, - ) + command = build_command(collection.name) + result = execute_command(collection, command) assertSuccessPartial( - result, {"ok": 1.0}, msg="Aggregate read should succeed while block active" + result, + {"ok": 1.0}, + msg="setUserWriteBlockMode should not block read operations", ) -def test_setUserWriteBlockMode_count_not_blocked(collection): - """Test count succeeds while write block is active.""" - collection.insert_one({"_id": "count_doc"}) - _enable_write_block(collection) - result = execute_command(collection, {"count": collection.name}) - assertSuccessPartial(result, {"ok": 1.0}, msg="Count should succeed while block active") - - -def test_setUserWriteBlockMode_distinct_not_blocked(collection): - """Test distinct succeeds while write block is active.""" - collection.insert_one({"_id": "dist_doc", "x": 1}) - _enable_write_block(collection) - result = execute_command(collection, {"distinct": collection.name, "key": "x"}) - assertSuccessPartial(result, {"ok": 1.0}, msg="Distinct should succeed while block active") - - -# --- Writes succeed when block is disabled --- - - -def test_setUserWriteBlockMode_insert_succeeds_when_disabled(collection): - """Test insert succeeds when write block is not active.""" - result = execute_command( - collection, {"insert": collection.name, "documents": [{"_id": "not_blocked"}]} - ) - assertSuccessPartial(result, {"ok": 1.0}, msg="Insert should succeed with no block") - - -def test_setUserWriteBlockMode_update_succeeds_when_disabled(collection): - """Test update succeeds when write block is not active.""" - collection.insert_one({"_id": "upd_ok", "x": 1}) - result = execute_command( - collection, - {"update": collection.name, "updates": [{"q": {"_id": "upd_ok"}, "u": {"$set": {"x": 2}}}]}, - ) - assertSuccessPartial(result, {"ok": 1.0}, msg="Update should succeed with no block") - - -def test_setUserWriteBlockMode_delete_succeeds_when_disabled(collection): - """Test delete succeeds when write block is not active.""" - collection.insert_one({"_id": "del_ok"}) - result = execute_command( - collection, - {"delete": collection.name, "deletes": [{"q": {"_id": "del_ok"}, "limit": 1}]}, +# Property [Writes Succeed When Disabled]: write operations succeed when no block is active. +_WRITE_SUCCEEDS_PARAMS = [ + pytest.param( + lambda name: {"insert": name, "documents": [{"_id": "ok"}]}, + id="insert", + ), + pytest.param( + lambda name: {"update": name, "updates": [{"q": {"_id": "upd"}, "u": {"$set": {"x": 2}}}]}, + id="update", + ), + pytest.param( + lambda name: {"delete": name, "deletes": [{"q": {"_id": "del"}, "limit": 1}]}, + id="delete", + ), +] + + +@pytest.mark.parametrize("build_command", _WRITE_SUCCEEDS_PARAMS) +def test_setUserWriteBlockMode_write_succeeds_when_disabled(collection, build_command): + """Test write operation succeeds when write block is not active.""" + collection.insert_one({"_id": "upd", "x": 1}) + collection.insert_one({"_id": "del"}) + command = build_command(collection.name) + result = execute_command(collection, command) + assertSuccessPartial( + result, + {"ok": 1.0}, + msg="setUserWriteBlockMode should allow writes when block is not active", ) - assertSuccessPartial(result, {"ok": 1.0}, msg="Delete should succeed with no block") From 575506945712b09a7f7e371de97b1770fe88c577 Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 14:02:43 -0700 Subject: [PATCH 03/10] add AdminTestCase Signed-off-by: Alina (Xi) Li --- ...tUserWriteBlockMode_argument_validation.py | 265 ++++++++++-------- .../test_setUserWriteBlockMode_errors.py | 60 ++-- .../administration/commands/utils/__init__.py | 0 .../commands/utils/admin_test_case.py | 25 ++ 4 files changed, 211 insertions(+), 139 deletions(-) create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/utils/__init__.py create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/utils/admin_test_case.py 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 index 1c7e7ef15..4ba640a3b 100644 --- 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 @@ -4,10 +4,18 @@ and command value handling. """ +from __future__ import annotations + import pytest from bson import Decimal128, Int64 -from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial +from documentdb_tests.compatibility.tests.core.utils.command_test_case import ( + CommandContext, +) +from documentdb_tests.compatibility.tests.system.administration.commands.utils.admin_test_case import ( # noqa: E501 + AdminTestCase, +) +from documentdb_tests.framework.assertions import assertResult from documentdb_tests.framework.error_codes import ( BAD_VALUE_ERROR, MISSING_FIELD_ERROR, @@ -15,6 +23,7 @@ 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)] @@ -50,146 +59,164 @@ def _manage_write_block(collection): # Property [Global Field Boolean Acceptance]: setUserWriteBlockMode accepts only boolean values # for the global field. -@pytest.mark.parametrize( - "value", - [pytest.param(True, id="global_true"), pytest.param(False, id="global_false")], -) -def test_setUserWriteBlockMode_global_valid(collection, value): - """Test setUserWriteBlockMode accepts boolean global field.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": value}) - assertSuccessPartial( - result, {"ok": 1.0}, msg="setUserWriteBlockMode should accept boolean 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 [Global Field Type Rejection]: setUserWriteBlockMode rejects all non-boolean types # for the global field with no coercion. -@pytest.mark.parametrize( - "value,error_code", - [ - pytest.param(1, TYPE_MISMATCH_ERROR, id="int32_1"), - pytest.param(0, TYPE_MISMATCH_ERROR, id="int32_0"), - pytest.param(1.0, TYPE_MISMATCH_ERROR, id="double_1"), - pytest.param(0.0, TYPE_MISMATCH_ERROR, id="double_0"), - pytest.param(Int64(1), TYPE_MISMATCH_ERROR, id="int64"), - pytest.param(Decimal128("1"), TYPE_MISMATCH_ERROR, id="decimal128"), - pytest.param(FLOAT_NAN, TYPE_MISMATCH_ERROR, id="nan"), - pytest.param(FLOAT_INFINITY, TYPE_MISMATCH_ERROR, id="infinity"), - pytest.param(float("-inf"), TYPE_MISMATCH_ERROR, id="negative_infinity"), - pytest.param(-0.0, TYPE_MISMATCH_ERROR, id="negative_zero"), - pytest.param("true", TYPE_MISMATCH_ERROR, id="string"), - pytest.param([], TYPE_MISMATCH_ERROR, id="array"), - pytest.param({}, TYPE_MISMATCH_ERROR, id="object"), - pytest.param(None, MISSING_FIELD_ERROR, id="null"), - ], -) -def test_setUserWriteBlockMode_global_type_rejected(collection, value, error_code): - """Test setUserWriteBlockMode rejects non-boolean global field.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": value}) - assertFailureCode( - result, error_code, msg="setUserWriteBlockMode should reject non-boolean global field" +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), + ("null", None, MISSING_FIELD_ERROR), + ] +] # Property [Missing Global Field]: setUserWriteBlockMode requires the global field. -def test_setUserWriteBlockMode_missing_global_fails(collection): - """Test setUserWriteBlockMode without global field fails.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1}) - assertFailureCode( - result, - MISSING_FIELD_ERROR, +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", - ) - + ), +] # Property [Reason Field Valid Values]: setUserWriteBlockMode accepts valid reason enum strings. -@pytest.mark.parametrize( - "reason", - [ - pytest.param("Unspecified", id="unspecified"), - pytest.param("ClusterToClusterMigrationInProgress", id="cluster_migration"), - pytest.param("DiskUseThresholdExceeded", id="disk_threshold"), - ], -) -def test_setUserWriteBlockMode_reason_valid(collection, reason): - """Test setUserWriteBlockMode accepts valid reason values.""" - result = execute_admin_command( - collection, - {"setUserWriteBlockMode": 1, "global": True, "reason": reason}, +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}", ) - assertSuccessPartial( - result, {"ok": 1.0}, msg="setUserWriteBlockMode should accept valid reason string" - ) - + for tid, reason in [ + ("unspecified", "Unspecified"), + ("cluster_migration", "ClusterToClusterMigrationInProgress"), + ("disk_threshold", "DiskUseThresholdExceeded"), + ] +] # Property [Reason Field Optional]: the reason field can be omitted or null. -def test_setUserWriteBlockMode_reason_omitted_succeeds(collection): - """Test setUserWriteBlockMode succeeds when reason field is omitted.""" - result = execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) - assertSuccessPartial( - result, {"ok": 1.0}, msg="setUserWriteBlockMode should succeed without reason field" - ) - - -def test_setUserWriteBlockMode_reason_null_succeeds(collection): - """Test setUserWriteBlockMode succeeds when reason is null.""" - result = execute_admin_command( - collection, {"setUserWriteBlockMode": 1, "global": True, "reason": None} - ) - assertSuccessPartial( - result, - {"ok": 1.0}, +REASON_OPTIONAL_TESTS: list[AdminTestCase] = [ + AdminTestCase( + "reason_omitted", + command=lambda ctx: {"setUserWriteBlockMode": 1, "global": True}, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should succeed without reason field", + ), + 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 [Reason Field Type Rejection]: setUserWriteBlockMode rejects non-string types for # the reason field. -@pytest.mark.parametrize( - "value,error_code", - [ - pytest.param(1, TYPE_MISMATCH_ERROR, id="int"), - pytest.param(True, TYPE_MISMATCH_ERROR, id="bool"), - pytest.param([], TYPE_MISMATCH_ERROR, id="array"), - pytest.param({}, TYPE_MISMATCH_ERROR, id="object"), - ], -) -def test_setUserWriteBlockMode_reason_type_rejected(collection, value, error_code): - """Test setUserWriteBlockMode rejects non-string reason field.""" - result = execute_admin_command( - collection, {"setUserWriteBlockMode": 1, "global": True, "reason": value} +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", ) - assertFailureCode( - result, - error_code, - msg="setUserWriteBlockMode should reject non-string reason field", - ) - + for tid, val in [ + ("int", 1), + ("bool", True), + ("array", []), + ("object", {}), + ] +] # Property [Reason Field Invalid Enum]: setUserWriteBlockMode rejects unrecognized reason # strings. -def test_setUserWriteBlockMode_reason_invalid_enum_fails(collection): - """Test setUserWriteBlockMode rejects unrecognized reason string.""" - result = execute_admin_command( - collection, - {"setUserWriteBlockMode": 1, "global": True, "reason": "InvalidReason"}, - ) - assertFailureCode( - result, - BAD_VALUE_ERROR, +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. -def test_setUserWriteBlockMode_unrecognized_field_fails(collection): - """Test setUserWriteBlockMode rejects extra unrecognized fields.""" - result = execute_admin_command( - collection, - {"setUserWriteBlockMode": 1, "global": False, "unknownField": 1}, - ) - assertFailureCode( - result, - UNRECOGNIZED_COMMAND_FIELD_ERROR, +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_VALIDATION_TESTS: list[AdminTestCase] = ( + GLOBAL_VALID_TESTS + + GLOBAL_TYPE_REJECTION_TESTS + + MISSING_GLOBAL_TESTS + + REASON_VALID_TESTS + + REASON_OPTIONAL_TESTS + + REASON_TYPE_REJECTION_TESTS + + REASON_INVALID_ENUM_TESTS + + UNRECOGNIZED_FIELD_TESTS +) + + +@pytest.mark.parametrize("test", pytest_params(ARGUMENT_VALIDATION_TESTS)) +def test_setUserWriteBlockMode_argument_validation(collection, test): + """Test setUserWriteBlockMode argument validation.""" + ctx = CommandContext.from_collection(collection) + result = execute_admin_command(collection, test.build_command(ctx)) + assertResult( + result, + expected=test.expected, + error_code=test.error_code, + msg=test.msg, + raw_res=True, ) 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 index 72ef5907a..280125b7f 100644 --- 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 @@ -3,11 +3,20 @@ Validates mismatched reason errors when changing the reason on an active block. """ +from __future__ import annotations + import pytest -from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial +from documentdb_tests.compatibility.tests.core.utils.command_test_case import ( + CommandContext, +) +from documentdb_tests.compatibility.tests.system.administration.commands.utils.admin_test_case import ( # noqa: E501 + AdminTestCase, +) +from documentdb_tests.framework.assertions import assertFailureCode, assertResult from documentdb_tests.framework.error_codes import ILLEGAL_OPERATION_ERROR from documentdb_tests.framework.executor import execute_admin_command +from documentdb_tests.framework.parametrize import pytest_params pytestmark = [pytest.mark.admin, pytest.mark.no_parallel, pytest.mark.requires(cluster_admin=True)] @@ -86,25 +95,36 @@ def test_setUserWriteBlockMode_disable_mismatched_reason_fails(collection): # Property [Same Reason Idempotent]: re-enabling with same explicit reason succeeds. -@pytest.mark.parametrize( - "reason", - [ - pytest.param("Unspecified", id="unspecified"), - pytest.param("ClusterToClusterMigrationInProgress", id="cluster_migration"), - ], -) -def test_setUserWriteBlockMode_same_reason_idempotent(collection, reason): - """Test setUserWriteBlockMode re-enable with same reason is idempotent.""" - execute_admin_command( - collection, - {"setUserWriteBlockMode": 1, "global": True, "reason": reason}, - ) - result = execute_admin_command( - collection, - {"setUserWriteBlockMode": 1, "global": True, "reason": reason}, +SAME_REASON_TESTS: list[AdminTestCase] = [ + AdminTestCase( + f"same_reason_{tid}", + command=lambda ctx, r=reason: { + "setUserWriteBlockMode": 1, + "global": True, + "reason": r, + }, + expected={"ok": 1.0}, + msg=f"setUserWriteBlockMode should be idempotent with same reason {reason}", ) - assertSuccessPartial( + for tid, reason in [ + ("unspecified", "Unspecified"), + ("cluster_migration", "ClusterToClusterMigrationInProgress"), + ] +] + + +@pytest.mark.parametrize("test", pytest_params(SAME_REASON_TESTS)) +def test_setUserWriteBlockMode_same_reason_idempotent(collection, test): + """Test setUserWriteBlockMode re-enable with same reason is idempotent.""" + ctx = CommandContext.from_collection(collection) + # First enable with the reason. + execute_admin_command(collection, test.build_command(ctx)) + # Re-enable with same reason should succeed. + result = execute_admin_command(collection, test.build_command(ctx)) + assertResult( result, - {"ok": 1.0}, - msg="setUserWriteBlockMode should be idempotent with same explicit reason", + expected=test.expected, + error_code=test.error_code, + msg=test.msg, + raw_res=True, ) 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..86178c438 --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/administration/commands/utils/admin_test_case.py @@ -0,0 +1,25 @@ +"""Shared test case for administration command tests.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from documentdb_tests.compatibility.tests.core.utils.command_test_case import ( + CommandTestCase, +) + + +@dataclass(frozen=True) +class AdminTestCase(CommandTestCase): + """Test case for administration command tests. + + Extends CommandTestCase with a ``use_admin`` flag that controls + whether the command is executed against the admin database. + + Attributes: + use_admin: If True (the default), execute against the admin + database via ``execute_admin_command``. If False, execute + against the test database via ``execute_command``. + """ + + use_admin: bool = True From 750b73e963cb89f8882a3f33f33ec76817df902f Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 14:06:43 -0700 Subject: [PATCH 04/10] convert _enforcement.py Signed-off-by: Alina (Xi) Li --- ...rWriteBlockMode_write_block_enforcement.py | 320 ++++++++++++------ .../commands/utils/admin_test_case.py | 34 +- 2 files changed, 245 insertions(+), 109 deletions(-) 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 index 18a7fd96b..5f121c24c 100644 --- 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 @@ -4,11 +4,24 @@ read operations are not affected, and operations succeed when block is disabled. """ +from __future__ import annotations + import pytest +from pymongo import IndexModel -from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial +from documentdb_tests.compatibility.tests.core.utils.command_test_case import ( + CommandContext, +) +from documentdb_tests.compatibility.tests.system.administration.commands.utils.admin_test_case import ( # noqa: E501 + AdminTestCase, +) +from documentdb_tests.framework.assertions import ( + assertFailureCode, + assertSuccessPartial, +) 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)] @@ -48,129 +61,228 @@ def _enable_write_block(collection): # Property [Write Operations Blocked]: all write operations are rejected while the block is # active. -_WRITE_BLOCKED_PARAMS = [ - pytest.param( - lambda name: {"insert": name, "documents": [{"_id": "blocked"}]}, - id="insert", - ), - pytest.param( - lambda name: {"update": name, "updates": [{"q": {}, "u": {"$set": {"x": 2}}}]}, - id="update", - ), - pytest.param( - lambda name: {"delete": name, "deletes": [{"q": {}, "limit": 1}]}, - id="delete", - ), - pytest.param( - lambda name: {"findAndModify": name, "query": {}, "update": {"$set": {"x": 2}}}, - id="findAndModify_update", - ), - pytest.param( - lambda name: {"findAndModify": name, "query": {}, "remove": True}, - id="findAndModify_remove", - ), - pytest.param( - lambda name: { - "createIndexes": name, +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=_enable_write_block, + 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=_enable_write_block, + 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=_enable_write_block, + 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=_enable_write_block, + 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=_enable_write_block, + 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=_enable_write_block, + command=lambda ctx: { + "createIndexes": ctx.collection, "indexes": [{"key": {"blocked_field": 1}, "name": "blocked_field_1"}], }, - id="createIndexes", + error_code=USER_WRITES_BLOCKED_ERROR, + msg="setUserWriteBlockMode should block createIndexes while active", ), - pytest.param( - lambda name: {"dropIndexes": name, "index": "a_1"}, - id="dropIndexes", + AdminTestCase( + "blocked_dropIndexes", + use_admin=False, + docs=[{"_id": "seed", "a": 1}], + indexes=[IndexModel([("a", 1)], name="a_1")], + pre_command=_enable_write_block, + command=lambda ctx: {"dropIndexes": ctx.collection, "index": "a_1"}, + error_code=USER_WRITES_BLOCKED_ERROR, + msg="setUserWriteBlockMode should block dropIndexes while active", ), - pytest.param( - lambda name: {"drop": name}, - id="drop_collection", + AdminTestCase( + "blocked_drop_collection", + use_admin=False, + docs=[{"_id": "seed"}], + pre_command=_enable_write_block, + command=lambda ctx: {"drop": ctx.collection}, + error_code=USER_WRITES_BLOCKED_ERROR, + msg="setUserWriteBlockMode should block drop collection while active", ), - pytest.param( - lambda name: {"create": f"{name}_blocked_new"}, - id="create_collection", + AdminTestCase( + "blocked_create_collection", + use_admin=False, + docs=[{"_id": "seed"}], + pre_command=_enable_write_block, + command=lambda ctx: {"create": f"{ctx.collection}_blocked_new"}, + error_code=USER_WRITES_BLOCKED_ERROR, + msg="setUserWriteBlockMode should block create collection while active", ), - pytest.param( - lambda name: {"dropDatabase": 1}, - id="dropDatabase", + AdminTestCase( + "blocked_dropDatabase", + use_admin=False, + docs=[{"_id": "seed"}], + pre_command=_enable_write_block, + command=lambda ctx: {"dropDatabase": 1}, + error_code=USER_WRITES_BLOCKED_ERROR, + msg="setUserWriteBlockMode should block dropDatabase while active", ), ] - -@pytest.mark.parametrize("build_command", _WRITE_BLOCKED_PARAMS) -def test_setUserWriteBlockMode_write_operation_blocked(collection, build_command): - """Test write operation is rejected while write block is active.""" - collection.insert_one({"_id": "seed", "a": 1}) - collection.create_index([("a", 1)], name="a_1") - _enable_write_block(collection) - command = build_command(collection.name) - result = execute_command(collection, command) - assertFailureCode( - result, - USER_WRITES_BLOCKED_ERROR, - msg="setUserWriteBlockMode should block write operations while active", - ) - - # Property [Read Operations Not Blocked]: read operations succeed while the block is active. -_READ_NOT_BLOCKED_PARAMS = [ - pytest.param( - lambda name: {"find": name, "filter": {}}, - id="find", +READ_NOT_BLOCKED_TESTS: list[AdminTestCase] = [ + AdminTestCase( + "read_find", + use_admin=False, + partial_success=True, + docs=[{"_id": "read_doc", "x": 1}], + pre_command=_enable_write_block, + command=lambda ctx: {"find": ctx.collection, "filter": {}}, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should not block find while active", ), - pytest.param( - lambda name: {"aggregate": name, "pipeline": [{"$match": {}}], "cursor": {}}, - id="aggregate", + AdminTestCase( + "read_aggregate", + use_admin=False, + partial_success=True, + docs=[{"_id": "read_doc", "x": 1}], + pre_command=_enable_write_block, + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$match": {}}], + "cursor": {}, + }, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should not block aggregate while active", ), - pytest.param( - lambda name: {"count": name}, - id="count", + AdminTestCase( + "read_count", + use_admin=False, + partial_success=True, + docs=[{"_id": "read_doc", "x": 1}], + pre_command=_enable_write_block, + command=lambda ctx: {"count": ctx.collection}, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should not block count while active", ), - pytest.param( - lambda name: {"distinct": name, "key": "x"}, - id="distinct", + AdminTestCase( + "read_distinct", + use_admin=False, + partial_success=True, + docs=[{"_id": "read_doc", "x": 1}], + pre_command=_enable_write_block, + command=lambda ctx: {"distinct": ctx.collection, "key": "x"}, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should not block distinct while active", ), ] - -@pytest.mark.parametrize("build_command", _READ_NOT_BLOCKED_PARAMS) -def test_setUserWriteBlockMode_read_operation_not_blocked(collection, build_command): - """Test read operation succeeds while write block is active.""" - collection.insert_one({"_id": "read_doc", "x": 1}) - _enable_write_block(collection) - command = build_command(collection.name) - result = execute_command(collection, command) - assertSuccessPartial( - result, - {"ok": 1.0}, - msg="setUserWriteBlockMode should not block read operations", - ) - - # Property [Writes Succeed When Disabled]: write operations succeed when no block is active. -_WRITE_SUCCEEDS_PARAMS = [ - pytest.param( - lambda name: {"insert": name, "documents": [{"_id": "ok"}]}, - id="insert", +WRITE_SUCCEEDS_TESTS: list[AdminTestCase] = [ + AdminTestCase( + "succeeds_insert", + use_admin=False, + partial_success=True, + docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], + command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": "ok"}]}, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should allow insert when block is not active", ), - pytest.param( - lambda name: {"update": name, "updates": [{"q": {"_id": "upd"}, "u": {"$set": {"x": 2}}}]}, - id="update", + AdminTestCase( + "succeeds_update", + use_admin=False, + partial_success=True, + docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], + command=lambda ctx: { + "update": ctx.collection, + "updates": [{"q": {"_id": "upd"}, "u": {"$set": {"x": 2}}}], + }, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should allow update when block is not active", ), - pytest.param( - lambda name: {"delete": name, "deletes": [{"q": {"_id": "del"}, "limit": 1}]}, - id="delete", + AdminTestCase( + "succeeds_delete", + use_admin=False, + partial_success=True, + docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], + command=lambda ctx: { + "delete": ctx.collection, + "deletes": [{"q": {"_id": "del"}, "limit": 1}], + }, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should allow delete when block is not active", ), ] +ENFORCEMENT_TESTS: list[AdminTestCase] = ( + WRITE_BLOCKED_TESTS + READ_NOT_BLOCKED_TESTS + WRITE_SUCCEEDS_TESTS +) + +ENFORCEMENT_ERROR_TESTS: list[AdminTestCase] = WRITE_BLOCKED_TESTS +ENFORCEMENT_SUCCESS_TESTS: list[AdminTestCase] = READ_NOT_BLOCKED_TESTS + WRITE_SUCCEEDS_TESTS + + +@pytest.mark.parametrize("test", pytest_params(ENFORCEMENT_ERROR_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) + -@pytest.mark.parametrize("build_command", _WRITE_SUCCEEDS_PARAMS) -def test_setUserWriteBlockMode_write_succeeds_when_disabled(collection, build_command): - """Test write operation succeeds when write block is not active.""" - collection.insert_one({"_id": "upd", "x": 1}) - collection.insert_one({"_id": "del"}) - command = build_command(collection.name) - result = execute_command(collection, command) - assertSuccessPartial( - result, - {"ok": 1.0}, - msg="setUserWriteBlockMode should allow writes when block is not active", - ) +@pytest.mark.parametrize("test", pytest_params(ENFORCEMENT_SUCCESS_TESTS)) +def test_setUserWriteBlockMode_allowed(database_client, collection, test): + """Test setUserWriteBlockMode allows reads and writes when appropriate.""" + collection = test.prepare(database_client, collection) + test.run_pre_command(collection) + ctx = CommandContext.from_collection(collection) + 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/utils/admin_test_case.py b/documentdb_tests/compatibility/tests/system/administration/commands/utils/admin_test_case.py index 86178c438..d35be87bb 100644 --- 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 @@ -2,9 +2,14 @@ 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, ) @@ -13,13 +18,32 @@ class AdminTestCase(CommandTestCase): """Test case for administration command tests. - Extends CommandTestCase with a ``use_admin`` flag that controls - whether the command is executed against the admin database. + Extends CommandTestCase with fields for admin-specific execution: Attributes: - use_admin: If True (the default), execute against the admin - database via ``execute_admin_command``. If False, execute - against the test database via ``execute_command``. + 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) From 95dd1163fbb12f9119ee6725dfee0a68e62a687f Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 14:20:15 -0700 Subject: [PATCH 05/10] move force_disable_write_block to a common place Signed-off-by: Alina (Xi) Li --- ...tUserWriteBlockMode_argument_validation.py | 27 ++---- ...est_setUserWriteBlockMode_core_behavior.py | 27 ++---- .../test_setUserWriteBlockMode_errors.py | 27 ++---- ...rWriteBlockMode_write_block_enforcement.py | 88 ++++++++++--------- .../setUserWriteBlockMode/utils/__init__.py | 0 .../utils/write_block_helpers.py | 21 +++++ 6 files changed, 83 insertions(+), 107 deletions(-) create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/utils/__init__.py create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/utils/write_block_helpers.py 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 index 4ba640a3b..779ab50e8 100644 --- 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 @@ -12,6 +12,9 @@ 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, ) @@ -29,32 +32,12 @@ pytestmark = [pytest.mark.admin, pytest.mark.no_parallel, pytest.mark.requires(cluster_admin=True)] -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 - - @pytest.fixture(autouse=True) def _manage_write_block(collection): """Ensure write block is disabled before and after each test.""" - _force_disable_write_block(collection) + force_disable_write_block(collection) yield - _force_disable_write_block(collection) + force_disable_write_block(collection) # Property [Global Field Boolean Acceptance]: setUserWriteBlockMode accepts only boolean values 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 index 43db8bf2f..c6ed1f0c3 100644 --- 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 @@ -5,38 +5,21 @@ 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)] -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 - - @pytest.fixture(autouse=True) def _manage_write_block(collection): """Ensure write block is disabled before and after each test.""" - _force_disable_write_block(collection) + force_disable_write_block(collection) yield - _force_disable_write_block(collection) + force_disable_write_block(collection) # Property [Idempotent Disable]: disabling write block when no block is active succeeds. 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 index 280125b7f..3e56f8726 100644 --- 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 @@ -10,6 +10,9 @@ 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, ) @@ -21,32 +24,12 @@ pytestmark = [pytest.mark.admin, pytest.mark.no_parallel, pytest.mark.requires(cluster_admin=True)] -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 - - @pytest.fixture(autouse=True) def _manage_write_block(collection): """Ensure write block is disabled before and after each test.""" - _force_disable_write_block(collection) + force_disable_write_block(collection) yield - _force_disable_write_block(collection) + force_disable_write_block(collection) # Property [Mismatched Reason on Enable]: re-enabling with a different reason fails. 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 index 5f121c24c..61a487d02 100644 --- 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 @@ -12,6 +12,9 @@ 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, ) @@ -26,37 +29,12 @@ pytestmark = [pytest.mark.admin, pytest.mark.no_parallel, pytest.mark.requires(cluster_admin=True)] -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 - - @pytest.fixture(autouse=True) def _manage_write_block(collection): """Ensure write block is disabled before and after each test.""" - _force_disable_write_block(collection) + force_disable_write_block(collection) yield - _force_disable_write_block(collection) - - -def _enable_write_block(collection): - """Enable write block.""" - execute_admin_command(collection, {"setUserWriteBlockMode": 1, "global": True}) + force_disable_write_block(collection) # Property [Write Operations Blocked]: all write operations are rejected while the block is @@ -67,7 +45,9 @@ def _enable_write_block(collection): use_admin=False, docs=[{"_id": "seed", "a": 1}], indexes=[IndexModel([("a", 1)], name="a_1")], - pre_command=_enable_write_block, + 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", @@ -77,7 +57,9 @@ def _enable_write_block(collection): use_admin=False, docs=[{"_id": "seed", "a": 1}], indexes=[IndexModel([("a", 1)], name="a_1")], - pre_command=_enable_write_block, + pre_command=lambda c: execute_admin_command( + c, {"setUserWriteBlockMode": 1, "global": True} + ), command=lambda ctx: { "update": ctx.collection, "updates": [{"q": {}, "u": {"$set": {"x": 2}}}], @@ -90,7 +72,9 @@ def _enable_write_block(collection): use_admin=False, docs=[{"_id": "seed", "a": 1}], indexes=[IndexModel([("a", 1)], name="a_1")], - pre_command=_enable_write_block, + pre_command=lambda c: execute_admin_command( + c, {"setUserWriteBlockMode": 1, "global": True} + ), command=lambda ctx: { "delete": ctx.collection, "deletes": [{"q": {}, "limit": 1}], @@ -102,7 +86,9 @@ def _enable_write_block(collection): "blocked_findAndModify_update", use_admin=False, docs=[{"_id": "seed", "a": 1}], - pre_command=_enable_write_block, + pre_command=lambda c: execute_admin_command( + c, {"setUserWriteBlockMode": 1, "global": True} + ), command=lambda ctx: { "findAndModify": ctx.collection, "query": {}, @@ -115,7 +101,9 @@ def _enable_write_block(collection): "blocked_findAndModify_remove", use_admin=False, docs=[{"_id": "seed", "a": 1}], - pre_command=_enable_write_block, + pre_command=lambda c: execute_admin_command( + c, {"setUserWriteBlockMode": 1, "global": True} + ), command=lambda ctx: { "findAndModify": ctx.collection, "query": {}, @@ -128,7 +116,9 @@ def _enable_write_block(collection): "blocked_createIndexes", use_admin=False, docs=[{"_id": "seed", "a": 1}], - pre_command=_enable_write_block, + 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"}], @@ -141,7 +131,9 @@ def _enable_write_block(collection): use_admin=False, docs=[{"_id": "seed", "a": 1}], indexes=[IndexModel([("a", 1)], name="a_1")], - pre_command=_enable_write_block, + 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", @@ -150,7 +142,9 @@ def _enable_write_block(collection): "blocked_drop_collection", use_admin=False, docs=[{"_id": "seed"}], - pre_command=_enable_write_block, + 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", @@ -159,7 +153,9 @@ def _enable_write_block(collection): "blocked_create_collection", use_admin=False, docs=[{"_id": "seed"}], - pre_command=_enable_write_block, + 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", @@ -168,7 +164,9 @@ def _enable_write_block(collection): "blocked_dropDatabase", use_admin=False, docs=[{"_id": "seed"}], - pre_command=_enable_write_block, + 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", @@ -182,7 +180,9 @@ def _enable_write_block(collection): use_admin=False, partial_success=True, docs=[{"_id": "read_doc", "x": 1}], - pre_command=_enable_write_block, + pre_command=lambda c: execute_admin_command( + c, {"setUserWriteBlockMode": 1, "global": True} + ), command=lambda ctx: {"find": ctx.collection, "filter": {}}, expected={"ok": 1.0}, msg="setUserWriteBlockMode should not block find while active", @@ -192,7 +192,9 @@ def _enable_write_block(collection): use_admin=False, partial_success=True, docs=[{"_id": "read_doc", "x": 1}], - pre_command=_enable_write_block, + pre_command=lambda c: execute_admin_command( + c, {"setUserWriteBlockMode": 1, "global": True} + ), command=lambda ctx: { "aggregate": ctx.collection, "pipeline": [{"$match": {}}], @@ -206,7 +208,9 @@ def _enable_write_block(collection): use_admin=False, partial_success=True, docs=[{"_id": "read_doc", "x": 1}], - pre_command=_enable_write_block, + pre_command=lambda c: execute_admin_command( + c, {"setUserWriteBlockMode": 1, "global": True} + ), command=lambda ctx: {"count": ctx.collection}, expected={"ok": 1.0}, msg="setUserWriteBlockMode should not block count while active", @@ -216,7 +220,9 @@ def _enable_write_block(collection): use_admin=False, partial_success=True, docs=[{"_id": "read_doc", "x": 1}], - pre_command=_enable_write_block, + pre_command=lambda c: execute_admin_command( + c, {"setUserWriteBlockMode": 1, "global": True} + ), command=lambda ctx: {"distinct": ctx.collection, "key": "x"}, expected={"ok": 1.0}, msg="setUserWriteBlockMode should not block distinct while active", 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 From 6f21130c916c82bd585bbd81b700b4a4d24e1110 Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 14:26:54 -0700 Subject: [PATCH 06/10] blocked_bulkWrite Signed-off-by: Alina (Xi) Li --- ...etUserWriteBlockMode_write_block_enforcement.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) 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 index 61a487d02..3aff6bc64 100644 --- 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 @@ -171,6 +171,20 @@ def _manage_write_block(collection): error_code=USER_WRITES_BLOCKED_ERROR, msg="setUserWriteBlockMode should block dropDatabase while active", ), + AdminTestCase( + "blocked_bulkWrite", + 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 batch insert while active", + ), ] # Property [Read Operations Not Blocked]: read operations succeed while the block is active. From 8731df196730bb596f7309564f7bd54a920b1e89 Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 14:30:37 -0700 Subject: [PATCH 07/10] split test cases Signed-off-by: Alina (Xi) Li --- ...tUserWriteBlockMode_argument_validation.py | 81 +------ ...est_setUserWriteBlockMode_core_behavior.py | 43 ++++ .../test_setUserWriteBlockMode_errors.py | 45 +--- .../test_setUserWriteBlockMode_success.py | 203 ++++++++++++++++++ ...rWriteBlockMode_write_block_enforcement.py | 122 +---------- 5 files changed, 261 insertions(+), 233 deletions(-) create mode 100644 documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_success.py 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 index 779ab50e8..a72876549 100644 --- 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 @@ -1,7 +1,7 @@ -"""Tests for setUserWriteBlockMode argument validation. +"""Tests for setUserWriteBlockMode argument validation errors. -Validates type checking for the global and reason fields, unrecognized fields, -and command value handling. +Validates type rejection for the global and reason fields, missing required fields, +invalid enum values, and unrecognized fields. """ from __future__ import annotations @@ -18,7 +18,7 @@ from documentdb_tests.compatibility.tests.system.administration.commands.utils.admin_test_case import ( # noqa: E501 AdminTestCase, ) -from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.assertions import assertFailureCode from documentdb_tests.framework.error_codes import ( BAD_VALUE_ERROR, MISSING_FIELD_ERROR, @@ -40,23 +40,6 @@ def _manage_write_block(collection): 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 [Global Field Type Rejection]: setUserWriteBlockMode rejects all non-boolean types # for the global field with no coercion. GLOBAL_TYPE_REJECTION_TESTS: list[AdminTestCase] = [ @@ -94,41 +77,6 @@ def _manage_write_block(collection): ), ] -# 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 omitted or null. -REASON_OPTIONAL_TESTS: list[AdminTestCase] = [ - AdminTestCase( - "reason_omitted", - command=lambda ctx: {"setUserWriteBlockMode": 1, "global": True}, - expected={"ok": 1.0}, - msg="setUserWriteBlockMode should succeed without reason field", - ), - 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 [Reason Field Type Rejection]: setUserWriteBlockMode rejects non-string types for # the reason field. REASON_TYPE_REJECTION_TESTS: list[AdminTestCase] = [ @@ -179,27 +127,18 @@ def _manage_write_block(collection): ), ] -ARGUMENT_VALIDATION_TESTS: list[AdminTestCase] = ( - GLOBAL_VALID_TESTS - + GLOBAL_TYPE_REJECTION_TESTS +ARGUMENT_ERROR_TESTS: list[AdminTestCase] = ( + GLOBAL_TYPE_REJECTION_TESTS + MISSING_GLOBAL_TESTS - + REASON_VALID_TESTS - + REASON_OPTIONAL_TESTS + REASON_TYPE_REJECTION_TESTS + REASON_INVALID_ENUM_TESTS + UNRECOGNIZED_FIELD_TESTS ) -@pytest.mark.parametrize("test", pytest_params(ARGUMENT_VALIDATION_TESTS)) -def test_setUserWriteBlockMode_argument_validation(collection, test): - """Test setUserWriteBlockMode argument validation.""" +@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)) - assertResult( - result, - expected=test.expected, - error_code=test.error_code, - msg=test.msg, - raw_res=True, - ) + 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 index c6ed1f0c3..d3048f64b 100644 --- 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 @@ -73,3 +73,46 @@ def test_setUserWriteBlockMode_enable_idempotent_same_reason(collection): {"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 index 3e56f8726..3f0056fdf 100644 --- 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 @@ -7,19 +7,12 @@ 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 assertFailureCode, assertResult +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 -from documentdb_tests.framework.parametrize import pytest_params pytestmark = [pytest.mark.admin, pytest.mark.no_parallel, pytest.mark.requires(cluster_admin=True)] @@ -75,39 +68,3 @@ def test_setUserWriteBlockMode_disable_mismatched_reason_fails(collection): ILLEGAL_OPERATION_ERROR, msg="setUserWriteBlockMode should reject mismatched reason on disable", ) - - -# Property [Same Reason Idempotent]: re-enabling with same explicit reason succeeds. -SAME_REASON_TESTS: list[AdminTestCase] = [ - AdminTestCase( - f"same_reason_{tid}", - command=lambda ctx, r=reason: { - "setUserWriteBlockMode": 1, - "global": True, - "reason": r, - }, - expected={"ok": 1.0}, - msg=f"setUserWriteBlockMode should be idempotent with same reason {reason}", - ) - for tid, reason in [ - ("unspecified", "Unspecified"), - ("cluster_migration", "ClusterToClusterMigrationInProgress"), - ] -] - - -@pytest.mark.parametrize("test", pytest_params(SAME_REASON_TESTS)) -def test_setUserWriteBlockMode_same_reason_idempotent(collection, test): - """Test setUserWriteBlockMode re-enable with same reason is idempotent.""" - ctx = CommandContext.from_collection(collection) - # First enable with the reason. - execute_admin_command(collection, test.build_command(ctx)) - # Re-enable with same reason should succeed. - result = execute_admin_command(collection, test.build_command(ctx)) - assertResult( - result, - expected=test.expected, - error_code=test.error_code, - msg=test.msg, - raw_res=True, - ) 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..93992b74e --- /dev/null +++ b/documentdb_tests/compatibility/tests/system/administration/commands/setUserWriteBlockMode/test_setUserWriteBlockMode_success.py @@ -0,0 +1,203 @@ +"""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 omitted or null. +REASON_OPTIONAL_TESTS: list[AdminTestCase] = [ + AdminTestCase( + "reason_omitted", + command=lambda ctx: {"setUserWriteBlockMode": 1, "global": True}, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should succeed without reason field", + ), + 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": {}}, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should not block find while active", + ), + AdminTestCase( + "read_aggregate", + 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: { + "aggregate": ctx.collection, + "pipeline": [{"$match": {}}], + "cursor": {}, + }, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should not block aggregate while active", + ), + AdminTestCase( + "read_count", + 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: {"count": ctx.collection}, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should not block count while active", + ), + AdminTestCase( + "read_distinct", + 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: {"distinct": ctx.collection, "key": "x"}, + expected={"ok": 1.0}, + 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( + "succeeds_insert", + use_admin=False, + partial_success=True, + docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], + command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": "ok"}]}, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should allow insert when block is not active", + ), + AdminTestCase( + "succeeds_update", + use_admin=False, + partial_success=True, + docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], + command=lambda ctx: { + "update": ctx.collection, + "updates": [{"q": {"_id": "upd"}, "u": {"$set": {"x": 2}}}], + }, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should allow update when block is not active", + ), + AdminTestCase( + "succeeds_delete", + use_admin=False, + partial_success=True, + docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], + command=lambda ctx: { + "delete": ctx.collection, + "deletes": [{"q": {"_id": "del"}, "limit": 1}], + }, + expected={"ok": 1.0}, + msg="setUserWriteBlockMode should allow delete when block is not active", + ), +] + +ACCEPTANCE_TESTS: list[AdminTestCase] = ( + GLOBAL_VALID_TESTS + REASON_VALID_TESTS + REASON_OPTIONAL_TESTS +) + +ADMIN_SUCCESS_TESTS: list[AdminTestCase] = ACCEPTANCE_TESTS +COLLECTION_SUCCESS_TESTS: list[AdminTestCase] = READ_NOT_BLOCKED_TESTS + WRITE_SUCCEEDS_TESTS + + +@pytest.mark.parametrize("test", pytest_params(ADMIN_SUCCESS_TESTS)) +def test_setUserWriteBlockMode_acceptance(collection, test): + """Test setUserWriteBlockMode accepts valid arguments.""" + ctx = CommandContext.from_collection(collection) + result = execute_admin_command(collection, test.build_command(ctx)) + assertSuccessPartial(result, test.expected, msg=test.msg) + + +@pytest.mark.parametrize("test", pytest_params(COLLECTION_SUCCESS_TESTS)) +def test_setUserWriteBlockMode_allowed(database_client, collection, test): + """Test setUserWriteBlockMode allows reads and writes when appropriate.""" + collection = test.prepare(database_client, collection) + test.run_pre_command(collection) + ctx = CommandContext.from_collection(collection) + 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 index 3aff6bc64..c21262e01 100644 --- 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 @@ -1,7 +1,6 @@ -"""Tests for setUserWriteBlockMode write block enforcement. +"""Tests for setUserWriteBlockMode write block enforcement errors. -Validates that write operations are blocked while the block is active, -read operations are not affected, and operations succeed when block is disabled. +Validates that write operations are rejected while the block is active. """ from __future__ import annotations @@ -18,10 +17,7 @@ from documentdb_tests.compatibility.tests.system.administration.commands.utils.admin_test_case import ( # noqa: E501 AdminTestCase, ) -from documentdb_tests.framework.assertions import ( - assertFailureCode, - assertSuccessPartial, -) +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 @@ -187,108 +183,8 @@ def _manage_write_block(collection): ), ] -# 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": {}}, - expected={"ok": 1.0}, - msg="setUserWriteBlockMode should not block find while active", - ), - AdminTestCase( - "read_aggregate", - 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: { - "aggregate": ctx.collection, - "pipeline": [{"$match": {}}], - "cursor": {}, - }, - expected={"ok": 1.0}, - msg="setUserWriteBlockMode should not block aggregate while active", - ), - AdminTestCase( - "read_count", - 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: {"count": ctx.collection}, - expected={"ok": 1.0}, - msg="setUserWriteBlockMode should not block count while active", - ), - AdminTestCase( - "read_distinct", - 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: {"distinct": ctx.collection, "key": "x"}, - expected={"ok": 1.0}, - 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( - "succeeds_insert", - use_admin=False, - partial_success=True, - docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], - command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": "ok"}]}, - expected={"ok": 1.0}, - msg="setUserWriteBlockMode should allow insert when block is not active", - ), - AdminTestCase( - "succeeds_update", - use_admin=False, - partial_success=True, - docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], - command=lambda ctx: { - "update": ctx.collection, - "updates": [{"q": {"_id": "upd"}, "u": {"$set": {"x": 2}}}], - }, - expected={"ok": 1.0}, - msg="setUserWriteBlockMode should allow update when block is not active", - ), - AdminTestCase( - "succeeds_delete", - use_admin=False, - partial_success=True, - docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], - command=lambda ctx: { - "delete": ctx.collection, - "deletes": [{"q": {"_id": "del"}, "limit": 1}], - }, - expected={"ok": 1.0}, - msg="setUserWriteBlockMode should allow delete when block is not active", - ), -] - -ENFORCEMENT_TESTS: list[AdminTestCase] = ( - WRITE_BLOCKED_TESTS + READ_NOT_BLOCKED_TESTS + WRITE_SUCCEEDS_TESTS -) - -ENFORCEMENT_ERROR_TESTS: list[AdminTestCase] = WRITE_BLOCKED_TESTS -ENFORCEMENT_SUCCESS_TESTS: list[AdminTestCase] = READ_NOT_BLOCKED_TESTS + WRITE_SUCCEEDS_TESTS - - -@pytest.mark.parametrize("test", pytest_params(ENFORCEMENT_ERROR_TESTS)) +@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) @@ -296,13 +192,3 @@ def test_setUserWriteBlockMode_blocked(database_client, collection, test): ctx = CommandContext.from_collection(collection) result = execute_command(collection, test.build_command(ctx)) assertFailureCode(result, test.error_code, msg=test.msg) - - -@pytest.mark.parametrize("test", pytest_params(ENFORCEMENT_SUCCESS_TESTS)) -def test_setUserWriteBlockMode_allowed(database_client, collection, test): - """Test setUserWriteBlockMode allows reads and writes when appropriate.""" - collection = test.prepare(database_client, collection) - test.run_pre_command(collection) - ctx = CommandContext.from_collection(collection) - result = execute_command(collection, test.build_command(ctx)) - assertSuccessPartial(result, test.expected, msg=test.msg) From dd693345d15684bf571328636c0ff05c13f44659 Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 14:36:29 -0700 Subject: [PATCH 08/10] remove duplicate Signed-off-by: Alina (Xi) Li --- .../test_setUserWriteBlockMode_success.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) 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 index 93992b74e..f8745653d 100644 --- 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 @@ -68,14 +68,8 @@ def _manage_write_block(collection): ] ] -# Property [Reason Field Optional]: the reason field can be omitted or null. +# Property [Reason Field Optional]: the reason field can be null (treated as omitted). REASON_OPTIONAL_TESTS: list[AdminTestCase] = [ - AdminTestCase( - "reason_omitted", - command=lambda ctx: {"setUserWriteBlockMode": 1, "global": True}, - expected={"ok": 1.0}, - msg="setUserWriteBlockMode should succeed without reason field", - ), AdminTestCase( "reason_null", command=lambda ctx: {"setUserWriteBlockMode": 1, "global": True, "reason": None}, From 1e420bb6d5b34b46728db811adc80ed0817182e3 Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 14:52:19 -0700 Subject: [PATCH 09/10] fix tests to check results Signed-off-by: Alina (Xi) Li --- ...tUserWriteBlockMode_argument_validation.py | 8 +++- .../test_setUserWriteBlockMode_success.py | 42 +++++++++---------- ...rWriteBlockMode_write_block_enforcement.py | 4 +- 3 files changed, 30 insertions(+), 24 deletions(-) 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 index a72876549..b29c2c161 100644 --- 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 @@ -63,11 +63,11 @@ def _manage_write_block(collection): ("string", "true", TYPE_MISMATCH_ERROR), ("array", [], TYPE_MISMATCH_ERROR), ("object", {}, TYPE_MISMATCH_ERROR), - ("null", None, MISSING_FIELD_ERROR), ] ] # Property [Missing Global Field]: setUserWriteBlockMode requires the global field. +# Null is treated as missing. MISSING_GLOBAL_TESTS: list[AdminTestCase] = [ AdminTestCase( "missing_global", @@ -75,6 +75,12 @@ def _manage_write_block(collection): 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 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 index f8745653d..5acd28903 100644 --- 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 @@ -88,48 +88,48 @@ def _manage_write_block(collection): pre_command=lambda c: execute_admin_command( c, {"setUserWriteBlockMode": 1, "global": True} ), - command=lambda ctx: {"find": ctx.collection, "filter": {}}, - expected={"ok": 1.0}, + 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": "read_doc", "x": 1}], + 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": {}}], + "pipeline": [{"$match": {"_id": "agg_doc"}}], "cursor": {}, }, - expected={"ok": 1.0}, + 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": "read_doc", "x": 1}], + 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}, + 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": "read_doc", "x": 1}], + 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}, + expected={"ok": 1.0, "values": [1, 2]}, msg="setUserWriteBlockMode should not block distinct while active", ), ] @@ -137,36 +137,36 @@ def _manage_write_block(collection): # Property [Writes Succeed When Disabled]: write operations succeed when no block is active. WRITE_SUCCEEDS_TESTS: list[AdminTestCase] = [ AdminTestCase( - "succeeds_insert", + "write_insert_no_block", use_admin=False, partial_success=True, - docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], - command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": "ok"}]}, - expected={"ok": 1.0}, + 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( - "succeeds_update", + "write_update_no_block", use_admin=False, partial_success=True, - docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], + docs=[{"_id": "target", "x": 1}], command=lambda ctx: { "update": ctx.collection, - "updates": [{"q": {"_id": "upd"}, "u": {"$set": {"x": 2}}}], + "updates": [{"q": {"_id": "target"}, "u": {"$set": {"x": 99}}}], }, - expected={"ok": 1.0}, + expected={"ok": 1.0, "n": 1, "nModified": 1}, msg="setUserWriteBlockMode should allow update when block is not active", ), AdminTestCase( - "succeeds_delete", + "write_delete_no_block", use_admin=False, partial_success=True, - docs=[{"_id": "upd", "x": 1}, {"_id": "del"}], + docs=[{"_id": "target"}], command=lambda ctx: { "delete": ctx.collection, - "deletes": [{"q": {"_id": "del"}, "limit": 1}], + "deletes": [{"q": {"_id": "target"}, "limit": 1}], }, - expected={"ok": 1.0}, + expected={"ok": 1.0, "n": 1}, msg="setUserWriteBlockMode should allow delete when block is not active", ), ] 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 index c21262e01..c6fce6074 100644 --- 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 @@ -168,7 +168,7 @@ def _manage_write_block(collection): msg="setUserWriteBlockMode should block dropDatabase while active", ), AdminTestCase( - "blocked_bulkWrite", + "blocked_batch_insert", use_admin=False, docs=[{"_id": "seed"}], pre_command=lambda c: execute_admin_command( @@ -179,7 +179,7 @@ def _manage_write_block(collection): "documents": [{"_id": "bulk1"}, {"_id": "bulk2"}], }, error_code=USER_WRITES_BLOCKED_ERROR, - msg="setUserWriteBlockMode should block batch insert while active", + msg="setUserWriteBlockMode should block multi-document insert while active", ), ] From 34389ff448a5c07293f513698be0947ce4511bf2 Mon Sep 17 00:00:00 2001 From: "Alina (Xi) Li" Date: Mon, 29 Jun 2026 14:56:54 -0700 Subject: [PATCH 10/10] merge test func Signed-off-by: Alina (Xi) Li --- .../test_setUserWriteBlockMode_success.py | 30 ++++++++----------- 1 file changed, 13 insertions(+), 17 deletions(-) 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 index 5acd28903..52756d162 100644 --- 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 @@ -171,27 +171,23 @@ def _manage_write_block(collection): ), ] -ACCEPTANCE_TESTS: list[AdminTestCase] = ( - GLOBAL_VALID_TESTS + REASON_VALID_TESTS + REASON_OPTIONAL_TESTS +SUCCESS_TESTS: list[AdminTestCase] = ( + GLOBAL_VALID_TESTS + + REASON_VALID_TESTS + + REASON_OPTIONAL_TESTS + + READ_NOT_BLOCKED_TESTS + + WRITE_SUCCEEDS_TESTS ) -ADMIN_SUCCESS_TESTS: list[AdminTestCase] = ACCEPTANCE_TESTS -COLLECTION_SUCCESS_TESTS: list[AdminTestCase] = READ_NOT_BLOCKED_TESTS + WRITE_SUCCEEDS_TESTS - -@pytest.mark.parametrize("test", pytest_params(ADMIN_SUCCESS_TESTS)) -def test_setUserWriteBlockMode_acceptance(collection, test): - """Test setUserWriteBlockMode accepts valid arguments.""" - ctx = CommandContext.from_collection(collection) - result = execute_admin_command(collection, test.build_command(ctx)) - assertSuccessPartial(result, test.expected, msg=test.msg) - - -@pytest.mark.parametrize("test", pytest_params(COLLECTION_SUCCESS_TESTS)) -def test_setUserWriteBlockMode_allowed(database_client, collection, test): - """Test setUserWriteBlockMode allows reads and writes when appropriate.""" +@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) - result = execute_command(collection, test.build_command(ctx)) + 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)