feat: add JobD helper

Author: paulschwoerer

sipgate_e2e_test_utils/jobd.py

@@ -0,0 +1,47 @@
+import os
+from typing import Any
+
+import aiohttp
+import socket
+from http_request_recorder import HttpRequestRecorder
+
+from sipgate_e2e_test_utils.rpc_matchers import xml_rpc
+from sipgate_e2e_test_utils.xml_rpc import XmlRpcRequest, XmlRpcResponse
+
+with open(os.path.join(os.path.dirname(__file__), 'jobd_functions.xml')) as file:
+    JOBD_FUNCTIONS_XML = file.read()
+
+
+class JobD:
+    def __init__(self, system_hostname: str, system_port: int) -> None:
+        self.notification_url = f"http://{socket.gethostname()}:8777/RPC2"
+        self.system_url = f"http://{system_hostname}:{system_port}/RPC2"
+
+    async def __aenter__(self) -> "JobD":
+        self.recorder = HttpRequestRecorder(name='JobD', port=8777)
+        self.recorder.expect_path(path='/functions.xml',
+                                  responses=(JOBD_FUNCTIONS_XML for _ in range(100)))
+
+        self.session = aiohttp.ClientSession()
+
+        await self.recorder.__aenter__()
+        return self
+
+    async def trigger_job_and_record_answer(self, job_name: str, timeout: int = 10) -> bytes:
+        expectation = self.recorder.expect(
+            xml_rpc('jobd.updateEvent'), responses=XmlRpcResponse(200, 'ok').serialize(), timeout=timeout)
+
+        response = await self.session.post(self.system_url, data=XmlRpcRequest('cron.triggerJob', {
+            'jobName': job_name,
+            'notificationUrl': self.notification_url,
+            'uniqueid': 42
+        }).serialize())
+        assert 200 == response.status
+
+        recorded_request: bytes = await expectation.wait()
+        return recorded_request
+
+    async def __aexit__(self, *args: tuple[Any]) -> None:
+        await self.session.close()
+        await_aexit__: None = await self.recorder.__aexit__(*args)
+        return await_aexit__

sipgate_e2e_test_utils/jobd_functions.xml

@@ -0,0 +1,68 @@
+<?xml version="1.0" encoding="utf-8"?>
+<collection version="0.01">
+
+    <!-- NAGIOS PING FUNCTION -->
+    <interface id="selfTest.testAll">
+        <description>veranlasst selbsttest und liefert fehler-/erfolgsprotokoll</description>
+        <input />
+        <output>
+            <scalar name="health" type="boolean" />
+            <array name="protocol">
+                <struct name="*">
+                    <scalar name="name" type="string" />
+                    <scalar name="exception" type="string" />
+                    <scalar name="success" type="boolean" />
+                    <scalar name="description" type="string" />
+                    <scalar name="critical" type="boolean" />
+                </struct>
+            </array>
+        </output>
+    </interface>
+
+    <interface id="daemon.shutdown">
+        <input />
+        <output />
+    </interface><interface id="daemon.uptime">
+    <input />
+    <output>
+        <struct name="uptime" optional="no">
+            <scalar name="seconds" type="int" optional="no" />
+        </struct>
+    </output>
+</interface><interface id="jobd.updateEvent">
+    <input>
+        <scalar name="systemName" type="string" optional="no" />
+        <scalar name="eventName" type="string" optional="no" />
+        <scalar name="uniqueid" type="string" optional="no" />
+        <scalar name="status" type="string" optional="no" />
+    </input>
+    <output />
+</interface><interface id="jobd.directTouch">
+    <input>
+        <scalar name="systemName" type="string" optional="no" />
+        <scalar name="uniqueid" type="string" optional="no" />
+        <scalar name="serviceIP" type="string" optional="yes" />
+    </input>
+    <output />
+</interface><interface id="jobd.listSystems">
+    <input>
+    </input>
+    <output>
+        <array name="systems" optional="no">
+            <scalar type="string" />
+        </array>
+    </output>
+</interface><interface id="jobd.listCriticalSystems">
+    <input>
+        <scalar type="boolean" name="confirmJobs" />
+    </input>
+    <output>
+        <struct name="systems" optional="no">
+            <struct name="*">
+                <array name="criticalJobs">
+                    <scalar type="string" />
+                </array>
+            </struct>
+        </struct>
+    </output>
+</interface></collection>

tests/test_jobd.py

@@ -0,0 +1,62 @@
+import asyncio
+import unittest
+from asyncio import create_task
+
+from aiohttp import web, ClientSession
+
+from sipgate_e2e_test_utils.jobd import JobD
+
+from sipgate_e2e_test_utils.xml_rpc import XmlRpcRequest, XmlRpcResponse
+
+
+class TestJobD(unittest.IsolatedAsyncioTestCase):
+    async def test_serves_functions_xml(self):
+        any_port = 42
+        async with (JobD('localhost', any_port), ClientSession() as http):
+            response = await http.get('http://localhost:8777/functions.xml')
+
+            self.assertIn('jobd.updateEvent', await response.text())
+
+    async def test_returns_success(self):
+        mock_service = MockService(46968)
+        await mock_service.start()
+
+        async with (JobD('localhost', mock_service.port) as jobd):
+            job_result = await jobd.trigger_job_and_record_answer('any_job')
+
+            self.assertIn(b'any_value', job_result)
+
+        await mock_service.stop()
+
+
+class MockService:
+    def __init__(self, port: int) -> None:
+        self.port = port
+
+        app = web.Application()
+        app.add_routes([web.post('/RPC2', self.__handle_request)])
+        self.runner = web.AppRunner(app)
+
+    async def start(self):
+        await self.runner.setup()
+        site = web.TCPSite(self.runner, '127.0.0.1', self.port)
+        await site.start()
+
+    async def __handle_request(self, request: web.BaseRequest) -> web.Response:
+        xml_rpc_request = XmlRpcRequest.parse(await request.text())
+        assert 'cron.triggerJob' == xml_rpc_request.method_name
+
+        create_task(self.__send_result_to_jobd_after_some_time(xml_rpc_request.members['notificationUrl']))
+
+        return web.Response(status=200, body=XmlRpcResponse(200, 'OK', {}).serialize())
+
+    async def __send_result_to_jobd_after_some_time(self, notification_url: str):
+        await asyncio.sleep(0.25)
+
+        async with ClientSession() as http:
+            await http.post(notification_url, data=XmlRpcRequest('jobd.updateEvent', {
+                'any_field': 'any_value'
+            }).serialize())
+
+    async def stop(self):
+        await self.runner.cleanup()