From 2ea062d800741f3e5de8e74ca6375f7aeb3e7e8b Mon Sep 17 00:00:00 2001
From: Daniel Frankcom <frankcom@amazon.com>
Date: Wed, 20 May 2026 11:57:50 -0700
Subject: [PATCH] Add capped collection tests Signed-off-by: Daniel Frankcom
 <frankcom@amazon.com>

---
 .../test_capped_collections_convert.py        |  84 +++++
 .../test_capped_collections_create.py         |  20 --
 .../test_capped_collections_deletes.py        |  96 ++++++
 .../test_capped_collections_eviction.py       | 186 +++++++++++
 .../test_capped_collections_ordering.py       | 311 ++++++++++++++++++
 .../test_capped_collections_write_position.py | 171 ++++++++++
 .../test_smoke_capped_collections.py}         |   4 +-
 7 files changed, 850 insertions(+), 22 deletions(-)
 create mode 100644 documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_convert.py
 delete mode 100644 documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_create.py
 create mode 100644 documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_deletes.py
 create mode 100644 documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_eviction.py
 create mode 100644 documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_ordering.py
 create mode 100644 documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_write_position.py
 rename documentdb_tests/compatibility/tests/core/collections/{capped-collections/test_smoke_capped-collections.py => capped_collections/test_smoke_capped_collections.py} (90%)

diff --git a/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_convert.py b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_convert.py
new file mode 100644
index 00000000..38e45bb7
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_convert.py
@@ -0,0 +1,84 @@
+"""Tests for capped collection convertToCapped behaviors."""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertProperties, assertSuccess
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.property_checks import Eq
+
+
+# Property [Convert Max Ignored]: convertToCapped does not store the max
+# parameter in collection metadata.
+@pytest.mark.collection_mgmt
+def test_capped_convert_max_ignored(database_client, collection):
+    """Test that convertToCapped ignores the max parameter."""
+    name = f"{collection.name}_conv"
+    database_client.create_collection(name)
+    database_client[name].insert_many([{"_id": i} for i in range(5)])
+    execute_command(database_client[name], {"convertToCapped": name, "size": 100_000, "max": 2})
+    result = execute_command(database_client[name], {"collStats": name})
+    assertProperties(
+        result,
+        {"max": Eq(0)},
+        msg="convertToCapped should not store the max parameter",
+        raw_res=True,
+    )
+
+
+# Property [Convert Max Not Enforced]: the max parameter passed to
+# convertToCapped is not enforced on subsequent inserts.
+@pytest.mark.collection_mgmt
+def test_capped_convert_max_not_enforced(database_client, collection):
+    """Test that max from convertToCapped is not enforced on inserts."""
+    name = f"{collection.name}_conv"
+    database_client.create_collection(name)
+    database_client[name].insert_many([{"_id": i} for i in range(5)])
+    execute_command(database_client[name], {"convertToCapped": name, "size": 100_000, "max": 2})
+    database_client[name].insert_many([{"_id": i} for i in range(5, 15)])
+    result = execute_command(database_client[name], {"find": name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": i} for i in range(15)],
+        msg="convertToCapped max should not limit subsequent inserts",
+    )
+
+
+# Property [Convert Drops Secondary Indexes]: all secondary indexes are dropped
+# during conversion and only the _id index survives.
+@pytest.mark.collection_mgmt
+def test_capped_convert_index_drop(database_client, collection):
+    """Test that convertToCapped drops secondary indexes."""
+    name = f"{collection.name}_conv"
+    database_client.create_collection(name)
+    coll = database_client[name]
+    coll.insert_many([{"_id": i, "x": i, "y": str(i)} for i in range(5)])
+    coll.create_index("x")
+    coll.create_index("y")
+    execute_command(coll, {"convertToCapped": name, "size": 100_000})
+    result = execute_command(coll, {"listIndexes": name})
+    assertSuccess(
+        result,
+        ["_id_"],
+        msg="convertToCapped should drop all secondary indexes",
+        raw_res=True,
+        transform=lambda r: [idx["name"] for idx in r["cursor"]["firstBatch"]],
+    )
+
+
+# Property [Convert Order Preservation]: insertion order is preserved for
+# retained documents after conversion.
+@pytest.mark.collection_mgmt
+def test_capped_convert_order(database_client, collection):
+    """Test that convertToCapped preserves insertion order."""
+    name = f"{collection.name}_conv"
+    database_client.create_collection(name)
+    database_client[name].insert_many([{"_id": 5}, {"_id": 2}, {"_id": 8}, {"_id": 1}, {"_id": 3}])
+    execute_command(database_client[name], {"convertToCapped": name, "size": 100_000})
+    result = execute_command(database_client[name], {"find": name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 5}, {"_id": 2}, {"_id": 8}, {"_id": 1}, {"_id": 3}],
+        msg="convertToCapped should preserve insertion order for retained documents",
+    )
diff --git a/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_create.py b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_create.py
deleted file mode 100644
index 815d59e9..00000000
--- a/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_create.py
+++ /dev/null
@@ -1,20 +0,0 @@
-"""
-Tests for capped collection operations.
-
-Capped collections are fixed-size collections that maintain insertion order.
-This feature may not be supported on all engines.
-"""
-
-import pytest
-
-from documentdb_tests.framework.assertions import assertSuccessPartial
-from documentdb_tests.framework.executor import execute_command
-
-
-@pytest.mark.collection_mgmt
-def test_create_capped_collection(collection):
-    """Test creating a capped collection."""
-    result = execute_command(
-        collection, {"create": collection.name + "_capped", "capped": True, "size": 100000}
-    )
-    assertSuccessPartial(result, {"ok": 1.0}, "Should create capped collection")
diff --git a/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_deletes.py b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_deletes.py
new file mode 100644
index 00000000..da7adffb
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_deletes.py
@@ -0,0 +1,96 @@
+"""Tests for capped collection delete behaviors."""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertProperties, assertSuccess
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.property_checks import Eq
+from documentdb_tests.framework.target_collection import CappedCollection
+
+
+# Property [Delete Order Preservation]: after deletes, remaining documents
+# maintain their relative insertion order.
+@pytest.mark.collection_mgmt
+def test_capped_delete_order(database_client, collection):
+    """Test that deletes preserve relative insertion order."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}, {"_id": 4}, {"_id": 5}])
+    execute_command(coll, {"delete": coll.name, "deletes": [{"q": {"_id": 3}, "limit": 1}]})
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}, {"_id": 2}, {"_id": 4}, {"_id": 5}],
+        msg="Remaining documents should maintain relative insertion order after delete",
+    )
+
+
+# Property [Delete Insert Append]: after deletes, new inserts append at the
+# end of natural order and deleted positions are not reused.
+@pytest.mark.collection_mgmt
+def test_capped_delete_insert_append(database_client, collection):
+    """Test that new inserts after delete append at the end."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}])
+    execute_command(coll, {"delete": coll.name, "deletes": [{"q": {"_id": 2}, "limit": 1}]})
+    coll.insert_one({"_id": 4})
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}, {"_id": 3}, {"_id": 4}],
+        msg="New inserts after delete should append at the end of natural order",
+    )
+
+
+# Property [Delete Preserves Capped]: the collection remains capped after any
+# delete operation.
+@pytest.mark.collection_mgmt
+def test_capped_delete_remains_capped(database_client, collection):
+    """Test that the collection remains capped after deletes."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}])
+    execute_command(coll, {"delete": coll.name, "deletes": [{"q": {}, "limit": 0}]})
+    result = execute_command(coll, {"collStats": coll.name})
+    assertProperties(
+        result,
+        {"capped": Eq(True)},
+        msg="Collection should remain capped after delete",
+        raw_res=True,
+    )
+
+
+# Property [Delete Natural Hint Forward]: $natural:1 hint in a delete targets
+# the oldest document.
+@pytest.mark.collection_mgmt
+def test_capped_delete_hint_natural_forward(database_client, collection):
+    """Test $natural:1 hint targets the oldest document for deletion."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}, {"_id": 4}, {"_id": 5}])
+    execute_command(
+        coll, {"delete": coll.name, "deletes": [{"q": {}, "limit": 1, "hint": {"$natural": 1}}]}
+    )
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 2}, {"_id": 3}, {"_id": 4}, {"_id": 5}],
+        msg="$natural:1 hint should target the oldest document for deletion",
+    )
+
+
+# Property [Delete Natural Hint Reverse]: $natural:-1 hint in a delete targets
+# the newest document.
+@pytest.mark.collection_mgmt
+def test_capped_delete_hint_natural_reverse(database_client, collection):
+    """Test $natural:-1 hint targets the newest document for deletion."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}, {"_id": 4}, {"_id": 5}])
+    execute_command(
+        coll, {"delete": coll.name, "deletes": [{"q": {}, "limit": 1, "hint": {"$natural": -1}}]}
+    )
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}, {"_id": 2}, {"_id": 3}, {"_id": 4}],
+        msg="$natural:-1 hint should target the newest document for deletion",
+    )
diff --git a/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_eviction.py b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_eviction.py
new file mode 100644
index 00000000..f9c7a252
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_eviction.py
@@ -0,0 +1,186 @@
+"""Tests for capped collection eviction behaviors."""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertProperties, assertSuccess
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.property_checks import Eq
+from documentdb_tests.framework.target_collection import CappedCollection
+
+
+# Property [Eviction Order Preservation]: surviving documents maintain their
+# relative insertion order after eviction has occurred.
+@pytest.mark.collection_mgmt
+def test_capped_eviction_order(database_client, collection):
+    """Test insertion order preservation after eviction."""
+    coll = CappedCollection(size=100_000, max=5).resolve(database_client, collection)
+    for doc in [{"_id": i} for i in [10, 7, 3, 8, 1, 5, 9, 2, 6, 4]]:
+        coll.insert_one(doc)
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 5}, {"_id": 9}, {"_id": 2}, {"_id": 6}, {"_id": 4}],
+        msg="Surviving documents should maintain relative insertion order after eviction",
+    )
+
+
+# Property [Evicted _id Reuse]: after eviction removes a document, its _id
+# value can be reused in a new insert.
+@pytest.mark.collection_mgmt
+def test_capped_duplicate_id_reuse(database_client, collection):
+    """Test that evicted _id values can be reused."""
+    coll = CappedCollection(size=100_000, max=2).resolve(database_client, collection)
+    for doc in [{"_id": 1}, {"_id": 2}, {"_id": 3}]:
+        coll.insert_one(doc)
+    # _id:1 was evicted (max=2 keeps only last 2). Reinsert it.
+    coll.insert_one({"_id": 1})
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 3}, {"_id": 1}],
+        msg="An evicted _id value should be reusable in a new insert",
+    )
+
+
+# Property [Eviction Removes Index Entries]: evicted documents are no longer
+# findable via secondary indexes.
+@pytest.mark.collection_mgmt
+def test_capped_eviction_index_cleanup(database_client, collection):
+    """Test that eviction removes entries from secondary indexes."""
+    coll = CappedCollection(size=100_000, max=3).resolve(database_client, collection)
+    coll.create_index("x")
+    coll.insert_many([{"_id": 1, "x": "findme"}, {"_id": 2, "x": "b"}, {"_id": 3, "x": "c"}])
+    # Trigger eviction by inserting beyond max.
+    coll.insert_one({"_id": 4, "x": "d"})
+    # The evicted document should no longer be findable via the index.
+    result = execute_command(
+        coll, {"find": coll.name, "filter": {"x": "findme"}, "projection": {"_id": 1}}
+    )
+    assertSuccess(result, [], msg="Evicted documents should not be findable via secondary index")
+
+
+# Property [Size-Based Eviction]: when the total data size exceeds the byte
+# size cap, the oldest documents are evicted to make room.
+@pytest.mark.collection_mgmt
+def test_capped_size_based_eviction(database_client, collection):
+    """Test that size-based eviction removes oldest documents."""
+    coll = CappedCollection(size=4096).resolve(database_client, collection)
+    for i in range(50):
+        coll.insert_one({"_id": i, "data": "x" * 500})
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+
+    def validate_size_eviction(r):
+        ids = [d["_id"] for d in r["cursor"]["firstBatch"]]
+        # Eviction occurred, most recent survives, in insertion order
+        return len(ids) < 50 and ids[-1] == 49 and ids == list(range(ids[0], ids[-1] + 1))
+
+    assertSuccess(
+        result,
+        True,
+        msg="Size-based eviction should remove oldest docs, preserving insertion order",
+        raw_res=True,
+        transform=validate_size_eviction,
+    )
+
+
+# Property [Dual Constraint Max First]: when both size and max are specified
+# and max is the tighter constraint, eviction is triggered by max.
+@pytest.mark.collection_mgmt
+def test_capped_dual_constraint_max_triggers(database_client, collection):
+    """Test that max triggers eviction when it is the tighter constraint."""
+    coll = CappedCollection(size=1_000_000, max=3).resolve(database_client, collection)
+    coll.insert_many([{"_id": i} for i in range(7)])
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 4}, {"_id": 5}, {"_id": 6}],
+        msg="max should trigger eviction when it is the tighter constraint",
+    )
+
+
+# Property [Dual Constraint Size First]: when both size and max are specified
+# and size is the tighter constraint, eviction is triggered by size.
+@pytest.mark.collection_mgmt
+def test_capped_dual_constraint_size_triggers(database_client, collection):
+    """Test that size triggers eviction when it is the tighter constraint."""
+    coll = CappedCollection(size=4096, max=1000).resolve(database_client, collection)
+    for i in range(50):
+        coll.insert_one({"_id": i, "data": "x" * 500})
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+
+    def validate_size_triggers_first(r):
+        ids = [d["_id"] for d in r["cursor"]["firstBatch"]]
+        # Size triggered (count well below max=1000), most recent survives, in order
+        return len(ids) < 50 and ids[-1] == 49 and ids == list(range(ids[0], ids[-1] + 1))
+
+    assertSuccess(
+        result,
+        True,
+        msg="Size should trigger eviction before max is reached",
+        raw_res=True,
+        transform=validate_size_triggers_first,
+    )
+
+
+# Property [Batch Insert Eviction]: a single insertMany that exceeds the max
+# document limit evicts oldest documents, keeping only the last max documents.
+@pytest.mark.collection_mgmt
+def test_capped_batch_insert_eviction(database_client, collection):
+    """Test eviction during a single insertMany call."""
+    coll = CappedCollection(size=100_000, max=5).resolve(database_client, collection)
+    coll.insert_many([{"_id": i} for i in range(10)])
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 5}, {"_id": 6}, {"_id": 7}, {"_id": 8}, {"_id": 9}],
+        msg="Batch insert should evict oldest documents when max is exceeded",
+    )
+
+
+# Property [Count Reflects Eviction]: the count command returns only the
+# number of surviving documents after eviction.
+@pytest.mark.collection_mgmt
+def test_capped_count_reflects_eviction(database_client, collection):
+    """Test that count reflects eviction."""
+    coll = CappedCollection(size=100_000, max=3).resolve(database_client, collection)
+    coll.insert_many([{"_id": i} for i in range(10)])
+    result = execute_command(coll, {"count": coll.name})
+    assertProperties(
+        result,
+        {"n": Eq(3)},
+        msg="Count should reflect only surviving documents after eviction",
+        raw_res=True,
+    )
+
+
+# Property [Oversize Document Accepted]: a single document larger than the
+# declared size cap is accepted.
+@pytest.mark.collection_mgmt
+def test_capped_oversize_document_accepted(database_client, collection):
+    """Test that a document larger than the size cap is accepted."""
+    coll = CappedCollection(size=4096).resolve(database_client, collection)
+    coll.insert_one({"_id": 1, "data": "x" * 10000})
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}],
+        msg="A document larger than the size cap should be accepted",
+    )
+
+
+# Property [Oversize Document Evicted]: after accepting an oversize document,
+# subsequent inserts evict it following normal oldest-first eviction.
+@pytest.mark.collection_mgmt
+def test_capped_oversize_document_evicted(database_client, collection):
+    """Test that an oversize document is evicted by subsequent inserts."""
+    coll = CappedCollection(size=4096).resolve(database_client, collection)
+    coll.insert_one({"_id": 1, "data": "x" * 10000})
+    coll.insert_one({"_id": 2, "data": "y" * 50})
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 2}],
+        msg="Oversize document should be evicted by subsequent inserts",
+    )
diff --git a/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_ordering.py b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_ordering.py
new file mode 100644
index 00000000..47bc3d2b
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_ordering.py
@@ -0,0 +1,311 @@
+"""Tests for capped collection ordering behaviors."""
+
+from __future__ import annotations
+
+import pytest
+from pymongo import IndexModel
+
+from documentdb_tests.compatibility.tests.core.collections.commands.utils.command_test_case import (
+    CommandContext,
+    CommandTestCase,
+)
+from documentdb_tests.framework.assertions import assertResult, assertSuccess
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.parametrize import pytest_params
+from documentdb_tests.framework.target_collection import CappedCollection
+
+# Property [Insertion Order]: documents inserted into a capped collection are
+# returned in insertion order when queried with find() without an explicit sort.
+CAPPED_INSERTION_ORDER_TESTS: list[CommandTestCase] = [
+    CommandTestCase(
+        "order_descending_ids",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 5}, {"_id": 3}, {"_id": 1}, {"_id": 4}],
+        command=lambda ctx: {"find": ctx.collection, "projection": {"_id": 1}},
+        expected=[{"_id": 5}, {"_id": 3}, {"_id": 1}, {"_id": 4}],
+        msg="find() should return documents in insertion order regardless of _id values",
+    ),
+    CommandTestCase(
+        "order_mixed_type_ids",
+        target_collection=CappedCollection(size=100_000),
+        docs=[
+            {"_id": "str"},
+            {"_id": 42},
+            {"_id": None},
+            {"_id": True},
+        ],
+        command=lambda ctx: {"find": ctx.collection, "projection": {"_id": 1}},
+        expected=[
+            {"_id": "str"},
+            {"_id": 42},
+            {"_id": None},
+            {"_id": True},
+        ],
+        msg="find() should preserve insertion order with mixed _id types",
+    ),
+    CommandTestCase(
+        "order_varying_doc_sizes",
+        target_collection=CappedCollection(size=100_000),
+        docs=[
+            {"_id": 3, "data": "x"},
+            {"_id": 1, "data": "y" * 100},
+            {"_id": 2, "data": "z" * 10},
+        ],
+        command=lambda ctx: {"find": ctx.collection, "projection": {"_id": 1}},
+        expected=[{"_id": 3}, {"_id": 1}, {"_id": 2}],
+        msg="find() should preserve insertion order regardless of document sizes",
+    ),
+    CommandTestCase(
+        "filter_preserves_order",
+        target_collection=CappedCollection(size=100_000),
+        docs=[
+            {"_id": 5, "x": "a"},
+            {"_id": 2, "x": "b"},
+            {"_id": 8, "x": "a"},
+            {"_id": 1, "x": "c"},
+        ],
+        command=lambda ctx: {
+            "find": ctx.collection,
+            "filter": {"x": "a"},
+            "projection": {"_id": 1},
+        },
+        expected=[{"_id": 5}, {"_id": 8}],
+        msg="find() with a filter should return matching documents in insertion order",
+    ),
+]
+
+# Property [Natural Sort Order]: sort({"$natural": 1}) returns documents in
+# insertion order (oldest first), sort({"$natural": -1}) returns reverse
+# insertion order (most recent first), and limit restricts to the N most recent.
+CAPPED_NATURAL_SORT_TESTS: list[CommandTestCase] = [
+    CommandTestCase(
+        "forward_insertion_order",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        command=lambda ctx: {
+            "find": ctx.collection,
+            "projection": {"_id": 1},
+            "sort": {"$natural": 1},
+        },
+        expected=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        msg="$natural: 1 should return documents in insertion order",
+    ),
+    CommandTestCase(
+        "reverse_insertion_order",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        command=lambda ctx: {
+            "find": ctx.collection,
+            "projection": {"_id": 1},
+            "sort": {"$natural": -1},
+        },
+        expected=[{"_id": 2}, {"_id": 4}, {"_id": 1}, {"_id": 3}],
+        msg="$natural: -1 should return documents in reverse insertion order",
+    ),
+    CommandTestCase(
+        "reverse_limit_3",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}, {"_id": 5}],
+        command=lambda ctx: {
+            "find": ctx.collection,
+            "projection": {"_id": 1},
+            "sort": {"$natural": -1},
+            "limit": 3,
+        },
+        expected=[{"_id": 5}, {"_id": 2}, {"_id": 4}],
+        msg="$natural: -1 with limit 3 should return the 3 most recent documents",
+    ),
+]
+
+# Property [Natural Hint Order]: hint({"$natural": 1}) forces forward
+# collection scan order, hint({"$natural": -1}) forces reverse.
+CAPPED_HINT_TESTS: list[CommandTestCase] = [
+    CommandTestCase(
+        "hint_forward_order",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        command=lambda ctx: {
+            "find": ctx.collection,
+            "projection": {"_id": 1},
+            "hint": {"$natural": 1},
+        },
+        expected=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        msg="hint $natural:1 should return documents in forward insertion order",
+    ),
+    CommandTestCase(
+        "hint_reverse_order",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        command=lambda ctx: {
+            "find": ctx.collection,
+            "projection": {"_id": 1},
+            "hint": {"$natural": -1},
+        },
+        expected=[{"_id": 2}, {"_id": 4}, {"_id": 1}, {"_id": 3}],
+        msg="hint $natural:-1 should return documents in reverse insertion order",
+    ),
+]
+
+# Property [Index Hint Overrides Natural]: hinting a secondary index returns
+# documents in index order rather than insertion order.
+CAPPED_INDEX_HINT_TESTS: list[CommandTestCase] = [
+    CommandTestCase(
+        "index_hint_overrides_natural",
+        target_collection=CappedCollection(size=100_000),
+        indexes=[IndexModel("x")],
+        docs=[{"_id": 1, "x": 30}, {"_id": 2, "x": 10}, {"_id": 3, "x": 20}],
+        command=lambda ctx: {
+            "find": ctx.collection,
+            "filter": {"x": {"$gte": 10}},
+            "hint": "x_1",
+            "projection": {"_id": 1},
+        },
+        expected=[{"_id": 2}, {"_id": 3}, {"_id": 1}],
+        msg="Secondary index hint should return documents in index order, not insertion order",
+    ),
+]
+
+# Property [Aggregate Natural Order]: aggregate on a capped collection preserves
+# insertion order unless an explicit $sort or reverse hint is applied.
+CAPPED_AGG_TESTS: list[CommandTestCase] = [
+    CommandTestCase(
+        "agg_empty_pipeline",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        command=lambda ctx: {"aggregate": ctx.collection, "pipeline": [], "cursor": {}},
+        expected=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        msg="Empty aggregation pipeline should return documents in insertion order",
+    ),
+    CommandTestCase(
+        "agg_hint_natural_forward",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        command=lambda ctx: {
+            "aggregate": ctx.collection,
+            "pipeline": [],
+            "cursor": {},
+            "hint": {"$natural": 1},
+        },
+        expected=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        msg="hint $natural:1 should return documents in insertion order",
+    ),
+    CommandTestCase(
+        "agg_hint_natural_reverse",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 3}, {"_id": 1}, {"_id": 4}, {"_id": 2}],
+        command=lambda ctx: {
+            "aggregate": ctx.collection,
+            "pipeline": [],
+            "cursor": {},
+            "hint": {"$natural": -1},
+        },
+        expected=[{"_id": 2}, {"_id": 4}, {"_id": 1}, {"_id": 3}],
+        msg="hint $natural:-1 should return documents in reverse insertion order",
+    ),
+    CommandTestCase(
+        "agg_match_preserves_order",
+        target_collection=CappedCollection(size=100_000),
+        docs=[
+            {"_id": 5, "x": "a"},
+            {"_id": 2, "x": "b"},
+            {"_id": 8, "x": "a"},
+            {"_id": 1, "x": "c"},
+        ],
+        command=lambda ctx: {
+            "aggregate": ctx.collection,
+            "pipeline": [{"$match": {"x": "a"}}, {"$project": {"_id": 1}}],
+            "cursor": {},
+        },
+        expected=[{"_id": 5}, {"_id": 8}],
+        msg="Aggregate $match should return matching documents in insertion order",
+    ),
+]
+
+# Property [Explicit Sort Overrides Natural]: find() with an explicit sort
+# field or aggregate $sort overrides the default natural insertion order.
+CAPPED_SORT_OVERRIDE_TESTS: list[CommandTestCase] = [
+    CommandTestCase(
+        "find_explicit_sort_overrides",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 5}, {"_id": 2}, {"_id": 8}, {"_id": 1}],
+        command=lambda ctx: {
+            "find": ctx.collection,
+            "projection": {"_id": 1},
+            "sort": {"_id": 1},
+        },
+        expected=[{"_id": 1}, {"_id": 2}, {"_id": 5}, {"_id": 8}],
+        msg="Explicit sort should override natural insertion order",
+    ),
+    CommandTestCase(
+        "agg_sort_overrides",
+        target_collection=CappedCollection(size=100_000),
+        docs=[{"_id": 5}, {"_id": 2}, {"_id": 8}, {"_id": 1}],
+        command=lambda ctx: {
+            "aggregate": ctx.collection,
+            "pipeline": [{"$sort": {"_id": 1}}, {"$project": {"_id": 1}}],
+            "cursor": {},
+        },
+        expected=[{"_id": 1}, {"_id": 2}, {"_id": 5}, {"_id": 8}],
+        msg="Aggregate $sort should override natural insertion order",
+    ),
+]
+
+CAPPED_SINGLE_COMMAND_TESTS = (
+    CAPPED_INSERTION_ORDER_TESTS
+    + CAPPED_NATURAL_SORT_TESTS
+    + CAPPED_HINT_TESTS
+    + CAPPED_INDEX_HINT_TESTS
+    + CAPPED_AGG_TESTS
+    + CAPPED_SORT_OVERRIDE_TESTS
+)
+
+
+@pytest.mark.collection_mgmt
+@pytest.mark.parametrize("test", pytest_params(CAPPED_SINGLE_COMMAND_TESTS))
+def test_capped_single_command(database_client, collection, test):
+    """Test capped single-command behaviors."""
+    collection = test.prepare(database_client, collection)
+    ctx = CommandContext.from_collection(collection)
+    result = execute_command(collection, test.build_command(ctx))
+    expected = test.build_expected(ctx)
+    assertResult(
+        result,
+        expected=expected,
+        error_code=test.error_code,
+        msg=test.msg,
+        raw_res=not isinstance(expected, list),
+    )
+
+
+# Property [Batch Boundary Independence]: insertion order is preserved across
+# multiple insertMany calls.
+@pytest.mark.collection_mgmt
+def test_capped_batch_boundaries(database_client, collection):
+    """Test insertion order across batch boundaries."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 4}, {"_id": 2}])
+    coll.insert_many([{"_id": 3}, {"_id": 1}])
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 4}, {"_id": 2}, {"_id": 3}, {"_id": 1}],
+        msg="find() should preserve order across multiple insertMany calls",
+    )
+
+
+# Property [Natural Sort Equivalence]: find() without sort produces the same
+# order as sort({"$natural": 1}).
+@pytest.mark.collection_mgmt
+def test_capped_natural_sort_equivalence(database_client, collection):
+    """Test that find() without sort equals $natural: 1."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 5}, {"_id": 2}, {"_id": 8}, {"_id": 1}])
+    unsorted = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    sorted_result = execute_command(
+        coll, {"find": coll.name, "projection": {"_id": 1}, "sort": {"$natural": 1}}
+    )
+    assertSuccess(
+        sorted_result,
+        unsorted["cursor"]["firstBatch"],
+        msg="find() without sort should produce same order as $natural: 1",
+    )
diff --git a/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_write_position.py b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_write_position.py
new file mode 100644
index 00000000..3869312f
--- /dev/null
+++ b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_capped_collections_write_position.py
@@ -0,0 +1,171 @@
+"""Tests for capped collection write position behaviors."""
+
+from __future__ import annotations
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertSuccess
+from documentdb_tests.framework.executor import execute_command
+from documentdb_tests.framework.target_collection import CappedCollection
+
+
+# Property [Upsert Append]: upserts that create new documents append at the
+# end of natural order.
+@pytest.mark.collection_mgmt
+def test_capped_upsert_append(database_client, collection):
+    """Test that upserts append at end of natural order."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 5}, {"_id": 2}])
+    coll.update_one({"_id": 1}, {"$set": {"v": "new"}}, upsert=True)
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 5}, {"_id": 2}, {"_id": 1}],
+        msg="Upserted document should appear at the end of natural order",
+    )
+
+
+# Property [Update Position Preservation]: updates preserve a document's
+# position in $natural order.
+@pytest.mark.collection_mgmt
+def test_capped_update_position(database_client, collection):
+    """Test that updates preserve document position in natural order."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1, "x": "a"}, {"_id": 2, "x": "b"}, {"_id": 3, "x": "c"}])
+    execute_command(
+        coll, {"update": coll.name, "updates": [{"q": {"_id": 2}, "u": {"$set": {"x": "z" * 50}}}]}
+    )
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}, {"_id": 2}, {"_id": 3}],
+        msg="Updates should preserve document position in natural order",
+    )
+
+
+# Property [Upsert Match No Eviction]: upserts that match an existing document
+# act as normal updates with no eviction and no position change.
+@pytest.mark.collection_mgmt
+def test_capped_upsert_match(database_client, collection):
+    """Test that upserts matching existing docs preserve order."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1, "x": "a"}, {"_id": 2, "x": "b"}, {"_id": 3, "x": "c"}])
+    execute_command(
+        coll,
+        {
+            "update": coll.name,
+            "updates": [{"q": {"_id": 2}, "u": {"$set": {"x": "updated"}}, "upsert": True}],
+        },
+    )
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}, {"_id": 2}, {"_id": 3}],
+        msg="Upsert matching existing doc should not evict or change position",
+    )
+
+
+# Property [FindAndModify Update Position]: findAndModify with update preserves
+# the document's position in natural order.
+@pytest.mark.collection_mgmt
+def test_capped_find_and_modify_position(database_client, collection):
+    """Test that findAndModify update preserves document position."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1, "x": "a"}, {"_id": 2, "x": "b"}, {"_id": 3, "x": "c"}])
+    execute_command(
+        coll,
+        {
+            "findAndModify": coll.name,
+            "query": {"_id": 2},
+            "update": {"$set": {"x": "z" * 50}},
+        },
+    )
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}, {"_id": 2}, {"_id": 3}],
+        msg="findAndModify update should preserve document position in natural order",
+    )
+
+
+# Property [FindAndModify Remove Order]: findAndModify with remove preserves
+# relative insertion order of remaining documents.
+@pytest.mark.collection_mgmt
+def test_capped_find_and_modify_remove_order(database_client, collection):
+    """Test that findAndModify remove preserves relative order."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1}, {"_id": 2}, {"_id": 3}, {"_id": 4}])
+    execute_command(coll, {"findAndModify": coll.name, "query": {"_id": 2}, "remove": True})
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}, {"_id": 3}, {"_id": 4}],
+        msg="findAndModify remove should preserve relative insertion order",
+    )
+
+
+# Property [FindAndModify Upsert Append]: findAndModify with upsert that creates
+# a new document appends it at the end of natural order.
+@pytest.mark.collection_mgmt
+def test_capped_find_and_modify_upsert_append(database_client, collection):
+    """Test that findAndModify upsert appends at end of natural order."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1}, {"_id": 2}])
+    execute_command(
+        coll,
+        {
+            "findAndModify": coll.name,
+            "query": {"_id": 5},
+            "update": {"$set": {"x": "new"}},
+            "upsert": True,
+        },
+    )
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}, {"_id": 2}, {"_id": 5}],
+        msg="findAndModify upsert should append new document at end of natural order",
+    )
+
+
+# Property [Multi Update Position Preservation]: update with multi:true
+# preserves all affected documents' positions in natural order.
+@pytest.mark.collection_mgmt
+def test_capped_multi_update_position(database_client, collection):
+    """Test that multi-update preserves all document positions."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many(
+        [{"_id": 5, "x": 1}, {"_id": 3, "x": 2}, {"_id": 7, "x": 1}, {"_id": 1, "x": 1}]
+    )
+    execute_command(
+        coll,
+        {
+            "update": coll.name,
+            "updates": [{"q": {"x": 1}, "u": {"$set": {"x": 99}}, "multi": True}],
+        },
+    )
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 5}, {"_id": 3}, {"_id": 7}, {"_id": 1}],
+        msg="Multi-update should preserve all document positions in natural order",
+    )
+
+
+# Property [Replace Position Preservation]: a replacement-style update preserves
+# the document's position in natural order.
+@pytest.mark.collection_mgmt
+def test_capped_replace_position(database_client, collection):
+    """Test that replacement updates preserve document position."""
+    coll = CappedCollection(size=100_000).resolve(database_client, collection)
+    coll.insert_many([{"_id": 1, "x": "a"}, {"_id": 2, "x": "b"}, {"_id": 3, "x": "c"}])
+    execute_command(
+        coll,
+        {"update": coll.name, "updates": [{"q": {"_id": 2}, "u": {"_id": 2, "y": "replaced"}}]},
+    )
+    result = execute_command(coll, {"find": coll.name, "projection": {"_id": 1}})
+    assertSuccess(
+        result,
+        [{"_id": 1}, {"_id": 2}, {"_id": 3}],
+        msg="Replacement update should preserve document position in natural order",
+    )
diff --git a/documentdb_tests/compatibility/tests/core/collections/capped-collections/test_smoke_capped-collections.py b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_smoke_capped_collections.py
similarity index 90%
rename from documentdb_tests/compatibility/tests/core/collections/capped-collections/test_smoke_capped-collections.py
rename to documentdb_tests/compatibility/tests/core/collections/capped_collections/test_smoke_capped_collections.py
index 8cddb5f5..895f7869 100644
--- a/documentdb_tests/compatibility/tests/core/collections/capped-collections/test_smoke_capped-collections.py
+++ b/documentdb_tests/compatibility/tests/core/collections/capped_collections/test_smoke_capped_collections.py
@@ -1,5 +1,5 @@
 """
-Smoke test for capped-collections.
+Smoke test for capped collections.
 
 Tests basic capped collection functionality with size limit.
 """
@@ -15,7 +15,7 @@
 def test_smoke_capped_collections(collection):
     """Test basic capped collection creation."""
     result = execute_command(
-        collection, {"create": f"{collection.name}_capped", "capped": True, "size": 1024.0}
+        collection, {"create": f"{collection.name}_capped", "capped": True, "size": 1024}
     )
 
     expected = {"ok": 1.0}