feat: add XML RPC (de-)serialization

paulschwoerer · paulschwoerer · commit 58f6a7563962 · 2025-08-22T11:57:55.000+02:00
diff --git a/pyproject.toml b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "sipgate_e2e_test_utils"
-version = "0.0.1"
+version = "0.1.0"
 authors = []
 description = "A library for common utils used in E2E testing (XML-, JSON-RPC, awaiting assertions, JobD helpers)"
 readme = "README.md"
diff --git a/sipgate_e2e_test_utils/xml_rpc.py b/sipgate_e2e_test_utils/xml_rpc.py
@@ -0,0 +1,193 @@
+import base64
+from dataclasses import dataclass, field
+from typing import Any, Callable
+from xml.etree import ElementTree
+from xml.etree.ElementTree import Element
+
+
+@dataclass
+class XmlRpcRequest:
+    """
+    types not implemented (as not widely used for requests):
+    - double
+    - datetime.iso8601
+    """
+
+    method_name: str
+    members: dict[str, Any] = field(default_factory=dict)
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} methodName='{self.method_name}' members={self.members}>"
+
+    @staticmethod
+    def parse(body: str | bytes) -> 'XmlRpcRequest':
+        return _parse_xml_rpc_request(body)
+
+    def serialize(self) -> str:
+        return _serialize_xml_rpc_request(self)
+
+
+@dataclass
+class XmlRpcResponse:
+    """
+    types not implemented (as not widely used for responses):
+    - double
+    - datetime.iso8601
+    """
+    fault_code: int
+    fault_string: str
+    members: dict[str, Any]
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} faultCode={self.fault_code} faultString={self.fault_string} members={self.members}>"
+
+    @staticmethod
+    def parse(body: str | bytes) -> 'XmlRpcResponse':
+        return _parse_xml_rpc_response(body)
+
+
+def _parse_xml_rpc_request(body: str | bytes) -> XmlRpcRequest:
+    root = ElementTree.fromstring(body)
+    if root.tag != 'methodCall':
+        raise ValueError("Expecting root tag to be '<methodCall>'")
+
+    method_name = root.find('methodName')
+    if method_name is None or method_name.text is None:
+        raise ValueError("Expected to find non-empty '<methodName>'")
+
+    members = {}
+    for member_node in root.findall('params/param/value/struct/member'):
+        key, val = __parse_member(member_node)
+        members[key] = val
+
+    return XmlRpcRequest(method_name.text, members)
+
+
+def _serialize_xml_rpc_request(request: XmlRpcRequest) -> str:
+    root_param = '' if request.members == {} else f'<param><value>{__serialize_struct(request.members)}</value></param>'
+
+    # TODO: omit whitespace?
+    return f"""<?xml version="1.0"?>
+        <methodCall>
+            <methodName>{request.method_name}</methodName>
+            <params>{root_param}</params>
+        </methodCall>"""
+
+
+def _parse_xml_rpc_response(body: str | bytes) -> XmlRpcResponse:
+    root = ElementTree.fromstring(body)
+    if root.tag != 'methodResponse':
+        raise ValueError("Expecting root tag to be '<methodResponse>'")
+
+    value = root.find('params/param/value/struct')
+    if value is None:
+        value = root.find('fault/value/struct')
+
+    if value is None:
+        raise ValueError("Expecting to find a value")
+
+    members = __parse_struct(value)
+    return XmlRpcResponse(int(members.pop('faultCode')), str(members.pop('faultString')), members)
+
+
+def __parse_member(node: Element) -> tuple[str, Any]:
+    assert node.tag == 'member', f"expected 'member', but got {node.tag=}"
+
+    name_node = node.find('name')
+    if name_node is None or name_node.text is None:
+        raise ValueError("Expected to find non-empty '<member><name>'")
+
+    value_node = node.find('value')
+    if value_node is None:
+        raise ValueError("Expected to find '<member><value>'")
+
+    return name_node.text, __parse_value(value_node)
+
+
+def __parse_value(node: Element) -> Any:
+    assert node.tag == 'value', f"expected 'value', but got {node.tag=}"
+
+    typed_value = node.find('*')
+    if typed_value is None:
+        # it is allowed to omit the <string></string> type
+        return '' if node.text is None else node.text
+
+    val_parsers: dict[str, Callable[[Element], Any]] = {
+        'int': lambda n: int('' if n.text is None else n.text),
+        'i4': lambda n: int('' if n.text is None else n.text),
+        'boolean': lambda n: __parse_boolean(n.text),
+        'string': lambda n: '' if n.text is None else n.text,
+        'struct': lambda n: __parse_struct(n),
+        'array': lambda n: __parse_array(n),
+        'base64': lambda n: b'' if n.text is None else base64.b64decode(n.text),
+    }
+
+    try:
+        parser = val_parsers[typed_value.tag]
+    except KeyError:
+        raise NotImplementedError(f"Unsupported type '<{typed_value.tag}>'")
+
+    try:
+        return parser(typed_value)
+    except Exception:
+        raise ValueError(f"Invalid value '{typed_value.text}' for <{typed_value.tag}>")
+
+
+def __parse_boolean(text: str | None) -> bool:
+    if text == '0':
+        return False
+
+    if text == '1':
+        return True
+
+    raise ValueError(f"expected '{text}' to be '0' or '1'")
+
+
+def __parse_array(node: Element) -> list[Any]:
+    assert node.tag == 'array', f"expected 'array', but got {node.tag=}"
+    array_values = node.findall('data/value')
+
+    return [__parse_value(struct_node) for struct_node in array_values]
+
+
+def __parse_struct(node: Element) -> dict[str, Any]:
+    assert node.tag == 'struct', f"expected 'struct', but got {node.tag=}"
+    member_nodes = node.findall('member')
+
+    return {k: v for k, v in map(__parse_member, member_nodes)}
+
+
+def __serialize_struct(struct: dict[str, Any]) -> str:
+    members = [__serialize_member(key, struct[key]) for key in struct]
+
+    return f'<struct>{"\n".join(members)}</struct>'
+
+
+def __serialize_array(array: list[Any]) -> str:
+    values = [__serialize_value(val) for val in array]
+
+    return f'<array><data>{"\n".join(values)}</data></array>'
+
+
+def __serialize_member(name: str, value: Any) -> str:
+    if value is None:
+        return ''  # TODO: test
+
+    return f"<member><name>{name}</name>{__serialize_value(value)}</member>"
+
+
+def __serialize_value(value: Any) -> str:
+    val_mappers: dict[type, Callable[[Any], str]] = {
+        int: lambda val: f'<i4>{val}</i4>',
+        str: lambda val: f'<string>{val}</string>',
+        bool: lambda val: f'<boolean>{val}</boolean>',
+        dict: lambda val: __serialize_struct(value),
+        list: lambda val: __serialize_array(value)
+    }
+
+    try:
+        val_mapper = val_mappers[type(value)]
+    except KeyError:
+        raise NotImplementedError(f'unsupported data type ({type(value)}) when serializing {value}')
+
+    return f"<value>{val_mapper(value)}</value>"
diff --git a/tests/test_xml_rpc_request.py b/tests/test_xml_rpc_request.py
@@ -0,0 +1,180 @@
+from unittest import TestCase
+from xml.etree.ElementTree import ParseError
+
+from sipgate_e2e_test_utils.xml_rpc import XmlRpcRequest
+
+
+class TestSipgateXmlRpcRequest(TestCase):
+    def test_empty_body(self) -> None:
+        with self.assertRaises(ParseError):
+            XmlRpcRequest.parse('')
+
+    def test_non_xml_body(self) -> None:
+        with self.assertRaises(ParseError):
+            XmlRpcRequest.parse("{'key': 'value'}")
+
+    def test_response_is_not_parsed_as_request(self) -> None:
+        body = """<?xml version="1.0" encoding="UTF-8"?>
+            <methodResponse>
+                <params><param><value>
+                    <struct>
+                        <member><name>faultString</name><value><string>OK</string></value></member>
+                        <member><name>faultCode</name><value><i4>200</i4></value></member>
+                    </struct>
+                </value></param></params>
+            </methodResponse>"""
+
+        with self.assertRaises(ValueError):
+            XmlRpcRequest.parse(body)
+
+    def test_method_name(self) -> None:
+        body = """<?xml version="1.0"?>
+                    <methodCall>
+                        <methodName>a_method_name</methodName>
+                        <params><param><value></value></param></params>
+                    </methodCall>"""
+
+        parsed = XmlRpcRequest.parse(body)
+        self.assertEqual('a_method_name', parsed.method_name)
+
+    def test_valid_members(self) -> None:
+        body = """<?xml version="1.0"?>
+            <methodCall>
+                <methodName>a_method_name</methodName>
+                <params><param><value>
+                    <struct>
+                        <member><name>TNB</name><value><string>D222</string></value></member>
+                        <member><name>port_date</name><value><string>2025-01-01</string></value></member>
+                        <member><name>emptyString</name><value><string></string></value></member>
+                        <member><name>stringWithoutType</name><value>thisIsAlsoAString</value></member>
+                        <member><name>bool_true</name><value><boolean>1</boolean></value></member>
+                        <member><name>bool_false</name><value><boolean>0</boolean></value></member>
+                        <member><name>i4</name><value><i4>-23</i4></value></member>
+                        <member><name>int</name><value><int>42</int></value></member>
+                        <member><name>base64_empty</name><value><base64></base64></value></member>
+                        <member><name>base64_data</name><value><base64>YW55X2RhdGE=</base64></value></member>
+                    </struct>
+                </value></param></params>
+            </methodCall>"""
+
+        parsed = XmlRpcRequest.parse(body)
+        self.assertEqual('D222', parsed.members['TNB'])
+        self.assertEqual('2025-01-01', parsed.members['port_date'])
+        self.assertEqual('', parsed.members['emptyString'])
+        self.assertEqual('thisIsAlsoAString', parsed.members['stringWithoutType'])
+        self.assertEqual(True, parsed.members['bool_true'])
+        self.assertEqual(False, parsed.members['bool_false'])
+        self.assertEqual(-23, parsed.members['i4'])
+        self.assertEqual(42, parsed.members['int'])
+        self.assertEqual(b'', parsed.members['base64_empty'])
+        self.assertEqual(b'any_data', parsed.members['base64_data'])
+
+    def test_invalid_members(self) -> None:
+        members = [
+            {'name': 'bool_empty', 'value': '<boolean></boolean>'},
+            {'name': 'bool_string', 'value': '<boolean>true</boolean>'},
+            {'name': 'int_text', 'value': '<int>fortytwo</int>'},
+            {'name': 'int_empty', 'value': '<i4></i4>'}
+        ]
+
+        for member in members:
+            with self.subTest(member['name']):
+                body = f"""<?xml version="1.0"?>
+                    <methodCall>
+                        <methodName>a_method_name</methodName>
+                        <params><param><value>
+                            <struct>
+                                <member><name>{member['name']}</name><value>{member['value']}</value></member>
+                            </struct>
+                        </value></param></params>
+                    </methodCall>"""
+
+                with self.assertRaises(ValueError):
+                    XmlRpcRequest.parse(body)
+
+    def test_unsupported_data_type(self) -> None:
+        body = """<?xml version="1.0"?>
+                    <methodCall>
+                        <methodName>a_method_name</methodName>
+                        <params><param><value>
+                            <struct>
+                                <member><name>PI</name><value><float>3.14</float></value></member>
+                            </struct>
+                        </value></param></params>
+                    </methodCall>"""
+
+        with self.assertRaises(NotImplementedError):
+            XmlRpcRequest.parse(body)
+
+    def test_has_string_representation(self) -> None:
+        body = """<?xml version="1.0"?>
+            <methodCall>
+                <methodName>another_method_name</methodName>
+                <params><param><value>
+                    <struct>
+                        <member><name>TNB</name><value><string>D111</string></value></member>
+                    </struct>
+                </value></param></params>
+            </methodCall>"""
+
+        parsed = XmlRpcRequest.parse(body)
+        self.assertRegex(f"{parsed}", "^<XmlRpcRequest.*methodName='another_method_name'.*>$")
+        self.assertIn('TNB', f"{parsed}")
+        self.assertIn('D111', f"{parsed}")
+
+    def test_also_accepts_bytes(self) -> None:
+        body = b"""<?xml version="1.0"?>
+                    <methodCall>
+                        <methodName>a_method_name</methodName>
+                        <params><param><value></value></param></params>
+                    </methodCall>"""
+
+        XmlRpcRequest.parse(body)
+
+    def test_serialization_empty_request(self) -> None:
+        request = XmlRpcRequest('a_method', {})
+
+        expected_body = """<?xml version="1.0"?>
+            <methodCall>
+                <methodName>a_method</methodName>
+                <params></params>
+            </methodCall>"""
+
+        # TODO: use better comparison, this would ignore spaces in values
+        self.assertEqual(''.join(expected_body.split()), ''.join(request.serialize().split()))
+
+    def test_serialization(self) -> None:
+        request = XmlRpcRequest('a_method', {
+            'an_int': 42,
+            'a_string': 'the_value',
+            'a_struct': {
+                'another_int': 23
+            },
+            'an_array': [{
+                'another_str': 'the_other_value'
+            }]
+        })
+
+        expected_body = """<?xml version="1.0"?>
+            <methodCall>
+                <methodName>a_method</methodName>
+                <params><param><value>
+                    <struct>
+                        <member><name>an_int</name><value><i4>42</i4></value></member>
+                        <member><name>a_string</name><value><string>the_value</string></value></member>
+                        <member><name>a_struct</name><value><struct>
+                            <member><name>another_int</name><value><i4>23</i4></value></member>
+                        </struct></value></member>
+                        <member><name>an_array</name><value><array>
+                            <data>
+                                <value><struct>
+                                    <member><name>another_str</name><value><string>the_other_value</string></value></member>
+                                </struct></value>
+                            </data>
+                        </array></value></member>
+                    </struct>
+                </value></param></params>
+            </methodCall>"""
+
+        # TODO: use better comparison, this would ignore spaces in values
+        self.assertEqual(''.join(expected_body.split()), ''.join(request.serialize().split()))
diff --git a/tests/test_xml_rpc_response.py b/tests/test_xml_rpc_response.py
