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,115 @@
"""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 error on non-existent system prefix collection",
),
CommandTestCase(
"dotted_name",
command=lambda ctx: {
"compactStructuredEncryptionData": "a.b.c",
"compactionTokens": {},
},
error_code=NAMESPACE_NOT_FOUND_ERROR,
msg="compactStructuredEncryptionData should error on non-existent dotted collection",
),
CommandTestCase(
"dollar_prefix",
command=lambda ctx: {
"compactStructuredEncryptionData": "$cmd",
"compactionTokens": {},
},
error_code=NAMESPACE_NOT_FOUND_ERROR,
msg="compactStructuredEncryptionData should error on non-existent"
" dollar-prefixed collection",
),
]

# Property [CompactionTokens Content Edge Cases]: compactStructuredEncryptionData
# handles various compactionTokens document content correctly.
COMPACTION_TOKENS_CONTENT_TESTS: list[CommandTestCase] = [
CommandTestCase(
"null_token_value",
docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {"ssn": None},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should handle null token values",
),
CommandTestCase(
"empty_string_key",
docs=[],
command=lambda ctx: {
"compactStructuredEncryptionData": ctx.collection,
"compactionTokens": {"": b"\x00\x01"},
},
error_code=NOT_ENCRYPTED_COLLECTION_ERROR,
msg="compactStructuredEncryptionData should handle empty string key in tokens",
),
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 handle dot-notation key in tokens",
),
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 handle nested document in token value",
),
]

EDGE_CASE_TESTS = COLLECTION_NAME_TESTS + COMPACTION_TOKENS_CONTENT_TESTS


@pytest.mark.parametrize("test", pytest_params(EDGE_CASE_TESTS))
def test_compactStructuredEncryptionData_edge_cases(database_client, collection, test):
"""Test compactStructuredEncryptionData edge cases."""
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,
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
"""Tests for compactStructuredEncryptionData command field validation.

Covers collection name type validation (§19 representative case),
compactionTokens BSON type rejection, and missing field errors.
"""

import pytest

from documentdb_tests.framework.assertions import assertFailureCode
from documentdb_tests.framework.bson_type_validator import (
BsonTypeTestCase,
generate_bson_acceptance_test_cases,
generate_bson_rejection_test_cases,
)
from documentdb_tests.framework.error_codes import (
INVALID_NAMESPACE_ERROR,
MISSING_FIELD_ERROR,
NOT_ENCRYPTED_COLLECTION_ERROR,
TYPE_MISMATCH_ERROR,
)
from documentdb_tests.framework.executor import execute_command
from documentdb_tests.framework.test_constants import BsonType

pytestmark = pytest.mark.admin

# Property [CompactionTokens Type Rejection]: compactStructuredEncryptionData rejects
# non-document types for the compactionTokens field.
BSON_TYPE_PARAMS = [
BsonTypeTestCase(
id="compactionTokens_type",
msg="compactionTokens should reject non-document types",
keyword="compactionTokens",
valid_types=[BsonType.OBJECT],
default_error_code=TYPE_MISMATCH_ERROR,
error_code_overrides={BsonType.NULL: MISSING_FIELD_ERROR},
),
]

REJECTION_CASES = generate_bson_rejection_test_cases(BSON_TYPE_PARAMS)
ACCEPTANCE_CASES = generate_bson_acceptance_test_cases(BSON_TYPE_PARAMS)


@pytest.mark.parametrize("bson_type,sample_value,spec", REJECTION_CASES)
def test_compactStructuredEncryptionData_rejects_invalid_compactionTokens_type(
collection, bson_type, sample_value, spec
):
"""Test compactStructuredEncryptionData rejects invalid BSON types for compactionTokens."""
cmd = {
"compactStructuredEncryptionData": collection.name,
"compactionTokens": sample_value,
}
result = execute_command(collection, cmd)
assertFailureCode(result, spec.expected_code(bson_type), msg=spec.msg)


@pytest.mark.parametrize("bson_type,sample_value,spec", ACCEPTANCE_CASES)
def test_compactStructuredEncryptionData_accepts_valid_compactionTokens_type(
collection, bson_type, sample_value, spec
):
"""Test compactStructuredEncryptionData accepts document type for compactionTokens.

The command accepts the type but fails because the collection is not encrypted.
Error 6346807 confirms the type was accepted and processing continued.
"""
collection.insert_one({"_id": 1})
cmd = {
"compactStructuredEncryptionData": collection.name,
"compactionTokens": sample_value,
}
result = execute_command(collection, cmd)
assertFailureCode(
result,
NOT_ENCRYPTED_COLLECTION_ERROR,
msg=spec.msg,
)


def test_compactStructuredEncryptionData_rejects_non_string_collection_name(collection):
"""Test compactStructuredEncryptionData rejects non-string collection name."""
cmd = {"compactStructuredEncryptionData": 1, "compactionTokens": {}}
result = execute_command(collection, cmd)
assertFailureCode(
result,
INVALID_NAMESPACE_ERROR,
msg="compactStructuredEncryptionData should reject non-string collection name",
)


def test_compactStructuredEncryptionData_rejects_empty_collection_name(collection):
"""Test compactStructuredEncryptionData rejects empty string collection name."""
cmd = {"compactStructuredEncryptionData": "", "compactionTokens": {}}
result = execute_command(collection, cmd)
assertFailureCode(
result,
INVALID_NAMESPACE_ERROR,
msg="compactStructuredEncryptionData should reject empty collection name",
)


def test_compactStructuredEncryptionData_missing_compactionTokens(collection):
"""Test compactStructuredEncryptionData requires compactionTokens field."""
collection.insert_one({"_id": 1})
cmd = {"compactStructuredEncryptionData": collection.name}
result = execute_command(collection, cmd)
assertFailureCode(
result,
MISSING_FIELD_ERROR,
msg="compactStructuredEncryptionData should error when compactionTokens is missing",
)
Loading
Loading