From ed4e21866102307bd3c96d27b5a0dc3cc6a61a20 Mon Sep 17 00:00:00 2001 From: Daniel Frankcom Date: Tue, 9 Jun 2026 16:20:20 -0700 Subject: [PATCH 1/4] Add changeStream stage tests Signed-off-by: Daniel Frankcom --- .../test_stages_position_changeStream.py | 290 ++++++++++ .../test_changeStream_command_options.py | 72 +++ .../test_changeStream_event_read_error.py | 49 ++ .../changeStream/test_changeStream_events.py | 464 +++++++++++++++ .../test_changeStream_expanded_events.py | 273 +++++++++ .../test_changeStream_history_lost.py | 129 +++++ .../test_changeStream_namespace_errors.py | 141 +++++ .../test_changeStream_namespace_scope.py | 98 ++++ .../changeStream/test_changeStream_resume.py | 547 ++++++++++++++++++ ..._changeStream_resume_mutual_exclusivity.py | 66 +++ .../test_changeStream_spec_acceptance.py | 227 ++++++++ .../test_changeStream_stable_api.py | 88 +++ .../test_changeStream_timestamp_boundary.py | 117 ++++ .../test_changeStream_validation_errors.py | 530 +++++++++++++++++ .../changeStream/utils/__init__.py | 0 .../changeStream/utils/changeStream_common.py | 60 ++ documentdb_tests/framework/error_codes.py | 11 + 17 files changed, 3162 insertions(+) create mode 100644 documentdb_tests/compatibility/tests/core/operator/stages/test_stages_position_changeStream.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_command_options.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_events.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_expanded_events.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_history_lost.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_errors.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_scope.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume_mutual_exclusivity.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_spec_acceptance.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_stable_api.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_timestamp_boundary.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_validation_errors.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/utils/__init__.py create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/utils/changeStream_common.py diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_position_changeStream.py b/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_position_changeStream.py new file mode 100644 index 000000000..49db2c66b --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_position_changeStream.py @@ -0,0 +1,290 @@ +"""Tests for $changeStream pipeline position constraints and stage composition.""" + +from __future__ import annotations + +import pytest +from pymongo.collection import Collection +from pymongo.database import Database + +from documentdb_tests.compatibility.tests.core.collections.commands.utils.command_test_case import ( + CommandContext, + CommandTestCase, +) +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.error_codes import ( + ILLEGAL_OPERATION_ERROR, + NOT_FIRST_STAGE_ERROR, +) +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Eq +from documentdb_tests.framework.target_collection import ExistingDatabase + +# Property [Following Stage Allow-List]: $changeStream opens as the first stage +# and the pipeline opens successfully when followed by any stage permitted in +# its allow-list. +CHANGESTREAM_FOLLOWING_STAGE_TESTS: list[CommandTestCase] = [ + CommandTestCase( + "following_match", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$match": {"operationType": "insert"}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $match", + ), + CommandTestCase( + "following_project", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$project": {"_id": 1}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $project", + ), + CommandTestCase( + "following_add_fields", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$addFields": {"x": 1}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $addFields", + ), + CommandTestCase( + "following_set", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$set": {"x": 1}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $set", + ), + CommandTestCase( + "following_replace_root", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$replaceRoot": {"newRoot": {"a": 1}}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $replaceRoot", + ), + CommandTestCase( + "following_replace_with", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$replaceWith": {"a": 1}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $replaceWith", + ), + CommandTestCase( + "following_redact", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$redact": "$$DESCEND"}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $redact", + ), + CommandTestCase( + "following_unset", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$unset": "x"}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $unset", + ), + CommandTestCase( + "following_fill", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$fill": {"output": {"x": {"value": 0}}}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $fill", + ), + CommandTestCase( + "following_change_stream_split_large_event", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$changeStreamSplitLargeEvent": {}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open when followed by $changeStreamSplitLargeEvent", + ), +] + +# Property [Disallowed Following Stage Rejection]: a stage outside the +# $changeStream allow-list placed after $changeStream is rejected, including +# stages that desugar to a disallowed system stage (e.g. $count and +# $sortByCount desugar to $group; $densify and $setWindowFields desugar to +# $sort; $fill with a sortBy desugars to $sort). +CHANGESTREAM_DISALLOWED_FOLLOWING_STAGE_TESTS: list[CommandTestCase] = [ + CommandTestCase( + "disallowed_group", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$group": {"_id": "$operationType"}}], + "cursor": {}, + }, + error_code=ILLEGAL_OPERATION_ERROR, + msg="$changeStream should reject a following $group stage", + ), + CommandTestCase( + "disallowed_sort", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$sort": {"_id": 1}}], + "cursor": {}, + }, + error_code=ILLEGAL_OPERATION_ERROR, + msg="$changeStream should reject a following $sort stage", + ), + CommandTestCase( + "disallowed_count", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}, {"$count": "n"}], + "cursor": {}, + }, + error_code=ILLEGAL_OPERATION_ERROR, + msg="$changeStream should reject a following $count stage", + ), + CommandTestCase( + "disallowed_bucket", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [ + {"$changeStream": {}}, + {"$bucket": {"groupBy": "$x", "boundaries": [0, 1, 2], "default": "other"}}, + ], + "cursor": {}, + }, + error_code=ILLEGAL_OPERATION_ERROR, + msg="$changeStream should reject a following $bucket stage", + ), + CommandTestCase( + "disallowed_densify", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [ + {"$changeStream": {}}, + {"$densify": {"field": "x", "range": {"step": 1, "bounds": "full"}}}, + ], + "cursor": {}, + }, + error_code=ILLEGAL_OPERATION_ERROR, + msg="$changeStream should reject a following $densify stage", + ), + CommandTestCase( + "disallowed_set_window_fields", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [ + {"$changeStream": {}}, + {"$setWindowFields": {"sortBy": {"x": 1}, "output": {"n": {"$sum": 1}}}}, + ], + "cursor": {}, + }, + error_code=ILLEGAL_OPERATION_ERROR, + msg="$changeStream should reject a following $setWindowFields stage", + ), + CommandTestCase( + "disallowed_fill_with_sort_by", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [ + {"$changeStream": {}}, + {"$fill": {"sortBy": {"x": 1}, "output": {"y": {"method": "linear"}}}}, + ], + "cursor": {}, + }, + error_code=ILLEGAL_OPERATION_ERROR, + msg="$changeStream should reject a following $fill stage that desugars to $sort", + ), +] + +# Property [Stage Position Rejection]: $changeStream placed anywhere other than +# the first stage of the pipeline is rejected with a not-first-stage error in +# every namespace scope. +# +# The collection-less (database/cluster) forms use $documents as the leading +# stage because it is valid in a collection-less pipeline; a collection- +# requiring stage such as $match would fail with an invalid-namespace error +# before $changeStream's position check runs, masking it. +CHANGESTREAM_STAGE_POSITION_ERROR_TESTS: list[CommandTestCase] = [ + CommandTestCase( + "not_first_collection", + docs=[{"_id": 1}], + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$match": {}}, {"$changeStream": {}}], + "cursor": {}, + }, + error_code=NOT_FIRST_STAGE_ERROR, + msg="$changeStream should reject a non-first stage position in a collection pipeline", + ), + CommandTestCase( + "not_first_database", + docs=[{"_id": 1}], + command={ + "aggregate": 1, + "pipeline": [{"$documents": [{"x": 1}]}, {"$changeStream": {}}], + "cursor": {}, + }, + error_code=NOT_FIRST_STAGE_ERROR, + msg="$changeStream should reject a non-first stage position in a database-scoped pipeline", + ), + CommandTestCase( + "not_first_cluster", + target_collection=ExistingDatabase(db_name="admin"), + docs=None, + command={ + "aggregate": 1, + "pipeline": [ + {"$documents": [{"x": 1}]}, + {"$changeStream": {"allChangesForCluster": True}}, + ], + "cursor": {}, + }, + error_code=NOT_FIRST_STAGE_ERROR, + msg="$changeStream should reject a non-first stage position in a cluster-scoped pipeline", + ), +] + +CHANGESTREAM_POSITION_TESTS = ( + CHANGESTREAM_FOLLOWING_STAGE_TESTS + + CHANGESTREAM_DISALLOWED_FOLLOWING_STAGE_TESTS + + CHANGESTREAM_STAGE_POSITION_ERROR_TESTS +) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test", pytest_params(CHANGESTREAM_POSITION_TESTS)) +def test_changeStream_position( + database_client: Database, collection: Collection, test: CommandTestCase +): + """Test $changeStream pipeline position constraints and stage composition.""" + target = test.prepare(database_client, collection) + ctx = CommandContext.from_collection(target) + result = execute_command(target, test.build_command(ctx)) + assertResult( + result, + expected=test.build_expected(ctx), + error_code=test.error_code, + msg=test.msg, + raw_res=True, + ) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_command_options.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_command_options.py new file mode 100644 index 000000000..eb0dc44c9 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_command_options.py @@ -0,0 +1,72 @@ +"""Tests for $changeStream top-level command option acceptance.""" + +from __future__ import annotations + +import pytest + +from documentdb_tests.compatibility.tests.core.collections.commands.utils.command_test_case import ( + CommandContext, + CommandTestCase, +) +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Eq + +# Property [Cursor and Command Options]: a stream opens when standard aggregate +# cursor and command options are supplied alongside the $changeStream pipeline. +CHANGESTREAM_COMMAND_OPTION_TESTS: list[CommandTestCase] = [ + CommandTestCase( + "cursor_batch_size_zero", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {"batchSize": 0}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open with cursor.batchSize 0", + ), + CommandTestCase( + "cursor_batch_size_five", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {"batchSize": 5}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open with cursor.batchSize 5", + ), + CommandTestCase( + "max_time_ms", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {}, + "maxTimeMS": 1000, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open with maxTimeMS", + ), + CommandTestCase( + "collation", + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {}, + "collation": {"locale": "en"}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open with a top-level collation", + ), +] + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test", pytest_params(CHANGESTREAM_COMMAND_OPTION_TESTS)) +def test_changeStream_command_options(database_client, collection, test): + """Test $changeStream opens with standard aggregate cursor and command options.""" + target = test.prepare(database_client, collection) + ctx = CommandContext.from_collection(target) + result = execute_command(target, test.build_command(ctx)) + assertResult(result, expected=test.build_expected(ctx), msg=test.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py new file mode 100644 index 000000000..ddbc368b2 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py @@ -0,0 +1,49 @@ +"""Tests for $changeStream event read errors when a required pre/post image is missing.""" + +from __future__ import annotations + +import pytest +from utils.changeStream_common import change_stream_command, get_more_command + +from documentdb_tests.compatibility.tests.core.operator.stages.utils.stage_test_case import ( + StageTestCase, +) +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.error_codes import NO_MATCHING_DOCUMENT_ERROR +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params + +# Property [Required Image Event-Read Error]: a required-image mode on a +# collection without pre/post images enabled fails when reading the update event. +CHANGESTREAM_EVENT_READ_ERROR_TESTS: list[StageTestCase] = [ + StageTestCase( + "full_document_required_no_images", + pipeline=[{"$changeStream": {"fullDocument": "required"}}], + error_code=NO_MATCHING_DOCUMENT_ERROR, + msg="$changeStream fullDocument required should open but fail reading an update" + " event without pre/post images enabled", + ), + StageTestCase( + "full_document_before_change_required_no_images", + pipeline=[{"$changeStream": {"fullDocumentBeforeChange": "required"}}], + error_code=NO_MATCHING_DOCUMENT_ERROR, + msg="$changeStream fullDocumentBeforeChange required should open but fail reading" + " an update event without pre/post images enabled", + ), +] + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EVENT_READ_ERROR_TESTS)) +def test_changeStream_required_image_event_read_error(collection, test_case): + """Test $changeStream defers required-image enforcement to event-read time.""" + collection.insert_one({"_id": 1, "a": 1}) + # The open must succeed: enforcement of the required image is deferred to the + # getMore that reads the update event, not raised at parse/open time. + opened = collection.database.command( + change_stream_command(collection, pipeline=test_case.pipeline) + ) + collection.update_one({"_id": 1}, {"$set": {"a": 2}}) + result = execute_command(collection, get_more_command(collection, opened["cursor"]["id"])) + assertResult(result, error_code=test_case.error_code, msg=test_case.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_events.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_events.py new file mode 100644 index 000000000..ea681ace8 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_events.py @@ -0,0 +1,464 @@ +"""Tests for $changeStream emitted event document structure.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any + +import pytest +from bson import Binary, Code, DBRef, Int64, MaxKey, MinKey, ObjectId, Timestamp +from utils.changeStream_common import change_stream_command, get_more_command + +from documentdb_tests.framework.assertions import assertProperties +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Eq, Exists, IsType, NotExists +from documentdb_tests.framework.test_case import BaseTestCase +from documentdb_tests.framework.test_constants import DECIMAL128_ONE_AND_HALF + + +@dataclass(frozen=True) +class Mutation: + """A change to perform against a collection, observed as one change event. + + Attributes: + operation_type: The ``operationType`` the resulting event carries, used + to pick the event out of the drained batch + apply: Performs the observed change after the stream is open + seed: Document inserted before the stream opens so the mutation has + something to act on (None for inserts) + """ + + operation_type: str + apply: Callable[[Any], None] + seed: dict[str, Any] | None = None + + +def _insert(doc: dict[str, Any]) -> Mutation: + """Insert ``doc`` after the stream opens; the insert is the observed event.""" + return Mutation("insert", lambda c: c.insert_one(dict(doc))) + + +def _update(seed: dict[str, Any], spec: Any) -> Mutation: + """Seed ``seed``, then apply update ``spec`` to it after the stream opens.""" + return Mutation("update", lambda c: c.update_one({"_id": seed["_id"]}, spec), seed=seed) + + +def _replace(seed: dict[str, Any], replacement: dict[str, Any]) -> Mutation: + """Seed ``seed``, then replace it with ``replacement`` after the stream opens.""" + return Mutation( + "replace", lambda c: c.replace_one({"_id": seed["_id"]}, dict(replacement)), seed=seed + ) + + +def _delete(seed: dict[str, Any]) -> Mutation: + """Seed ``seed``, then delete it after the stream opens.""" + return Mutation("delete", lambda c: c.delete_one({"_id": seed["_id"]}), seed=seed) + + +@dataclass(frozen=True) +class ChangeStreamEventTestCase(BaseTestCase): + """Test case for $changeStream emitted event structure. + + Drives an imperative open-mutate-drain sequence: the stream is opened with + ``pipeline`` (the full ``[{"$changeStream": ...}]`` pipeline), ``mutation`` + is performed, and the inherited ``expected`` (a property-check map) asserts + envelope/payload fields on the resulting event. + + Attributes: + pipeline: The full ``[{"$changeStream": ...}]`` pipeline to open with + (required) + mutation: Fully describes the observed write, see ``Mutation`` (required) + pre_and_post_images: When true, enables pre/post image capture before + the stream opens + post_mutation: An operation run after the observed mutation and before + the stream is drained, e.g. deleting the document so a later + updateLookup finds nothing (None when not needed) + """ + + pipeline: list[dict[str, Any]] | None = None + mutation: Mutation | None = None + pre_and_post_images: bool = False + post_mutation: Callable[[Any], None] | None = None + + def __post_init__(self): + super().__post_init__() + if self.pipeline is None: + raise ValueError(f"ChangeStreamEventTestCase '{self.id}' must set pipeline") + if self.mutation is None: + raise ValueError(f"ChangeStreamEventTestCase '{self.id}' must set mutation") + + +# Property [Event Envelope Fields]: every change event carries the common +# envelope fields with their documented BSON types, and operationType equals the +# name of the operation that produced the event. +CHANGESTREAM_EVENT_ENVELOPE_TESTS: list[ChangeStreamEventTestCase] = [ + ChangeStreamEventTestCase( + "envelope_insert", + pipeline=[{"$changeStream": {}}], + mutation=_insert({"_id": 1, "a": 1}), + expected={ + "_id._data": IsType("string"), + "operationType": Eq("insert"), + "clusterTime": IsType("timestamp"), + "wallTime": IsType("date"), + "ns.db": IsType("string"), + "ns.coll": IsType("string"), + "documentKey": IsType("object"), + }, + msg="$changeStream insert event should carry the common envelope fields", + ), + ChangeStreamEventTestCase( + "envelope_update", + pipeline=[{"$changeStream": {}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={ + "_id._data": IsType("string"), + "operationType": Eq("update"), + "clusterTime": IsType("timestamp"), + "wallTime": IsType("date"), + "ns.db": IsType("string"), + "ns.coll": IsType("string"), + "documentKey": IsType("object"), + }, + msg="$changeStream update event should carry the common envelope fields", + ), + ChangeStreamEventTestCase( + "envelope_replace", + pipeline=[{"$changeStream": {}}], + mutation=_replace({"_id": 1, "a": 1}, {"b": 2}), + expected={ + "_id._data": IsType("string"), + "operationType": Eq("replace"), + "clusterTime": IsType("timestamp"), + "wallTime": IsType("date"), + "ns.db": IsType("string"), + "ns.coll": IsType("string"), + "documentKey": IsType("object"), + }, + msg="$changeStream replace event should carry the common envelope fields", + ), + ChangeStreamEventTestCase( + "envelope_delete", + pipeline=[{"$changeStream": {}}], + mutation=_delete({"_id": 1, "a": 1}), + expected={ + "_id._data": IsType("string"), + "operationType": Eq("delete"), + "clusterTime": IsType("timestamp"), + "wallTime": IsType("date"), + "ns.db": IsType("string"), + "ns.coll": IsType("string"), + "documentKey": IsType("object"), + }, + msg="$changeStream delete event should carry the common envelope fields", + ), +] + +# Property [documentKey Id Type Preservation]: documentKey._id preserves the BSON +# type and value of the source document's _id. +CHANGESTREAM_EVENT_DOCUMENT_KEY_TESTS: list[ChangeStreamEventTestCase] = [ + ChangeStreamEventTestCase( + f"document_key_{tid}", + pipeline=[{"$changeStream": {}}], + mutation=_insert({"_id": val, "marker": 1}), + expected={"documentKey._id": Eq(val)}, + msg=f"$changeStream should preserve a {tid} _id in documentKey", + ) + for tid, val in [ + ("int32", 7), + ("int64", Int64(7)), + ("double", 3.5), + ("decimal128", DECIMAL128_ONE_AND_HALF), + ("bool", True), + ("string", "abc"), + ("objectid", ObjectId("507f1f77bcf86cd799439011")), + ("datetime", datetime(2024, 1, 1, tzinfo=timezone.utc)), + ("timestamp", Timestamp(1, 1)), + ("binary", Binary(b"\x01" * 16, 4)), + ("code", Code("function(){}")), + ("minkey", MinKey()), + ("maxkey", MaxKey()), + ("dbref", DBRef("c", 1)), + ("object", {"x": 1}), + ("null", None), + ] +] + +# Property [Operation Payload Fields]: each operationType carries exactly the +# payload fields documented for it, and insert carries fullDocument regardless +# of the fullDocument option. +CHANGESTREAM_EVENT_PAYLOAD_TESTS: list[ChangeStreamEventTestCase] = [ + ChangeStreamEventTestCase( + "payload_insert", + pipeline=[{"$changeStream": {}}], + mutation=_insert({"_id": 1, "a": 1}), + expected={ + "fullDocument": Eq({"_id": 1, "a": 1}), + "updateDescription": NotExists(), + }, + msg="$changeStream insert event should carry fullDocument and no updateDescription", + ), + ChangeStreamEventTestCase( + "payload_insert_full_document_mode", + pipeline=[{"$changeStream": {"fullDocument": "updateLookup"}}], + mutation=_insert({"_id": 1, "a": 1}), + expected={ + "fullDocument": Eq({"_id": 1, "a": 1}), + "updateDescription": NotExists(), + }, + msg=( + "$changeStream insert event should carry fullDocument regardless of" + " the fullDocument option" + ), + ), + ChangeStreamEventTestCase( + "payload_replace", + pipeline=[{"$changeStream": {}}], + mutation=_replace({"_id": 1, "a": 1}, {"b": 2}), + expected={ + "fullDocument": Eq({"_id": 1, "b": 2}), + "updateDescription": NotExists(), + }, + msg="$changeStream replace event should carry fullDocument and no updateDescription", + ), + ChangeStreamEventTestCase( + "payload_delete", + pipeline=[{"$changeStream": {}}], + mutation=_delete({"_id": 1, "a": 1}), + expected={ + "fullDocument": NotExists(), + "updateDescription": NotExists(), + }, + msg="$changeStream delete event should carry neither fullDocument nor updateDescription", + ), + ChangeStreamEventTestCase( + "payload_update", + pipeline=[{"$changeStream": {}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={ + "updateDescription": Exists(), + }, + msg="$changeStream update event should always carry updateDescription", + ), +] + +# Property [updateDescription Structure]: an update event's updateDescription +# records changed paths in updatedFields (new nested paths as nested objects), +# removed paths in removedFields, and end-truncations in truncatedArrays, +# independently of the fullDocument mode. +CHANGESTREAM_EVENT_UPDATE_DESCRIPTION_TESTS: list[ChangeStreamEventTestCase] = [ + ChangeStreamEventTestCase( + "update_description_set_unset", + pipeline=[{"$changeStream": {}}], + mutation=_update({"_id": 1, "a": 1, "b": 1}, {"$set": {"b": 2}, "$unset": {"a": ""}}), + expected={ + "updateDescription": Eq( + {"updatedFields": {"b": 2}, "removedFields": ["a"], "truncatedArrays": []} + ) + }, + msg="$changeStream updateDescription should record set paths and removed fields", + ), + ChangeStreamEventTestCase( + "update_description_nested_set", + pipeline=[{"$changeStream": {}}], + mutation=_update({"_id": 1}, {"$set": {"nested.x": 5}}), + expected={"updateDescription.updatedFields": Eq({"nested": {"x": 5}})}, + msg="$changeStream updateDescription should represent a new nested path as a nested object", + ), + ChangeStreamEventTestCase( + "update_description_truncated_arrays", + pipeline=[{"$changeStream": {}}], + mutation=_update( + {"_id": 1, "arr": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]}, + [{"$set": {"arr": {"$slice": ["$arr", 5]}}}], + ), + expected={"updateDescription.truncatedArrays": Eq([{"field": "arr", "newSize": 5}])}, + msg="$changeStream updateDescription should record end-truncation in truncatedArrays", + ), + ChangeStreamEventTestCase( + "update_description_with_full_document_mode", + pipeline=[{"$changeStream": {"fullDocument": "updateLookup"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={ + "updateDescription": Eq( + {"updatedFields": {"a": 2}, "removedFields": [], "truncatedArrays": []} + ) + }, + msg="$changeStream updateDescription should be emitted regardless of the fullDocument mode", + ), +] + +# Property [fullDocument Event Behavior]: an update event's fullDocument field +# tracks the fullDocument mode and pre/post-image availability. +CHANGESTREAM_EVENT_FULL_DOCUMENT_TESTS: list[ChangeStreamEventTestCase] = [ + ChangeStreamEventTestCase( + "full_document_default_omitted", + pipeline=[{"$changeStream": {}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={"fullDocument": NotExists()}, + msg="$changeStream update event should omit fullDocument when the mode is omitted", + ), + ChangeStreamEventTestCase( + "full_document_default_string", + pipeline=[{"$changeStream": {"fullDocument": "default"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={"fullDocument": NotExists()}, + msg="$changeStream update event should omit fullDocument under the default mode", + ), + ChangeStreamEventTestCase( + "full_document_null", + pipeline=[{"$changeStream": {"fullDocument": None}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={"fullDocument": NotExists()}, + msg="$changeStream update event should omit fullDocument when the mode is null", + ), + ChangeStreamEventTestCase( + "full_document_update_lookup", + pipeline=[{"$changeStream": {"fullDocument": "updateLookup"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={"fullDocument": Eq({"_id": 1, "a": 2})}, + msg="$changeStream updateLookup should carry the current document", + ), + ChangeStreamEventTestCase( + "full_document_update_lookup_deleted", + pipeline=[{"$changeStream": {"fullDocument": "updateLookup"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + post_mutation=lambda collection: collection.delete_one({"_id": 1}), + expected={"fullDocument": Eq(None)}, + msg="$changeStream updateLookup should carry null when the document no longer exists", + ), + ChangeStreamEventTestCase( + "full_document_when_available_no_images", + pipeline=[{"$changeStream": {"fullDocument": "whenAvailable"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={"fullDocument": Eq(None)}, + msg="$changeStream whenAvailable should carry null without pre/post images enabled", + ), + ChangeStreamEventTestCase( + "full_document_when_available_with_images", + pipeline=[{"$changeStream": {"fullDocument": "whenAvailable"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + pre_and_post_images=True, + expected={"fullDocument": Eq({"_id": 1, "a": 2})}, + msg="$changeStream whenAvailable should carry the post-image with pre/post images enabled", + ), + ChangeStreamEventTestCase( + "full_document_required_with_images", + pipeline=[{"$changeStream": {"fullDocument": "required"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + pre_and_post_images=True, + expected={"fullDocument": Eq({"_id": 1, "a": 2})}, + msg="$changeStream required should carry the post-image with pre/post images enabled", + ), +] + +# Property [fullDocumentBeforeChange Event Behavior]: an update event's +# fullDocumentBeforeChange tracks the fullDocumentBeforeChange mode and +# pre-image availability, independently of any fullDocument post-image. +CHANGESTREAM_EVENT_FULL_DOCUMENT_BEFORE_CHANGE_TESTS: list[ChangeStreamEventTestCase] = [ + ChangeStreamEventTestCase( + "before_change_off", + pipeline=[{"$changeStream": {"fullDocumentBeforeChange": "off"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={"fullDocumentBeforeChange": NotExists()}, + msg="$changeStream update event should omit fullDocumentBeforeChange under the off mode", + ), + ChangeStreamEventTestCase( + "before_change_when_available_no_images", + pipeline=[{"$changeStream": {"fullDocumentBeforeChange": "whenAvailable"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + expected={"fullDocumentBeforeChange": Eq(None)}, + msg=( + "$changeStream whenAvailable should carry a null fullDocumentBeforeChange" + " without pre/post images enabled" + ), + ), + ChangeStreamEventTestCase( + "before_change_when_available_with_images", + pipeline=[{"$changeStream": {"fullDocumentBeforeChange": "whenAvailable"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + pre_and_post_images=True, + expected={"fullDocumentBeforeChange": Eq({"_id": 1, "a": 1})}, + msg="$changeStream whenAvailable should carry the pre-image with pre/post images enabled", + ), + ChangeStreamEventTestCase( + "before_change_required_with_images", + pipeline=[{"$changeStream": {"fullDocumentBeforeChange": "required"}}], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + pre_and_post_images=True, + expected={"fullDocumentBeforeChange": Eq({"_id": 1, "a": 1})}, + msg="$changeStream required should carry the pre-image with pre/post images enabled", + ), + ChangeStreamEventTestCase( + "before_change_and_full_document_both_images", + pipeline=[ + { + "$changeStream": { + "fullDocument": "whenAvailable", + "fullDocumentBeforeChange": "whenAvailable", + } + } + ], + mutation=_update({"_id": 1, "a": 1}, {"$set": {"a": 2}}), + pre_and_post_images=True, + expected={ + "fullDocument": Eq({"_id": 1, "a": 2}), + "fullDocumentBeforeChange": Eq({"_id": 1, "a": 1}), + }, + msg=( + "$changeStream should carry the post-image in fullDocument and the pre-image" + " in fullDocumentBeforeChange independently when both modes are set" + ), + ), +] + +CHANGESTREAM_EVENT_STRUCTURE_TESTS = ( + CHANGESTREAM_EVENT_ENVELOPE_TESTS + + CHANGESTREAM_EVENT_DOCUMENT_KEY_TESTS + + CHANGESTREAM_EVENT_PAYLOAD_TESTS + + CHANGESTREAM_EVENT_UPDATE_DESCRIPTION_TESTS + + CHANGESTREAM_EVENT_FULL_DOCUMENT_TESTS + + CHANGESTREAM_EVENT_FULL_DOCUMENT_BEFORE_CHANGE_TESTS +) + + +def _emit_event(collection, test_case: ChangeStreamEventTestCase) -> dict[str, Any]: + """Open a stream, perform the test case mutation, and return the matching event. + + The mutation happens after the stream is open, so its event is delivered by + the first getMore. + """ + mutation = test_case.mutation + assert mutation is not None # guaranteed by __post_init__ + if mutation.seed is not None: + collection.insert_one(dict(mutation.seed)) + if test_case.pre_and_post_images: + execute_command( + collection, + {"collMod": collection.name, "changeStreamPreAndPostImages": {"enabled": True}}, + ) + opened = execute_command( + collection, change_stream_command(collection, pipeline=test_case.pipeline) + ) + mutation.apply(collection) + if test_case.post_mutation is not None: + test_case.post_mutation(collection) + batch = execute_command(collection, get_more_command(collection, opened["cursor"]["id"]))[ + "cursor" + ]["nextBatch"] + matching = [e for e in batch if e["operationType"] == mutation.operation_type] + assert matching, f"no {mutation.operation_type} event in batch: {batch!r}" + event: dict[str, Any] = matching[0] + return event + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EVENT_STRUCTURE_TESTS)) +def test_changeStream_event_structure(collection, test_case: ChangeStreamEventTestCase): + """Test $changeStream emitted event structure.""" + event = _emit_event(collection, test_case) + assertProperties(event, test_case.expected, msg=test_case.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_expanded_events.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_expanded_events.py new file mode 100644 index 000000000..f10382c31 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_expanded_events.py @@ -0,0 +1,273 @@ +"""Tests for $changeStream showExpandedEvents operationType gating.""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +import pytest +from utils.changeStream_common import change_stream_command, get_more_command + +from documentdb_tests.framework.assertions import assertProperties +from documentdb_tests.framework.executor import execute_admin_command, execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Contains, NotContains +from documentdb_tests.framework.test_case import BaseTestCase + + +@dataclass(frozen=True) +class DdlOperation: + """A DDL/metadata operation observed through a change stream. + + Attributes: + operation_type: The ``operationType`` the resulting event carries, used + to find the event among the drained events + command: Builds the command document from the fixture collection + scope: The change-stream scope the operation must be observed at + (``collection`` or ``database``) + admin: Routes ``command`` through ``execute_admin_command`` rather than + ``execute_command`` (e.g. rename) + setup: Builds a precondition command run before the stream opens, e.g. + creating an index so it can later be dropped (None when not needed) + """ + + operation_type: str + command: Callable[[Any], dict[str, Any]] + scope: str = "collection" + admin: bool = False + setup: Callable[[Any], dict[str, Any]] | None = None + + +def _create_index_command(collection) -> dict[str, Any]: + return {"createIndexes": collection.name, "indexes": [{"key": {"a": 1}, "name": "a_1"}]} + + +def _create_index() -> DdlOperation: + """Create an index; emits a createIndexes event.""" + return DdlOperation("createIndexes", _create_index_command) + + +def _drop_index() -> DdlOperation: + """Drop an index (created during setup); emits a dropIndexes event.""" + return DdlOperation( + "dropIndexes", + lambda c: {"dropIndexes": c.name, "index": "a_1"}, + setup=_create_index_command, + ) + + +def _coll_mod() -> DdlOperation: + """Modify collection options; emits a modify event.""" + return DdlOperation("modify", lambda c: {"collMod": c.name, "validationLevel": "moderate"}) + + +def _create() -> DdlOperation: + """Create a sibling collection; emits a create event on a database-scoped stream.""" + return DdlOperation("create", lambda c: {"create": f"{c.name}_ddl"}, scope="database") + + +def _drop() -> DdlOperation: + """Drop the collection; emits a drop event.""" + return DdlOperation("drop", lambda c: {"drop": c.name}) + + +def _rename() -> DdlOperation: + """Rename the collection; emits a rename event.""" + return DdlOperation( + "rename", + lambda c: { + "renameCollection": f"{c.database.name}.{c.name}", + "to": f"{c.database.name}.{c.name}_renamed", + }, + admin=True, + ) + + +@dataclass(frozen=True) +class ChangeStreamGatingTestCase(BaseTestCase): + """Test case for $changeStream showExpandedEvents operationType gating. + + Drives an open-DDL-drain sequence: the stream is opened with ``pipeline`` + (the full ``[{"$changeStream": ...}]`` pipeline) at the scope ``ddl`` + requires, the DDL operation is performed, and the drained command result is + checked against the inherited ``expected`` property map. + + Attributes: + pipeline: The full ``[{"$changeStream": ...}]`` pipeline to open with, + at the scope ``ddl`` requires (required) + ddl: The DDL operation to perform, see ``DdlOperation`` (required) + expected: A property-check map over the raw getMore result; for a + gating case it asserts the cursor batch ``Contains`` (shown) or + ``NotContains`` (suppressed) an event with the ddl's operationType + (inherited from ``BaseTestCase``) + """ + + pipeline: list[dict[str, Any]] | None = None + ddl: DdlOperation | None = None + + def __post_init__(self): + super().__post_init__() + if self.pipeline is None: + raise ValueError(f"ChangeStreamGatingTestCase '{self.id}' must set pipeline") + if self.ddl is None: + raise ValueError(f"ChangeStreamGatingTestCase '{self.id}' must set ddl") + + +# Property [Expanded Event Gating]: showExpandedEvents true surfaces the +# expanded DDL/index events that showExpandedEvents false or omission suppresses. +CHANGESTREAM_EXPANDED_EVENT_GATING_TESTS: list[ChangeStreamGatingTestCase] = [ + ChangeStreamGatingTestCase( + "create_index_shown_when_true", + pipeline=[{"$changeStream": {"showExpandedEvents": True}}], + ddl=_create_index(), + expected={"cursor": {"nextBatch": Contains("operationType", "createIndexes")}}, + msg="$changeStream should emit a createIndexes event when showExpandedEvents is true", + ), + ChangeStreamGatingTestCase( + "create_index_suppressed_when_false", + pipeline=[{"$changeStream": {"showExpandedEvents": False}}], + ddl=_create_index(), + expected={"cursor": {"nextBatch": NotContains("operationType", "createIndexes")}}, + msg=( + "$changeStream should suppress the createIndexes event when" + " showExpandedEvents is false" + ), + ), + ChangeStreamGatingTestCase( + "create_index_suppressed_when_omitted", + pipeline=[{"$changeStream": {}}], + ddl=_create_index(), + expected={"cursor": {"nextBatch": NotContains("operationType", "createIndexes")}}, + msg=( + "$changeStream should suppress the createIndexes event when" + " showExpandedEvents is omitted" + ), + ), + ChangeStreamGatingTestCase( + "coll_mod_shown_when_true", + pipeline=[{"$changeStream": {"showExpandedEvents": True}}], + ddl=_coll_mod(), + expected={"cursor": {"nextBatch": Contains("operationType", "modify")}}, + msg="$changeStream should emit a modify event when showExpandedEvents is true", + ), + ChangeStreamGatingTestCase( + "coll_mod_suppressed_when_false", + pipeline=[{"$changeStream": {"showExpandedEvents": False}}], + ddl=_coll_mod(), + expected={"cursor": {"nextBatch": NotContains("operationType", "modify")}}, + msg="$changeStream should suppress the modify event when showExpandedEvents is false", + ), + ChangeStreamGatingTestCase( + "drop_index_shown_when_true", + pipeline=[{"$changeStream": {"showExpandedEvents": True}}], + ddl=_drop_index(), + expected={"cursor": {"nextBatch": Contains("operationType", "dropIndexes")}}, + msg="$changeStream should emit a dropIndexes event when showExpandedEvents is true", + ), + ChangeStreamGatingTestCase( + "drop_index_suppressed_when_false", + pipeline=[{"$changeStream": {"showExpandedEvents": False}}], + ddl=_drop_index(), + expected={"cursor": {"nextBatch": NotContains("operationType", "dropIndexes")}}, + msg="$changeStream should suppress the dropIndexes event when showExpandedEvents is false", + ), + ChangeStreamGatingTestCase( + "create_shown_when_true", + pipeline=[{"$changeStream": {"showExpandedEvents": True}}], + ddl=_create(), + expected={"cursor": {"nextBatch": Contains("operationType", "create")}}, + msg="$changeStream should emit a create event when showExpandedEvents is true", + ), + ChangeStreamGatingTestCase( + "create_suppressed_when_false", + pipeline=[{"$changeStream": {"showExpandedEvents": False}}], + ddl=_create(), + expected={"cursor": {"nextBatch": NotContains("operationType", "create")}}, + msg="$changeStream should suppress the create event when showExpandedEvents is false", + ), +] + +# Property [DDL Event Emission Regardless Of showExpandedEvents]: drop and +# rename events are emitted whether showExpandedEvents is true or false. +CHANGESTREAM_DDL_ALWAYS_EMITTED_TESTS: list[ChangeStreamGatingTestCase] = [ + ChangeStreamGatingTestCase( + "drop_emitted_true", + pipeline=[{"$changeStream": {"showExpandedEvents": True}}], + ddl=_drop(), + expected={"cursor": {"nextBatch": Contains("operationType", "drop")}}, + msg="$changeStream should emit a drop event when showExpandedEvents is true", + ), + ChangeStreamGatingTestCase( + "drop_emitted_false", + pipeline=[{"$changeStream": {"showExpandedEvents": False}}], + ddl=_drop(), + expected={"cursor": {"nextBatch": Contains("operationType", "drop")}}, + msg="$changeStream should emit a drop event when showExpandedEvents is false", + ), + ChangeStreamGatingTestCase( + "rename_emitted_true", + pipeline=[{"$changeStream": {"showExpandedEvents": True}}], + ddl=_rename(), + expected={"cursor": {"nextBatch": Contains("operationType", "rename")}}, + msg="$changeStream should emit a rename event when showExpandedEvents is true", + ), + ChangeStreamGatingTestCase( + "rename_emitted_false", + pipeline=[{"$changeStream": {"showExpandedEvents": False}}], + ddl=_rename(), + expected={"cursor": {"nextBatch": Contains("operationType", "rename")}}, + msg="$changeStream should emit a rename event when showExpandedEvents is false", + ), +] + +CHANGESTREAM_EXPANDED_EVENT_TESTS = ( + CHANGESTREAM_EXPANDED_EVENT_GATING_TESTS + CHANGESTREAM_DDL_ALWAYS_EMITTED_TESTS +) + + +def _emit_ddl_result(collection, database_client, test_case: ChangeStreamGatingTestCase) -> Any: + """Open a stream, perform the DDL operation, and return the raw getMore result. + + For a suppression case a marker insert follows the DDL so the getMore + returns an event instead of blocking until maxTimeMS. + """ + ddl = test_case.ddl + assert ddl is not None # guaranteed by __post_init__ + suppression = isinstance(test_case.expected["cursor"]["nextBatch"], NotContains) + database_scope = ddl.scope == "database" + + if not database_scope: + collection.insert_one({"_id": 1}) + if ddl.setup is not None: + execute_command(collection, ddl.setup(collection)) + + open_kwargs = {"aggregate": 1} if database_scope else {} + opened = execute_command( + collection, change_stream_command(collection, pipeline=test_case.pipeline, **open_kwargs) + ) + + run = execute_admin_command if ddl.admin else execute_command + run(collection, ddl.command(collection)) + + if suppression: + if database_scope: + database_client[f"{collection.name}_ddl"].insert_one({"_id": 1}) + else: + collection.insert_one({"_id": 99}) + + getmore_kwargs = {"name": "$cmd.aggregate"} if database_scope else {} + return execute_command( + collection, get_more_command(collection, opened["cursor"]["id"], **getmore_kwargs) + ) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EXPANDED_EVENT_TESTS)) +def test_changeStream_showExpandedEvents_event_gating( + collection, database_client, test_case: ChangeStreamGatingTestCase +): + """Test $changeStream showExpandedEvents gating of expanded DDL/index events.""" + result = _emit_ddl_result(collection, database_client, test_case) + assertProperties(result, test_case.expected, msg=test_case.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_history_lost.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_history_lost.py new file mode 100644 index 000000000..e7e13f0e6 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_history_lost.py @@ -0,0 +1,129 @@ +"""Tests for $changeStream history-lost errors when the start point predates the oplog.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import pytest +from bson import Timestamp +from utils.changeStream_common import change_stream_command + +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.error_codes import CHANGE_STREAM_HISTORY_LOST_ERROR +from documentdb_tests.framework.executor import execute_admin_command, execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.test_case import BaseTestCase + + +# Return the timestamp of the oldest retained oplog entry, used to express the +# history-lost boundary relative to the live oplog rather than as a static value. +def _oldest_oplog_ts(collection) -> Timestamp: + oldest = ( + collection.database.client["local"]["oplog.rs"] + .find() + .sort("$natural", 1) + .limit(1) + .next()["ts"] + ) + return oldest + + +@dataclass(frozen=True) +class ChangeStreamHistoryLostTestCase(BaseTestCase): + """Test case for a pre-oplog startAtOperationTime rejected as history lost. + + Attributes: + compute_start: Receives the timestamp of the oldest retained oplog + entry captured at run time and returns the startAtOperationTime to + test, so the boundary case is expressed relative to the live oplog + rather than as a static value + """ + + compute_start: Any = None + + +# Property [Timestamp History Lost]: a startAtOperationTime strictly before the +# oldest retained oplog entry is rejected at open with a ChangeStreamHistoryLost +# error, including the boundary five seconds before the oldest retained entry. +# This is verified identically across collection-, database-, and cluster-scoped +# streams. +CHANGESTREAM_HISTORY_LOST_TESTS: list[ChangeStreamHistoryLostTestCase] = [ + ChangeStreamHistoryLostTestCase( + "zero", + compute_start=lambda oldest: Timestamp(0, 0), + error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, + msg="$changeStream should reject a zero startAtOperationTime as history lost", + ), + ChangeStreamHistoryLostTestCase( + "one_zero", + compute_start=lambda oldest: Timestamp(1, 0), + error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, + msg="$changeStream should reject a pre-oplog startAtOperationTime as history lost", + ), + ChangeStreamHistoryLostTestCase( + "max_increment", + compute_start=lambda oldest: Timestamp(1, 4_294_967_295), + error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, + msg="$changeStream should reject a pre-oplog startAtOperationTime as history lost", + ), + ChangeStreamHistoryLostTestCase( + "oldest_minus_5s", + compute_start=lambda oldest: Timestamp(oldest.time - 5, oldest.inc), + error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, + msg="$changeStream should reject a startAtOperationTime five seconds before" + " the oldest retained oplog entry as history lost", + ), +] + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_HISTORY_LOST_TESTS)) +def test_changeStream_history_lost_collection_scope( + collection, test_case: ChangeStreamHistoryLostTestCase +): + """Test $changeStream rejects a pre-oplog startAtOperationTime on a collection-scoped stream.""" + start = test_case.compute_start(_oldest_oplog_ts(collection)) + result = execute_command( + collection, + change_stream_command( + collection, pipeline=[{"$changeStream": {"startAtOperationTime": start}}] + ), + ) + assertResult(result, error_code=test_case.error_code, msg=test_case.msg) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_HISTORY_LOST_TESTS)) +def test_changeStream_history_lost_database_scope( + collection, test_case: ChangeStreamHistoryLostTestCase +): + """Test $changeStream rejects a pre-oplog startAtOperationTime on a database-scoped stream.""" + start = test_case.compute_start(_oldest_oplog_ts(collection)) + result = execute_command( + collection, + change_stream_command( + collection, + pipeline=[{"$changeStream": {"startAtOperationTime": start}}], + aggregate=1, + ), + ) + assertResult(result, error_code=test_case.error_code, msg=test_case.msg) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_HISTORY_LOST_TESTS)) +def test_changeStream_history_lost_cluster_scope( + collection, test_case: ChangeStreamHistoryLostTestCase +): + """Test $changeStream rejects a pre-oplog startAtOperationTime on a cluster-wide stream.""" + start = test_case.compute_start(_oldest_oplog_ts(collection)) + spec = {"startAtOperationTime": start, "allChangesForCluster": True} + result = execute_admin_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": spec}], aggregate=1), + ) + assertResult(result, error_code=test_case.error_code, msg=test_case.msg) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_errors.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_errors.py new file mode 100644 index 000000000..07fe23be9 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_errors.py @@ -0,0 +1,141 @@ +"""Tests for $changeStream namespace-related rejections (view, reserved, cluster).""" + +from __future__ import annotations + +import pytest + +from documentdb_tests.compatibility.tests.core.collections.commands.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, + INVALID_NAMESPACE_ERROR, + INVALID_OPTIONS_ERROR, +) +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.target_collection import ExistingDatabase, ViewCollection + +# Property [View Namespace Rejection]: opening a stream on a view is rejected +# because change streams are not allowed on views. +CHANGESTREAM_VIEW_NAMESPACE_ERROR_TESTS: list[CommandTestCase] = [ + CommandTestCase( + "view", + target_collection=ViewCollection(), + docs=[{"_id": 1}], + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {}, + }, + error_code=COMMAND_NOT_SUPPORTED_ON_VIEW_ERROR, + msg="$changeStream should reject opening a stream on a view", + ), +] + +# Property [Reserved Namespace Rejection]: opening a stream on a reserved +# database or a reserved system collection is rejected as an invalid namespace. +CHANGESTREAM_RESERVED_NAMESPACE_ERROR_TESTS: list[CommandTestCase] = [ + CommandTestCase( + "reserved_db_admin", + target_collection=ExistingDatabase(db_name="admin"), + docs=None, + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {}, + }, + error_code=INVALID_NAMESPACE_ERROR, + msg="$changeStream should reject a collection-scoped stream on the admin database", + ), + CommandTestCase( + "reserved_db_local", + target_collection=ExistingDatabase(db_name="local"), + docs=None, + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {}, + }, + error_code=INVALID_NAMESPACE_ERROR, + msg="$changeStream should reject a collection-scoped stream on the local database", + ), + CommandTestCase( + "reserved_db_config", + target_collection=ExistingDatabase(db_name="config"), + docs=None, + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {}, + }, + error_code=INVALID_NAMESPACE_ERROR, + msg="$changeStream should reject a collection-scoped stream on the config database", + ), + CommandTestCase( + "system_views", + docs=[{"_id": 1}], + command={"aggregate": "system.views", "pipeline": [{"$changeStream": {}}], "cursor": {}}, + error_code=INVALID_NAMESPACE_ERROR, + msg="$changeStream should reject a stream on the system.views collection", + ), + CommandTestCase( + "system_profile", + docs=[{"_id": 1}], + command={"aggregate": "system.profile", "pipeline": [{"$changeStream": {}}], "cursor": {}}, + error_code=INVALID_NAMESPACE_ERROR, + msg="$changeStream should reject a stream on the system.profile collection", + ), +] + +# Property [allChangesForCluster Namespace Rejection]: allChangesForCluster true +# is rejected as an invalid option anywhere other than a collection-less stream +# on the admin database, including a non-admin database and the admin database +# with a collection name present. +CHANGESTREAM_ALL_CHANGES_NAMESPACE_ERROR_TESTS: list[CommandTestCase] = [ + CommandTestCase( + "all_changes_true_non_admin", + docs=[{"_id": 1}], + command={ + "aggregate": 1, + "pipeline": [{"$changeStream": {"allChangesForCluster": True}}], + "cursor": {}, + }, + error_code=INVALID_OPTIONS_ERROR, + msg="$changeStream should reject allChangesForCluster true on a non-admin database", + ), + CommandTestCase( + "all_changes_true_admin_collection", + target_collection=ExistingDatabase(db_name="admin"), + docs=None, + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {"allChangesForCluster": True}}], + "cursor": {}, + }, + error_code=INVALID_OPTIONS_ERROR, + msg=( + "$changeStream should reject allChangesForCluster true on the admin" + " database with a collection name" + ), + ), +] + +CHANGESTREAM_NAMESPACE_ERROR_TESTS = ( + CHANGESTREAM_VIEW_NAMESPACE_ERROR_TESTS + + CHANGESTREAM_RESERVED_NAMESPACE_ERROR_TESTS + + CHANGESTREAM_ALL_CHANGES_NAMESPACE_ERROR_TESTS +) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test", pytest_params(CHANGESTREAM_NAMESPACE_ERROR_TESTS)) +def test_changeStream_namespace_errors(database_client, collection, test): + """Test $changeStream rejects disallowed namespaces and option/namespace combinations.""" + target = test.prepare(database_client, collection) + ctx = CommandContext.from_collection(target) + result = execute_command(target, test.build_command(ctx)) + assertResult(result, error_code=test.error_code, msg=test.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_scope.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_scope.py new file mode 100644 index 000000000..4289b81d8 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_scope.py @@ -0,0 +1,98 @@ +"""Tests for $changeStream namespace scope (collection, database, cluster).""" + +from __future__ import annotations + +import pytest + +from documentdb_tests.compatibility.tests.core.collections.commands.utils.command_test_case import ( + CommandContext, + CommandTestCase, +) +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Eq +from documentdb_tests.framework.target_collection import ( + CappedCollection, + ExistingDatabase, + TargetDatabase, +) + +# Property [Namespace Scope]: a stream opens on a collection-scoped, a +# database-scoped, and the cluster-wide namespace. +CHANGESTREAM_NAMESPACE_SCOPE_TESTS: list[CommandTestCase] = [ + CommandTestCase( + "collection_existing", + docs=[{"_id": 1}], + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open a collection-scoped stream on an existing collection", + ), + CommandTestCase( + "collection_nonexistent", + docs=None, + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open a collection-scoped stream on a non-existent collection", + ), + CommandTestCase( + "collection_capped", + target_collection=CappedCollection(), + docs=[], + command=lambda ctx: { + "aggregate": ctx.collection, + "pipeline": [{"$changeStream": {}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open a collection-scoped stream on a capped collection", + ), + CommandTestCase( + "database_existing", + docs=[{"_id": 1}], + command={"aggregate": 1, "pipeline": [{"$changeStream": {}}], "cursor": {}}, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open a database-scoped stream on an existing database", + ), + CommandTestCase( + "database_nonexistent", + target_collection=TargetDatabase(suffix="absent"), + docs=None, + command={"aggregate": 1, "pipeline": [{"$changeStream": {}}], "cursor": {}}, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open a database-scoped stream on a non-existent database", + ), + CommandTestCase( + "cluster_admin", + target_collection=ExistingDatabase(db_name="admin"), + docs=None, + command={ + "aggregate": 1, + "pipeline": [{"$changeStream": {"allChangesForCluster": True}}], + "cursor": {}, + }, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open a cluster-wide stream on the admin database", + ), +] + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test", pytest_params(CHANGESTREAM_NAMESPACE_SCOPE_TESTS)) +def test_changeStream_namespace_scope(database_client, collection, register_db_cleanup, test): + """Test $changeStream opens across collection, database, and cluster namespace scopes.""" + target = test.prepare(database_client, collection) + if isinstance(test.target_collection, TargetDatabase): + register_db_cleanup(target.database.name) + ctx = CommandContext.from_collection(target) + result = execute_command(target, test.build_command(ctx)) + assertResult(result, expected=test.build_expected(ctx), msg=test.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py new file mode 100644 index 000000000..39f8c1e6a --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py @@ -0,0 +1,547 @@ +"""Tests for $changeStream resume-token repositioning behavior. + +Each behavior is written as a standalone, self-contained test. Each test +captures the token it needs inline, opens the resumed stream inline, and asserts +inline, so the precondition, action, and expectation are all visible in one +place. +""" + +from __future__ import annotations + +import pytest +from bson import Binary +from utils.changeStream_common import change_stream_command, get_more_command + +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.error_codes import ( + CHANGE_STREAM_FATAL_ERROR, + CHANGE_STREAM_HISTORY_LOST_ERROR, + INVALID_RESUME_TOKEN_ERROR, + OVERFLOW_ERROR, +) +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.property_checks import Eq, Len + +pytestmark = [pytest.mark.replica_set, pytest.mark.aggregate] + + +# Token capture helpers. A resume token must be captured from a real stream +# before it can be used, so each helper opens a stream, optionally seeds events, +# and returns a token from the response. + + +def _open_stream(collection, spec=None): + """Open a $changeStream and return the open response.""" + return execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": spec or {}}]), + ) + + +def _drain(collection, opened): + """Return the next batch of events from an already-open stream.""" + result = execute_command(collection, get_more_command(collection, opened["cursor"]["id"])) + return result["cursor"]["nextBatch"] + + +def _first_event_token(collection, seed, spec=None): + """Open a stream, seed events, and return the first event's resume token.""" + opened = _open_stream(collection, spec) + seed(collection) + return _drain(collection, opened)[0]["_id"] + + +def _event_token_of_type(collection, operation_type, seed, spec=None): + """Open a stream, seed events, and return the token of the first matching event.""" + opened = _open_stream(collection, spec) + seed(collection) + batch = _drain(collection, opened) + return next(e["_id"] for e in batch if e["operationType"] == operation_type) + + +def _high_water_mark(collection): + """Return the postBatchResumeToken of a freshly opened, idle stream.""" + return _open_stream(collection)["cursor"]["postBatchResumeToken"] + + +def _evicted_token(collection): + """Return a token whose captured cluster time predates the oplog window. + + Rewrites the cluster-time seconds (four bytes after the version byte) of a + high-water-mark token's _data hex to a near-zero value. + """ + data = _high_water_mark(collection)["_data"] + return {"_data": data[:2] + "00000001" + data[10:]} + + +# Property [Resume After Event]: resumeAfter an event token delivers the events that follow it. +def test_resume_after_event_token(collection): + """Test $changeStream resumeAfter from an event token.""" + token = _first_event_token( + collection, lambda c: c.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}]) + ) + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"resumeAfter": token}}]), + ) + + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(2)}, + "cursor.firstBatch.0.documentKey._id": Eq(2), + "cursor.firstBatch.1.documentKey._id": Eq(3), + }, + msg="$changeStream resumeAfter should resume after the captured event token", + raw_res=True, + ) + + +# Property [Start After Event]: startAfter an event token delivers the events that follow it. +def test_start_after_event_token(collection): + """Test $changeStream startAfter from an event token.""" + token = _first_event_token( + collection, lambda c: c.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}]) + ) + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"startAfter": token}}]), + ) + + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(2)}, + "cursor.firstBatch.0.documentKey._id": Eq(2), + "cursor.firstBatch.1.documentKey._id": Eq(3), + }, + msg="$changeStream startAfter should resume after the captured event token", + raw_res=True, + ) + + +# Property [Resume After Drop]: resumeAfter a drop token positions the stream at the invalidate. +def test_resume_after_drop_token(collection): + """Test $changeStream resumeAfter from a drop-event token.""" + + def seed(c): + c.insert_one({"_id": 1}) + c.drop() + + token = _event_token_of_type(collection, "drop", seed) + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"resumeAfter": token}}]), + ) + + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(1)}, + "cursor.firstBatch.0.operationType": Eq("invalidate"), + }, + msg="$changeStream resumeAfter should accept a drop-event token, positioned at invalidate", + raw_res=True, + ) + + +# Property [Start After Drop]: startAfter a drop token positions the stream at the invalidate. +def test_start_after_drop_token(collection): + """Test $changeStream startAfter from a drop-event token.""" + + def seed(c): + c.insert_one({"_id": 1}) + c.drop() + + token = _event_token_of_type(collection, "drop", seed) + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"startAfter": token}}]), + ) + + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(1)}, + "cursor.firstBatch.0.operationType": Eq("invalidate"), + }, + msg="$changeStream startAfter should accept a drop-event token, positioned at invalidate", + raw_res=True, + ) + + +# Property [Resume Regular Under Expanded]: a regular-event token resumes under showExpandedEvents. +def test_resume_regular_token_under_show_expanded_events(collection): + """Test $changeStream resuming a regular-event token under showExpandedEvents true.""" + token = _first_event_token( + collection, lambda c: c.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}]) + ) + + result = execute_command( + collection, + change_stream_command( + collection, + pipeline=[{"$changeStream": {"resumeAfter": token, "showExpandedEvents": True}}], + ), + ) + + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(2)}, + "cursor.firstBatch.0.documentKey._id": Eq(2), + "cursor.firstBatch.1.documentKey._id": Eq(3), + }, + msg="$changeStream should resume a regular-event token under showExpandedEvents true", + raw_res=True, + ) + + +# Property [Resume After Evicted]: resumeAfter an evicted token fails as history lost. +def test_resume_after_evicted_token(collection): + """Test $changeStream resumeAfter from an evicted token.""" + token = _evicted_token(collection) + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"resumeAfter": token}}]), + ) + + assertResult( + result, + error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, + msg="$changeStream resumeAfter should reject an evicted token as history lost", + raw_res=True, + ) + + +# Property [Start After Evicted]: startAfter an evicted token fails as history lost. +def test_start_after_evicted_token(collection): + """Test $changeStream startAfter from an evicted token.""" + token = _evicted_token(collection) + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"startAfter": token}}]), + ) + + assertResult( + result, + error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, + msg="$changeStream startAfter should reject an evicted token as history lost", + raw_res=True, + ) + + +# Property [Resume After Invalidate]: resumeAfter an invalidate token is rejected. +def test_resume_after_invalidate_token(collection): + """Test $changeStream resumeAfter from an invalidate-event token.""" + + def seed(c): + c.insert_one({"_id": 1}) + c.drop() + + token = _event_token_of_type(collection, "invalidate", seed) + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"resumeAfter": token}}]), + ) + + assertResult( + result, + error_code=INVALID_RESUME_TOKEN_ERROR, + msg="$changeStream resumeAfter should reject an invalidate-event token", + raw_res=True, + ) + + +# Property [Resume After Two-Byte Truncated]: resumeAfter a two-byte-truncated token fails. +def test_resume_after_two_byte_truncated_token(collection): + """Test $changeStream resumeAfter from a two-byte-truncated token.""" + token = dict(_first_event_token(collection, lambda c: c.insert_one({"_id": 1}))) + # One byte is tolerated; two bytes leave the KeyString unable to decode. + token["_data"] = token["_data"][:-4] + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"resumeAfter": token}}]), + ) + + assertResult( + result, + error_code=OVERFLOW_ERROR, + msg="$changeStream resumeAfter should reject a two-byte-truncated token as overflow", + raw_res=True, + ) + + +# Property [Start After Two-Byte Truncated]: startAfter a two-byte-truncated token fails. +def test_start_after_two_byte_truncated_token(collection): + """Test $changeStream startAfter from a two-byte-truncated token.""" + token = dict(_first_event_token(collection, lambda c: c.insert_one({"_id": 1}))) + token["_data"] = token["_data"][:-4] + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"startAfter": token}}]), + ) + + assertResult( + result, + error_code=OVERFLOW_ERROR, + msg="$changeStream startAfter should reject a two-byte-truncated token as overflow", + raw_res=True, + ) + + +# Property [Start After Invalidate Advances]: startAfter an invalidate delivers later events. +def test_start_after_resumes_after_invalidate(collection): + """Test $changeStream startAfter past an invalidate event.""" + + def seed(c): + c.insert_one({"_id": 1}) + c.drop() + + token = _event_token_of_type(collection, "invalidate", seed) + + opened = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"startAfter": token}}]), + ) + collection.insert_one({"_id": 2}) + result = execute_command(collection, get_more_command(collection, opened["cursor"]["id"])) + + assertResult( + result, + expected={ + "cursor": {"nextBatch": Len(1)}, + "cursor.nextBatch.0.operationType": Eq("insert"), + "cursor.nextBatch.0.documentKey._id": Eq(2), + }, + msg="$changeStream startAfter should deliver events after an invalidate", + raw_res=True, + ) + + +# Property [Expanded-Only Token Fails Advance]: an expanded-only token fails to advance unexpanded. +def test_expanded_only_token_fails_on_advance(collection): + """Test $changeStream resumeAfter from an expanded-only event token without expansion.""" + + def seed(c): + c.insert_one({"_id": 0}) + execute_command(c, {"createIndexes": c.name, "indexes": [{"key": {"a": 1}, "name": "a_1"}]}) + + token = _event_token_of_type( + collection, "createIndexes", seed, spec={"showExpandedEvents": True} + ) + + opened = execute_command( + collection, + change_stream_command( + collection, + pipeline=[{"$changeStream": {"resumeAfter": token, "showExpandedEvents": False}}], + ), + ) + collection.insert_one({"_id": 1}) + result = execute_command(collection, get_more_command(collection, opened["cursor"]["id"])) + + assertResult( + result, + error_code=CHANGE_STREAM_FATAL_ERROR, + msg="$changeStream should fail to advance past an expanded-only token without expansion", + raw_res=True, + ) + + +# Property [Resume After Foreign Token]: resumeAfter a foreign-collection token fails to advance. +def test_resume_after_foreign_collection_token(collection, database_client): + """Test $changeStream resumeAfter from a foreign-collection token.""" + other = database_client[f"{collection.name}_other"] + database_client.create_collection(other.name) + token = _first_event_token(other, lambda c: c.insert_one({"_id": 1})) + + opened = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"resumeAfter": token}}]), + ) + collection.insert_one({"_id": 99}) + result = execute_command(collection, get_more_command(collection, opened["cursor"]["id"])) + + assertResult( + result, + error_code=CHANGE_STREAM_FATAL_ERROR, + msg="$changeStream resumeAfter should fail to advance past a foreign-collection token", + raw_res=True, + ) + + +# Property [Start After Foreign Token]: startAfter a foreign-collection token fails to advance. +def test_start_after_foreign_collection_token(collection, database_client): + """Test $changeStream startAfter from a foreign-collection token.""" + other = database_client[f"{collection.name}_other"] + database_client.create_collection(other.name) + token = _first_event_token(other, lambda c: c.insert_one({"_id": 1})) + + opened = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"startAfter": token}}]), + ) + collection.insert_one({"_id": 99}) + result = execute_command(collection, get_more_command(collection, opened["cursor"]["id"])) + + assertResult( + result, + error_code=CHANGE_STREAM_FATAL_ERROR, + msg="$changeStream startAfter should fail to advance past a foreign-collection token", + raw_res=True, + ) + + +def _resume_after_mutated(collection, mutate): + """Capture a first-event token, apply ``mutate``, and resume from it. + + The tolerance tests mutate the token in ways that do not change its + interpreted position, so each resume should behave identically to resuming + from the unmutated token. + """ + token = dict( + _first_event_token( + collection, lambda c: c.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}]) + ) + ) + mutated = mutate(token) + return execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {"resumeAfter": mutated}}]), + ) + + +# Property [Tolerance Lowercase Hex]: a lower-cased _data hex resumes identically. +def test_tolerance_lowercase_hex(collection): + """Test $changeStream resumeAfter from a lower-cased _data token.""" + result = _resume_after_mutated(collection, lambda t: {**t, "_data": t["_data"].lower()}) + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(2)}, + "cursor.firstBatch.0.documentKey._id": Eq(2), + "cursor.firstBatch.1.documentKey._id": Eq(3), + }, + msg="$changeStream resumeAfter should resume identically when the token is lower-cased hex", + raw_res=True, + ) + + +# Property [Tolerance Truncate One Byte]: a one-byte-truncated _data resumes identically. +def test_tolerance_truncate_one_byte(collection): + """Test $changeStream resumeAfter from a one-byte-truncated token.""" + result = _resume_after_mutated(collection, lambda t: {**t, "_data": t["_data"][:-2]}) + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(2)}, + "cursor.firstBatch.0.documentKey._id": Eq(2), + "cursor.firstBatch.1.documentKey._id": Eq(3), + }, + msg="$changeStream resumeAfter should resume identically when the token is truncated by" + " one byte", + raw_res=True, + ) + + +# Property [Tolerance Append FF]: a _data extended with an FF byte resumes identically. +def test_tolerance_append_ff_byte(collection): + """Test $changeStream resumeAfter from a token extended with an FF byte.""" + result = _resume_after_mutated(collection, lambda t: {**t, "_data": t["_data"] + "FF"}) + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(2)}, + "cursor.firstBatch.0.documentKey._id": Eq(2), + "cursor.firstBatch.1.documentKey._id": Eq(3), + }, + msg="$changeStream resumeAfter should resume identically when the token is extended with an" + " FF byte", + raw_res=True, + ) + + +# Property [Tolerance Append 00]: a _data extended with a 00 byte resumes identically. +def test_tolerance_append_00_byte(collection): + """Test $changeStream resumeAfter from a token extended with a 00 byte.""" + result = _resume_after_mutated(collection, lambda t: {**t, "_data": t["_data"] + "00"}) + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(2)}, + "cursor.firstBatch.0.documentKey._id": Eq(2), + "cursor.firstBatch.1.documentKey._id": Eq(3), + }, + msg="$changeStream resumeAfter should resume identically when the token is extended with a" + " 00 byte", + raw_res=True, + ) + + +# Property [Tolerance Extra Fields]: a token with unknown extra fields resumes identically. +def test_tolerance_extra_fields(collection): + """Test $changeStream resumeAfter from a token with unknown extra fields.""" + result = _resume_after_mutated(collection, lambda t: {**t, "unknownField": 1, "another": "x"}) + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(2)}, + "cursor.firstBatch.0.documentKey._id": Eq(2), + "cursor.firstBatch.1.documentKey._id": Eq(3), + }, + msg="$changeStream resumeAfter should resume identically when the token is augmented with" + " unknown fields", + raw_res=True, + ) + + +# Property [Tolerance Empty TypeBits]: a token with an empty _typeBits binary resumes identically. +def test_tolerance_empty_typebits(collection): + """Test $changeStream resumeAfter from a token with an empty _typeBits binary.""" + result = _resume_after_mutated(collection, lambda t: {**t, "_typeBits": Binary(b"\x00")}) + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(2)}, + "cursor.firstBatch.0.documentKey._id": Eq(2), + "cursor.firstBatch.1.documentKey._id": Eq(3), + }, + msg="$changeStream resumeAfter should resume identically when the token is given an empty" + " _typeBits binary", + raw_res=True, + ) + + +# Property [High-Water-Mark Resume]: a high-water-mark token resumes, delivering later events. +@pytest.mark.parametrize("option", ["resumeAfter", "startAfter"]) +def test_resume_from_high_water_mark(collection, option): + """Test $changeStream resuming from a high-water-mark token.""" + mark = _high_water_mark(collection) + collection.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}]) + + result = execute_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": {option: mark}}]), + ) + + assertResult( + result, + expected={ + "cursor": {"firstBatch": Len(3)}, + "cursor.firstBatch.0.documentKey._id": Eq(1), + "cursor.firstBatch.1.documentKey._id": Eq(2), + "cursor.firstBatch.2.documentKey._id": Eq(3), + }, + msg=f"$changeStream {option!r} should resume from a high-water-mark token", + raw_res=True, + ) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume_mutual_exclusivity.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume_mutual_exclusivity.py new file mode 100644 index 000000000..69ab8518a --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume_mutual_exclusivity.py @@ -0,0 +1,66 @@ +"""Tests for $changeStream mutual exclusivity of resume options.""" + +from __future__ import annotations + +import pytest +from utils.changeStream_common import OPERATION_TIME, RESUME_TOKEN, change_stream_command + +from documentdb_tests.compatibility.tests.core.operator.stages.utils.stage_test_case import ( + StageTestCase, +) +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.error_codes import ( + MULTIPLE_RESUME_OPTIONS_ERROR, + RESUME_AFTER_START_AFTER_CONFLICT_ERROR, +) +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params + +# Property [Resume-Option Mutual Exclusivity]: specifying any pair of resume +# options in a single spec is rejected at open; the conflict is a parse-time +# check, so the same token may fill both token options. +CHANGESTREAM_RESUME_MUTUAL_EXCLUSIVITY_TESTS: list[StageTestCase] = [ + StageTestCase( + "resume_after_and_start_after", + pipeline=[{"$changeStream": {"resumeAfter": RESUME_TOKEN, "startAfter": RESUME_TOKEN}}], + error_code=RESUME_AFTER_START_AFTER_CONFLICT_ERROR, + msg="$changeStream should reject both resumeAfter and startAfter together", + ), + StageTestCase( + "resume_after_and_operation_time", + pipeline=[ + {"$changeStream": {"resumeAfter": RESUME_TOKEN, "startAtOperationTime": OPERATION_TIME}} + ], + error_code=MULTIPLE_RESUME_OPTIONS_ERROR, + msg="$changeStream should reject both resumeAfter and startAtOperationTime together", + ), + StageTestCase( + "start_after_and_operation_time", + pipeline=[ + {"$changeStream": {"startAfter": RESUME_TOKEN, "startAtOperationTime": OPERATION_TIME}} + ], + error_code=MULTIPLE_RESUME_OPTIONS_ERROR, + msg="$changeStream should reject both startAfter and startAtOperationTime together", + ), +] + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_RESUME_MUTUAL_EXCLUSIVITY_TESTS)) +def test_changeStream_resume_option_mutual_exclusivity(collection, test_case): + """Test $changeStream rejects specifying more than one resume option.""" + opened = execute_command( + collection, change_stream_command(collection, pipeline=[{"$changeStream": {}}]) + ) + token = opened["cursor"]["postBatchResumeToken"] + operation_time = opened["operationTime"] + substitutions = {id(RESUME_TOKEN): token, id(OPERATION_TIME): operation_time} + spec = { + key: substitutions.get(id(value), value) + for key, value in test_case.pipeline[0]["$changeStream"].items() + } + result = execute_command( + collection, change_stream_command(collection, pipeline=[{"$changeStream": spec}]) + ) + assertResult(result, error_code=test_case.error_code, msg=test_case.msg) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_spec_acceptance.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_spec_acceptance.py new file mode 100644 index 000000000..5fc190635 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_spec_acceptance.py @@ -0,0 +1,227 @@ +"""Tests for $changeStream spec and option acceptance for valid specs.""" + +from __future__ import annotations + +import pytest +from utils.changeStream_common import change_stream_command + +from documentdb_tests.compatibility.tests.core.operator.stages.utils.stage_test_case import ( + StageTestCase, +) +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Eq + +# Property [Empty Spec Acceptance]: an empty spec document opens a change stream +# because every option field is optional. +CHANGESTREAM_EMPTY_SPEC_TESTS: list[StageTestCase] = [ + StageTestCase( + "empty_spec", + pipeline=[{"$changeStream": {}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should open with an empty spec document", + ), +] + +# Property [Boolean Option Acceptance]: showExpandedEvents accepts both boolean +# values, and allChangesForCluster accepts false on a collection-scoped stream. +# allChangesForCluster true is not accepted here because it requires a +# cluster-scoped stream on the admin database; that accept path is covered by +# the namespace-scope tests. +CHANGESTREAM_BOOLEAN_OPTION_TESTS: list[StageTestCase] = [ + StageTestCase( + "all_changes_for_cluster_false", + pipeline=[{"$changeStream": {"allChangesForCluster": False}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept allChangesForCluster false", + ), + StageTestCase( + "show_expanded_events_true", + pipeline=[{"$changeStream": {"showExpandedEvents": True}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept showExpandedEvents true", + ), + StageTestCase( + "show_expanded_events_false", + pipeline=[{"$changeStream": {"showExpandedEvents": False}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept showExpandedEvents false", + ), +] + +# Property [fullDocument Enum Acceptance]: fullDocument accepts each of its +# documented enum strings. +CHANGESTREAM_FULL_DOCUMENT_TESTS: list[StageTestCase] = [ + StageTestCase( + f"full_document_{value}", + pipeline=[{"$changeStream": {"fullDocument": value}}], + expected={"ok": Eq(1.0)}, + msg=f"$changeStream should accept fullDocument '{value}'", + ) + for value in ["default", "required", "updateLookup", "whenAvailable"] +] + +# Property [fullDocumentBeforeChange Enum Acceptance]: fullDocumentBeforeChange +# accepts each of its documented enum strings. +CHANGESTREAM_FULL_DOCUMENT_BEFORE_CHANGE_TESTS: list[StageTestCase] = [ + StageTestCase( + f"full_document_before_change_{value}", + pipeline=[{"$changeStream": {"fullDocumentBeforeChange": value}}], + expected={"ok": Eq(1.0)}, + msg=f"$changeStream should accept fullDocumentBeforeChange '{value}'", + ) + for value in ["off", "whenAvailable", "required"] +] + +# Property [Option Combination]: independent option fields combine in one spec +# without a parse-time conflict, and the two enum options are parsed +# independently of each other. +CHANGESTREAM_OPTION_COMBINATION_TESTS: list[StageTestCase] = [ + StageTestCase( + "combo_all_options", + pipeline=[ + { + "$changeStream": { + "allChangesForCluster": False, + "showExpandedEvents": True, + "fullDocument": "updateLookup", + "fullDocumentBeforeChange": "whenAvailable", + } + } + ], + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept all independent options combined in one spec", + ), +] + [ + StageTestCase( + f"combo_{full_document}_{before_change}", + pipeline=[ + { + "$changeStream": { + "fullDocument": full_document, + "fullDocumentBeforeChange": before_change, + } + } + ], + expected={"ok": Eq(1.0)}, + msg=( + f"$changeStream should accept fullDocument '{full_document}' with " + f"fullDocumentBeforeChange '{before_change}'" + ), + ) + for full_document, before_change in [ + ("default", "off"), + ("required", "required"), + ("whenAvailable", "whenAvailable"), + ] +] + +# Property [Null Option Acceptance]: an explicit null value for any of the five +# non-boolean-typed options (fullDocument, fullDocumentBeforeChange, +# resumeAfter, startAfter, startAtOperationTime) is accepted and opens a stream +# as if the field were unset, and a null resume option does not trigger +# resume-option mutual exclusivity. This group exercises only the null value; +# rejection of other BSON types per option is covered in +# test_changeStream_validation_errors.py. +CHANGESTREAM_NULL_OPTION_TESTS: list[StageTestCase] = [ + StageTestCase( + "null_full_document", + pipeline=[{"$changeStream": {"fullDocument": None}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept a null fullDocument", + ), + StageTestCase( + "null_full_document_before_change", + pipeline=[{"$changeStream": {"fullDocumentBeforeChange": None}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept a null fullDocumentBeforeChange", + ), + StageTestCase( + "null_resume_after", + pipeline=[{"$changeStream": {"resumeAfter": None}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept a null resumeAfter", + ), + StageTestCase( + "null_start_after", + pipeline=[{"$changeStream": {"startAfter": None}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept a null startAfter", + ), + StageTestCase( + "null_start_at_operation_time", + pipeline=[{"$changeStream": {"startAtOperationTime": None}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept a null startAtOperationTime", + ), + StageTestCase( + "null_all_resume_options", + pipeline=[ + { + "$changeStream": { + "resumeAfter": None, + "startAfter": None, + "startAtOperationTime": None, + } + } + ], + expected={"ok": Eq(1.0)}, + msg="$changeStream should not treat null resume options as a mutual exclusivity conflict", + ), +] + +CHANGESTREAM_SUCCESS_TESTS = ( + CHANGESTREAM_EMPTY_SPEC_TESTS + + CHANGESTREAM_BOOLEAN_OPTION_TESTS + + CHANGESTREAM_FULL_DOCUMENT_TESTS + + CHANGESTREAM_FULL_DOCUMENT_BEFORE_CHANGE_TESTS + + CHANGESTREAM_OPTION_COMBINATION_TESTS + + CHANGESTREAM_NULL_OPTION_TESTS +) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_SUCCESS_TESTS)) +def test_changeStream_cases(collection, test_case: StageTestCase): + """Test $changeStream spec document and option acceptance.""" + result = execute_command( + collection, + {"aggregate": collection.name, "pipeline": test_case.pipeline, "cursor": {}}, + ) + assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) + + +# Property [Null Resume-Option Non-Conflict]: a resume option set to an explicit +# null does not count as a second resume point, so pairing it with another +# present resume token opens successfully rather than tripping resume-option +# mutual exclusivity. This holds for every (present token, null option) pairing. +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize( + "present_field,null_field", + [ + ("resumeAfter", "startAtOperationTime"), + ("startAfter", "startAtOperationTime"), + ("resumeAfter", "startAfter"), + ("startAfter", "resumeAfter"), + ], +) +def test_changeStream_null_resume_option_non_conflict(collection, present_field, null_field): + """Test $changeStream treats a null resume option as absent, not a second resume point.""" + opened = execute_command( + collection, change_stream_command(collection, pipeline=[{"$changeStream": {}}]) + ) + token = opened["cursor"]["postBatchResumeToken"] + + spec = {present_field: token, null_field: None} + result = execute_command( + collection, change_stream_command(collection, pipeline=[{"$changeStream": spec}]) + ) + assertResult( + result, + expected={"ok": Eq(1.0)}, + msg=f"$changeStream should accept {present_field!r} token with null {null_field!r}", + raw_res=True, + ) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_stable_api.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_stable_api.py new file mode 100644 index 000000000..a75788015 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_stable_api.py @@ -0,0 +1,88 @@ +"""Tests for $changeStream behavior under Stable API v1 (apiStrict).""" + +from __future__ import annotations + +import pytest +from utils.changeStream_common import OPERATION_TIME, change_stream_command + +from documentdb_tests.compatibility.tests.core.operator.stages.utils.stage_test_case import ( + StageTestCase, +) +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.error_codes import API_STRICT_ERROR +from documentdb_tests.framework.executor import execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Eq + +# Property [Stable API V1 Acceptance]: under apiStrict true (Stable API V1), a +# spec that omits showExpandedEvents opens the stream; specifying +# showExpandedEvents under apiStrict true is rejected because the parameter is +# not part of API Version 1, regardless of its boolean value. +CHANGESTREAM_STABLE_API_V1_TESTS: list[StageTestCase] = [ + StageTestCase( + "stable_api_empty_spec", + pipeline=[{"$changeStream": {}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should open with an empty spec under apiStrict true", + ), + StageTestCase( + "stable_api_full_document", + pipeline=[{"$changeStream": {"fullDocument": "updateLookup"}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should open with fullDocument under apiStrict true", + ), + StageTestCase( + "stable_api_full_document_before_change", + pipeline=[{"$changeStream": {"fullDocumentBeforeChange": "whenAvailable"}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should open with fullDocumentBeforeChange under apiStrict true", + ), + StageTestCase( + "stable_api_start_at_operation_time", + pipeline=[{"$changeStream": {"startAtOperationTime": OPERATION_TIME}}], + expected={"ok": Eq(1.0)}, + msg="$changeStream should open with startAtOperationTime under apiStrict true", + ), + StageTestCase( + "stable_api_show_expanded_events_true", + pipeline=[{"$changeStream": {"showExpandedEvents": True}}], + error_code=API_STRICT_ERROR, + msg="$changeStream should reject showExpandedEvents true under apiStrict true", + ), + StageTestCase( + "stable_api_show_expanded_events_false", + pipeline=[{"$changeStream": {"showExpandedEvents": False}}], + error_code=API_STRICT_ERROR, + msg="$changeStream should reject showExpandedEvents false under apiStrict true", + ), +] + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_STABLE_API_V1_TESTS)) +def test_changeStream_stable_api_v1(collection, test_case): + """Test $changeStream Stable API V1 (apiStrict true) acceptance and rejection.""" + stage = dict(test_case.pipeline[0]["$changeStream"]) + if stage.get("startAtOperationTime") is OPERATION_TIME: + base = execute_command( + collection, change_stream_command(collection, pipeline=[{"$changeStream": {}}]) + ) + stage["startAtOperationTime"] = base["operationTime"] + result = execute_command( + collection, + { + "aggregate": collection.name, + "pipeline": [{"$changeStream": stage}], + "cursor": {}, + "apiVersion": "1", + "apiStrict": True, + }, + ) + assertResult( + result, + expected=test_case.expected, + error_code=test_case.error_code, + msg=test_case.msg, + raw_res=True, + ) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_timestamp_boundary.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_timestamp_boundary.py new file mode 100644 index 000000000..18182fbf7 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_timestamp_boundary.py @@ -0,0 +1,117 @@ +"""Tests for $changeStream startAtOperationTime boundary across scopes.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import pytest +from bson import Timestamp +from utils.changeStream_common import change_stream_command + +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.executor import execute_admin_command, execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Eq +from documentdb_tests.framework.test_case import BaseTestCase + + +@dataclass(frozen=True) +class ChangeStreamTimestampTestCase(BaseTestCase): + """Test case for opening a $changeStream at a startAtOperationTime boundary. + + Attributes: + compute_start: Receives the current operationTime captured at run time + and returns the timestamp to pass as startAtOperationTime, so the + boundary is expressed relative to the live oplog rather than as a + static value + """ + + compute_start: Any = None + + +# Property [Timestamp Start Boundary]: a startAtOperationTime at the current +# operationTime, at a zero-increment timestamp for the current second, and at a +# far-future timestamp each opens the stream, identically across collection-, +# database-, and cluster-scoped streams. +CHANGESTREAM_TIMESTAMP_BOUNDARY_TESTS: list[ChangeStreamTimestampTestCase] = [ + ChangeStreamTimestampTestCase( + "current", + compute_start=lambda operation_time: operation_time, + expected={"ok": Eq(1.0)}, + msg="$changeStream should open at the current operationTime", + ), + ChangeStreamTimestampTestCase( + "zero_increment", + compute_start=lambda operation_time: Timestamp(operation_time.time, 0), + expected={"ok": Eq(1.0)}, + msg="$changeStream should open at a current-second zero-increment timestamp", + ), + ChangeStreamTimestampTestCase( + "future", + compute_start=lambda operation_time: Timestamp(operation_time.time + 1_000_000, 0), + expected={"ok": Eq(1.0)}, + msg="$changeStream should open at a far-future timestamp", + ), +] + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_TIMESTAMP_BOUNDARY_TESTS)) +def test_changeStream_timestamp_boundary_collection_scope( + collection, test_case: ChangeStreamTimestampTestCase +): + """Test $changeStream opens a collection-scoped stream at startAtOperationTime boundaries.""" + base = execute_command( + collection, change_stream_command(collection, pipeline=[{"$changeStream": {}}]) + ) + start = test_case.compute_start(base["operationTime"]) + result = execute_command( + collection, + change_stream_command( + collection, pipeline=[{"$changeStream": {"startAtOperationTime": start}}] + ), + ) + assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_TIMESTAMP_BOUNDARY_TESTS)) +def test_changeStream_timestamp_boundary_database_scope( + collection, test_case: ChangeStreamTimestampTestCase +): + """Test $changeStream opens a database-scoped stream at startAtOperationTime boundaries.""" + base = execute_command( + collection, change_stream_command(collection, pipeline=[{"$changeStream": {}}]) + ) + start = test_case.compute_start(base["operationTime"]) + result = execute_command( + collection, + change_stream_command( + collection, + pipeline=[{"$changeStream": {"startAtOperationTime": start}}], + aggregate=1, + ), + ) + assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_TIMESTAMP_BOUNDARY_TESTS)) +def test_changeStream_timestamp_boundary_cluster_scope( + collection, test_case: ChangeStreamTimestampTestCase +): + """Test $changeStream opens a cluster-wide stream at startAtOperationTime boundaries.""" + base = execute_command( + collection, change_stream_command(collection, pipeline=[{"$changeStream": {}}]) + ) + start = test_case.compute_start(base["operationTime"]) + spec = {"startAtOperationTime": start, "allChangesForCluster": True} + result = execute_admin_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": spec}], aggregate=1), + ) + assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_validation_errors.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_validation_errors.py new file mode 100644 index 000000000..be5644138 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_validation_errors.py @@ -0,0 +1,530 @@ +"""Tests for $changeStream spec validation errors (malformed specs and option values).""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest +from bson import Binary, Code, DBRef, Int64, MaxKey, MinKey, ObjectId, Regex, Timestamp + +from documentdb_tests.compatibility.tests.core.operator.stages.utils.stage_test_case import ( + StageTestCase, +) +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.error_codes import ( + BAD_VALUE_ERROR, + CHANGE_STREAM_SPEC_NOT_OBJECT_ERROR, + FAILED_TO_PARSE_ERROR, + KEYSTRING_UNKNOWN_TYPE_ERROR, + RESUME_TOKEN_EMPTY_ERROR, + RESUME_TOKEN_MALFORMED_ERROR, + RESUME_TOKEN_TYPEBITS_WRONG_TYPE_ERROR, + TYPE_MISMATCH_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.test_constants import ( + DECIMAL128_ONE_AND_HALF, + DOUBLE_ZERO, + FLOAT_INFINITY, + FLOAT_NAN, + FLOAT_NEGATIVE_INFINITY, +) + +# Property [Spec-Level Non-Object Rejection]: a spec value of any non-object +# BSON type, including null and an array, produces an error; an array spec is +# not unwrapped into its elements. +CHANGESTREAM_SPEC_NON_OBJECT_TESTS: list[StageTestCase] = [ + StageTestCase( + f"spec_non_object_{tid}", + pipeline=[{"$changeStream": val}], + error_code=CHANGE_STREAM_SPEC_NOT_OBJECT_ERROR, + msg=f"$changeStream should reject a {tid} spec as not a nested object", + ) + for tid, val in [ + ("int32", 1), + ("int64", Int64(1)), + ("double", 1.5), + ("decimal128", DECIMAL128_ONE_AND_HALF), + ("bool", True), + ("string", "x"), + ("objectid", ObjectId("507f1f77bcf86cd799439011")), + ("datetime", datetime(2024, 1, 1, tzinfo=timezone.utc)), + ("timestamp", Timestamp(1, 1)), + ("binary", Binary(b"\x01\x02\x03")), + ("regex", Regex(".*", "i")), + ("code", Code("function(){}")), + ("minkey", MinKey()), + ("maxkey", MaxKey()), + ("null", None), + ("empty_array", []), + ("array_of_spec", [{"fullDocument": "default"}]), + ] +] + +# Property [Spec-Level Unknown Field Rejection]: an option field name that is +# not a recognized option produces an unknown-field error; option names are +# case-sensitive, are not trimmed, and a DBRef spec is treated as a nested +# object whose reserved keys are unrecognized fields. +CHANGESTREAM_SPEC_UNKNOWN_FIELD_TESTS: list[StageTestCase] = [ + StageTestCase( + "unknown_field", + pipeline=[{"$changeStream": {"bogus": 1}}], + error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="$changeStream should reject an unknown option field", + ), + StageTestCase( + "wrong_case_capitalized", + pipeline=[{"$changeStream": {"FullDocument": "default"}}], + error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="$changeStream should reject a capitalized option name as unknown", + ), + StageTestCase( + "wrong_case_upper", + pipeline=[{"$changeStream": {"FULLDOCUMENT": "default"}}], + error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="$changeStream should reject an upper-case option name as unknown", + ), + StageTestCase( + "wrong_case_lower", + pipeline=[{"$changeStream": {"fulldocument": "default"}}], + error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="$changeStream should reject a lower-case option name as unknown", + ), + StageTestCase( + "leading_whitespace", + pipeline=[{"$changeStream": {" fullDocument": "default"}}], + error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="$changeStream should reject an option name with leading whitespace as unknown", + ), + StageTestCase( + "dollar_prefixed_name", + pipeline=[{"$changeStream": {"$fullDocument": "default"}}], + error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="$changeStream should reject a dollar-prefixed option name as unknown", + ), + StageTestCase( + "empty_name", + pipeline=[{"$changeStream": {"": "x"}}], + error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="$changeStream should reject an empty option name as unknown", + ), + StageTestCase( + "dbref_object", + pipeline=[{"$changeStream": DBRef("c", 1)}], + error_code=UNRECOGNIZED_COMMAND_FIELD_ERROR, + msg="$changeStream should treat a DBRef spec as an object with unknown fields", + ), +] + +# Property [Expression Arguments Are Not Evaluated]: no option value is +# evaluated as an aggregation expression; an expression-shaped value is read as +# literal BSON at parse time and falls through to the type/enum/token validation +# for its literal shape rather than producing its evaluated result. +CHANGESTREAM_EXPRESSION_NOT_EVALUATED_TESTS: list[StageTestCase] = [ + StageTestCase( + "all_changes_for_cluster_literal_true", + pipeline=[{"$changeStream": {"allChangesForCluster": {"$literal": True}}}], + error_code=TYPE_MISMATCH_ERROR, + msg=( + "$changeStream should reject an expression object for" + " allChangesForCluster as wrong type" + ), + ), + StageTestCase( + "show_expanded_events_cond", + pipeline=[{"$changeStream": {"showExpandedEvents": {"$cond": [True, True, False]}}}], + error_code=TYPE_MISMATCH_ERROR, + msg="$changeStream should reject an expression object for showExpandedEvents as wrong type", + ), + StageTestCase( + "full_document_literal", + pipeline=[{"$changeStream": {"fullDocument": {"$literal": "updateLookup"}}}], + error_code=TYPE_MISMATCH_ERROR, + msg="$changeStream should reject an expression object for fullDocument as wrong type", + ), + StageTestCase( + "full_document_before_change_concat", + pipeline=[ + {"$changeStream": {"fullDocumentBeforeChange": {"$concat": ["when", "Available"]}}} + ], + error_code=TYPE_MISMATCH_ERROR, + msg=( + "$changeStream should reject an expression object for" + " fullDocumentBeforeChange as wrong type" + ), + ), + StageTestCase( + "start_at_operation_time_literal", + pipeline=[{"$changeStream": {"startAtOperationTime": {"$literal": Timestamp(1, 1)}}}], + error_code=TYPE_MISMATCH_ERROR, + msg=( + "$changeStream should reject an expression object for" + " startAtOperationTime as wrong type" + ), + ), + StageTestCase( + "resume_after_literal", + pipeline=[{"$changeStream": {"resumeAfter": {"$literal": {"_data": "8264000000"}}}}], + error_code=RESUME_TOKEN_MALFORMED_ERROR, + msg="$changeStream should reject an expression object for resumeAfter as a malformed token", + ), + StageTestCase( + "start_after_concat", + pipeline=[{"$changeStream": {"startAfter": {"$concat": ["8264", "000000"]}}}], + error_code=RESUME_TOKEN_MALFORMED_ERROR, + msg="$changeStream should reject an expression object for startAfter as a malformed token", + ), + StageTestCase( + "full_document_field_path", + pipeline=[{"$changeStream": {"fullDocument": "$someField"}}], + error_code=BAD_VALUE_ERROR, + msg=( + "$changeStream should reject a field-path string for" + " fullDocument as an invalid enum value" + ), + ), + StageTestCase( + "full_document_variable", + pipeline=[{"$changeStream": {"fullDocument": "$$ROOT"}}], + error_code=BAD_VALUE_ERROR, + msg=( + "$changeStream should reject a variable string for" + " fullDocument as an invalid enum value" + ), + ), + StageTestCase( + "full_document_before_change_field_path", + pipeline=[{"$changeStream": {"fullDocumentBeforeChange": "$someField"}}], + error_code=BAD_VALUE_ERROR, + msg=( + "$changeStream should reject a field-path string for" + " fullDocumentBeforeChange as an invalid enum value" + ), + ), +] + +# Property [Boolean Option Type Rejection]: each boolean option +# (allChangesForCluster, showExpandedEvents) rejects any non-boolean BSON type +# with a TypeMismatch error; there is no coercion from numbers, numeric +# strings, arrays, objects, or any other type to a boolean. +CHANGESTREAM_BOOLEAN_TYPE_ERROR_TESTS: list[StageTestCase] = [ + StageTestCase( + f"bool_type_{opt_id}_{tid}", + pipeline=[{"$changeStream": {opt: val}}], + error_code=TYPE_MISMATCH_ERROR, + msg=f"$changeStream should reject a non-boolean value for {opt}", + ) + for opt, opt_id in [ + ("allChangesForCluster", "all_changes"), + ("showExpandedEvents", "show_expanded"), + ] + for tid, val in [ + ("int32_zero", 0), + ("int32_one", 1), + ("double_zero", DOUBLE_ZERO), + ("double_one", 1.0), + ("int64", Int64(1)), + ("decimal128", DECIMAL128_ONE_AND_HALF), + ("nan", FLOAT_NAN), + ("positive_infinity", FLOAT_INFINITY), + ("negative_infinity", FLOAT_NEGATIVE_INFINITY), + ("string_true", "true"), + ("string_false", "false"), + ("objectid", ObjectId("507f1f77bcf86cd799439011")), + ("datetime", datetime(2024, 1, 1, tzinfo=timezone.utc)), + ("timestamp", Timestamp(1, 1)), + ("binary", Binary(b"\x01\x02\x03")), + ("regex", Regex(".*", "i")), + ("code", Code("function(){}")), + # DBRef encodes as a BSON object and is rejected as a non-boolean object. + ("dbref", DBRef("c", 1)), + ("minkey", MinKey()), + ("maxkey", MaxKey()), + ("empty_array", []), + ("array_true", [True]), + ("array_false", [False]), + ("empty_object", {}), + ] +] + +# Property [Boolean Option Null Rejection]: each boolean option rejects an +# explicit null with a TypeMismatch error, so only field omission yields the +# default; this contrasts with the five non-boolean options that accept null as +# unset. +CHANGESTREAM_BOOLEAN_NULL_ERROR_TESTS: list[StageTestCase] = [ + StageTestCase( + f"bool_null_{opt_id}", + pipeline=[{"$changeStream": {opt: None}}], + error_code=TYPE_MISMATCH_ERROR, + msg=f"$changeStream should reject an explicit null for {opt}", + ) + for opt, opt_id in [ + ("allChangesForCluster", "all_changes"), + ("showExpandedEvents", "show_expanded"), + ] +] + +# Property [String Enum Option Type Rejection]: each string-enum option +# (fullDocument, fullDocumentBeforeChange) rejects any non-string, non-null BSON +# type with a TypeMismatch error; an array is not unwrapped to its element. +CHANGESTREAM_ENUM_TYPE_ERROR_TESTS: list[StageTestCase] = [ + StageTestCase( + f"enum_type_{opt_id}_{tid}", + pipeline=[{"$changeStream": {opt: val}}], + error_code=TYPE_MISMATCH_ERROR, + msg=f"$changeStream should reject a non-string value for {opt}", + ) + for opt, opt_id in [ + ("fullDocument", "full_document"), + ("fullDocumentBeforeChange", "before_change"), + ] + for tid, val in [ + ("int32", 1), + ("int64", Int64(1)), + ("double", 1.5), + ("decimal128", DECIMAL128_ONE_AND_HALF), + ("bool", True), + ("objectid", ObjectId("507f1f77bcf86cd799439011")), + ("datetime", datetime(2024, 1, 1, tzinfo=timezone.utc)), + ("timestamp", Timestamp(1, 1)), + ("binary", Binary(b"\x01\x02\x03")), + ("regex", Regex(".*", "i")), + ("code", Code("function(){}")), + # DBRef encodes as a BSON object and is rejected as a non-string object. + ("dbref", DBRef("c", 1)), + ("minkey", MinKey()), + ("maxkey", MaxKey()), + ("empty_object", {}), + ("object", {"a": 1}), + ("empty_array", []), + ("array_single", ["default"]), + ("array_multi", ["default", "off"]), + ("array_nested", [["default"]]), + ] +] + +# Property [String Enum Option Value Rejection]: a string that is not a member +# of the option's enum set produces a BadValue error; comparison is exact, with +# no case folding, whitespace trimming, NUL truncation, Unicode normalization, +# or stripping of invisible or marker characters, and a sibling option's value +# or a dollar-prefixed string is treated as a plain invalid enum string. +CHANGESTREAM_ENUM_VALUE_ERROR_TESTS: list[StageTestCase] = [ + StageTestCase( + f"enum_value_{opt_id}_{suffix}", + pipeline=[{"$changeStream": {opt: val}}], + error_code=BAD_VALUE_ERROR, + msg=f"$changeStream should reject an invalid {opt} enum string", + ) + for opt, opt_id, suffix, val in [ + ("fullDocument", "full_document", "empty", ""), + ("fullDocument", "full_document", "sibling_off", "off"), + ("fullDocument", "full_document", "case_capitalized", "Default"), + ("fullDocument", "full_document", "case_upper", "DEFAULT"), + ("fullDocument", "full_document", "case_update_lookup", "UpdateLookup"), + ("fullDocument", "full_document", "leading_space", " default"), + ("fullDocument", "full_document", "trailing_space", "default "), + ("fullDocument", "full_document", "trailing_tab", "default\t"), + ("fullDocument", "full_document", "trailing_cr", "default\r"), + ("fullDocument", "full_document", "trailing_lf", "updateLookup\n"), + ("fullDocument", "full_document", "interior_space", "when available"), + ("fullDocument", "full_document", "trailing_nul", "default\x00"), + ("fullDocument", "full_document", "leading_nul", "\x00default"), + ("fullDocument", "full_document", "interior_nul", "default\x00default"), + # Fullwidth latin small d (U+FF44) in place of the ASCII d. + ("fullDocument", "full_document", "fullwidth_d", "\uff44efault"), + # Fullwidth latin small d (U+FF44) in place of the trailing ASCII d. + ("fullDocument", "full_document", "fullwidth_require_d", "require\uff44"), + # No-break space (U+00A0) prefix. + ("fullDocument", "full_document", "nbsp", "\u00a0default"), + # En space (U+2000) prefix. + ("fullDocument", "full_document", "en_space", "\u2000default"), + # Em space (U+2003) suffix. + ("fullDocument", "full_document", "em_space", "default\u2003"), + # Byte order mark (U+FEFF) prefix. + ("fullDocument", "full_document", "bom", "\ufeffdefault"), + # Zero-width space (U+200B) prefix. + ("fullDocument", "full_document", "zwsp", "\u200bdefault"), + # Zero-width joiner (U+200D) prefix. + ("fullDocument", "full_document", "zwj", "\u200ddefault"), + # Left-to-right mark (U+200E) prefix. + ("fullDocument", "full_document", "ltr_mark", "\u200edefault"), + # Right-to-left mark (U+200F) prefix. + ("fullDocument", "full_document", "rtl_mark", "\u200fdefault"), + ("fullDocument", "full_document", "dollar", "$"), + ("fullDocument", "full_document", "double_dollar", "$$"), + ("fullDocument", "full_document", "dollar_value", "$default"), + ("fullDocument", "full_document", "double_dollar_value", "$$default"), + ("fullDocumentBeforeChange", "before_change", "empty", ""), + ("fullDocumentBeforeChange", "before_change", "sibling_default", "default"), + ("fullDocumentBeforeChange", "before_change", "sibling_update_lookup", "updateLookup"), + ("fullDocumentBeforeChange", "before_change", "case_capitalized", "Off"), + ("fullDocumentBeforeChange", "before_change", "case_upper", "OFF"), + ("fullDocumentBeforeChange", "before_change", "case_when_available", "WhenAvailable"), + ("fullDocumentBeforeChange", "before_change", "leading_mixed_ws", " \t\n\r off"), + ("fullDocumentBeforeChange", "before_change", "trailing_space", "off "), + ("fullDocumentBeforeChange", "before_change", "trailing_tab", "off\t"), + ("fullDocumentBeforeChange", "before_change", "trailing_cr", "off\r"), + ("fullDocumentBeforeChange", "before_change", "trailing_lf", "whenAvailable\n"), + ("fullDocumentBeforeChange", "before_change", "interior_space", "when available"), + ("fullDocumentBeforeChange", "before_change", "trailing_nul", "off\x00"), + ("fullDocumentBeforeChange", "before_change", "leading_nul", "\x00off"), + ("fullDocumentBeforeChange", "before_change", "interior_nul", "off\x00off"), + # Fullwidth latin small o (U+FF4F) in place of the leading ASCII o. + ("fullDocumentBeforeChange", "before_change", "fullwidth_o", "\uff4fff"), + # No-break space (U+00A0) prefix. + ("fullDocumentBeforeChange", "before_change", "nbsp", "\u00a0off"), + # En space (U+2000) prefix. + ("fullDocumentBeforeChange", "before_change", "en_space", "\u2000off"), + # Em space (U+2003) suffix. + ("fullDocumentBeforeChange", "before_change", "em_space", "off\u2003"), + # Byte order mark (U+FEFF) prefix. + ("fullDocumentBeforeChange", "before_change", "bom", "\ufeffoff"), + # Zero-width space (U+200B) prefix. + ("fullDocumentBeforeChange", "before_change", "zwsp", "\u200boff"), + # Zero-width joiner (U+200D) prefix. + ("fullDocumentBeforeChange", "before_change", "zwj", "\u200doff"), + # Left-to-right mark (U+200E) prefix. + ("fullDocumentBeforeChange", "before_change", "ltr_mark", "\u200eoff"), + # Right-to-left mark (U+200F) prefix. + ("fullDocumentBeforeChange", "before_change", "rtl_mark", "\u200foff"), + ("fullDocumentBeforeChange", "before_change", "dollar", "$"), + ("fullDocumentBeforeChange", "before_change", "dollar_value", "$off"), + ] +] + +# Property [Resume Token Type Rejection]: each resume-token option (resumeAfter, +# startAfter) rejects any non-object, non-null BSON type with a TypeMismatch +# error; there is no coercion of a scalar, string, or array to a resume-token +# object. +CHANGESTREAM_RESUME_TOKEN_TYPE_ERROR_TESTS: list[StageTestCase] = [ + StageTestCase( + f"resume_token_type_{opt_id}_{tid}", + pipeline=[{"$changeStream": {opt: val}}], + error_code=TYPE_MISMATCH_ERROR, + msg=f"$changeStream should reject a non-object value for {opt}", + ) + for opt, opt_id in [ + ("resumeAfter", "resume_after"), + ("startAfter", "start_after"), + ] + for tid, val in [ + ("string", "x"), + ("int32", 1), + ("int64", Int64(1)), + ("double", 1.5), + ("decimal128", DECIMAL128_ONE_AND_HALF), + ("bool", True), + ("array", []), + ("objectid", ObjectId("507f1f77bcf86cd799439011")), + ("datetime", datetime(2024, 1, 1, tzinfo=timezone.utc)), + ("timestamp", Timestamp(1, 1)), + ("binary", Binary(b"\x01\x02\x03")), + ("regex", Regex(".*", "i")), + ("code", Code("function(){}")), + ("minkey", MinKey()), + ("maxkey", MaxKey()), + ] +] + +# Property [Resume Token Structure Rejection]: a resume-token object that fails +# structure validation is rejected with an error reflecting the failure mode; a +# missing or non-string _data, an empty-string _data, a non-hex or odd-length +# _data, a well-formed-hex _data that decodes to a malformed KeyString, and a +# wrong-type _typeBits each produce their own distinct error. +CHANGESTREAM_RESUME_TOKEN_STRUCTURE_ERROR_TESTS: list[StageTestCase] = [ + StageTestCase( + f"resume_token_structure_{opt_id}_{suffix}", + pipeline=[{"$changeStream": {opt: token}}], + error_code=error_code, + msg=f"$changeStream should reject a malformed {opt} resume token ({suffix})", + ) + for opt, opt_id in [ + ("resumeAfter", "resume_after"), + ("startAfter", "start_after"), + ] + for token, error_code, suffix in [ + ({}, RESUME_TOKEN_MALFORMED_ERROR, "empty_object"), + ({"a": 1}, RESUME_TOKEN_MALFORMED_ERROR, "no_data_key"), + ({"_data": 42}, RESUME_TOKEN_MALFORMED_ERROR, "data_int"), + ({"_data": None}, RESUME_TOKEN_MALFORMED_ERROR, "data_null"), + ({"_data": True}, RESUME_TOKEN_MALFORMED_ERROR, "data_bool"), + ({"_data": Binary(b"\x01")}, RESUME_TOKEN_MALFORMED_ERROR, "data_binary"), + ({"_data": [1]}, RESUME_TOKEN_MALFORMED_ERROR, "data_array"), + ({"_data": {"x": 1}}, RESUME_TOKEN_MALFORMED_ERROR, "data_object"), + # A DBRef encodes as a BSON object, so it passes the type check and is + # then rejected as a token with no string _data field. + (DBRef("c", 1), RESUME_TOKEN_MALFORMED_ERROR, "dbref"), + ({"_data": ""}, RESUME_TOKEN_EMPTY_ERROR, "data_empty"), + ({"_data": "xyz"}, FAILED_TO_PARSE_ERROR, "data_nonhex"), + ({"_data": "ZZ"}, FAILED_TO_PARSE_ERROR, "data_nonhex_letters"), + ({"_data": "abc"}, FAILED_TO_PARSE_ERROR, "data_odd_length"), + ({"_data": "00"}, KEYSTRING_UNKNOWN_TYPE_ERROR, "data_unknown_keystring_type"), + ( + {"_data": "00", "_typeBits": "x"}, + RESUME_TOKEN_TYPEBITS_WRONG_TYPE_ERROR, + "typebits_string", + ), + ] +] + +# Property [Timestamp Option Type Rejection]: startAtOperationTime rejects any +# non-timestamp, non-null BSON type with a TypeMismatch error; there is no +# coercion from a number, string, datetime, array, or object to a timestamp, and +# an array is not unwrapped to a contained timestamp. +CHANGESTREAM_TIMESTAMP_TYPE_ERROR_TESTS: list[StageTestCase] = [ + StageTestCase( + f"timestamp_type_{tid}", + pipeline=[{"$changeStream": {"startAtOperationTime": val}}], + error_code=TYPE_MISMATCH_ERROR, + msg="$changeStream should reject a non-timestamp value for startAtOperationTime", + ) + for tid, val in [ + ("string", "x"), + ("int32", 1), + ("int64", Int64(1)), + ("double", 1.5), + ("decimal128", DECIMAL128_ONE_AND_HALF), + ("bool", True), + ("objectid", ObjectId("507f1f77bcf86cd799439011")), + ("datetime", datetime(2024, 1, 1, tzinfo=timezone.utc)), + ("binary", Binary(b"\x01\x02\x03")), + ("regex", Regex(".*", "i")), + ("code", Code("function(){}")), + # DBRef encodes as a BSON object and is rejected as a non-timestamp object. + ("dbref", DBRef("c", 1)), + ("minkey", MinKey()), + ("maxkey", MaxKey()), + ("empty_object", {}), + ("object", {"a": 1}), + ("empty_array", []), + ("array_single", [Timestamp(1, 1)]), + ("array_multi", [Timestamp(1, 1), Timestamp(2, 2)]), + ] +] + +CHANGESTREAM_ERROR_TESTS = ( + CHANGESTREAM_SPEC_NON_OBJECT_TESTS + + CHANGESTREAM_SPEC_UNKNOWN_FIELD_TESTS + + CHANGESTREAM_EXPRESSION_NOT_EVALUATED_TESTS + + CHANGESTREAM_BOOLEAN_TYPE_ERROR_TESTS + + CHANGESTREAM_BOOLEAN_NULL_ERROR_TESTS + + CHANGESTREAM_ENUM_TYPE_ERROR_TESTS + + CHANGESTREAM_ENUM_VALUE_ERROR_TESTS + + CHANGESTREAM_RESUME_TOKEN_TYPE_ERROR_TESTS + + CHANGESTREAM_RESUME_TOKEN_STRUCTURE_ERROR_TESTS + + CHANGESTREAM_TIMESTAMP_TYPE_ERROR_TESTS +) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_ERROR_TESTS)) +def test_changeStream_error_cases(collection, test_case: StageTestCase): + """Test $changeStream rejects malformed specs and reads option values as literal BSON.""" + result = execute_command( + collection, + {"aggregate": collection.name, "pipeline": test_case.pipeline, "cursor": {}}, + ) + assertResult(result, error_code=test_case.error_code, msg=test_case.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/utils/__init__.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/utils/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/utils/changeStream_common.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/utils/changeStream_common.py new file mode 100644 index 000000000..ece437195 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/utils/changeStream_common.py @@ -0,0 +1,60 @@ +"""Shared $changeStream command builders and resume-token sentinels. + +The builders return command documents; callers issue them with +``execute_command`` at the callsite so the database call is explicit in each +test file. +""" + +from __future__ import annotations + +from typing import Any + + +def change_stream_command( + collection, *, pipeline, aggregate=None, **command_options +) -> dict[str, Any]: + """Build the aggregate command document that wraps a $changeStream pipeline. + + Args: + collection: The fixture collection the stream targets + pipeline: The full ``[{"$changeStream": ...}]`` pipeline to run + (required) + aggregate: Overrides the aggregate target; defaults to the collection + name, pass ``1`` for a database-scoped stream + command_options: Extra keyword arguments merged as top-level command + fields + """ + command: dict[str, Any] = { + "aggregate": collection.name if aggregate is None else aggregate, + "pipeline": pipeline, + "cursor": {}, + } + command.update(command_options) + return command + + +def get_more_command(collection, cursor_id, *, name: str | None = None) -> dict[str, Any]: + """Build a getMore command document for an open change-stream cursor. + + Args: + collection: The fixture collection whose name is the default namespace + cursor_id: The id of the open change-stream cursor to advance + name: The getMore ``collection`` field; defaults to the fixture + collection name, set to ``$cmd.aggregate`` for a database- or + cluster-scoped stream + """ + return { + "getMore": cursor_id, + "collection": collection.name if name is None else name, + "maxTimeMS": 500, + } + + +# Placeholder for the produced resume token inside a $changeStream spec. The +# harness substitutes the token captured by ``produce_token`` before sending +# the spec, so each case can be written as a full $changeStream document. +RESUME_TOKEN = object() + +# Placeholder for a source stream's operationTime, used by the mutual-exclusivity +# cases that pair a resume token with startAtOperationTime. +OPERATION_TIME = object() diff --git a/documentdb_tests/framework/error_codes.py b/documentdb_tests/framework/error_codes.py index 96beece3d..4a58e07cc 100644 --- a/documentdb_tests/framework/error_codes.py +++ b/documentdb_tests/framework/error_codes.py @@ -15,6 +15,7 @@ INDEX_NOT_FOUND_ERROR = 27 CONFLICTING_UPDATE_OPERATORS_ERROR = 40 CURSOR_NOT_FOUND_ERROR = 43 +NO_MATCHING_DOCUMENT_ERROR = 47 NAMESPACE_EXISTS_ERROR = 48 DOLLAR_PREFIXED_FIELD_NAME_ERROR = 52 INVALID_BSON_ID_ERROR = 53 @@ -45,6 +46,9 @@ QUERY_FEATURE_NOT_ALLOWED = 224 MAX_NESTED_SUB_PIPELINE_ERROR = 232 CONVERSION_FAILURE_ERROR = 241 +INVALID_RESUME_TOKEN_ERROR = 260 +CHANGE_STREAM_FATAL_ERROR = 280 +CHANGE_STREAM_HISTORY_LOST_ERROR = 286 NO_QUERY_EXECUTION_PLANS_ERROR = 291 QUERY_EXCEEDED_MEMORY_NO_DISK_USE_ERROR = 292 API_VERSION_ERROR = 322 @@ -333,6 +337,10 @@ OUT_NOT_LAST_STAGE_ERROR = 40601 NOT_FIRST_STAGE_ERROR = 40602 GEO_NEAR_NOT_FIRST_STAGE_ERROR = 40603 +RESUME_TOKEN_MALFORMED_ERROR = 40647 +RESUME_TOKEN_TYPEBITS_WRONG_TYPE_ERROR = 40648 +RESUME_TOKEN_EMPTY_ERROR = 40649 +MULTIPLE_RESUME_OPTIONS_ERROR = 40674 DATEFROMSTRING_INVALID_FORMAT_TYPE_ERROR = 40684 DATEFROMSTRING_INVALID_FORMAT_ERROR = 40685 TRIM_UNKNOWN_FIELD_ERROR = 50694 @@ -343,8 +351,10 @@ TO_TYPE_ARITY_ERROR = 50723 CURSOR_SESSION_MISMATCH_ERROR = 50738 SUBSTR_NEGATIVE_START_ERROR = 50752 +KEYSTRING_UNKNOWN_TYPE_ERROR = 50811 CLUSTERED_NAN_DUPLICATE_ERROR = 50819 CLUSTERED_INFINITY_DUPLICATE_ERROR = 50826 +RESUME_AFTER_START_AFTER_CONFLICT_ERROR = 50865 PLAN_CACHE_STATS_COLLECTION_NOT_FOUND_ERROR = 50933 LOOKUP_SUB_PIPELINE_NOT_ALLOWED_ERROR = 51047 REGEX_OPTIONS_BEFORE_REGEX_FLAGS_ERROR = 51074 @@ -477,6 +487,7 @@ DATEADD_INT64_MIN_NEGATE_ERROR = 6045000 CLUSTERED_INDEX_CAPPED_CONFLICT_ERROR = 6049200 CLUSTERED_INDEX_MAX_CONFLICT_ERROR = 6049204 +CHANGE_STREAM_SPEC_NOT_OBJECT_ERROR = 6188500 ENCRYPTED_FIELD_EMPTY_PATH_ERROR = 6316402 ENCRYPTED_FIELD_ID_PATH_ERROR = 6316403 ENCRYPTED_FIELD_DUPLICATE_PATH_ERROR = 6338402 From bf1558d11b1f0b64cf01ac041933f3443b7bf8fb Mon Sep 17 00:00:00 2001 From: Daniel Frankcom Date: Fri, 12 Jun 2026 15:27:39 -0700 Subject: [PATCH 2/4] Use execute_command Signed-off-by: Daniel Frankcom --- .../changeStream/test_changeStream_event_read_error.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py index ddbc368b2..30db12a73 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py @@ -41,8 +41,8 @@ def test_changeStream_required_image_event_read_error(collection, test_case): collection.insert_one({"_id": 1, "a": 1}) # The open must succeed: enforcement of the required image is deferred to the # getMore that reads the update event, not raised at parse/open time. - opened = collection.database.command( - change_stream_command(collection, pipeline=test_case.pipeline) + opened = execute_command( + collection, change_stream_command(collection, pipeline=test_case.pipeline) ) collection.update_one({"_id": 1}, {"$set": {"a": 2}}) result = execute_command(collection, get_more_command(collection, opened["cursor"]["id"])) From 37e28ea670263d22f8bb7328e9065271af67d37b Mon Sep 17 00:00:00 2001 From: Daniel Frankcom Date: Wed, 17 Jun 2026 10:22:07 -0700 Subject: [PATCH 3/4] Rework change-stream pre-oplog tests to assert observed behavior The history-lost tests asserted that a startAtOperationTime (or a resume/start token) at or before the oldest retained oplog entry is rejected with ChangeStreamHistoryLost. The server does not do this: such a start point is accepted and the stream opens from the earliest available event. The ChangeStreamHistoryLost error is reserved for resuming from a point that was retained and has since been evicted, which requires sustained oplog rollover that a controlled test environment cannot reliably force. As written the tests never validated anything: they asserted an error the server does not return, and only ever ran where no replica set was present, so they were always skipped. Rework them to assert the behavior the server actually exhibits - an early start point is accepted across collection-, database-, and cluster-scoped streams, and resume/start tokens predating the oplog window are accepted too - so they verify real behavior on a replica set. Rename the file and the helper to match. Signed-off-by: Daniel Frankcom --- .../test_changeStream_early_start.py | 136 ++++++++++++++++++ .../test_changeStream_history_lost.py | 129 ----------------- .../changeStream/test_changeStream_resume.py | 31 ++-- 3 files changed, 153 insertions(+), 143 deletions(-) create mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_early_start.py delete mode 100644 documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_history_lost.py diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_early_start.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_early_start.py new file mode 100644 index 000000000..785f630f4 --- /dev/null +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_early_start.py @@ -0,0 +1,136 @@ +"""Tests for $changeStream startAtOperationTime at or before the oldest oplog entry. + +A startAtOperationTime at or before the oldest retained oplog entry is accepted: +the stream opens and begins from the earliest available event. (It is not +rejected as history lost; that error is reserved for resuming from a point that +was retained and has since been evicted, which a controlled test environment +cannot reliably force.) +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import pytest +from bson import Timestamp +from utils.changeStream_common import change_stream_command + +from documentdb_tests.framework.assertions import assertResult +from documentdb_tests.framework.executor import execute_admin_command, execute_command +from documentdb_tests.framework.parametrize import pytest_params +from documentdb_tests.framework.property_checks import Eq +from documentdb_tests.framework.test_case import BaseTestCase + + +# Return the timestamp of the oldest retained oplog entry, so a start point can +# be expressed relative to the live oplog rather than as a static value. +def _oldest_oplog_ts(collection) -> Timestamp: + oldest = ( + collection.database.client["local"]["oplog.rs"] + .find() + .sort("$natural", 1) + .limit(1) + .next()["ts"] + ) + return oldest + + +@dataclass(frozen=True) +class ChangeStreamEarlyStartTestCase(BaseTestCase): + """Test case for a startAtOperationTime at or before the oldest oplog entry. + + Attributes: + compute_start: Receives the timestamp of the oldest retained oplog entry + captured at run time and returns the startAtOperationTime to test, so + the boundary case is expressed relative to the live oplog rather than + as a static value. + """ + + compute_start: Any = None + + +# Property [Early Start Accepted]: a startAtOperationTime at or before the oldest +# retained oplog entry opens the stream from the earliest available event rather +# than being rejected, including the boundary five seconds before the oldest +# retained entry. This is verified identically across collection-, database-, and +# cluster-scoped streams. +CHANGESTREAM_EARLY_START_TESTS: list[ChangeStreamEarlyStartTestCase] = [ + ChangeStreamEarlyStartTestCase( + "zero", + compute_start=lambda oldest: Timestamp(0, 0), + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept a zero startAtOperationTime", + ), + ChangeStreamEarlyStartTestCase( + "one_zero", + compute_start=lambda oldest: Timestamp(1, 0), + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept a startAtOperationTime before the oldest oplog entry", + ), + ChangeStreamEarlyStartTestCase( + "max_increment", + compute_start=lambda oldest: Timestamp(1, 4_294_967_295), + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept a startAtOperationTime before the oldest oplog entry", + ), + ChangeStreamEarlyStartTestCase( + "oldest_minus_5s", + compute_start=lambda oldest: Timestamp(oldest.time - 5, oldest.inc), + expected={"ok": Eq(1.0)}, + msg="$changeStream should accept a startAtOperationTime five seconds before" + " the oldest retained oplog entry", + ), +] + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EARLY_START_TESTS)) +def test_changeStream_early_start_collection_scope( + collection, test_case: ChangeStreamEarlyStartTestCase +): + """Test $changeStream accepts an early startAtOperationTime on a collection-scoped stream.""" + start = test_case.compute_start(_oldest_oplog_ts(collection)) + result = execute_command( + collection, + change_stream_command( + collection, pipeline=[{"$changeStream": {"startAtOperationTime": start}}] + ), + ) + assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EARLY_START_TESTS)) +def test_changeStream_early_start_database_scope( + collection, test_case: ChangeStreamEarlyStartTestCase +): + """Test $changeStream accepts an early startAtOperationTime on a database-scoped stream.""" + start = test_case.compute_start(_oldest_oplog_ts(collection)) + result = execute_command( + collection, + change_stream_command( + collection, + pipeline=[{"$changeStream": {"startAtOperationTime": start}}], + aggregate=1, + ), + ) + assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) + + +@pytest.mark.replica_set +@pytest.mark.aggregate +@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EARLY_START_TESTS)) +def test_changeStream_early_start_cluster_scope( + collection, test_case: ChangeStreamEarlyStartTestCase +): + """Test $changeStream accepts an early startAtOperationTime on a cluster-wide stream.""" + start = test_case.compute_start(_oldest_oplog_ts(collection)) + spec = {"startAtOperationTime": start, "allChangesForCluster": True} + result = execute_admin_command( + collection, + change_stream_command(collection, pipeline=[{"$changeStream": spec}], aggregate=1), + ) + assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_history_lost.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_history_lost.py deleted file mode 100644 index e7e13f0e6..000000000 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_history_lost.py +++ /dev/null @@ -1,129 +0,0 @@ -"""Tests for $changeStream history-lost errors when the start point predates the oplog.""" - -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any - -import pytest -from bson import Timestamp -from utils.changeStream_common import change_stream_command - -from documentdb_tests.framework.assertions import assertResult -from documentdb_tests.framework.error_codes import CHANGE_STREAM_HISTORY_LOST_ERROR -from documentdb_tests.framework.executor import execute_admin_command, execute_command -from documentdb_tests.framework.parametrize import pytest_params -from documentdb_tests.framework.test_case import BaseTestCase - - -# Return the timestamp of the oldest retained oplog entry, used to express the -# history-lost boundary relative to the live oplog rather than as a static value. -def _oldest_oplog_ts(collection) -> Timestamp: - oldest = ( - collection.database.client["local"]["oplog.rs"] - .find() - .sort("$natural", 1) - .limit(1) - .next()["ts"] - ) - return oldest - - -@dataclass(frozen=True) -class ChangeStreamHistoryLostTestCase(BaseTestCase): - """Test case for a pre-oplog startAtOperationTime rejected as history lost. - - Attributes: - compute_start: Receives the timestamp of the oldest retained oplog - entry captured at run time and returns the startAtOperationTime to - test, so the boundary case is expressed relative to the live oplog - rather than as a static value - """ - - compute_start: Any = None - - -# Property [Timestamp History Lost]: a startAtOperationTime strictly before the -# oldest retained oplog entry is rejected at open with a ChangeStreamHistoryLost -# error, including the boundary five seconds before the oldest retained entry. -# This is verified identically across collection-, database-, and cluster-scoped -# streams. -CHANGESTREAM_HISTORY_LOST_TESTS: list[ChangeStreamHistoryLostTestCase] = [ - ChangeStreamHistoryLostTestCase( - "zero", - compute_start=lambda oldest: Timestamp(0, 0), - error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, - msg="$changeStream should reject a zero startAtOperationTime as history lost", - ), - ChangeStreamHistoryLostTestCase( - "one_zero", - compute_start=lambda oldest: Timestamp(1, 0), - error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, - msg="$changeStream should reject a pre-oplog startAtOperationTime as history lost", - ), - ChangeStreamHistoryLostTestCase( - "max_increment", - compute_start=lambda oldest: Timestamp(1, 4_294_967_295), - error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, - msg="$changeStream should reject a pre-oplog startAtOperationTime as history lost", - ), - ChangeStreamHistoryLostTestCase( - "oldest_minus_5s", - compute_start=lambda oldest: Timestamp(oldest.time - 5, oldest.inc), - error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, - msg="$changeStream should reject a startAtOperationTime five seconds before" - " the oldest retained oplog entry as history lost", - ), -] - - -@pytest.mark.replica_set -@pytest.mark.aggregate -@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_HISTORY_LOST_TESTS)) -def test_changeStream_history_lost_collection_scope( - collection, test_case: ChangeStreamHistoryLostTestCase -): - """Test $changeStream rejects a pre-oplog startAtOperationTime on a collection-scoped stream.""" - start = test_case.compute_start(_oldest_oplog_ts(collection)) - result = execute_command( - collection, - change_stream_command( - collection, pipeline=[{"$changeStream": {"startAtOperationTime": start}}] - ), - ) - assertResult(result, error_code=test_case.error_code, msg=test_case.msg) - - -@pytest.mark.replica_set -@pytest.mark.aggregate -@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_HISTORY_LOST_TESTS)) -def test_changeStream_history_lost_database_scope( - collection, test_case: ChangeStreamHistoryLostTestCase -): - """Test $changeStream rejects a pre-oplog startAtOperationTime on a database-scoped stream.""" - start = test_case.compute_start(_oldest_oplog_ts(collection)) - result = execute_command( - collection, - change_stream_command( - collection, - pipeline=[{"$changeStream": {"startAtOperationTime": start}}], - aggregate=1, - ), - ) - assertResult(result, error_code=test_case.error_code, msg=test_case.msg) - - -@pytest.mark.replica_set -@pytest.mark.aggregate -@pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_HISTORY_LOST_TESTS)) -def test_changeStream_history_lost_cluster_scope( - collection, test_case: ChangeStreamHistoryLostTestCase -): - """Test $changeStream rejects a pre-oplog startAtOperationTime on a cluster-wide stream.""" - start = test_case.compute_start(_oldest_oplog_ts(collection)) - spec = {"startAtOperationTime": start, "allChangesForCluster": True} - result = execute_admin_command( - collection, - change_stream_command(collection, pipeline=[{"$changeStream": spec}], aggregate=1), - ) - assertResult(result, error_code=test_case.error_code, msg=test_case.msg) diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py index 39f8c1e6a..5042d824b 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py @@ -15,7 +15,6 @@ from documentdb_tests.framework.assertions import assertResult from documentdb_tests.framework.error_codes import ( CHANGE_STREAM_FATAL_ERROR, - CHANGE_STREAM_HISTORY_LOST_ERROR, INVALID_RESUME_TOKEN_ERROR, OVERFLOW_ERROR, ) @@ -64,7 +63,7 @@ def _high_water_mark(collection): return _open_stream(collection)["cursor"]["postBatchResumeToken"] -def _evicted_token(collection): +def _pre_oplog_token(collection): """Return a token whose captured cluster time predates the oplog window. Rewrites the cluster-time seconds (four bytes after the version byte) of a @@ -201,10 +200,12 @@ def test_resume_regular_token_under_show_expanded_events(collection): ) -# Property [Resume After Evicted]: resumeAfter an evicted token fails as history lost. -def test_resume_after_evicted_token(collection): - """Test $changeStream resumeAfter from an evicted token.""" - token = _evicted_token(collection) +# Property [Resume After Pre-Oplog]: resumeAfter a token whose cluster time +# predates the oplog window is accepted, opening from the earliest available +# event rather than being rejected. +def test_resume_after_pre_oplog_token(collection): + """Test $changeStream resumeAfter from a token predating the oplog window.""" + token = _pre_oplog_token(collection) result = execute_command( collection, @@ -213,16 +214,18 @@ def test_resume_after_evicted_token(collection): assertResult( result, - error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, - msg="$changeStream resumeAfter should reject an evicted token as history lost", + expected={"ok": Eq(1.0)}, + msg="$changeStream resumeAfter should accept a token predating the oplog window", raw_res=True, ) -# Property [Start After Evicted]: startAfter an evicted token fails as history lost. -def test_start_after_evicted_token(collection): - """Test $changeStream startAfter from an evicted token.""" - token = _evicted_token(collection) +# Property [Start After Pre-Oplog]: startAfter a token whose cluster time +# predates the oplog window is accepted, opening from the earliest available +# event rather than being rejected. +def test_start_after_pre_oplog_token(collection): + """Test $changeStream startAfter from a token predating the oplog window.""" + token = _pre_oplog_token(collection) result = execute_command( collection, @@ -231,8 +234,8 @@ def test_start_after_evicted_token(collection): assertResult( result, - error_code=CHANGE_STREAM_HISTORY_LOST_ERROR, - msg="$changeStream startAfter should reject an evicted token as history lost", + expected={"ok": Eq(1.0)}, + msg="$changeStream startAfter should accept a token predating the oplog window", raw_res=True, ) From c45eb0cdabc40ca0c0e7c205cff40e30eae20e5d Mon Sep 17 00:00:00 2001 From: Daniel Frankcom Date: Tue, 16 Jun 2026 17:44:01 -0700 Subject: [PATCH 4/4] Gate change-stream tests on the change_streams capability Migrate the $changeStream stage and system-stage tests from replica_set markers to requires(change_streams=...) markers. Signed-off-by: Daniel Frankcom --- .../stages/test_stages_position_changeStream.py | 2 +- .../changeStream/test_changeStream_command_options.py | 2 +- .../changeStream/test_changeStream_early_start.py | 10 +++++----- .../changeStream/test_changeStream_event_read_error.py | 2 +- .../changeStream/test_changeStream_events.py | 2 +- .../changeStream/test_changeStream_expanded_events.py | 2 +- .../changeStream/test_changeStream_namespace_errors.py | 2 +- .../changeStream/test_changeStream_namespace_scope.py | 2 +- .../changeStream/test_changeStream_resume.py | 2 +- .../test_changeStream_resume_mutual_exclusivity.py | 2 +- .../changeStream/test_changeStream_spec_acceptance.py | 4 ++-- .../changeStream/test_changeStream_stable_api.py | 2 +- .../test_changeStream_timestamp_boundary.py | 6 +++--- .../test_changeStream_validation_errors.py | 2 +- 14 files changed, 21 insertions(+), 21 deletions(-) diff --git a/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_position_changeStream.py b/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_position_changeStream.py index 6646e984b..2d05f0a82 100644 --- a/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_position_changeStream.py +++ b/documentdb_tests/compatibility/tests/core/operator/stages/test_stages_position_changeStream.py @@ -271,7 +271,7 @@ ) -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test", pytest_params(CHANGESTREAM_POSITION_TESTS)) def test_changeStream_position( diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_command_options.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_command_options.py index f9e6897bd..4da64cc50 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_command_options.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_command_options.py @@ -61,7 +61,7 @@ ] -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test", pytest_params(CHANGESTREAM_COMMAND_OPTION_TESTS)) def test_changeStream_command_options(database_client, collection, test): diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_early_start.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_early_start.py index 785f630f4..f0f53ebc0 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_early_start.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_early_start.py @@ -10,7 +10,7 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any +from typing import Any, cast import pytest from bson import Timestamp @@ -33,7 +33,7 @@ def _oldest_oplog_ts(collection) -> Timestamp: .limit(1) .next()["ts"] ) - return oldest + return cast(Timestamp, oldest) @dataclass(frozen=True) @@ -84,7 +84,7 @@ class ChangeStreamEarlyStartTestCase(BaseTestCase): ] -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EARLY_START_TESTS)) def test_changeStream_early_start_collection_scope( @@ -101,7 +101,7 @@ def test_changeStream_early_start_collection_scope( assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EARLY_START_TESTS)) def test_changeStream_early_start_database_scope( @@ -120,7 +120,7 @@ def test_changeStream_early_start_database_scope( assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EARLY_START_TESTS)) def test_changeStream_early_start_cluster_scope( diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py index 30db12a73..870cd7014 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_event_read_error.py @@ -33,7 +33,7 @@ ] -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EVENT_READ_ERROR_TESTS)) def test_changeStream_required_image_event_read_error(collection, test_case): diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_events.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_events.py index ea681ace8..bdb9eba66 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_events.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_events.py @@ -455,7 +455,7 @@ def _emit_event(collection, test_case: ChangeStreamEventTestCase) -> dict[str, A return event -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EVENT_STRUCTURE_TESTS)) def test_changeStream_event_structure(collection, test_case: ChangeStreamEventTestCase): diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_expanded_events.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_expanded_events.py index f10382c31..026d0e7cc 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_expanded_events.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_expanded_events.py @@ -262,7 +262,7 @@ def _emit_ddl_result(collection, database_client, test_case: ChangeStreamGatingT ) -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_EXPANDED_EVENT_TESTS)) def test_changeStream_showExpandedEvents_event_gating( diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_errors.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_errors.py index 9eb5a1ad6..b84800516 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_errors.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_errors.py @@ -130,7 +130,7 @@ ) -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test", pytest_params(CHANGESTREAM_NAMESPACE_ERROR_TESTS)) def test_changeStream_namespace_errors(database_client, collection, test): diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_scope.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_scope.py index 8bb43d4eb..23e2b375e 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_scope.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_namespace_scope.py @@ -85,7 +85,7 @@ ] -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test", pytest_params(CHANGESTREAM_NAMESPACE_SCOPE_TESTS)) def test_changeStream_namespace_scope(database_client, collection, register_db_cleanup, test): diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py index 5042d824b..dfb4bc6f5 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume.py @@ -21,7 +21,7 @@ from documentdb_tests.framework.executor import execute_command from documentdb_tests.framework.property_checks import Eq, Len -pytestmark = [pytest.mark.replica_set, pytest.mark.aggregate] +pytestmark = [pytest.mark.requires(change_streams=True), pytest.mark.aggregate] # Token capture helpers. A resume token must be captured from a real stream diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume_mutual_exclusivity.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume_mutual_exclusivity.py index 69ab8518a..52528e793 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume_mutual_exclusivity.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_resume_mutual_exclusivity.py @@ -45,7 +45,7 @@ ] -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_RESUME_MUTUAL_EXCLUSIVITY_TESTS)) def test_changeStream_resume_option_mutual_exclusivity(collection, test_case): diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_spec_acceptance.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_spec_acceptance.py index 5fc190635..d30312287 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_spec_acceptance.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_spec_acceptance.py @@ -181,7 +181,7 @@ ) -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_SUCCESS_TESTS)) def test_changeStream_cases(collection, test_case: StageTestCase): @@ -197,7 +197,7 @@ def test_changeStream_cases(collection, test_case: StageTestCase): # null does not count as a second resume point, so pairing it with another # present resume token opens successfully rather than tripping resume-option # mutual exclusivity. This holds for every (present token, null option) pairing. -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize( "present_field,null_field", diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_stable_api.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_stable_api.py index a75788015..d77cb152d 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_stable_api.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_stable_api.py @@ -58,7 +58,7 @@ ] -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_STABLE_API_V1_TESTS)) def test_changeStream_stable_api_v1(collection, test_case): diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_timestamp_boundary.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_timestamp_boundary.py index 18182fbf7..e136e6009 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_timestamp_boundary.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_timestamp_boundary.py @@ -56,7 +56,7 @@ class ChangeStreamTimestampTestCase(BaseTestCase): ] -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_TIMESTAMP_BOUNDARY_TESTS)) def test_changeStream_timestamp_boundary_collection_scope( @@ -76,7 +76,7 @@ def test_changeStream_timestamp_boundary_collection_scope( assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_TIMESTAMP_BOUNDARY_TESTS)) def test_changeStream_timestamp_boundary_database_scope( @@ -98,7 +98,7 @@ def test_changeStream_timestamp_boundary_database_scope( assertResult(result, expected=test_case.expected, msg=test_case.msg, raw_res=True) -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_TIMESTAMP_BOUNDARY_TESTS)) def test_changeStream_timestamp_boundary_cluster_scope( diff --git a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_validation_errors.py b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_validation_errors.py index be5644138..0a42fc7bd 100644 --- a/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_validation_errors.py +++ b/documentdb_tests/compatibility/tests/core/operator/system-stages/changeStream/test_changeStream_validation_errors.py @@ -518,7 +518,7 @@ ) -@pytest.mark.replica_set +@pytest.mark.requires(change_streams=True) @pytest.mark.aggregate @pytest.mark.parametrize("test_case", pytest_params(CHANGESTREAM_ERROR_TESTS)) def test_changeStream_error_cases(collection, test_case: StageTestCase):