mcaxl/tests/test_risport.py

"""Unit tests for the RisPort70 SOAP envelope construction and parser.

Live-cluster integration is verified separately via the smoke-test
script — these tests are pure: envelope shape, response-XML parsing,
edge cases. No network.
"""

import xml.etree.ElementTree as ET

import pytest

from mcp_cucm_axl.risport import (
    DEVICE_STATUS_VALUES,
    RisPortClient,
    _build_select_envelope,
    _escape_xml,
    _parse_response,
)


class TestEscapeXml:
    def test_basic_escapes(self):
        assert _escape_xml("<bad>") == "&lt;bad&gt;"
        assert _escape_xml("a&b") == "a&amp;b"
        assert _escape_xml('"x"') == "&quot;x&quot;"
        assert _escape_xml("'y'") == "&apos;y&apos;"

    def test_passthrough_safe_text(self):
        assert _escape_xml("Phone-1234") == "Phone-1234"


class TestSelectEnvelope:
    def test_envelope_is_well_formed_xml(self):
        env = _build_select_envelope()
        # Must parse — if not, RisPort will reject it
        root = ET.fromstring(env)
        assert root is not None

    def test_default_envelope_includes_required_fields(self):
        env = _build_select_envelope()
        # Cisco's WSDL requires every CmSelectionCriteria child
        for required in [
            "MaxReturnedDevices",
            "DeviceClass",
            "Model",
            "Status",
            "NodeName",
            "SelectBy",
            "SelectItems",
            "Protocol",
            "DownloadStatus",
        ]:
            assert f"<soap:{required}>" in env, (
                f"missing required CmSelectionCriteria field <soap:{required}>"
            )

    def test_state_info_threaded_into_envelope(self):
        env = _build_select_envelope(state_info="cursor-abc-123")
        assert "cursor-abc-123" in env
        assert "<soap:StateInfo>cursor-abc-123</soap:StateInfo>" in env

    def test_select_items_default_wildcard(self):
        env = _build_select_envelope()
        # Default: ['*'] — all devices
        assert "<soap:Item>*</soap:Item>" in env

    def test_select_items_explicit_list(self):
        env = _build_select_envelope(select_items=["SEP1", "SEP2"])
        assert "<soap:Item>SEP1</soap:Item>" in env
        assert "<soap:Item>SEP2</soap:Item>" in env

    def test_max_devices_int_coerced(self):
        env = _build_select_envelope(max_devices=500)
        assert "<soap:MaxReturnedDevices>500</soap:MaxReturnedDevices>" in env

    def test_xml_special_chars_in_filter_escaped(self):
        # SOAP injection defense — single quote and angle bracket must be escaped
        env = _build_select_envelope(select_items=["O'Brien <test>"])
        assert "&apos;" in env
        assert "&lt;" in env
        assert "&gt;" in env
        # The raw chars must NOT appear
        assert "'Brien <test>" not in env


class TestParseResponse:
    # Match CUCM 15's actual response shape: an extra <SelectCmDeviceResult>
    # wrapper inside <selectCmDeviceReturn>. Discovered while live-verifying
    # against cucm-pub.binghammemorial.org 2026-04-26 — the parser must
    # descend through this wrapper.
    SAMPLE_RESPONSE = """<?xml version="1.0"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
<soapenv:Body>
<ns:selectCmDeviceResponse xmlns:ns="http://schemas.cisco.com/ast/soap">
<selectCmDeviceReturn>
<SelectCmDeviceResult>
  <TotalDevicesFound>2</TotalDevicesFound>
  <StateInfo></StateInfo>
  <CmNodes>
    <item>
      <Name>cucm-pub</Name>
      <ReturnCode>Ok</ReturnCode>
      <CmDevices>
        <item>
          <Name>SEP1234567890AB</Name>
          <IpAddress><item><IP>10.0.0.5</IP><IPAddrType>ipv4</IPAddrType></item></IpAddress>
          <DirNumber>2001</DirNumber>
          <Status>Registered</Status>
          <StatusReason>0</StatusReason>
          <Protocol>SIP</Protocol>
          <Description>Patient Room 101</Description>
          <Model>Cisco 7841</Model>
          <ActiveLoadID>sip78xx.12-5-1SR4-3</ActiveLoadID>
          <TimeStamp>1700000000</TimeStamp>
        </item>
        <item>
          <Name>SEPABCDEF123456</Name>
          <IpAddress></IpAddress>
          <DirNumber></DirNumber>
          <Status>UnRegistered</Status>
          <StatusReason>1</StatusReason>
          <Description>Decommissioned phone</Description>
        </item>
      </CmDevices>
    </item>
  </CmNodes>
</SelectCmDeviceResult>
</selectCmDeviceReturn>
</ns:selectCmDeviceResponse>
</soapenv:Body>
</soapenv:Envelope>"""

    # Pre-CUCM-15 shape (no SelectCmDeviceResult wrapper). The parser falls
    # back to fields directly under selectCmDeviceReturn for backward compat.
    LEGACY_RESPONSE = """<?xml version="1.0"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
<soapenv:Body>
<ns:selectCmDeviceResponse xmlns:ns="http://schemas.cisco.com/ast/soap">
<selectCmDeviceReturn>
  <TotalDevicesFound>1</TotalDevicesFound>
  <StateInfo></StateInfo>
  <CmNodes>
    <item>
      <Name>cucm-old</Name>
      <ReturnCode>Ok</ReturnCode>
      <CmDevices>
        <item>
          <Name>SEPLEGACY</Name>
          <Status>Registered</Status>
        </item>
      </CmDevices>
    </item>
  </CmNodes>
</selectCmDeviceReturn>
</ns:selectCmDeviceResponse>
</soapenv:Body>
</soapenv:Envelope>"""

    def test_legacy_response_shape_still_parses(self):
        """Backward compat with pre-CUCM-15 RisPort responses."""
        result = _parse_response(self.LEGACY_RESPONSE)
        assert result["total_devices_found"] == 1
        assert result["cm_nodes"][0]["devices"][0]["name"] == "SEPLEGACY"

    def test_parse_total_count(self):
        result = _parse_response(self.SAMPLE_RESPONSE)
        assert result["total_devices_found"] == 2

    def test_parse_node_count(self):
        result = _parse_response(self.SAMPLE_RESPONSE)
        assert len(result["cm_nodes"]) == 1
        assert result["cm_nodes"][0]["name"] == "cucm-pub"

    def test_parse_device_fields(self):
        result = _parse_response(self.SAMPLE_RESPONSE)
        devices = result["cm_nodes"][0]["devices"]
        assert len(devices) == 2

        first = devices[0]
        assert first["name"] == "SEP1234567890AB"
        assert first["status"] == "Registered"
        assert first["dir_number"] == "2001"
        assert first["description"] == "Patient Room 101"
        assert first["protocol"] == "SIP"
        # Nested IP extraction
        assert first["ip_address"] == "10.0.0.5"

    def test_parse_unregistered_device(self):
        result = _parse_response(self.SAMPLE_RESPONSE)
        second = result["cm_nodes"][0]["devices"][1]
        assert second["status"] == "UnRegistered"
        assert second["ip_address"] == ""  # missing IP renders empty

    def test_parse_state_info_for_pagination(self):
        result = _parse_response(self.SAMPLE_RESPONSE)
        assert result["state_info"] == ""  # last page

    def test_parse_soap_fault_raises(self):
        fault_response = """<?xml version="1.0"?>
<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
<soapenv:Body>
<soapenv:Fault>
<faultstring>Authentication failed</faultstring>
</soapenv:Fault>
</soapenv:Body>
</soapenv:Envelope>"""
        with pytest.raises(RuntimeError, match="Authentication failed"):
            _parse_response(fault_response)


class TestStatusValidation:
    def test_known_status_values(self):
        assert "Registered" in DEVICE_STATUS_VALUES
        assert "UnRegistered" in DEVICE_STATUS_VALUES
        assert "PartiallyRegistered" in DEVICE_STATUS_VALUES

    def test_select_cm_device_rejects_invalid_status(self):
        client = RisPortClient()
        # No env vars set, so _ensure_session would fail first;
        # but the validation should run BEFORE that on bad input.
        with pytest.raises(ValueError, match="status must be"):
            client.select_cm_device(status="not-a-real-status")