feat: add RPC matchers

Author: paulschwoerer

.pre-commit-config.yaml

@@ -9,12 +9,3 @@ repos:
     rev: 'v2.3.2'
     hooks:
       - id: autopep8
-
-  - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: 'v1.17.1'
-    hooks:
-      - id: mypy
-        args: ['--strict']
-        additional_dependencies:
-          - types-setuptools
-          - typing_extensions

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "sipgate_e2e_test_utils"
-version = "0.1.0"
+version = "0.2.0"
 authors = []
 description = "A library for common utils used in E2E testing (XML-, JSON-RPC, awaiting assertions, JobD helpers)"
 readme = "README.md"

requirements.txt

@@ -1 +1,2 @@
+http_request_recorder @ git+https://github.com/sipgate/http-request-recorder.git@main
 pre-commit

sipgate_e2e_test_utils/rpc_matchers.py

@@ -0,0 +1,29 @@
+import json
+from typing import Callable
+
+from http_request_recorder import RecordedRequest
+
+
+def json_rpc(method: str) -> Callable[[RecordedRequest], bool]:
+    def matcher(request: RecordedRequest) -> bool:
+        if request.method != 'POST' or '/jsonrpc' != request.path.lower():
+            return False
+
+        try:
+            parsed = json.loads(request.body)
+        except json.decoder.JSONDecodeError:
+            return False
+
+        return 'method' in parsed and method == parsed['method']
+
+    return matcher
+
+
+def xml_rpc(method: str) -> Callable[[RecordedRequest], bool]:
+    def matcher(request: RecordedRequest) -> bool:
+        return (
+            'POST' == request.method and
+            '/rpc2' == request.path.lower() and
+            b'<methodName>' + method.encode() + b'</methodName>' in request.body)
+
+    return matcher

tests/test_rpc_matchers.py

@@ -0,0 +1,49 @@
+import json
+
+from http_request_recorder import RecordedRequest
+
+
+from unittest import TestCase
+from sipgate_e2e_test_utils.rpc_matchers import xml_rpc, json_rpc
+
+
+class TestSipgateRpcMatchers(TestCase):
+    def test_xml_rpc(self) -> None:
+        assertions = [
+            (False, ('POST', '/jsonrpc', json.dumps({'method': 'test_method', 'version': '1.1', 'params': [], 'id': 42}).encode())),
+            (False, ('POST', '/jsonrpc', b'<?xml version="1.0"?><methodCall><methodName>test_method</methodName><params><param><value></value></param></params></methodCall>')),
+            (False, ('GET', '/rpc2', b'<?xml version="1.0"?><methodCall><methodName>test_method</methodName><params><param><value></value></param></params></methodCall>')),
+            (False, ('POST', '/rpc2', b'anydata')),
+            (False, ('POST', '/rpc2', b'<?xml version="1.0"?><methodCall><methodName>another_method</methodName><params><param><value></value></param></params></methodCall>')),
+            (True, ('POST', '/rpc2', b'<?xml version="1.0"?><methodCall><methodName>test_method</methodName><params><param><value></value></param></params></methodCall>')),
+            (True, ('POST', '/RPC2', b'<?xml version="1.0"?><methodCall><methodName>test_method</methodName><params><param><value></value></param></params></methodCall>')),
+        ]
+
+        for (expected, (method, path, body)) in assertions:
+            with self.subTest(f'expect {expected} for method={method} path={path} body={body.decode()}'):
+                request = RecordedRequest()
+                request.method = method
+                request.path = path
+                request.body = body
+
+                self.assertEqual(expected, xml_rpc('test_method')(request))
+
+    def test_json_rpc(self) -> None:
+        assertions = [
+            (False, ('POST', '/rpc2', b'<?xml version="1.0"?><methodCall><methodName>test_method</methodName><params><param><value></value></param></params></methodCall>')),
+            (False, ('POST', '/rpc2', json.dumps({'method': 'test_method', 'version': '1.1', 'params': [], 'id': 42}).encode())),
+            (False, ('GET', '/jsonrpc', json.dumps({'method': 'test_method', 'version': '1.1', 'params': [], 'id': 42}).encode())),
+            (False, ('POST', '/jsonrpc', b'anydata')),
+            (False, ('POST', '/jsonrpc', json.dumps({'method': 'another_method', 'version': '1.1', 'params': [], 'id': 42}).encode())),
+            (True, ('POST', '/jsonrpc', json.dumps({'method': 'test_method', 'version': '1.1', 'params': [], 'id': 42}).encode())),
+            (True, ('POST', '/JSONRPC', json.dumps({'method': 'test_method', 'version': '1.1', 'params': [], 'id': 42}).encode())),
+        ]
+
+        for (expected, (method, path, body)) in assertions:
+            with self.subTest(f'expect {expected} for method={method} path={path} body={body.decode()}'):
+                request = RecordedRequest()
+                request.method = method
+                request.path = path
+                request.body = body
+
+                self.assertEqual(expected, json_rpc('test_method')(request))