"""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, ...]} used by the _suggest_failsafe_fix's partition-lookup query """ def __init__( self, cti_rp_rows: list[dict], reachable_destinations: set[tuple[str, str]] | None = None, destination_partitions: dict[str, list[str]] | None = None, ): self._cti_rows = cti_rp_rows self._reachable = reachable_destinations or set() self._dest_partitions = destination_partitions 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: # Extract the destination + CSS from the SQL to figure out # whether to return a "match" row or no rows. The destination # appears in the called-side filter; the CSS appears in the # callingsearchspace WHERE clause. # # Simplest dispatch: scan the query for the (dest, css) pairs # we know are reachable. If any match, return a fake matching # pattern row. for dest, css in self._reachable: if f"name = '{css}'" in sql: # For each reachable destination, the test fake returns # a single pattern that exactly equals the destination # so translation_chain's wildcard matcher resolves it. 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 3: _suggest_failsafe_fix's partition-lookup query if "rp.name IS NOT NULL" in sql and "np.dnorpattern" in sql: # Extract the dnorpattern literal from the 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 # (might match a wildcard, but not an exact-literal). Suggest_fix # falls back to a 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' ) result = cti_failsafe_reachability(client) fix = result["findings"][0]["suggested_fix"] assert "matches no exact-literal 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"]