"""Tests for cti_failsafe_reachability — find broken CFNA/CFUR forwards. Source: cucx-docs handoff at ``axl/agent-threads/cti-audit-prompts/001-cucx-cfna-reachability-audit.md`` documenting a real life-safety bug at Bingham (912-CTI-RP CFNA → '10911' under 911CER-CSS, where '10.911' lives in CER911-PT which 911CER-CSS doesn't reach). The tool composes three SQL queries per broken forward: 1. Top-level forwards SQL (fetch CTI RPs with CFNA/CFUR set) 2. translation_chain's SQL (per-forward reachability check) 3. _suggest_failsafe_fix's partition-lookup SQL (one per finding) The FakeAxlClient dispatches by query content rather than sequence because the order of (2) and (3) interleaves across multiple findings. """ import pytest from mcaxl.route_plan import ( _LIFE_SAFETY_TOKENS, _is_life_safety_cti, cti_failsafe_reachability, ) class FakeAxlClient: """Dispatching fake — returns canned responses keyed on SQL content. Constructor takes: - cti_rp_rows: rows for the top-level "find CTI RPs with forwards" query - reachable_destinations: set of (destination, css) pairs that have a matching pattern (translation_chain returns match_count > 0 for these) - destination_partitions: dict {destination: [partition_name, ...]} for the exact-literal partition-lookup query in _suggest_failsafe_fix - dotted_patterns: list of (pattern, partition) tuples for the dot-stripped lookup. The pattern includes literal dots (e.g. ``"10.911"``); _suggest_failsafe_fix strips dots and compares to the destination """ def __init__( self, cti_rp_rows: list[dict], reachable_destinations: set[tuple[str, str]] | None = None, destination_partitions: dict[str, list[str]] | None = None, dotted_patterns: list[tuple[str, str]] | None = None, ): self._cti_rows = cti_rp_rows self._reachable = reachable_destinations or set() self._dest_partitions = destination_partitions or {} self._dotted_patterns = dotted_patterns or [] self.queries: list[str] = [] def execute_sql_query(self, sql: str) -> dict: self.queries.append(sql) # Dispatch 1: top-level "find CTI RPs with CFNA/CFUR" query if "tc.name = 'CTI Route Point'" in sql and "cfnadestination" in sql: return {"row_count": len(self._cti_rows), "rows": self._cti_rows} # Dispatch 2: translation_chain's reachability check # Recognizable by `tkpatternusage IN (3, 5, 7)` from route_plan.py if "tkpatternusage IN (3, 5, 7)" in sql: for dest, css in self._reachable: if f"name = '{css}'" in sql: return { "row_count": 1, "rows": [{ "pattern": dest, "pattern_type": "Translation", "partition_name": "Reachable-PT", "calling_party_xform_mask": None, "called_party_xform_mask": None, "prefix_digits_out": None, "digit_discard_instructions": None, "route_filter": None, "description": "fake-reachable", }], } return {"row_count": 0, "rows": []} # Dispatch 3a: _suggest_failsafe_fix's dot-stripped lookup # (Stage 2 of the fix-suggestion logic — pulls all dot-containing # patterns and filters Python-side) if "np.dnorpattern LIKE '%.%'" in sql: rows = [ {"pattern": pat, "partition": part} for pat, part in self._dotted_patterns ] return {"row_count": len(rows), "rows": rows} # Dispatch 3b: _suggest_failsafe_fix's exact-literal lookup # (Stage 1 — exact match on np.dnorpattern) if "rp.name IS NOT NULL" in sql and "np.dnorpattern" in sql: for dest, parts in self._dest_partitions.items(): if f"np.dnorpattern = '{dest}'" in sql: rows = [{"partition": p} for p in parts] return {"row_count": len(rows), "rows": rows} return {"row_count": 0, "rows": []} # Anything else — empty (unexpected query path; fail loud later) return {"row_count": 0, "rows": []} def _cti_row(name, description, cfna=None, cfur=None, cfna_css=None, cfur_css=None): return { "name": name, "description": description, "cfnadestination": cfna, "cfurdestination": cfur, "cfna_css_name": cfna_css, "cfur_css_name": cfur_css, } # ─── Life-safety token detection (helper in isolation) ──────────────── class TestLifeSafetyDetection: @pytest.mark.parametrize("description", [ "Primary CER Server", "911 CTI Route Point", "Emergency CER", "PSAP gateway", "PANIC button receiver", "Code BLUE Alert", ]) def test_life_safety_tokens_match(self, description): assert _is_life_safety_cti("some-name", description) is True @pytest.mark.parametrize("name", [ "911-CTI-RP", "EMERGENCY-RP", "CER-Primary", "psap-gateway", ]) def test_token_matched_in_name_field(self, name): # Tokens match against name OR description — some clusters tag # the role in the name field rather than the description assert _is_life_safety_cti(name, "Generic CTI Route Point") is True @pytest.mark.parametrize("description", [ "Patient Intake CTI Route Point", "Voicemail Pilot", "Receptionist Hunt Pilot", "Generic application route point", ]) def test_non_life_safety_descriptions(self, description): assert _is_life_safety_cti("regular-rp", description) is False def test_null_name_and_description_does_not_match(self): assert _is_life_safety_cti(None, None) is False assert _is_life_safety_cti("", "") is False def test_advertised_token_list_is_what_we_implement(self): # If the token list grows or shrinks, the docstring + agent-thread # reply must be updated alongside. Catches accidental drift. assert _LIFE_SAFETY_TOKENS == ( "emergency", "911", "cer", "psap", "panic", "alert", ) # ─── Tool-level integration ────────────────────────────────────────── class TestCtiFailsafeReachability: def test_no_cti_route_points_returns_empty_findings(self): client = FakeAxlClient(cti_rp_rows=[]) result = cti_failsafe_reachability(client) assert result["total_cti_route_points"] == 0 assert result["broken_cfna"] == 0 assert result["broken_cfur"] == 0 assert result["findings"] == [] def test_working_cfna_produces_no_finding(self): client = FakeAxlClient( cti_rp_rows=[ _cti_row("Working-RP", "Patient intake", cfna="5550100", cfna_css="Internal-CSS"), ], reachable_destinations={("5550100", "Internal-CSS")}, ) result = cti_failsafe_reachability(client) assert result["broken_cfna"] == 0 assert result["findings"] == [] def test_broken_cfna_non_life_safety_is_medium(self): client = FakeAxlClient( cti_rp_rows=[ _cti_row("Generic-RP", "Patient intake", cfna="5550100", cfna_css="BadCSS"), ], reachable_destinations=set(), # nothing reachable destination_partitions={"5550100": ["Internal-PT"]}, ) result = cti_failsafe_reachability(client) assert result["broken_cfna"] == 1 assert len(result["findings"]) == 1 finding = result["findings"][0] assert finding["device"] == "Generic-RP" assert finding["forward_kind"] == "cfna" assert finding["destination"] == "5550100" assert finding["css"] == "BadCSS" assert finding["match_count"] == 0 assert finding["severity"] == "MEDIUM" assert "Internal-PT" in finding["suggested_fix"] assert "BadCSS" in finding["suggested_fix"] def test_broken_cfna_life_safety_is_high(self): client = FakeAxlClient( cti_rp_rows=[ _cti_row("911-CTI-RP", "Emergency dispatch", cfna="10911", cfna_css="911CER-CSS"), ], destination_partitions={"10911": ["CER911-PT"]}, ) result = cti_failsafe_reachability(client) assert result["findings"][0]["severity"] == "HIGH" def test_broken_cfna_and_cfur_produce_two_findings(self): # Same device with both forwards broken — should produce TWO entries # (per-forward, not per-device, per the design decision) client = FakeAxlClient( cti_rp_rows=[ _cti_row( "912-CTI-RP", "CTI RP for Secondary CER Server", cfna="10911", cfna_css="911CER-CSS", cfur="10911", cfur_css="911CER-CSS", ), ], destination_partitions={"10911": ["CER911-PT"]}, ) result = cti_failsafe_reachability(client) assert result["broken_cfna"] == 1 assert result["broken_cfur"] == 1 assert len(result["findings"]) == 2 kinds = {f["forward_kind"] for f in result["findings"]} assert kinds == {"cfna", "cfur"} # Both should be HIGH (description contains "CER") assert all(f["severity"] == "HIGH" for f in result["findings"]) def test_only_cfna_set_does_not_check_cfur(self): # CFUR null → don't check it (not a finding) client = FakeAxlClient( cti_rp_rows=[ _cti_row("Half-RP", "Generic", cfna="9999", cfna_css="BadCSS"), ], destination_partitions={"9999": ["Some-PT"]}, ) result = cti_failsafe_reachability(client) assert result["broken_cfna"] == 1 assert result["broken_cfur"] == 0 def test_canonical_bingham_bug_reproduced(self): """The canary scenario from cucx-docs's 001 — verifies the tool produces exactly the expected output for the motivating bug.""" client = FakeAxlClient( cti_rp_rows=[ _cti_row( "912-CTI-RP", "CTI RP for Secondary CER Server", cfna="10911", cfna_css="911CER-CSS", cfur="10911", cfur_css="911CER-CSS", ), ], destination_partitions={"10911": ["CER911-PT"]}, ) result = cti_failsafe_reachability(client) cfna_finding = next(f for f in result["findings"] if f["forward_kind"] == "cfna") assert cfna_finding == { "device": "912-CTI-RP", "description": "CTI RP for Secondary CER Server", "forward_kind": "cfna", "destination": "10911", "css": "911CER-CSS", "match_count": 0, "severity": "HIGH", # description contains "CER" "suggested_fix": ( "Pattern '10911' lives in partition 'CER911-PT'. " "Either add 'CER911-PT' to CSS '911CER-CSS', " "OR change the forward CSS to a CSS that already " "contains 'CER911-PT'." ), } def test_suggested_fix_when_no_partition_holds_destination(self): # Edge case: destination doesn't match any literal pattern # OR any dot-stripped variant (might match a wildcard, but not # something exact). Falls back to the wildcard-investigation # generic message. client = FakeAxlClient( cti_rp_rows=[ _cti_row("Wild-RP", "Generic", cfna="orphan-dest", cfna_css="BadCSS"), ], destination_partitions={}, # no partition holds 'orphan-dest' # dotted_patterns defaults to [] → no dot-stripped match either ) result = cti_failsafe_reachability(client) fix = result["findings"][0]["suggested_fix"] assert "no exact-literal or dot-stripped pattern" in fix assert "wildcard" in fix.lower() def test_suggested_fix_when_destination_in_multiple_partitions(self): # Edge case: destination matches in multiple partitions; the # fix message lists them and asks the operator to pick. client = FakeAxlClient( cti_rp_rows=[ _cti_row("Multi-RP", "Generic", cfna="5555", cfna_css="BadCSS"), ], destination_partitions={"5555": ["Site-A-PT", "Site-B-PT"]}, ) result = cti_failsafe_reachability(client) fix = result["findings"][0]["suggested_fix"] assert "multiple partitions" in fix assert "Site-A-PT" in fix assert "Site-B-PT" in fix def test_response_includes_scope_note(self): client = FakeAxlClient(cti_rp_rows=[]) result = cti_failsafe_reachability(client) assert "_note" in result # Scope discipline visible at the call site — CFB exclusion is # documented, and the life-safety token list is named. assert "CFB" in result["_note"] assert "emergency" in result["_note"] # ─── Dot-stripped fix-suggestion (cti-audit-prompts/004 limitation) ──── # # The CUCM separator-dot in patterns like `10.911` is purely visual — # it represents access-code boundary, not a digit. A destination string # `10911` (no dot) should match a configured pattern `10.911`. The # original _suggest_failsafe_fix only did exact-literal lookups and # missed this; the live Bingham smoke-test surfaced the limitation on # `912-CTI-RP`. These tests pin the dot-stripped fallback behavior. class TestDotStrippedFixSuggestion: def test_dot_stripped_match_cites_dotted_pattern(self): # Destination "10911" should match pattern "10.911" via dot-strip client = FakeAxlClient( cti_rp_rows=[ _cti_row("Test-RP", "Generic", cfna="10911", cfna_css="BadCSS"), ], destination_partitions={}, # no exact-literal match dotted_patterns=[("10.911", "CER911-PT")], ) result = cti_failsafe_reachability(client) fix = result["findings"][0]["suggested_fix"] # The fix message names BOTH the pattern form and the destination # so the operator sees what the dot-strip matched assert "'10.911'" in fix assert "'10911'" in fix assert "CER911-PT" in fix assert "BadCSS" in fix def test_exact_literal_takes_precedence_over_dotted(self): # If both an exact-literal match and a dotted match exist, the # exact-literal wins — no need to mention the dotted form client = FakeAxlClient( cti_rp_rows=[ _cti_row("Test-RP", "Generic", cfna="912", cfna_css="BadCSS"), ], destination_partitions={"912": ["911CER-PT"]}, dotted_patterns=[("9.12", "Decoy-PT")], # would match if dotted ran ) result = cti_failsafe_reachability(client) fix = result["findings"][0]["suggested_fix"] assert "911CER-PT" in fix # Decoy-PT shouldn't appear — exact-literal should short-circuit assert "Decoy-PT" not in fix def test_dotted_match_with_multiple_partitions(self): # If the same dotted pattern exists in multiple partitions, the # multi-partition message format applies — same as exact-literal client = FakeAxlClient( cti_rp_rows=[ _cti_row("Test-RP", "Generic", cfna="10911", cfna_css="BadCSS"), ], destination_partitions={}, dotted_patterns=[ ("10.911", "Site-A-PT"), ("10.911", "Site-B-PT"), ], ) result = cti_failsafe_reachability(client) fix = result["findings"][0]["suggested_fix"] assert "multiple partitions" in fix assert "Site-A-PT" in fix assert "Site-B-PT" in fix def test_no_exact_no_dotted_falls_back_to_generic(self): # Neither exact-literal nor dot-stripped lookup finds a match # → fall back to the wildcard-investigation generic message client = FakeAxlClient( cti_rp_rows=[ _cti_row("Test-RP", "Generic", cfna="60003", cfna_css="BadCSS"), ], destination_partitions={}, dotted_patterns=[], ) result = cti_failsafe_reachability(client) fix = result["findings"][0]["suggested_fix"] assert "no exact-literal or dot-stripped pattern" in fix assert "wildcard pattern" in fix.lower() def test_dotted_pattern_with_irrelevant_dot_does_not_match(self): # Pattern "1.0911" has a dot but its dot-stripped form is "10911" # — should match. Pattern "1.0912" stripped is "10912" — should NOT. # This exercises the substring-equality logic. client = FakeAxlClient( cti_rp_rows=[ _cti_row("Test-RP", "Generic", cfna="10911", cfna_css="BadCSS"), ], destination_partitions={}, dotted_patterns=[ ("1.0911", "Match-PT"), # strips to "10911" → matches ("1.0912", "Nonmatch-PT"), # strips to "10912" → no match ("10.91", "AnotherMatch-PT"), # strips to "1091" → no match ], ) result = cti_failsafe_reachability(client) fix = result["findings"][0]["suggested_fix"] assert "Match-PT" in fix assert "Nonmatch-PT" not in fix assert "AnotherMatch-PT" not in fix