mcaxl/tests/test_css_impact.py

"""Hamilton review MAJOR #3: find_devices_using_css must surface partial failures.

The function is per-category resilient by design — if one schema query fails,
the others still produce results. But the top-level summary previously hid
that some categories errored out: `total_returned` and `any_truncated` only
reflected the SUCCESSFUL categories. An LLM consuming "47 references, low
impact" wouldn't know that 5 categories errored and the real number is
likely much higher.

After the fix: the response includes `complete: bool`, `categories_with_errors`,
and `error_categories`, so an LLM (or human auditor) can see the partial-failure
state and act on it.
"""

import pytest

from mcaxl.route_plan import find_devices_using_css


class FakeAxlClient:
    """Minimal stand-in for AxlClient that lets us simulate per-query failures.

    Returns a fake CSS pkid for the lookup query, then either a single fake row
    or an exception based on substring matching.
    """

    def __init__(self, error_on_columns: list[str] | None = None):
        self.error_on_columns = error_on_columns or []
        self.queries: list[str] = []

    def execute_sql_query(self, sql: str) -> dict:
        self.queries.append(sql)
        # The CSS lookup query — return a fake pkid
        if "callingsearchspace WHERE name" in sql:
            return {"row_count": 1, "rows": [{"pkid": "fake-css-pkid"}]}
        # Any query referencing an "error trigger" column → simulate failure
        for trigger in self.error_on_columns:
            if trigger in sql:
                raise RuntimeError(f"simulated cluster failure on {trigger}")
        # Otherwise return one fake reference row so the category isn't empty
        return {
            "row_count": 1,
            "rows": [{"name": "FakeRef", "context": "FakePart", "description": "fake"}],
        }


def test_no_errors_reports_complete():
    """Baseline: when every category succeeds, complete=True and no error fields populated."""
    client = FakeAxlClient()
    result = find_devices_using_css(client, "Some-CSS")
    assert result["complete"] is True
    assert result["categories_with_errors"] == 0
    assert result["error_categories"] == []
    # And total_returned reflects the successful categories
    assert result["total_returned"] >= 1


def test_one_errored_category_marks_incomplete():
    """The audit-trust failure mode: one category errors out and the summary lies.
    Fix: complete=False, categories_with_errors >= 1.
    """
    client = FakeAxlClient(error_on_columns=["fkcallingsearchspace_cgpnunknown"])
    result = find_devices_using_css(client, "Some-CSS")
    assert result["complete"] is False, (
        "complete must be False when any category errored"
    )
    assert result["categories_with_errors"] >= 1
    assert "device_cgpn_unknown_css" in result["error_categories"]


def test_multiple_errors_all_listed():
    """All errored categories must be enumerated in error_categories.

    After issue #1 fix, several column suffixes (`_cgpnunknown`,
    `_reroute`) appear on BOTH the device and devicepool tables, so
    a single suffix in error_on_columns hits multiple categories.
    The test verifies the relevant categories are surfaced.
    """
    client = FakeAxlClient(
        error_on_columns=[
            "fkcallingsearchspace_pilotqueuefull",  # huntpilotqueue only
        ]
    )
    result = find_devices_using_css(client, "Some-CSS")
    assert result["complete"] is False
    assert result["categories_with_errors"] >= 1
    assert "huntpilot_queue_full_css" in result["error_categories"]


def test_error_in_multiple_tables_propagates():
    """A column suffix shared across tables (e.g., `_cgpnunknown` on both
    device AND devicepool) errors out in BOTH categories — both must
    appear in error_categories."""
    client = FakeAxlClient(
        error_on_columns=["fkcallingsearchspace_cgpnunknown"],
    )
    result = find_devices_using_css(client, "Some-CSS")
    assert "device_cgpn_unknown_css" in result["error_categories"]
    assert "devicepool_cgpn_unknown_css" in result["error_categories"]


def test_total_returned_does_not_include_error_categories():
    """An errored category contributes 0 to total_returned (correct behavior).
    What's NEW: the response also flags that the count is partial.
    """
    client = FakeAxlClient(error_on_columns=["fkcallingsearchspace_cgpnunknown"])
    result = find_devices_using_css(client, "Some-CSS")
    # The count itself is unchanged from before — what's new is the warning
    assert result["complete"] is False
    # The error category has no rows in references_by_category
    err_cat = result["references_by_category"].get("device_cgpn_unknown_css", {})
    assert "error" in err_cat


def test_css_not_found_returns_error_not_partial():
    """If the CSS lookup itself fails (CSS doesn't exist), we return the
    'not found' error early, NOT a partial-failure response. Distinct
    failure modes deserve distinct shapes.
    """

    class CssNotFoundClient:
        def execute_sql_query(self, sql):
            if "callingsearchspace WHERE name" in sql:
                return {"row_count": 0, "rows": []}
            return {"row_count": 1, "rows": [{}]}

    result = find_devices_using_css(CssNotFoundClient(), "Nonexistent-CSS")
    assert "error" in result
    assert "complete" not in result, (
        "CSS-not-found is a hard error; we shouldn't dress it up as partial"
    )


# ---- Issue #1 regression tests --------------------------------------------
# https://git.supported.systems/bingham/mcp-cucm-axl/issues/1
#
# Pre-fix: route_devices_using_css missed device.fkcallingsearchspace_cgpntransform
# and device.fkcallingsearchspace_cdpntransform. These are the columns trunks
# use to attach calling-party / called-party transformation CSSs. A CSS only
# referenced via these columns showed up as "0 references" — operator running
# impact analysis would conclude safe-to-delete and break outbound transforms.

from mcaxl.route_plan import _CSS_REFERENCE_QUERIES


def test_issue_1_cgpntransform_column_enumerated():
    """The specific column that triggered the issue is in our reference set."""
    columns = {
        (spec["table"], spec["column"])
        for spec in _CSS_REFERENCE_QUERIES.values()
    }
    assert ("device", "fkcallingsearchspace_cgpntransform") in columns, (
        "device.fkcallingsearchspace_cgpntransform must be enumerated; "
        "see Gitea issue #1 — false-zero impact analysis on calling-party "
        "transformation CSSs (e.g., XFORM-Outbound-ANI)"
    )


def test_issue_1_cdpntransform_column_enumerated():
    """The sibling column (called-party transformation) is also enumerated."""
    columns = {
        (spec["table"], spec["column"])
        for spec in _CSS_REFERENCE_QUERIES.values()
    }
    assert ("device", "fkcallingsearchspace_cdpntransform") in columns, (
        "device.fkcallingsearchspace_cdpntransform must be enumerated; "
        "same bug pattern as _cgpntransform (issue #1)"
    )


def test_finds_trunk_via_cgpntransform_reference():
    """End-to-end: a trunk referencing a CSS via _cgpntransform should
    appear in the impact analysis."""

    class TrunkRefClient:
        """Returns 1 row only when queried for fkcallingsearchspace_cgpntransform."""
        def execute_sql_query(self, sql):
            if "callingsearchspace WHERE name" in sql:
                return {"row_count": 1, "rows": [{"pkid": "fake-css-pkid"}]}
            if "fkcallingsearchspace_cgpntransform" in sql:
                return {
                    "row_count": 1,
                    "rows": [{
                        "name": "PSTN-Router-SIP-Trk",
                        "context": "Trunk",
                        "description": "the trunk that references this CSS",
                    }],
                }
            return {"row_count": 0, "rows": []}

    result = find_devices_using_css(TrunkRefClient(), "XFORM-Outbound-ANI")
    # Total must be ≥ 1 (the trunk reference), not 0
    assert result["total_returned"] >= 1, (
        "trunk referenced via _cgpntransform must surface in total_returned"
    )
    # And the specific category should be populated
    cgpn_cat = result["references_by_category"].get("device_cgpn_xform_css")
    assert cgpn_cat is not None and cgpn_cat.get("returned_count") == 1, (
        f"device_cgpn_xform_css category should have 1 row; "
        f"got: {result['references_by_category']}"
    )


# ---- Comprehensive coverage --------------------------------------------
# The 71-column snapshot from CUCM 15.0.1.12900-234. If a future CUCM
# version adds a new fkcallingsearchspace_* column, this test fires red
# so the contributor knows to add it to _CSS_REFERENCE_QUERIES.

# Format: (table, column). Sourced from a SELECT against syscolumns
# 2026-04-26. Update when a new CUCM release lands.
_KNOWN_CSS_COLUMNS_FROM_CUCM_15 = frozenset({
    # device — primary + 16 variants
    ("device", "fkcallingsearchspace"),
    ("device", "fkcallingsearchspace_aar"),
    ("device", "fkcallingsearchspace_calledintl"),
    ("device", "fkcallingsearchspace_callednational"),
    ("device", "fkcallingsearchspace_calledsubscriber"),
    ("device", "fkcallingsearchspace_calledunknown"),
    ("device", "fkcallingsearchspace_cdpntransform"),
    ("device", "fkcallingsearchspace_cgpningressdn"),
    ("device", "fkcallingsearchspace_cgpnintl"),
    ("device", "fkcallingsearchspace_cgpnnational"),
    ("device", "fkcallingsearchspace_cgpnsubscriber"),
    ("device", "fkcallingsearchspace_cgpntransform"),
    ("device", "fkcallingsearchspace_cgpnunknown"),
    ("device", "fkcallingsearchspace_rdntransform"),
    ("device", "fkcallingsearchspace_refer"),
    ("device", "fkcallingsearchspace_reroute"),
    ("device", "fkcallingsearchspace_restrict"),
    # devicenumplanmap
    ("devicenumplanmap", "fkcallingsearchspace_monitoring"),
    # devicepool — 16 variants (DP-level inheritance)
    ("devicepool", "fkcallingsearchspace_aar"),
    ("devicepool", "fkcallingsearchspace_adjunct"),
    ("devicepool", "fkcallingsearchspace_autoregistration"),
    ("devicepool", "fkcallingsearchspace_calledintl"),
    ("devicepool", "fkcallingsearchspace_callednational"),
    ("devicepool", "fkcallingsearchspace_calledsubscriber"),
    ("devicepool", "fkcallingsearchspace_calledunknown"),
    ("devicepool", "fkcallingsearchspace_cdpntransform"),
    ("devicepool", "fkcallingsearchspace_cgpningressdn"),
    ("devicepool", "fkcallingsearchspace_cgpnintl"),
    ("devicepool", "fkcallingsearchspace_cgpnnational"),
    ("devicepool", "fkcallingsearchspace_cgpnsubscriber"),
    ("devicepool", "fkcallingsearchspace_cgpntransform"),
    ("devicepool", "fkcallingsearchspace_cgpnunknown"),
    ("devicepool", "fkcallingsearchspace_cntdpntransform"),
    ("devicepool", "fkcallingsearchspace_mobility"),
    ("devicepool", "fkcallingsearchspace_rdntransform"),
    # External call control + h323 + sipdevice + huntpilotqueue
    ("externalcallcontrolprofile", "fkcallingsearchspace_diversionrerouting"),
    ("h323device", "fkcallingsearchspace_cntdpntransform"),
    ("sipdevice", "fkcallingsearchspace_cntdpntransform"),
    ("huntpilotqueue", "fkcallingsearchspace_maxwaittime"),
    ("huntpilotqueue", "fkcallingsearchspace_noagent"),
    ("huntpilotqueue", "fkcallingsearchspace_pilotqueuefull"),
    # incomingtransformationprofile (4)
    ("incomingtransformationprofile", "fkcallingsearchspace_intl"),
    ("incomingtransformationprofile", "fkcallingsearchspace_national"),
    ("incomingtransformationprofile", "fkcallingsearchspace_subscriber"),
    ("incomingtransformationprofile", "fkcallingsearchspace_unknown"),
    # numplan — 18 forwarding/transformation CSSs
    ("numplan", "fkcallingsearchspace_cfapt"),
    ("numplan", "fkcallingsearchspace_cfb"),
    ("numplan", "fkcallingsearchspace_cfbint"),
    ("numplan", "fkcallingsearchspace_cfhr"),
    ("numplan", "fkcallingsearchspace_cfhrint"),
    ("numplan", "fkcallingsearchspace_cfna"),
    ("numplan", "fkcallingsearchspace_cfnaint"),
    ("numplan", "fkcallingsearchspace_cfur"),
    ("numplan", "fkcallingsearchspace_cfurint"),
    ("numplan", "fkcallingsearchspace_devicefailure"),
    ("numplan", "fkcallingsearchspace_mwi"),
    ("numplan", "fkcallingsearchspace_pff"),
    ("numplan", "fkcallingsearchspace_pffint"),
    ("numplan", "fkcallingsearchspace_pkmonfwdnoret"),
    ("numplan", "fkcallingsearchspace_pkmonfwdnoretint"),
    ("numplan", "fkcallingsearchspace_reroute"),
    ("numplan", "fkcallingsearchspace_revert"),
    ("numplan", "fkcallingsearchspace_sharedlineappear"),
    ("numplan", "fkcallingsearchspace_translation"),
    # Profile tables + simple primary fkcallingsearchspace
    ("recordingprofile", "fkcallingsearchspace_callrecording"),
    ("routelist", "fkcallingsearchspace"),
    ("site", "fkcallingsearchspace"),
    ("usageprofile", "fkcallingsearchspace_blocking"),
    ("vipre164transformation", "fkcallingsearchspace_outgoingcdpntranf"),
    ("vipre164transformation", "fkcallingsearchspace_outgoingcgpntranf"),
    ("voicemessagingpilot", "fkcallingsearchspace"),
})


def test_complete_schema_coverage_against_known_columns():
    """If CUCM adds a new column type or we missed one, this test surfaces it.
    Counts: 71 columns total in the CUCM 15.0.1.12900-234 snapshot.
    """
    actual = {
        (spec["table"], spec["column"])
        for spec in _CSS_REFERENCE_QUERIES.values()
    }
    missing = _KNOWN_CSS_COLUMNS_FROM_CUCM_15 - actual
    assert not missing, (
        f"_CSS_REFERENCE_QUERIES is missing {len(missing)} known columns:\n"
        + "\n".join(f"  {t}.{c}" for t, c in sorted(missing))
    )


def test_no_duplicate_table_column_pairs():
    """Each (table, column) pair should map to exactly one category label.
    Two categories pointing at the same column would double-count references."""
    seen: dict[tuple, list[str]] = {}
    for label, spec in _CSS_REFERENCE_QUERIES.items():
        key = (spec["table"], spec["column"])
        seen.setdefault(key, []).append(label)
    duplicates = {k: v for k, v in seen.items() if len(v) > 1}
    assert not duplicates, (
        f"duplicate (table, column) pairs would double-count:\n"
        + "\n".join(f"  {k}: {v}" for k, v in duplicates.items())
    )