Ryan Malloy
8b3da9d729
Read-only MCP server for Cisco Unified CM 15 AXL — built for LLM-driven
cluster auditing, with a particular focus on the Route Plan Report:
partitions, calling search spaces, route patterns, translation patterns,
called/calling party transformations, and digit-discard instructions.
Pairs intentionally with the sibling mcp-cisco-docs server (live
cluster state + vendor docs in one LLM context).
Architecture:
- zeep SOAP client to CUCM AXL
- WSDL bootstrap from Cisco's axlsqltoolkit.zip (auto-extract on
first launch; zip is gitignored, vendor-licensed)
- SQLite response cache at ~/.cache/mcp-cucm-axl/responses/
- Schema-grounded prompts that pull chunks from the sibling
cisco-docs index (docs_loader.py)
Read-only by structural guarantee — never registers AXL write methods
(no executeSQLUpdate, no add*/update*/remove*/apply*/reset*/restart*
tools). SQL queries also client-side validated (sql_validator.py) to
begin with SELECT or WITH.
Tools exposed:
Foundational: axl_version, axl_sql, axl_list_tables,
axl_describe_table, cache_stats, cache_clear
Route plan: route_partitions, route_calling_search_spaces,
route_patterns, route_inspect_pattern,
route_lists_and_groups, route_translation_chain,
route_digit_discard_instructions
Prompts (schema-grounded):
route_plan_overview, investigate_pattern, audit_routing,
cucm_sql_help
Tests cover cache, docs_loader, normalize, sql_validator, wildcard.
88 lines
2.6 KiB
Python
88 lines
2.6 KiB
Python
"""Tests for the SQLite TTL cache."""
|
|
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from mcp_cucm_axl.cache import AxlCache
|
|
|
|
|
|
@pytest.fixture
|
|
def cache(tmp_path: Path) -> AxlCache:
|
|
return AxlCache(tmp_path / "test.sqlite", default_ttl=60)
|
|
|
|
|
|
def test_set_and_get(cache: AxlCache):
|
|
cache.set("getCCMVersion", {}, {"version": "15.0.1"})
|
|
assert cache.get("getCCMVersion", {}) == {"version": "15.0.1"}
|
|
|
|
|
|
def test_miss_returns_none(cache: AxlCache):
|
|
assert cache.get("nonexistent", {"x": 1}) is None
|
|
|
|
|
|
def test_kwargs_order_independent(cache: AxlCache):
|
|
cache.set("listPhone", {"name": "SEP1", "limit": 10}, {"rows": ["p1"]})
|
|
# Different order should still hit
|
|
assert cache.get("listPhone", {"limit": 10, "name": "SEP1"}) == {"rows": ["p1"]}
|
|
|
|
|
|
def test_different_args_different_keys(cache: AxlCache):
|
|
cache.set("listPhone", {"name": "SEP1"}, {"rows": ["a"]})
|
|
cache.set("listPhone", {"name": "SEP2"}, {"rows": ["b"]})
|
|
assert cache.get("listPhone", {"name": "SEP1"}) == {"rows": ["a"]}
|
|
assert cache.get("listPhone", {"name": "SEP2"}) == {"rows": ["b"]}
|
|
|
|
|
|
def test_expired_entries_not_returned(tmp_path: Path):
|
|
c = AxlCache(tmp_path / "ttl.sqlite", default_ttl=60)
|
|
c.set("foo", {}, {"x": 1}, ttl=1)
|
|
time.sleep(1.1)
|
|
assert c.get("foo", {}) is None
|
|
|
|
|
|
def test_ttl_zero_disables_caching(tmp_path: Path):
|
|
c = AxlCache(tmp_path / "off.sqlite", default_ttl=0)
|
|
c.set("foo", {}, {"x": 1})
|
|
# default_ttl=0 means writes are no-ops
|
|
assert c.get("foo", {}) is None
|
|
|
|
|
|
def test_stats_reports_breakdown(cache: AxlCache):
|
|
cache.set("listPhone", {}, {"x": 1})
|
|
cache.set("listPhone", {"a": 1}, {"x": 2})
|
|
cache.set("getCCMVersion", {}, {"v": "15"})
|
|
|
|
stats = cache.stats()
|
|
assert stats["live_entries"] == 3
|
|
assert stats["by_method"]["listPhone"] == 2
|
|
assert stats["by_method"]["getCCMVersion"] == 1
|
|
|
|
|
|
def test_clear_all(cache: AxlCache):
|
|
cache.set("a", {}, "x")
|
|
cache.set("b", {}, "y")
|
|
deleted = cache.clear()
|
|
assert deleted == 2
|
|
assert cache.stats()["live_entries"] == 0
|
|
|
|
|
|
def test_clear_by_pattern(cache: AxlCache):
|
|
cache.set("listPhone", {}, "p")
|
|
cache.set("listLine", {}, "l")
|
|
cache.set("getCCMVersion", {}, "v")
|
|
deleted = cache.clear("list*")
|
|
assert deleted == 2
|
|
assert cache.get("getCCMVersion", {}) == "v"
|
|
|
|
|
|
def test_purge_expired(tmp_path: Path):
|
|
c = AxlCache(tmp_path / "p.sqlite", default_ttl=60)
|
|
c.set("a", {}, "x", ttl=1)
|
|
c.set("b", {}, "y", ttl=60)
|
|
time.sleep(1.1)
|
|
purged = c.purge_expired()
|
|
assert purged == 1
|
|
assert c.stats()["live_entries"] == 1
|