Add encryption compatibility tests

documentdb_tests/compatibility/tests/system/security/encryption/conftest.py

@@ -0,0 +1,9 @@
+"""Shared fixtures for Queryable Encryption tests in this directory."""
+
+from documentdb_tests.compatibility.tests.system.security.encryption.utils.qe_collections import (
+    qe_collection,
+    qe_collection_multi,
+    qe_collection_nested,
+)
+
+__all__ = ["qe_collection", "qe_collection_multi", "qe_collection_nested"]

documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_crud.py

@@ -0,0 +1,135 @@
+"""Tests for CRUD operations against Queryable Encryption collections.
+
+Verifies insert, update, and delete operations with encryption feature. Any
+plaintext value at an encrypted path is rejected, including null and regardless
+of bsonType. Absent fields are unaffected.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.compatibility.tests.core.utils.command_test_case import (
+    CommandContext,
+    CommandTestCase,
+)
+from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial
+from documentdb_tests.framework.error_codes import DOCUMENT_VALIDATION_FAILURE_ERROR
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+
+pytestmark = pytest.mark.requires(queryable_encryption=True)
+
+
+# Property [Missing Field Acceptance]: insert succeeds when an encrypted field is absent.
+@pytest.mark.insert
+def test_encryption_insert_missing_field_succeeds(qe_collection):
+    """Test insert succeeds when the encrypted field is entirely absent."""
+    result = execute_command(
+        qe_collection, {"insert": qe_collection.name, "documents": [{"_id": 1}]}
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0, "n": 1},
+        msg="insert should accept a document missing the encrypted field.",
+    )
+
+
+# Property [Plaintext Rejection]: insert rejects null and any plaintext value at an
+# encrypted path, regardless of bsonType.
+INSERT_PLAINTEXT_REJECTION_TESTS: list[CommandTestCase] = [
+    CommandTestCase(
+        "null_value",
+        command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": 1, "ssn": None}]},
+        error_code=DOCUMENT_VALIDATION_FAILURE_ERROR,
+        msg="insert should reject an explicit null at an encrypted path.",
+    ),
+    CommandTestCase(
+        "plaintext_matching_bsontype",
+        command=lambda ctx: {
+            "insert": ctx.collection,
+            "documents": [{"_id": 2, "ssn": "123-45-6789"}],
+        },
+        error_code=DOCUMENT_VALIDATION_FAILURE_ERROR,
+        msg="insert should reject a plaintext value even when it matches the declared bsonType.",
+    ),
+    CommandTestCase(
+        "plaintext_wrong_bsontype",
+        command=lambda ctx: {"insert": ctx.collection, "documents": [{"_id": 3, "ssn": 123}]},
+        error_code=DOCUMENT_VALIDATION_FAILURE_ERROR,
+        msg="insert should reject a plaintext value of the wrong bsonType.",
+    ),
+]
+
+
+@pytest.mark.insert
+@pytest.mark.parametrize("test", pytest_params(INSERT_PLAINTEXT_REJECTION_TESTS))
+def test_encryption_insert_rejects_plaintext(qe_collection, test: CommandTestCase):
+    """Test insert rejects plaintext values at an encrypted path."""
+    ctx = CommandContext.from_collection(qe_collection)
+    result = execute_command(qe_collection, test.build_command(ctx))
+    assertFailureCode(result, test.error_code, msg=test.msg)
+
+
+# Property [Multi-Field Independence]: each declared encrypted field is validated
+# independently of the others.
+@pytest.mark.insert
+def test_encryption_insert_multiple_fields_all_absent(qe_collection_multi):
+    """Test insert succeeds when every declared encrypted field is absent."""
+    result = execute_command(
+        qe_collection_multi,
+        {"insert": qe_collection_multi.name, "documents": [{"_id": 1, "name": "a"}]},
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0, "n": 1},
+        msg="insert should accept a document missing every encrypted field.",
+    )
+
+
+@pytest.mark.insert
+def test_encryption_insert_one_of_multiple_fields_plaintext_rejected(qe_collection_multi):
+    """Test insert rejects a plaintext value on one of several encrypted fields."""
+    result = execute_command(
+        qe_collection_multi,
+        {"insert": qe_collection_multi.name, "documents": [{"_id": 1, "dob": "2000-01-01"}]},
+    )
+    assertFailureCode(
+        result,
+        DOCUMENT_VALIDATION_FAILURE_ERROR,
+        msg="insert should reject a plaintext value on any one of several encrypted fields.",
+    )
+
+
+# Property [Update Enforcement]: update applies the same validation as insert.
+@pytest.mark.update
+def test_encryption_update_rejects_plaintext(qe_collection):
+    """Test update rejects setting an encrypted field to a plaintext value."""
+    qe_collection.insert_one({"_id": 1})
+    result = execute_command(
+        qe_collection,
+        {
+            "update": qe_collection.name,
+            "updates": [{"q": {"_id": 1}, "u": {"$set": {"ssn": "123-45-6789"}}}],
+        },
+    )
+    assertFailureCode(
+        result,
+        DOCUMENT_VALIDATION_FAILURE_ERROR,
+        msg="update should reject a plaintext value written to an encrypted path.",
+    )
+
+
+# Property [Delete Unaffected]: delete is unaffected by the collection's encryption schema.
+@pytest.mark.delete
+def test_encryption_delete_succeeds(qe_collection):
+    """Test delete succeeds for a document with an absent encrypted field."""
+    qe_collection.insert_one({"_id": 1})
+    result = execute_command(
+        qe_collection, {"delete": qe_collection.name, "deletes": [{"q": {"_id": 1}, "limit": 1}]}
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0, "n": 1},
+        msg="delete should succeed on a Queryable Encryption collection.",
+    )

documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_edge_cases.py

@@ -0,0 +1,112 @@
+"""Tests for encryption edge cases: nested paths, value size, and multiplicity.
+
+Verifies that no value shape other than absence can be written to an encrypted
+path without a client FLE driver, regardless of nesting, size, or document
+count.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertFailureCode, assertSuccessPartial
+from documentdb_tests.framework.error_codes import DOCUMENT_VALIDATION_FAILURE_ERROR
+from documentdb_tests.framework.executor import execute_command
+
+pytestmark = pytest.mark.requires(queryable_encryption=True)
+
+
+# Property [Nested Path Independence]: a nested encrypted path is unaffected by
+# insert as long as it is absent, regardless of whether its parent object is present.
+@pytest.mark.insert
+def test_encryption_insert_nested_path_missing_parent_succeeds(qe_collection_nested):
+    """Test insert succeeds when the parent of a nested encrypted path is absent."""
+    result = execute_command(
+        qe_collection_nested, {"insert": qe_collection_nested.name, "documents": [{"_id": 1}]}
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0, "n": 1},
+        msg="insert should succeed when the parent object of a nested encrypted path is absent.",
+    )
+
+
+@pytest.mark.insert
+def test_encryption_insert_nested_path_parent_present_child_absent_succeeds(qe_collection_nested):
+    """Test insert succeeds when the parent object is present but the encrypted child is absent."""
+    result = execute_command(
+        qe_collection_nested,
+        {"insert": qe_collection_nested.name, "documents": [{"_id": 1, "address": {"city": "x"}}]},
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0, "n": 1},
+        msg="insert should succeed when a nested encrypted field is absent"
+        " even if its parent object is present.",
+    )
+
+
+@pytest.mark.insert
+def test_encryption_insert_nested_path_plaintext_rejected(qe_collection_nested):
+    """Test insert rejects a plaintext value at a nested encrypted path."""
+    result = execute_command(
+        qe_collection_nested,
+        {
+            "insert": qe_collection_nested.name,
+            "documents": [{"_id": 1, "address": {"ssn": "123-45-6789"}}],
+        },
+    )
+    assertFailureCode(
+        result,
+        DOCUMENT_VALIDATION_FAILURE_ERROR,
+        msg="insert should reject a plaintext value at a nested encrypted path.",
+    )
+
+
+# Property [Validation Ignores Size]: a plaintext value at an encrypted path is
+# rejected regardless of its size.
+@pytest.mark.insert
+def test_encryption_insert_large_plaintext_value_rejected(qe_collection):
+    """Test insert rejects a large plaintext value at an encrypted path."""
+    result = execute_command(
+        qe_collection,
+        {"insert": qe_collection.name, "documents": [{"_id": 1, "ssn": "x" * 4_000}]},
+    )
+    assertFailureCode(
+        result,
+        DOCUMENT_VALIDATION_FAILURE_ERROR,
+        msg="insert should reject an oversized plaintext value at an encrypted path"
+        " just as it would a small one.",
+    )
+
+
+# Property [Absence Has No Multiplicity Constraint]: several documents may each omit
+# the same encrypted field in one insert; equality does not impose uniqueness on absence.
+@pytest.mark.insert
+def test_encryption_insert_multiple_documents_all_missing_field(qe_collection):
+    """Test a batch insert succeeds when every document omits the same encrypted field."""
+    result = execute_command(
+        qe_collection,
+        {"insert": qe_collection.name, "documents": [{"_id": 1}, {"_id": 2}]},
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0, "n": 2},
+        msg="a batch insert should succeed when every document omits the same encrypted field.",
+    )
+
+
+# Property [Explain Compatibility]: explain runs normally against a
+# Queryable Encryption collection.
+@pytest.mark.find
+def test_encryption_explain_find_on_encrypted_collection(qe_collection):
+    """Test explain succeeds for a find against a Queryable Encryption collection."""
+    result = execute_command(
+        qe_collection,
+        {"explain": {"find": qe_collection.name, "filter": {"ssn": "x"}}},
+    )
+    assertSuccessPartial(
+        result,
+        {"ok": 1.0},
+        msg="explain should succeed for a find against a Queryable Encryption collection.",
+    )

documentdb_tests/compatibility/tests/system/security/encryption/test_encryption_query.py

@@ -0,0 +1,92 @@
+"""Tests for query and aggregation behavior against Queryable Encryption collections.
+
+Verifies that a raw filter against an encrypted path, including a shape a real
+client would never send such as a range comparison on an equality-only field, is
+evaluated as an ordinary filter and never raises an encryption-specific error.
+This repo has no client-side FLE driver, so query rewriting never happens.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertResult
+from documentdb_tests.framework.executor import execute_command
+
+pytestmark = pytest.mark.requires(queryable_encryption=True)
+
+
+@pytest.fixture()
+def qe_collection_seeded(qe_collection):
+    """A qe_collection populated with two documents that both omit the encrypted field."""
+    qe_collection.insert_many([{"_id": 1, "name": "a"}, {"_id": 2, "name": "b"}])
+    return qe_collection
+
+
+@pytest.mark.find
+def test_encryption_find_equality_filter_returns_empty(qe_collection_seeded):
+    """Test an equality filter on an encrypted field returns no matches, not an error."""
+    result = execute_command(
+        qe_collection_seeded, {"find": qe_collection_seeded.name, "filter": {"ssn": "123-45-6789"}}
+    )
+    assertResult(
+        result,
+        expected=[],
+        msg="an equality filter on an encrypted field should return no matches"
+        " when no document has that field set.",
+    )
+
+
+@pytest.mark.find
+def test_encryption_find_range_operator_on_equality_field_no_error(qe_collection_seeded):
+    """Test a range operator on an equality-only encrypted field executes without error."""
+    result = execute_command(
+        qe_collection_seeded, {"find": qe_collection_seeded.name, "filter": {"ssn": {"$gt": "a"}}}
+    )
+    assertResult(
+        result,
+        expected=[],
+        msg="a range filter on an equality-only encrypted field should execute as an"
+        " ordinary filter rather than raise a query-type error, since raw commands"
+        " bypass client-side query rewriting.",
+    )
+
+
+@pytest.mark.find
+def test_encryption_find_exists_false_matches_absent_field(qe_collection_seeded):
+    """Test $exists:false on an encrypted field matches documents where it is absent."""
+    result = execute_command(
+        qe_collection_seeded,
+        {
+            "find": qe_collection_seeded.name,
+            "filter": {"ssn": {"$exists": False}},
+            "sort": {"_id": 1},
+        },
+    )
+    assertResult(
+        result,
+        expected=[{"_id": 1, "name": "a"}, {"_id": 2, "name": "b"}],
+        msg="$exists:false on an encrypted field should match documents where the field is absent.",
+    )
+
+
+@pytest.mark.aggregate
+def test_encryption_aggregate_match_expr_on_encrypted_field(qe_collection_seeded):
+    """Test $match+$expr referencing an encrypted field in aggregation."""
+    result = execute_command(
+        qe_collection_seeded,
+        {
+            "aggregate": qe_collection_seeded.name,
+            "pipeline": [
+                {"$match": {"$expr": {"$eq": [{"$type": "$ssn"}, "missing"]}}},
+                {"$project": {"_id": 1}},
+                {"$sort": {"_id": 1}},
+            ],
+            "cursor": {},
+        },
+    )
+    assertResult(
+        result,
+        expected=[{"_id": 1}, {"_id": 2}],
+        msg="$match+$expr referencing an encrypted field should evaluate normally in aggregation.",
+    )