Add top tests

documentdb_tests/compatibility/tests/system/diagnostic/commands/top/test_top_argument_handling.py

@@ -0,0 +1,157 @@
+"""Tests for top command argument handling.
+
+Validates that top accepts any BSON type as its argument value and
+accepts unrecognized fields.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+from bson import Binary, Code, Decimal128, Int64, MaxKey, MinKey, ObjectId, Regex, Timestamp
+
+from documentdb_tests.compatibility.tests.system.diagnostic.utils.diagnostic_test_case import (
+    DiagnosticTestCase,
+)
+from documentdb_tests.framework.assertions import assertProperties
+from documentdb_tests.framework.executor import execute_admin_command
+from documentdb_tests.framework.parametrize import pytest_params
+from documentdb_tests.framework.property_checks import Eq
+
+pytestmark = pytest.mark.admin
+
+# Property [BSON Type Acceptance]: top accepts any non-deprecated BSON type as command value.
+ARGUMENT_TYPE_TESTS: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        "int_1", command={"top": 1}, checks={"ok": Eq(1.0)}, msg="Should accept int 1"
+    ),
+    DiagnosticTestCase(
+        "int_0", command={"top": 0}, checks={"ok": Eq(1.0)}, msg="Should accept int 0"
+    ),
+    DiagnosticTestCase(
+        "int_neg1", command={"top": -1}, checks={"ok": Eq(1.0)}, msg="Should accept int -1"
+    ),
+    DiagnosticTestCase(
+        "bool_true", command={"top": True}, checks={"ok": Eq(1.0)}, msg="Should accept true"
+    ),
+    DiagnosticTestCase(
+        "bool_false",
+        command={"top": False},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept false",
+    ),
+    DiagnosticTestCase(
+        "string", command={"top": "hello"}, checks={"ok": Eq(1.0)}, msg="Should accept string"
+    ),
+    DiagnosticTestCase(
+        "null", command={"top": None}, checks={"ok": Eq(1.0)}, msg="Should accept null"
+    ),
+    DiagnosticTestCase(
+        "empty_object",
+        command={"top": {}},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept empty object",
+    ),
+    DiagnosticTestCase(
+        "empty_array",
+        command={"top": []},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept empty array",
+    ),
+    DiagnosticTestCase(
+        "double", command={"top": 1.5}, checks={"ok": Eq(1.0)}, msg="Should accept double"
+    ),
+    DiagnosticTestCase(
+        "int64", command={"top": Int64(1)}, checks={"ok": Eq(1.0)}, msg="Should accept int64"
+    ),
+    DiagnosticTestCase(
+        "decimal128",
+        command={"top": Decimal128("1")},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept decimal128",
+    ),
+    DiagnosticTestCase(
+        "decimal128_nan",
+        command={"top": Decimal128("NaN")},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept decimal128 NaN",
+    ),
+    DiagnosticTestCase(
+        "infinity",
+        command={"top": float("inf")},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept infinity",
+    ),
+    DiagnosticTestCase(
+        "date",
+        command={"top": datetime(2024, 1, 1, tzinfo=timezone.utc)},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept date",
+    ),
+    DiagnosticTestCase(
+        "binData",
+        command={"top": Binary(b"")},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept binData",
+    ),
+    DiagnosticTestCase(
+        "objectId",
+        command={"top": ObjectId()},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept objectId",
+    ),
+    DiagnosticTestCase(
+        "regex",
+        command={"top": Regex("test")},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept regex",
+    ),
+    DiagnosticTestCase(
+        "timestamp",
+        command={"top": Timestamp(0, 0)},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept timestamp",
+    ),
+    DiagnosticTestCase(
+        "minKey",
+        command={"top": MinKey()},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept minKey",
+    ),
+    DiagnosticTestCase(
+        "maxKey",
+        command={"top": MaxKey()},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept maxKey",
+    ),
+    DiagnosticTestCase(
+        "code",
+        command={"top": Code("function(){}")},
+        checks={"ok": Eq(1.0)},
+        msg="Should accept JavaScript code",
+    ),
+]
+
+# Property [Unrecognized Fields]: top accepts and ignores unrecognized fields.
+UNRECOGNIZED_FIELD_TESTS: list[DiagnosticTestCase] = [
+    DiagnosticTestCase(
+        "single_unrecognized_field",
+        command={"top": 1, "unknownField": 1},
+        checks={"ok": Eq(1.0)},
+        msg="top should accept a single unrecognized field",
+    ),
+    DiagnosticTestCase(
+        "multiple_unrecognized_fields",
+        command={"top": 1, "foo": 1, "bar": "baz", "qux": []},
+        checks={"ok": Eq(1.0)},
+        msg="top should accept multiple unrecognized fields",
+    ),
+]
+
+ARGUMENT_HANDLING_TESTS = ARGUMENT_TYPE_TESTS + UNRECOGNIZED_FIELD_TESTS
+
+
+@pytest.mark.parametrize("test", pytest_params(ARGUMENT_HANDLING_TESTS))
+def test_top_argument_handling(collection, test):
+    """Test that top accepts various BSON types and unrecognized fields."""
+    result = execute_admin_command(collection, test.command)
+    assertProperties(result, test.checks, msg=test.msg, raw_res=True)

documentdb_tests/compatibility/tests/system/diagnostic/commands/top/test_top_consistency.py

@@ -0,0 +1,156 @@
+"""Tests for top command consistency, visibility, and special collection types.
+
+Validates idempotency, namespace visibility, system namespace structure,
+and behavior with capped collections and views.
+
+Standalone functions are used throughout because every test requires runtime
+logic that DiagnosticTestCase cannot express: multi-call comparisons with
+dynamic thresholds, namespace key extraction from the response, conditional
+skips, or ad-hoc collection creation (capped, views).
+"""
+
+import pytest
+
+from documentdb_tests.framework.assertions import assertProperties, assertSuccessPartial
+from documentdb_tests.framework.executor import execute_admin_command
+from documentdb_tests.framework.property_checks import Exists, Gte, IsType
+
+pytestmark = pytest.mark.admin
+
+
+# Property [Idempotency]: repeated top calls succeed and counters are non-decreasing.
+
+
+def test_top_repeated_calls_return_ok(collection):
+    """Test that calling top returns ok after multiple calls."""
+    for _ in range(5):
+        execute_admin_command(collection, {"top": 1})
+    result = execute_admin_command(collection, {"top": 1})
+    assertSuccessPartial(result, {"ok": 1.0}, msg="top should succeed after repeated calls")
+
+
+def test_top_counters_non_decreasing_count(collection):
+    """Test that total.count is non-decreasing across two consecutive calls."""
+    collection.insert_one({"_id": 1})
+    ns = f"{collection.database.name}.{collection.name}"
+    result1 = execute_admin_command(collection, {"top": 1})
+    count1 = result1["totals"][ns]["total"]["count"]
+    result2 = execute_admin_command(collection, {"top": 1})
+    ns_data2 = result2["totals"][ns]
+    assertProperties(
+        ns_data2,
+        {"total.count": Gte(count1)},
+        msg="total.count should be non-decreasing",
+        raw_res=True,
+    )
+
+
+def test_top_counters_non_decreasing_time(collection):
+    """Test that total.time is non-decreasing across two consecutive calls."""
+    collection.insert_one({"_id": 1})
+    ns = f"{collection.database.name}.{collection.name}"
+    result1 = execute_admin_command(collection, {"top": 1})
+    time1 = result1["totals"][ns]["total"]["time"]
+    result2 = execute_admin_command(collection, {"top": 1})
+    ns_data2 = result2["totals"][ns]
+    assertProperties(
+        ns_data2,
+        {"total.time": Gte(time1)},
+        msg="total.time should be non-decreasing",
+        raw_res=True,
+    )
+
+
+# Property [Collection Visibility]: active collections appear in totals as db.collection keys.
+
+
+def test_top_newly_created_collection_appears(collection):
+    """Test that a newly created collection appears in top totals."""
+    collection.insert_one({"_id": 1})
+    result = execute_admin_command(collection, {"top": 1})
+    ns = f"{collection.database.name}.{collection.name}"
+    ns_data = result["totals"].get(ns)
+    assertProperties(
+        {"ns_entry": ns_data},
+        {"ns_entry": Exists()},
+        msg=f"Namespace {ns} should appear in top totals",
+        raw_res=True,
+    )
+
+
+def test_top_multiple_collections_appear(collection):
+    """Test that multiple collections appear in top totals."""
+    db = collection.database
+    coll1 = db.create_collection(f"{collection.name}_multi1")
+    coll2 = db.create_collection(f"{collection.name}_multi2")
+    coll1.insert_one({"_id": 1})
+    coll2.insert_one({"_id": 1})
+    result = execute_admin_command(coll1, {"top": 1})
+    ns1 = f"{db.name}.{coll1.name}"
+    ns2 = f"{db.name}.{coll2.name}"
+    assertProperties(
+        {"ns1": result["totals"].get(ns1), "ns2": result["totals"].get(ns2)},
+        {"ns1": Exists(), "ns2": Exists()},
+        msg="Both namespaces should appear in top totals",
+        raw_res=True,
+    )
+
+
+# Property [System Collections]: system namespaces have the standard event field structure.
+
+
+def test_top_system_collections_have_event_structure(collection):
+    """Test that a system namespace in totals has the expected event field structure."""
+    collection.insert_one({"_id": 1})
+    result = execute_admin_command(collection, {"top": 1})
+    system_ns = None
+    for ns_key in result["totals"]:
+        if ".system." in ns_key or ns_key.startswith("admin.") or ns_key.startswith("local."):
+            system_ns = ns_key
+            break
+    if system_ns is None:
+        pytest.skip("No system namespace found in top totals")
+    ns_data = result["totals"][system_ns]
+    assertProperties(
+        ns_data,
+        {"total": IsType("object"), "total.time": Gte(0), "total.count": Gte(0)},
+        msg=f"System namespace {system_ns} should have event fields with time/count",
+        raw_res=True,
+    )
+
+
+# Property [Special Collection Types]: capped collections and views are handled by top.
+
+
+def test_top_tracks_capped_collection(collection):
+    """Test that a capped collection appears in top totals with expected structure."""
+    db = collection.database
+    coll = db.create_collection(f"{collection.name}_capped", capped=True, size=4096)
+    coll.insert_one({"_id": 1})
+    result = execute_admin_command(coll, {"top": 1})
+    ns = f"{db.name}.{coll.name}"
+    ns_data = result["totals"][ns]
+    assertProperties(
+        ns_data,
+        {"total": IsType("object"), "total.time": Gte(0), "total.count": Gte(0)},
+        msg="Capped collection should appear in top totals with expected structure",
+        raw_res=True,
+    )
+
+
+def test_top_tracks_view(collection):
+    """Test that a view namespace appears in top totals."""
+    db = collection.database
+    source_coll = db.create_collection(f"{collection.name}_view_src")
+    source_coll.insert_one({"_id": 1})
+    view_name = f"{collection.name}_view"
+    db.command("create", view_name, viewOn=source_coll.name, pipeline=[])
+    result = execute_admin_command(source_coll, {"top": 1})
+    view_ns = f"{db.name}.{view_name}"
+    view_data = result["totals"].get(view_ns)
+    assertProperties(
+        {"ns_entry": view_data},
+        {"ns_entry": Exists()},
+        msg=f"View namespace {view_ns} should appear in top totals",
+        raw_res=True,
+    )