Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Empty file.
Empty file.
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
"""Tests for compactStructuredEncryptionData core behavior.

Verifies the command correctly rejects non-encrypted collections with error 6346807
and handles non-existent collections.
"""

import pytest

from documentdb_tests.compatibility.tests.core.utils.command_test_case import (
CommandContext,
CommandTestCase,
)
from documentdb_tests.framework.assertions import assertResult
from documentdb_tests.framework.error_codes import (
NAMESPACE_NOT_FOUND_ERROR,
NOT_ENCRYPTED_COLLECTION_ERROR,
)
from documentdb_tests.framework.executor import execute_command
from documentdb_tests.framework.parametrize import pytest_params

pytestmark = pytest.mark.admin

# Property [Non-Encrypted Rejection]: compactStructuredEncryptionData rejects
# collections that are not configured for Queryable Encryption with error 6346807.
CORE_BEHAVIOR_TESTS: list[CommandTestCase] = [
CommandTestCase(
"empty_compaction_tokens",
docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should reject non-encrypted collection"
" with empty tokens",
),
CommandTestCase(
"non_empty_compaction_tokens",
docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {"field": b"\x00\x01\x02"},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should reject non-encrypted collection with tokens",
),
CommandTestCase(
"collection_with_documents",
docs=[{"_id": 1, "name": "test"}, {"_id": 2, "name": "data"}],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should reject non-encrypted collection with documents",
),
CommandTestCase(
"nonexistent_collection",
command=lambda ctx: {
"compactStructuredEncryptionData": "nonexistent_collection_xyz",
"compactionTokens": {},
},
error_code=NAMESPACE_NOT_FOUND_ERROR,
msg="compactStructuredEncryptionData should error on non-existent collection",
),
]


@pytest.mark.parametrize("test", pytest_params(CORE_BEHAVIOR_TESTS))
def test_compactStructuredEncryptionData_core_behavior(database_client, collection, test):
"""Test compactStructuredEncryptionData core behavior on non-encrypted collections."""
collection = test.prepare(database_client, collection)
ctx = CommandContext.from_collection(collection)
result = execute_command(collection, test.build_command(ctx))
assertResult(
result,
error_code=test.error_code,
msg=test.msg,
raw_res=True,
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,137 @@
"""Tests for compactStructuredEncryptionData edge cases.

Covers collection name edge cases and compactionTokens document content
edge cases.
"""

import pytest

from documentdb_tests.compatibility.tests.core.utils.command_test_case import (
CommandContext,
CommandTestCase,
)
from documentdb_tests.framework.assertions import assertResult
from documentdb_tests.framework.error_codes import (
NAMESPACE_NOT_FOUND_ERROR,
NOT_ENCRYPTED_COLLECTION_ERROR,
)
from documentdb_tests.framework.executor import execute_command
from documentdb_tests.framework.parametrize import pytest_params

pytestmark = pytest.mark.admin

# Property [Collection Name Edge Cases]: compactStructuredEncryptionData handles
# special collection name patterns correctly.
COLLECTION_NAME_TESTS: list[CommandTestCase] = [
CommandTestCase(
"system_prefix",
command=lambda ctx: {
"compactStructuredEncryptionData": "system.buckets.test",
"compactionTokens": {},
},
error_code=NAMESPACE_NOT_FOUND_ERROR,
msg="compactStructuredEncryptionData should reject system.* prefix"
" collection names with namespace-not-found",
),
CommandTestCase(
"dotted_name",
command=lambda ctx: {
"compactStructuredEncryptionData": "a.b.c",
"compactionTokens": {},
},
error_code=NAMESPACE_NOT_FOUND_ERROR,
msg="compactStructuredEncryptionData should reject multi-segment"
" dotted names with namespace-not-found",
),
CommandTestCase(
"dollar_prefix",
command=lambda ctx: {
"compactStructuredEncryptionData": "$myCollection",
"compactionTokens": {},
},
error_code=NAMESPACE_NOT_FOUND_ERROR,
msg="compactStructuredEncryptionData should reject dollar-prefixed"
" collection names with namespace-not-found",
),
]

# Property [CompactionTokens Content Edge Cases]: compactStructuredEncryptionData
# handles various compactionTokens document content correctly.
COMPACTION_TOKENS_CONTENT_TESTS: list[CommandTestCase] = [
CommandTestCase(
"null_token_value",

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The four compactionTokens content "edge case" tests (null_token_value, empty_string_key, dot_notation_key, nested_document_value) in edge_cases.py still all resolve to the same NOT_ENCRYPTED_COLLECTION_ERROR on standalone — they short-circuit before token parsing, so they prove the same thing. Now that the QE file exercises real token validation, consider move to the QE-gated file where the content is actually parsed.

docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {"ssn": None},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should reject null token value"
" with non-encrypted error",
),
CommandTestCase(
"empty_string_key",
docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {"": b"\x00\x01"},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should reject empty-string"
" token key with non-encrypted error",
),
CommandTestCase(
"dot_notation_key",
docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {"a.b": b"\x00\x01"},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should reject dot-notation"
" token key with non-encrypted error",
),
CommandTestCase(
"nested_document_value",
docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {"field": {"nested": b"\x00\x01"}},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should reject nested document"
" token value with non-encrypted error",
),
]


@pytest.mark.parametrize("test", pytest_params(COLLECTION_NAME_TESTS))
def test_compactStructuredEncryptionData_collection_name_edge_cases(
database_client, collection, test
):
"""Test compactStructuredEncryptionData rejects special collection name patterns."""
collection = test.prepare(database_client, collection)
ctx = CommandContext.from_collection(collection)
result = execute_command(collection, test.build_command(ctx))
assertResult(
result,
error_code=test.error_code,
msg=test.msg,
raw_res=True,
)


@pytest.mark.parametrize("test", pytest_params(COMPACTION_TOKENS_CONTENT_TESTS))
def test_compactStructuredEncryptionData_compactionTokens_content_edge_cases(
database_client, collection, test
):
"""Test compactStructuredEncryptionData rejects edge-case compactionTokens document content."""
collection = test.prepare(database_client, collection)
ctx = CommandContext.from_collection(collection)
result = execute_command(collection, test.build_command(ctx))
assertResult(
result,
error_code=test.error_code,
msg=test.msg,
raw_res=True,
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
"""Tests for compactStructuredEncryptionData error cases.

Covers unrecognized fields and collection type variants (views, capped).
"""

import pytest

from documentdb_tests.compatibility.tests.core.utils.command_test_case import (
CommandContext,
CommandTestCase,
)
from documentdb_tests.framework.assertions import assertResult
from documentdb_tests.framework.error_codes import (
COMMAND_NOT_SUPPORTED_ON_VIEW_ERROR,
NOT_ENCRYPTED_COLLECTION_ERROR,
UNRECOGNIZED_COMMAND_FIELD_ERROR,
)
from documentdb_tests.framework.executor import execute_command
from documentdb_tests.framework.parametrize import pytest_params
from documentdb_tests.framework.target_collection import CappedCollection, ViewCollection

pytestmark = pytest.mark.admin

# Property [Unrecognized Field Rejection]: compactStructuredEncryptionData rejects
# commands with unrecognized fields.
UNRECOGNIZED_FIELD_TESTS: list[CommandTestCase] = [
CommandTestCase(
"extra_field",
docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {},
"unknownField": 1,
},
error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR,
msg="compactStructuredEncryptionData should reject unrecognized fields",
),
CommandTestCase(
"similar_field_name",
docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {},
"compactionToken": {},
},
error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR,
msg="compactStructuredEncryptionData should reject fields with similar names",
),
]

# Property [Collection Type Rejection]: compactStructuredEncryptionData rejects
# views and returns non-encrypted error for capped collections.
COLLECTION_VARIANT_TESTS: list[CommandTestCase] = [
CommandTestCase(
"on_view",
docs=[{"_id": 1}],
target_collection=ViewCollection(),
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {},
},
error_code=COMMAND_NOT_SUPPORTED_ON_VIEW_ERROR,
msg="compactStructuredEncryptionData should reject views",
),
CommandTestCase(
"on_capped_collection",
docs=[{"_id": 1}],
target_collection=CappedCollection(),
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should reject non-encrypted capped collection",
),
]

ERROR_TESTS = UNRECOGNIZED_FIELD_TESTS + COLLECTION_VARIANT_TESTS


@pytest.mark.parametrize("test", pytest_params(ERROR_TESTS))
def test_compactStructuredEncryptionData_errors(database_client, collection, test):
"""Test compactStructuredEncryptionData error conditions."""
collection = test.prepare(database_client, collection)
ctx = CommandContext.from_collection(collection)
result = execute_command(collection, test.build_command(ctx))
assertResult(
result,
error_code=test.error_code,
msg=test.msg,
raw_res=True,
)
Loading
Loading