From 9f56db8438cf6190c675530d23ecf5dbce9610f3 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Fri, 24 Jul 2026 03:47:38 -0400 Subject: [PATCH] test: refine replacement property strategy --- tests/test_properties.py | 321 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 320 insertions(+), 1 deletion(-) diff --git a/tests/test_properties.py b/tests/test_properties.py index 59939e0..ace3633 100644 --- a/tests/test_properties.py +++ b/tests/test_properties.py @@ -1,12 +1,15 @@ +import json import re +from copy import deepcopy from unicodedata import normalize as unicode_normalize from hypothesis import assume, given from hypothesis import strategies as st from context_compiler import create_engine +from context_compiler.controller import preview, state_diff from context_compiler.engine import _CANONICAL_DIRECTIVE_STARTS, State -from context_compiler.grammar import validate_directive +from context_compiler.grammar import DirectiveKind, render_directive, validate_directive def _run_sequence(inputs: list[str]) -> State: @@ -45,11 +48,235 @@ def _contains_canonical_start_fragment(value: str) -> bool: return False +def _sanitize_premise_like_engine(value: str) -> str: + sanitized = unicode_normalize("NFKC", value) + sanitized = sanitized.replace("’", "'").replace("`", "'") + return re.sub(r"\s+", " ", sanitized).strip() + + +NORMALIZATION_SENSITIVE_TEXT = st.text( + alphabet=st.characters( + blacklist_categories=("Cs",), + blacklist_characters="\x00", + ) + | st.sampled_from([" ", "\t", "’", "`"]), + min_size=1, + max_size=20, +) + +POLICY_VALUE = st.sampled_from(["use", "prohibit"]) + +VALID_STATE_PAYLOADS = st.builds( + lambda premise, pairs: { + "premise": premise, + "policies": dict(pairs), + "version": 2, + }, + premise=st.one_of(st.none(), NORMALIZATION_SENSITIVE_TEXT), + pairs=st.lists( + st.tuples(NORMALIZATION_SENSITIVE_TEXT, POLICY_VALUE), + min_size=0, + max_size=8, + ), +).filter(lambda payload: all(_normalize_item_like_engine(key) != "" for key in payload["policies"])) + + +VALID_NONEMPTY_ITEM_TEXT = NORMALIZATION_SENSITIVE_TEXT.filter( + lambda value: ( + _normalize_item_like_engine(value) != "" + and not _contains_canonical_start_fragment(value) + and " instead of " not in value + and not value.startswith("instead of ") + and not value.endswith(" instead of") + ) +) + +VALID_USE_ITEM_TEXT = VALID_NONEMPTY_ITEM_TEXT.filter( + lambda value: validate_directive(f"use {value}") is not None +) + +VALID_PROHIBIT_ITEM_TEXT = VALID_NONEMPTY_ITEM_TEXT.filter( + lambda value: validate_directive(f"prohibit {value}") is not None +) + +VALID_PREMISE_TEXT = NORMALIZATION_SENSITIVE_TEXT.filter( + lambda value: _sanitize_premise_like_engine(value) != "" +) + +CANONICAL_GRAMMAR_PREMISE_TEXT = NORMALIZATION_SENSITIVE_TEXT.map( + _sanitize_premise_like_engine +).filter( + lambda value: ( + value != "" + and validate_directive(f"set premise {value}") is not None + and validate_directive(f"change premise to {value}") is not None + ) +) + +CANONICAL_GRAMMAR_ITEM_TEXT = NORMALIZATION_SENSITIVE_TEXT.map(_normalize_item_like_engine).filter( + lambda value: ( + value != "" + and validate_directive(f"use {value}") is not None + and validate_directive(f"prohibit {value}") is not None + and validate_directive(f"remove policy {value}") is not None + ) +) + +PREVIEW_INPUTS = st.one_of( + st.text(max_size=40), + VALID_USE_ITEM_TEXT.map(lambda item: f"use {item}"), + VALID_PROHIBIT_ITEM_TEXT.map(lambda item: f"prohibit {item}"), + VALID_NONEMPTY_ITEM_TEXT.map(lambda item: f"remove policy {item}").filter( + lambda text: validate_directive(text) is not None + ), + VALID_PREMISE_TEXT.map(lambda value: f"set premise {value}").filter( + lambda text: validate_directive(text) is not None + ), + VALID_PREMISE_TEXT.map(lambda value: f"change premise to {value}").filter( + lambda text: validate_directive(text) is not None + ), + st.sampled_from(["clear premise", "reset policies", "clear state"]), + st.tuples(VALID_USE_ITEM_TEXT, VALID_NONEMPTY_ITEM_TEXT) + .filter( + lambda pair: _normalize_item_like_engine(pair[0]) != _normalize_item_like_engine(pair[1]) + ) + .map(lambda pair: f"use {pair[0]} instead of {pair[1]}") + .filter(lambda text: validate_directive(text) is not None), +) + + +def _build_deterministic_replacement_case( + premise: str | None, + unrelated_pairs: list[tuple[str, str]], + new_item: str, + old_item: str, + old_present: bool, + new_present: bool, +) -> dict[str, object]: + new_key = _normalize_item_like_engine(new_item) + old_key = _normalize_item_like_engine(old_item) + + policies = {key: value for key, value in unrelated_pairs if key not in {new_key, old_key}} + if old_present: + policies[old_key] = "use" + if new_present: + policies[new_key] = "use" + + return { + "initial_state": { + "premise": premise, + "policies": dict(sorted(policies.items())), + "version": 2, + }, + "new_item": new_item, + "old_item": old_item, + "old_present": old_present, + } + + +DETERMINISTIC_REPLACEMENT_CASES = ( + st.tuples( + st.one_of(st.none(), VALID_PREMISE_TEXT.map(_sanitize_premise_like_engine)), + st.lists( + st.tuples(CANONICAL_GRAMMAR_ITEM_TEXT, POLICY_VALUE), + min_size=0, + max_size=6, + ), + VALID_USE_ITEM_TEXT, + VALID_NONEMPTY_ITEM_TEXT, + st.booleans(), + st.booleans(), + ) + .filter( + lambda args: ( + _normalize_item_like_engine(args[2]) != _normalize_item_like_engine(args[3]) + and validate_directive(f"use {args[2]} instead of {args[3]}") is not None + and not any( + key in {_normalize_item_like_engine(args[2]), _normalize_item_like_engine(args[3])} + and value == "prohibit" + for key, value in args[1] + ) + ) + ) + .map(lambda args: _build_deterministic_replacement_case(*args)) +) + + +def _payload_has_stable_export_import_cycle(payload: dict[str, object]) -> bool: + engine = create_engine() + engine.import_json(json.dumps(payload)) + exported = engine.export_json() + + restored = create_engine() + try: + restored.import_json(exported) + except ValueError: + return False + + return restored.state == engine.state + + +GRAMMAR_RENDER_CASES = st.one_of( + CANONICAL_GRAMMAR_PREMISE_TEXT.map( + lambda value: {"kind": DirectiveKind.SET_PREMISE, "operands": {"value": value}} + ), + CANONICAL_GRAMMAR_PREMISE_TEXT.map( + lambda value: {"kind": DirectiveKind.CHANGE_PREMISE, "operands": {"value": value}} + ), + CANONICAL_GRAMMAR_ITEM_TEXT.map( + lambda item: {"kind": DirectiveKind.USE_ITEM, "operands": {"item": item}} + ), + CANONICAL_GRAMMAR_ITEM_TEXT.map( + lambda item: {"kind": DirectiveKind.PROHIBIT_ITEM, "operands": {"item": item}} + ), + CANONICAL_GRAMMAR_ITEM_TEXT.map( + lambda item: {"kind": DirectiveKind.REMOVE_POLICY, "operands": {"item": item}} + ), + st.tuples(CANONICAL_GRAMMAR_ITEM_TEXT, CANONICAL_GRAMMAR_ITEM_TEXT) + .filter( + lambda pair: _normalize_item_like_engine(pair[0]) != _normalize_item_like_engine(pair[1]) + ) + .map( + lambda pair: { + "kind": DirectiveKind.REPLACE_USE, + "operands": {"new_item": pair[0], "old_item": pair[1]}, + } + ), + st.sampled_from( + [ + {"kind": DirectiveKind.CLEAR_PREMISE, "operands": {}}, + {"kind": DirectiveKind.RESET_POLICIES, "operands": {}}, + {"kind": DirectiveKind.CLEAR_STATE, "operands": {}}, + ] + ), +) + + @given(st.lists(st.text(max_size=40), min_size=0, max_size=20)) def test_determinism_same_input_sequence_same_state(inputs: list[str]) -> None: assert _run_sequence(inputs) == _run_sequence(inputs) +@given(GRAMMAR_RENDER_CASES) +def test_grammar_helper_render_validate_round_trip_is_stable( + case: dict[str, DirectiveKind | dict[str, str]], +) -> None: + kind = case["kind"] + operands = case["operands"] + + assert isinstance(kind, DirectiveKind) + assert isinstance(operands, dict) + + rendered = render_directive(kind, **operands) + validated = validate_directive(rendered) + + assert validated is not None + assert validated.kind is kind + assert validated.text == rendered + assert validate_directive(validated.text) == validated + assert render_directive(kind, **operands) == rendered + + @given(st.text(min_size=1, max_size=30)) def test_idempotent_use_item_is_update_and_stable_state(item: str) -> None: assume(" instead of " not in item) @@ -177,3 +404,95 @@ def test_contradiction_prohibit_after_use_always_clarifies(item: str) -> None: decision = engine.step(f"prohibit {item}") assert decision["kind"] == "clarify" assert engine.state == before + + +@given(VALID_STATE_PAYLOADS) +def test_export_import_round_trip_preserves_authoritative_state_for_generated_payloads( + payload: dict[str, object], +) -> None: + source = create_engine() + source.import_json(json.dumps(payload)) + canonical_state = source.state + + target = create_engine() + target.import_json(source.export_json()) + + assert target.state == canonical_state + + +@given(VALID_STATE_PAYLOADS, st.integers(min_value=1, max_value=5)) +def test_repeated_export_import_cycles_remain_stable( + payload: dict[str, object], cycles: int +) -> None: + engine = create_engine() + engine.import_json(json.dumps(payload)) + + expected_state = engine.state + expected_json = engine.export_json() + + for _ in range(cycles): + next_engine = create_engine() + next_engine.import_json(expected_json) + assert next_engine.state == expected_state + assert next_engine.export_json() == expected_json + expected_state = next_engine.state + expected_json = next_engine.export_json() + + +@given(DETERMINISTIC_REPLACEMENT_CASES) +def test_deterministic_replacement_matches_equivalent_explicit_transition( + case: dict[str, object], +) -> None: + initial_state = case["initial_state"] + new_item = case["new_item"] + old_item = case["old_item"] + old_present = case["old_present"] + + assert isinstance(initial_state, dict) + assert isinstance(new_item, str) + assert isinstance(old_item, str) + assert isinstance(old_present, bool) + + oracle_engine = create_engine(state=deepcopy(initial_state)) + oracle_engine.step(f"remove policy {old_item}") + expected_decision = oracle_engine.step(f"use {new_item}") + expected_state = oracle_engine.state + + engine = create_engine(state=initial_state) + decision = engine.step(f"use {new_item} instead of {old_item}") + + assert expected_decision == {"kind": "update", "state": expected_state, "prompt_to_user": None} + assert decision == expected_decision + assert engine.state == expected_state + assert engine.has_pending_clarification() is False + + if not old_present: + followup = engine.step("yes") + assert followup == {"kind": "passthrough", "state": None, "prompt_to_user": None} + assert engine.has_pending_clarification() is False + assert engine.state == expected_state + + +@given(VALID_STATE_PAYLOADS.filter(_payload_has_stable_export_import_cycle), PREVIEW_INPUTS) +def test_preview_matches_isolated_execution_without_mutating_live_engine( + payload: dict[str, object], user_input: str +) -> None: + live_engine = create_engine() + live_engine.import_json(json.dumps(payload)) + before = deepcopy(live_engine.state) + + preview_result = preview(live_engine, user_input) + + isolated_engine = create_engine() + isolated_engine.import_json(json.dumps(before)) + isolated_decision = isolated_engine.step(user_input) + isolated_after = isolated_engine.state + isolated_diff = state_diff(before, isolated_after) + + assert live_engine.state == before + assert preview_result["decision"] == isolated_decision + assert preview_result["state_before"] == before + assert preview_result["state_after"] == isolated_after + assert preview_result["diff"] == isolated_diff + assert preview_result["would_mutate"] is (before != isolated_after) + assert preview_result["would_mutate"] is preview_result["diff"]["changed"]