"""Audit batch 3 (I-3, I-4): capability verdict unification and policy merge unification. Pure refactor: no behaviour change. Run: python3 -m unittest discover -s cloud/tests -p "test_audit_batch3.py" -v """ import copy import json import os import random import sys import tempfile import unittest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) from cloud.models import ( check_model_capabilities, decide_routing, evaluate_capability, load_routing_policy, merge_policy_documents, parse_routing_policy, split_model_capability_missing, ) from cloud.routing_dryrun import dry_run_from_env, dry_run_policy_document REQUIRES = {"tool_calling": True, "vision": True, "min_context_tokens": 1000} CAPS = { "good": {"tool_calling": True, "vision": True, "context_tokens": 200000}, "no-tool": {"tool_calling": False, "vision": True, "context_tokens": 200000}, "no-vision-field": {"tool_calling": True, "context_tokens": 200000}, "small-ctx": {"tool_calling": True, "vision": True, "context_tokens": 100}, "no-ctx": {"tool_calling": True, "vision": True}, "mixed": {"tool_calling": False, "context_tokens": 200000}, } def _policy(mode, tier_model, requires, caps, worker_model=None): doc = { "version": "batch3", "mode": mode, "tiers": {"default": {"model": tier_model, "requires": requires}}, "model_capabilities": caps, } return parse_routing_policy(doc) class EvaluateCapabilityTestCase(unittest.TestCase): def test_matrix_known_unknown_mixed_absent_empty(self): cases = [ # (model, ok, missing, unknown, known, unknown_missing, # status, verb) ("good", True, [], False, [], [], "ok", None), ("no-tool", False, ["tool_calling"], False, ["tool_calling"], [], "fail", "lacks"), ("no-vision-field", False, ["vision"], True, [], ["vision"], "unknown", "cannot verify"), ("small-ctx", False, ["min_context_tokens"], False, ["min_context_tokens"], [], "fail", "lacks"), ("no-ctx", False, ["min_context_tokens"], True, [], ["min_context_tokens"], "unknown", "cannot verify"), ("mixed", False, ["tool_calling", "vision"], True, ["tool_calling"], ["vision"], "unknown", "cannot verify"), ("absent-model", False, ["min_context_tokens", "tool_calling", "vision"], True, [], ["min_context_tokens", "tool_calling", "vision"], "unknown", "cannot verify"), ] for (model, ok, missing, unknown, known, unknown_missing, status, verb) in cases: with self.subTest(model=model): verdict = evaluate_capability(CAPS, model, REQUIRES) self.assertEqual(verdict.ok, ok) self.assertEqual(verdict.missing, missing) self.assertEqual(verdict.unknown, unknown) self.assertEqual(verdict.known_missing, known) self.assertEqual(verdict.unknown_missing, unknown_missing) self.assertEqual(verdict.status, status) if verb is not None: self.assertEqual(verdict.verb, verb) def test_empty_requires_always_ok_even_for_absent_model(self): verdict = evaluate_capability(CAPS, "absent-model", {}) self.assertTrue(verdict.ok) self.assertEqual(verdict.missing, []) self.assertFalse(verdict.unknown) self.assertEqual(verdict.status, "ok") def test_parity_with_old_check_and_split(self): # The refactor contract: every verdict field equals what the # two old functions returned for the same inputs, across a # matrix of directories, models and requires — including the # preserved edge where an absent model plus an uncleaned # requires key makes the total check list the invalid key # while the split lists only recognised keys. That edge is # documented here as preserved parity, not as desired # semantics; changing it would be a behaviour change. requires_sets = [ REQUIRES, {"tool_calling": True}, {"min_context_tokens": 500}, {}, {"tool_calling": True, "bogus_key": True}, ] models = list(CAPS) + ["absent-model", "", None] for requires in requires_sets: for model in models: with self.subTest(model=model, requires=requires): verdict = evaluate_capability( CAPS, model, requires) ok, missing, unknown = check_model_capabilities( CAPS, model, requires) known, unknown_missing = \ split_model_capability_missing( CAPS, model, requires) self.assertEqual(verdict.ok, ok) self.assertEqual(verdict.missing, missing) self.assertEqual(verdict.unknown, unknown) self.assertEqual(verdict.known_missing, known) self.assertEqual( verdict.unknown_missing, unknown_missing) class DecideRoutingWordingTestCase(unittest.TestCase): def test_active_known_missing_falls_back_to_worker_default(self): policy = _policy("active", "no-tool", REQUIRES, CAPS) decision = decide_routing({}, "good", policy=policy) self.assertEqual(decision.source, "worker_default") self.assertTrue(decision.capability_fallback) self.assertEqual(decision.chosen_model, "good") self.assertIn("capability lacks: tool_calling", decision.reason) def test_active_both_fail_blocks_with_exact_reason_shape(self): policy = _policy("active", "no-tool", REQUIRES, CAPS) decision = decide_routing({}, "no-tool", policy=policy) self.assertEqual(decision.source, "capability_blocked") self.assertIsNone(decision.chosen_model) self.assertIn("refusing to dispatch an incapable or" " unverifiable model", decision.reason) def test_shadow_unknown_and_known_notes(self): policy = _policy("shadow", "no-vision-field", REQUIRES, CAPS) decision = decide_routing({}, "good", policy=policy) self.assertEqual(decision.source, "shadow") self.assertFalse(decision.applied) self.assertIn("capability unknown: vision", decision.reason) policy = _policy("shadow", "no-tool", REQUIRES, CAPS) decision = decide_routing({}, "good", policy=policy) self.assertIn("capability lacks: tool_calling", decision.reason) class DryRunWordingTestCase(unittest.TestCase): def _doc(self, mode, tier_model): return { "version": "batch3", "mode": mode, "tiers": {"default": {"model": tier_model, "requires": REQUIRES}}, "model_capabilities": CAPS, } def test_tier_check_notes_and_block_flags(self): report = dry_run_policy_document( self._doc("active", "no-tool")) tier = report["tiers"]["default"] self.assertEqual(tier["capabilities"]["status"], "fail") self.assertTrue(report["blocked"]) self.assertTrue(any( "lacks capability: tool_calling" in w and "would block 'default' tasks at dispatch" in w for w in report["warnings"])) report = dry_run_policy_document( self._doc("shadow", "no-vision-field")) tier = report["tiers"]["default"] self.assertEqual(tier["capabilities"]["status"], "unknown") self.assertFalse(report["blocked"]) self.assertTrue(any( "cannot verify capability: vision" in w and "shadow mode does not block dispatch" in w for w in report["warnings"])) def test_gateway_check_notes_and_block_flags(self): report = dry_run_policy_document( self._doc("active", "good"), gateway_model="no-tool") gw = report["gateway"]["tiers"]["default"] self.assertEqual(gw["status"], "fail") self.assertTrue(report["blocked"]) self.assertTrue(any( "gateway model 'no-tool' lacks capability: tool_calling" in w and "would block 'default' tasks at the gateway gate" in w for w in report["warnings"])) report = dry_run_policy_document( self._doc("shadow", "good"), gateway_model=None) gw = report["gateway"]["tiers"]["default"] self.assertEqual(gw["status"], "unknown") self.assertTrue(gw["unknown"]) self.assertTrue(any( "gateway model is not set in CLOUD_SUB2API_DEFAULT_MODEL" in w for w in report["warnings"])) class MergePolicyDocumentsTestCase(unittest.TestCase): def test_boundary_cases(self): self.assertIsNone(merge_policy_documents(None, None)) self.assertEqual(merge_policy_documents(None, {"a": 1}), {"a": 1}) self.assertEqual(merge_policy_documents({"a": 1}, None), {"a": 1}) self.assertEqual(merge_policy_documents({"a": 1}, ["x"]), {"a": 1}) self.assertEqual(merge_policy_documents(["x"], {"a": 1}), {"a": 1}) # No tiers anywhere: plain shallow update. self.assertEqual( merge_policy_documents({"mode": "shadow", "version": "v1"}, {"mode": "active"}), {"mode": "active", "version": "v1"}) def test_json_replaces_whole_tiers_map_not_per_tier(self): # Preserved (not fixed) semantics, documented in # merge_policy_documents: JSON tiers replace the whole map. # File tiers absent from JSON are dropped, not preserved. base = {"mode": "shadow", "tiers": {"plan": "m-file-plan", "default": "m-file-default"}} override = {"tiers": {"plan": "m-json-plan"}} self.assertEqual(merge_policy_documents(base, override), {"mode": "shadow", "tiers": {"plan": "m-json-plan"}}) def test_inputs_not_mutated(self): base = {"mode": "shadow", "tiers": {"plan": "m-file-plan", "default": "m-file-default"}} override = {"tiers": {"plan": "m-json-plan"}} base_before, override_before = copy.deepcopy(base), \ copy.deepcopy(override) merge_policy_documents(base, override) self.assertEqual(base, base_before) self.assertEqual(override, override_before) def test_property_random_pairs_match_shallow_update(self): rng = random.Random(20261102) tier_values = ["m1", "m2", {"model": "m3"}, {"model": "m4", "mode": "shadow"}, {"model": "m5", "mode": "bogus"}, {"model": ""}, " ", 7, None] top_values = ["shadow", "active", "v9", 3, None, {"x": 1}, ["a"]] def _rand_doc(): doc = {} if rng.random() < 0.8: doc["mode"] = rng.choice(["shadow", "active", "off"]) if rng.random() < 0.5: doc["version"] = rng.choice(top_values) if rng.random() < 0.8: tiers = {} for tier in rng.sample( ["plan", "goal", "default", "weird"], rng.randint(0, 3)): tiers[tier] = rng.choice(tier_values) if rng.random() < 0.1: doc["tiers"] = rng.choice(["x", 5, None]) else: doc["tiers"] = tiers return doc for _ in range(300): base = _rand_doc() override = _rand_doc() got = merge_policy_documents(base, override) expected = dict(base) expected.update(override) self.assertEqual(got, expected) def test_loader_and_dryrun_agree_on_same_env(self): # Loader and dry-run now share merge_policy_documents: for # the same file+JSON env, the parsed policy tiers and the # dry-run report tiers (and fingerprints) must agree. cases = [ ({"mode": "shadow", "tiers": {"plan": "m-file-plan", "default": "m-file-default"}}, {"tiers": {"plan": "m-json-plan"}}), ({"mode": "active", "tiers": {"default": {"model": "m1", "requires": {"tool_calling": True}}}, "model_capabilities": {"m1": {"tool_calling": True}}}, {"mode": "shadow"}), ] for base_doc, override_doc in cases: with tempfile.NamedTemporaryFile( "w", suffix=".json", delete=False) as fh: json.dump(base_doc, fh) path = fh.name try: env = {"CLOUD_ROUTING_POLICY_FILE": path, "CLOUD_ROUTING_POLICY_JSON": json.dumps(override_doc)} policy = load_routing_policy(env) report = dry_run_from_env(env) self.assertIsNotNone(policy) self.assertTrue(report["valid"]) for tier, model in policy.tiers.items(): self.assertEqual( report["tiers"][tier]["model"], model) self.assertEqual(report["content_fingerprint"], policy.content_fingerprint()) finally: os.unlink(path) if __name__ == "__main__": unittest.main()