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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,230 @@
"""Tests for whatsmyuri command argument handling.

Validates that whatsmyuri accepts any BSON type as its argument value
and ignores 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 [Type Acceptance]: whatsmyuri accepts all BSON types as the command field value.
ARGUMENT_TYPE_TESTS: list[DiagnosticTestCase] = [
DiagnosticTestCase(
"int_1",
command={"whatsmyuri": 1},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept int 1",
),
DiagnosticTestCase(
"int_0",
command={"whatsmyuri": 0},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept int 0",
),
DiagnosticTestCase(
"int_neg1",
command={"whatsmyuri": -1},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept int -1",
),
DiagnosticTestCase(
"bool_true",
command={"whatsmyuri": True},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept true",
),
DiagnosticTestCase(
"bool_false",
command={"whatsmyuri": False},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept false",
),
DiagnosticTestCase(
"string",
command={"whatsmyuri": "hello"},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept string",
),
DiagnosticTestCase(
"empty_string",
command={"whatsmyuri": ""},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept empty string",
),
DiagnosticTestCase(
"null",
command={"whatsmyuri": None},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept null",
),
DiagnosticTestCase(
"empty_object",
command={"whatsmyuri": {}},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept empty object",
),
DiagnosticTestCase(
"nested_object",
command={"whatsmyuri": {"a": {"b": 1}}},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept nested object",
),
DiagnosticTestCase(
"empty_array",
command={"whatsmyuri": []},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept empty array",
),
DiagnosticTestCase(
"array_with_elements",
command={"whatsmyuri": [1, 2, 3]},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept array with elements",
),
DiagnosticTestCase(
"double",
command={"whatsmyuri": 1.5},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept double",
),
DiagnosticTestCase(
"negative_double",
command={"whatsmyuri": -1.5},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept negative double",
),
DiagnosticTestCase(
"large_int",
command={"whatsmyuri": 999_999_999},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept large int",
),
DiagnosticTestCase(
"int64",
command={"whatsmyuri": Int64(1)},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept int64",
),
DiagnosticTestCase(
"decimal128",
command={"whatsmyuri": Decimal128("1")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept decimal128",
),
DiagnosticTestCase(
"decimal128_nan",
command={"whatsmyuri": Decimal128("NaN")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept decimal128 NaN",
),
DiagnosticTestCase(
"decimal128_infinity",
command={"whatsmyuri": Decimal128("Infinity")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept decimal128 Infinity",
),
DiagnosticTestCase(
"decimal128_neg_zero",
command={"whatsmyuri": Decimal128("-0")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept decimal128 negative zero",
),
DiagnosticTestCase(
"infinity",
command={"whatsmyuri": float("inf")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept infinity",
),
DiagnosticTestCase(
"neg_infinity",
command={"whatsmyuri": float("-inf")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept negative infinity",
),
DiagnosticTestCase(
"nan",
command={"whatsmyuri": float("nan")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept NaN",
),
DiagnosticTestCase(
"date",
command={"whatsmyuri": datetime(2024, 1, 1, tzinfo=timezone.utc)},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept date",
),
DiagnosticTestCase(
"binData",
command={"whatsmyuri": Binary(b"")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept binData",
),
DiagnosticTestCase(
"objectId",
command={"whatsmyuri": ObjectId()},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept objectId",
),
DiagnosticTestCase(
"regex",
command={"whatsmyuri": Regex("test")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept regex",
),
DiagnosticTestCase(
"timestamp",
command={"whatsmyuri": Timestamp(0, 0)},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept timestamp",
),
DiagnosticTestCase(
"minKey",
command={"whatsmyuri": MinKey()},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept minKey",
),
DiagnosticTestCase(
"maxKey",
command={"whatsmyuri": MaxKey()},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept maxKey",
),
DiagnosticTestCase(
"code",
command={"whatsmyuri": Code("function(){}")},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should accept JavaScript code",
),
]

# Property [Extra Fields Ignored]: whatsmyuri ignores unrecognized fields.
EXTRA_FIELD_TESTS: list[DiagnosticTestCase] = [
DiagnosticTestCase(
"extra_field_ignored",
command={"whatsmyuri": 1, "unknownField": 1},
checks={"ok": Eq(1.0)},
msg="whatsmyuri should succeed even with unrecognized fields",
),
]

ALL_TESTS = ARGUMENT_TYPE_TESTS + EXTRA_FIELD_TESTS


@pytest.mark.parametrize("test", pytest_params(ALL_TESTS))
def test_whatsmyuri_argument_handling(collection, test):
"""Test whatsmyuri argument handling."""
result = execute_admin_command(collection, test.command)
assertProperties(result, test.checks, msg=test.msg, raw_res=True)
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
"""Tests for whatsmyuri command consistency and database independence.

Validates that whatsmyuri returns consistent results across calls,
databases, and is unaffected by server settings.
"""

import pytest

from documentdb_tests.compatibility.tests.system.diagnostic.utils.diagnostic_test_case import (
DiagnosticTestCase,
)
from documentdb_tests.framework.assertions import (
assertProperties,
assertSuccess,
assertSuccessPartial,
)
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

pytestmark = pytest.mark.admin


# Property [Database Independence]: whatsmyuri succeeds on any database.
DATABASE_INDEPENDENCE_TESTS: list[DiagnosticTestCase] = [
DiagnosticTestCase(
"any_database",
command={"whatsmyuri": 1},
use_admin=False,
checks={"ok": Eq(1.0)},
msg="whatsmyuri should succeed on non-admin database",
),
]


@pytest.mark.parametrize("test", pytest_params(DATABASE_INDEPENDENCE_TESTS))
def test_whatsmyuri_consistency(collection, test):
"""Test whatsmyuri consistency."""
result = execute_command(collection, test.command)
assertProperties(result, test.checks, msg=test.msg, raw_res=True)


def test_whatsmyuri_idempotent(collection):
"""Test whatsmyuri idempotency."""
result1 = execute_admin_command(collection, {"whatsmyuri": 1})
result2 = execute_admin_command(collection, {"whatsmyuri": 1})
assertSuccess(
result2,
expected=result1,
msg="whatsmyuri should return identical results across calls",
raw_res=True,
)


def test_whatsmyuri_same_result_any_database(collection):
"""Test whatsmyuri returns same result from admin and non-admin database."""
admin_result = execute_admin_command(collection, {"whatsmyuri": 1})
db_result = execute_command(collection, {"whatsmyuri": 1})
assertSuccess(
db_result,
expected=admin_result,
msg="whatsmyuri should return same result from any database",
raw_res=True,
)


def test_whatsmyuri_nonexistent_database(collection):
"""Test whatsmyuri on a non-existent database."""
other_db = f"{collection.name}_nonexistent_db"
other_col = collection.database.client[other_db][collection.name]
result = execute_command(other_col, {"whatsmyuri": 1})
assertSuccessPartial(
result,
{"ok": 1.0},
msg="whatsmyuri should succeed on non-existent database",
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
"""Tests for whatsmyuri command error conditions.

Validates that invalid usages of whatsmyuri produce appropriate errors.
Uses CommandTestCase because the aggregation stage test needs ctx.collection
for the aggregate command.
"""

import pytest

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

pytestmark = pytest.mark.admin


# Property [Case Sensitivity]: whatsmyuri is case-sensitive and rejects mismatched casing.
CASE_SENSITIVITY_TESTS: list[CommandTestCase] = [
CommandTestCase(
"case_sensitive_capital_w",
command={"WhatsMyUri": 1},
error_code=COMMAND_NOT_FOUND_ERROR,
msg="whatsmyuri should reject camel-cased command name",
),
CommandTestCase(
"case_sensitive_all_upper",
command={"WHATSMYURI": 1},
error_code=COMMAND_NOT_FOUND_ERROR,
msg="whatsmyuri should reject all-uppercase command name",
),
]

# Property [Not a Pipeline Stage]: whatsmyuri is not usable as an aggregation stage.
PIPELINE_STAGE_TESTS: list[CommandTestCase] = [
CommandTestCase(
"as_aggregation_stage",
command=lambda ctx: {
"aggregate": ctx.collection,
"pipeline": [{"$whatsmyuri": {}}],
"cursor": {},
},
error_code=UNKNOWN_PIPELINE_STAGE_ERROR,
msg="whatsmyuri should not be usable as an aggregation stage",
),
]

ALL_TESTS = CASE_SENSITIVITY_TESTS + PIPELINE_STAGE_TESTS


@pytest.mark.parametrize("test", pytest_params(ALL_TESTS))
def test_whatsmyuri_error_conditions(collection, test):
"""Test whatsmyuri error conditions."""
ctx = CommandContext.from_collection(collection)
cmd = test.build_command(ctx)
# Case sensitivity tests target the admin db; the aggregate test does not.
if next(iter(cmd)).lower() == "whatsmyuri":
result = execute_admin_command(collection, cmd)
else:
result = execute_command(collection, cmd)
assertResult(result, error_code=test.error_code, msg=test.msg)
Loading
Loading