Ryan Malloy
9427e3d4df
Two complementary additions from cucx-docs's prompt-suggestions handoff
(see axl/agent-threads/cucx-prompt-suggestions/ for the source thread).
device_grep(pattern, classes=None) — fuzzy device discovery by name OR
description, optionally filtered by tkclass.name. Surfaces "wait, there
are TWO of these?" findings (parallel fax servers, duplicate CUBEs,
vestigial conference bridges) by grouping matches by class so the
structure of what matched is visible at a glance. CUCM-style % wildcards
work; case-insensitive matching via UPPER(); single quotes properly
escaped via _esc.
axl_sql error hints — when AXL returns an error AND the query contains
the trigger phrase, append a path-correction hint to the error message.
Two patterns shipped:
- "Column (fkdevice) not found" + numplan in query → suggest the
devicenumplanmap M:N join (the literal multi-attempt schema-discovery
experience cucx-docs hit at Bingham — numplan has no direct fkdevice)
- "not in database" + sipdestination in query → suggest sipdestinationgroup
+ sipprofile + axl_list_tables(pattern='sip%') for discovery (the
`sipdestination` table is reasonable-sounding but doesn't exist)
Hints are surgical (both error fragment AND query trigger must match)
to keep false-positive risk near zero. Validator behavior unchanged —
this is post-execution error augmentation, not gate enhancement. Failing
queries now raise RuntimeError(augmented) when a hint applies; otherwise
the original exception passes through unchanged.
Tests: +19 (8 device_grep + 11 error-hints with end-to-end mock through
execute_sql_query). Full suite 219 → 238 passing.
Live-cluster smoke test still pending (TLS handshake intermittent
this session). Sequencing nit from cucx-docs's msg 003 (move error-hint
earlier) honored — bundled with device_grep in this single commit.
95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
"""Tests for device_grep — fuzzy device discovery by name/description.
|
|
|
|
The "wait there are TWO of these?" finding shape from cucx-docs's
|
|
ZetaFax-vs-RightFax discovery (msg 001 in
|
|
axl/agent-threads/cucx-prompt-suggestions/).
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from mcaxl.route_plan import device_grep
|
|
|
|
|
|
class FakeAxlClient:
|
|
def __init__(self, rows):
|
|
self._rows = rows
|
|
self.queries = []
|
|
|
|
def execute_sql_query(self, sql):
|
|
self.queries.append(sql)
|
|
return {"row_count": len(self._rows), "rows": self._rows}
|
|
|
|
|
|
def _make_row(name, description, class_name, device_type="SIP Trunk", pool="DP-1"):
|
|
return {
|
|
"name": name,
|
|
"description": description,
|
|
"class_name": class_name,
|
|
"device_type": device_type,
|
|
"pool_name": pool,
|
|
}
|
|
|
|
|
|
class TestDeviceGrepBasics:
|
|
def test_groups_by_class(self):
|
|
rows = [
|
|
_make_row("RightFax-Trunk", "RightFax inbound", "Trunk"),
|
|
_make_row("ZetaFax-Trunk", "ZetaFax internal", "Trunk"),
|
|
_make_row("SEPABCDFAXX01", "RightFax desk test", "Phone", "Cisco 8841"),
|
|
_make_row("RightFax-RL", "RightFax route list", "Route List"),
|
|
]
|
|
client = FakeAxlClient(rows)
|
|
result = device_grep(client, "FAX")
|
|
assert result["match_count"] == 4
|
|
assert set(result["groups"].keys()) == {"Trunk", "Phone", "Route List"}
|
|
assert len(result["groups"]["Trunk"]) == 2
|
|
|
|
def test_class_filter_passed_to_sql(self):
|
|
client = FakeAxlClient([])
|
|
device_grep(client, "FAX", classes=["Trunk", "Route List"])
|
|
# Both class names appear escaped in the SQL IN clause
|
|
sql = client.queries[0]
|
|
assert "tc.name IN" in sql
|
|
assert "'Trunk'" in sql
|
|
assert "'Route List'" in sql
|
|
|
|
def test_no_class_filter_omits_in_clause(self):
|
|
client = FakeAxlClient([])
|
|
device_grep(client, "FAX")
|
|
assert "tc.name IN" not in client.queries[0]
|
|
|
|
def test_empty_pattern_raises(self):
|
|
client = FakeAxlClient([])
|
|
with pytest.raises(ValueError, match="non-empty"):
|
|
device_grep(client, "")
|
|
with pytest.raises(ValueError, match="non-empty"):
|
|
device_grep(client, " ")
|
|
|
|
def test_pattern_quote_escaped(self):
|
|
client = FakeAxlClient([])
|
|
device_grep(client, "fake'; DROP TABLE device --")
|
|
# SQL injection via pattern is escaped (doubled single quotes)
|
|
assert "fake''" in client.queries[0]
|
|
|
|
def test_class_filter_quote_escaped(self):
|
|
client = FakeAxlClient([])
|
|
device_grep(client, "FAX", classes=["Phone'; DROP TABLE device --"])
|
|
assert "Phone''" in client.queries[0]
|
|
|
|
def test_unknown_class_grouped_separately(self):
|
|
# If tkclass enum doesn't resolve (LEFT JOIN miss), class_name is NULL
|
|
rows = [
|
|
{"name": "WeirdDevice", "description": "?", "class_name": None,
|
|
"device_type": None, "pool_name": None},
|
|
]
|
|
client = FakeAxlClient(rows)
|
|
result = device_grep(client, "weird")
|
|
assert "Unknown" in result["groups"]
|
|
assert result["groups"]["Unknown"][0]["name"] == "WeirdDevice"
|
|
|
|
def test_response_includes_filter_metadata(self):
|
|
client = FakeAxlClient([])
|
|
result = device_grep(client, "FAX", classes=["Trunk"])
|
|
assert result["pattern"] == "FAX"
|
|
assert result["classes_filter"] == ["Trunk"]
|