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
46 changes: 45 additions & 1 deletion conformance/src/trace_tests/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,44 @@ def _run_core(
return results


_VALID_FORMATS = frozenset({"trace", "cmcp-runtime"})


def _validate_runner_inputs(
record: Any,
fmt: Any,
max_age_seconds: Any,
expected_nonce: Any,
receipt: Any,
entry: Any,
policy_resolver: Any,
) -> None:
if not isinstance(record, dict):
raise ValueError(f"record must be a JSON object, got {type(record).__name__}")
if not isinstance(fmt, str) or fmt not in _VALID_FORMATS:
raise ValueError(
f"Unknown record format {fmt!r}; valid: 'trace', 'cmcp-runtime'"
)
if type(max_age_seconds) is not int or max_age_seconds < 1:
raise ValueError(
f"max_age_seconds must be a positive integer, got {max_age_seconds!r}"
)
if expected_nonce is not None and not isinstance(expected_nonce, str):
raise ValueError(
f"expected_nonce must be a string or None, got {type(expected_nonce).__name__}"
)
if receipt is not None and not isinstance(receipt, dict):
raise ValueError(
f"receipt must be a JSON object or None, got {type(receipt).__name__}"
)
if entry is not None and not isinstance(entry, dict):
raise ValueError(
f"entry must be a JSON object or None, got {type(entry).__name__}"
)
if policy_resolver is not None and not callable(policy_resolver):
raise ValueError("policy_resolver must be callable or None")


def run(
record: dict[str, Any],
fmt: str,
Expand All @@ -125,8 +163,11 @@ def run(
expected_policy_hash: str | None = None,
) -> dict[str, list[Finding]]:
"""Run all modules required for *level* and return findings keyed by module ID."""
if level not in _LEVEL_MODULES:
if type(level) is not int or level not in _LEVEL_MODULES:
raise ValueError(f"Unknown conformance level {level!r}; valid: 0, 1, 2")
_validate_runner_inputs(
record, fmt, max_age_seconds, expected_nonce, receipt, entry, policy_resolver
)
tr_pol.validate_expected_policy_hash(expected_policy_hash)
with accounting._without_capture():
trace = extract_trace(record, fmt)
Expand Down Expand Up @@ -165,6 +206,9 @@ def _run_levels(
or any(type(level) is not int for level in attempted)
):
raise ValueError("attempted levels must be exactly (0,), (0, 1), or (0, 1, 2)")
_validate_runner_inputs(
record, fmt, max_age_seconds, expected_nonce, receipt, entry, policy_resolver
)
tr_pol.validate_expected_policy_hash(expected_policy_hash)

planned = tuple(
Expand Down
57 changes: 57 additions & 0 deletions conformance/tests/unit/test_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,3 +97,60 @@ def test_cmcp_runtime_vector_signature_verifies():
def test_invalid_level_raises():
with pytest.raises(ValueError, match="Unknown conformance level"):
run({}, "trace", level=99)


@pytest.mark.parametrize("bad_level", [True, False, 1.0, "0", None, -1])
def test_boolean_or_non_int_level_raises(bad_level):
with pytest.raises(ValueError, match="Unknown conformance level"):
run({}, "trace", level=bad_level)


@pytest.mark.parametrize("bad_record", [None, [], "not-a-dict", 123])
def test_non_dict_record_raises(bad_record):
from trace_tests.runner import _run_levels

with pytest.raises(ValueError, match="record must be a JSON object"):
run(bad_record, "trace", level=0)
with pytest.raises(ValueError, match="record must be a JSON object"):
_run_levels(bad_record, "trace", (0,))


@pytest.mark.parametrize("bad_fmt", ["", "json", "plain", None, 123, [], {}])
def test_invalid_format_raises(bad_fmt):
from trace_tests.runner import _run_levels

with pytest.raises(ValueError, match="Unknown record format"):
run({}, bad_fmt, level=0)
with pytest.raises(ValueError, match="Unknown record format"):
_run_levels({}, bad_fmt, (0,))


@pytest.mark.parametrize("bad_max_age", [True, False, 0, -1, 86400.0, "86400", None])
def test_invalid_max_age_seconds_raises(bad_max_age):
from trace_tests.runner import _run_levels

with pytest.raises(ValueError, match="max_age_seconds must be a positive integer"):
run({}, "trace", level=0, max_age_seconds=bad_max_age)
with pytest.raises(ValueError, match="max_age_seconds must be a positive integer"):
_run_levels({}, "trace", (0,), max_age_seconds=bad_max_age)


def test_invalid_optional_runner_arguments_raise():
from trace_tests.runner import _run_levels

with pytest.raises(ValueError, match="expected_nonce must be a string"):
run({}, "trace", level=0, expected_nonce=123) # type: ignore[arg-type]
with pytest.raises(ValueError, match="expected_nonce must be a string"):
_run_levels({}, "trace", (0,), expected_nonce=123) # type: ignore[arg-type]
with pytest.raises(ValueError, match="receipt must be a JSON object"):
run({}, "trace", level=0, receipt=[]) # type: ignore[arg-type]
with pytest.raises(ValueError, match="receipt must be a JSON object"):
_run_levels({}, "trace", (0,), receipt=[]) # type: ignore[arg-type]
with pytest.raises(ValueError, match="entry must be a JSON object"):
run({}, "trace", level=0, entry="bad") # type: ignore[arg-type]
with pytest.raises(ValueError, match="entry must be a JSON object"):
_run_levels({}, "trace", (0,), entry="bad") # type: ignore[arg-type]
with pytest.raises(ValueError, match="policy_resolver must be callable"):
run({}, "trace", level=0, policy_resolver="not-callable") # type: ignore[arg-type]
with pytest.raises(ValueError, match="policy_resolver must be callable"):
_run_levels({}, "trace", (0,), policy_resolver=123) # type: ignore[arg-type]
64 changes: 64 additions & 0 deletions tests/test_conformance_runner_option_boundaries.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
"""Behavioral boundary checks for conformance runner entry validation (#626)."""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

import pytest

_CONFORMANCE_SRC = str(Path(__file__).resolve().parents[1] / "conformance" / "src")
if sys.path[:1] != [_CONFORMANCE_SRC]:
sys.path.insert(0, _CONFORMANCE_SRC)

from trace_tests.runner import _run_levels, run # type: ignore[import-untyped] # noqa: E402


@pytest.mark.parametrize("bad_level", [True, False, 1.0, "0", None, -1])
def test_run_rejects_boolean_and_non_int_level(bad_level: Any) -> None:
with pytest.raises(ValueError, match="Unknown conformance level"):
run({}, "trace", level=bad_level)


@pytest.mark.parametrize("bad_record", [None, [], "not-a-dict", 123])
def test_runner_rejects_non_dict_record(bad_record: Any) -> None:
with pytest.raises(ValueError, match="record must be a JSON object"):
run(bad_record, "trace", level=0)
with pytest.raises(ValueError, match="record must be a JSON object"):
_run_levels(bad_record, "trace", (0,))


@pytest.mark.parametrize("bad_fmt", ["", "json", "plain", None, 123, [], {}])
def test_runner_rejects_unknown_format(bad_fmt: Any) -> None:
with pytest.raises(ValueError, match="Unknown record format"):
run({}, bad_fmt, level=0)
with pytest.raises(ValueError, match="Unknown record format"):
_run_levels({}, bad_fmt, (0,))


@pytest.mark.parametrize("bad_max_age", [True, False, 0, -1, 86400.0, "86400", None])
def test_runner_rejects_invalid_max_age_seconds(bad_max_age: Any) -> None:
with pytest.raises(ValueError, match="max_age_seconds must be a positive integer"):
run({}, "trace", level=0, max_age_seconds=bad_max_age)
with pytest.raises(ValueError, match="max_age_seconds must be a positive integer"):
_run_levels({}, "trace", (0,), max_age_seconds=bad_max_age)


def test_runner_rejects_invalid_optional_arguments() -> None:
with pytest.raises(ValueError, match="expected_nonce must be a string"):
run({}, "trace", level=0, expected_nonce=123) # type: ignore[arg-type]
with pytest.raises(ValueError, match="expected_nonce must be a string"):
_run_levels({}, "trace", (0,), expected_nonce=123) # type: ignore[arg-type]
with pytest.raises(ValueError, match="receipt must be a JSON object"):
run({}, "trace", level=0, receipt=[]) # type: ignore[arg-type]
with pytest.raises(ValueError, match="receipt must be a JSON object"):
_run_levels({}, "trace", (0,), receipt=[]) # type: ignore[arg-type]
with pytest.raises(ValueError, match="entry must be a JSON object"):
run({}, "trace", level=0, entry="bad") # type: ignore[arg-type]
with pytest.raises(ValueError, match="entry must be a JSON object"):
_run_levels({}, "trace", (0,), entry="bad") # type: ignore[arg-type]
with pytest.raises(ValueError, match="policy_resolver must be callable"):
run({}, "trace", level=0, policy_resolver="not-callable") # type: ignore[arg-type]
with pytest.raises(ValueError, match="policy_resolver must be callable"):
_run_levels({}, "trace", (0,), policy_resolver="not-callable") # type: ignore[arg-type]
Loading