Fix #353, added minor_servos minimal REST Api

simulators/minor_servos/__init__.py

@@ -16,8 +16,9 @@
 from multiprocessing import Value
 from bisect import bisect_left
 from socketserver import ThreadingTCPServer
+from http.server import HTTPServer
 from simulators.common import ListeningSystem
-from simulators.minor_servos.helpers import setup_import
+from simulators.minor_servos.helpers import setup_import, VBrainRequestHandler
 
 
 # Each system module (like active_surface.py, acu.py, etc.) has to
@@ -28,6 +29,7 @@
 # subscribe and unsibscribe methods, while kwargs is a dict of optional
 # extra arguments.
 servers = [(('0.0.0.0', 12800), (), ThreadingTCPServer, {})]
+httpserver_address = ('0.0.0.0', 12799)
 DEFAULT_TIMER_VALUE = 5
 
 
@@ -76,7 +78,7 @@ def plc_time(now=None):
         'BWG4': {'ID': 24},
     }
 
-    def __init__(self, timer_value=DEFAULT_TIMER_VALUE):
+    def __init__(self, timer_value=DEFAULT_TIMER_VALUE, rest_api=True):
         self.msg = ''
         self.configuration = 0
         self.simulation = 1
@@ -108,6 +110,16 @@ def __init__(self, timer_value=DEFAULT_TIMER_VALUE):
         )
         self.update_thread.daemon = True
         self.update_thread.start()
+        self.rest_api = rest_api
+        if self.rest_api:
+            self.httpserver = HTTPServer(
+                httpserver_address,
+                VBrainRequestHandler
+            )
+            self.server_thread = Thread(
+                target=self.httpserver.serve_forever
+            )
+            self.server_thread.start()
         self.cover_timer = Timer(1, lambda: None)
 
     def __del__(self):
@@ -123,6 +135,9 @@ def system_stop(self):
                 self.cover_timer.join()
             except RuntimeError:
                 pass
+        if self.rest_api:
+            self.httpserver.shutdown()
+            self.server_thread.join()
         return super().system_stop()
 
     @staticmethod

simulators/minor_servos/helpers.py

@@ -1,5 +1,7 @@
 import os
 import csv
+import json
+from http.server import BaseHTTPRequestHandler
 import pkg_resources
 
 
@@ -40,3 +42,31 @@ def setup_import(servos, configurations):
                         coord = None
                     coordinates.append(coord)
                 configurations[line[0]][servo] = coordinates
+
+
+class VBrainRequestHandler(BaseHTTPRequestHandler):
+
+    emergency = 'INAF_SRT_OR7_EMG_RESET_CMD'
+    alarm = 'INAF_SRT_OR7_RESET_CMD'
+    baseurl = '/Exporting/json/ExecuteCommand?name'
+    urls = [
+        f'{baseurl}={emergency}',
+        f'{baseurl}={alarm}'
+    ]
+    answer = {'Message': 'OUTPUT:GOOD', 'Status': 'Good'}
+
+    def do_GET(self):
+        try:
+            if self.path in self.urls:
+                self.send_response(200)
+                self.send_header('Content-type', 'application/json')
+                self.end_headers()
+                self.wfile.write(json.dumps(self.answer).encode())
+            else:
+                self.send_response(404)
+                self.end_headers()
+        except BrokenPipeError:  # skip coverage
+            pass
+
+    def log_message(self, *_):
+        pass

tests/test_minor_servos.py

@@ -1,7 +1,9 @@
 import unittest
 import random
 import time
-from simulators.minor_servos import System
+import requests
+from simulators.minor_servos import System, httpserver_address
+from simulators.minor_servos.helpers import VBrainRequestHandler
 
 TIMER_VALUE = 0.01
 tail = '\r\n'
@@ -12,7 +14,10 @@
 class TestMinorServos(unittest.TestCase):
 
     def setUp(self):
-        self.system = System(timer_value=TIMER_VALUE)
+        self.system = System(
+            timer_value=TIMER_VALUE,
+            rest_api=False
+        )
 
     def tearDown(self):
         del self.system
@@ -554,5 +559,36 @@ def test_offset_non_numeric_coordinates(self):
             self.assertRegex(self.system.parse(cmd[-1]), bad)
 
 
+class TestRESTApi(unittest.TestCase):
+
+    def setUp(self):
+        self.system = System(
+            rest_api=True
+        )
+
+    def tearDown(self):
+        del self.system
+
+    def test_rest_GET(self):
+        for url in VBrainRequestHandler.urls:
+            baseurl = f'http://{httpserver_address[0]}:{httpserver_address[1]}'
+            url = baseurl + url
+            try:
+                response = requests.get(url, timeout=TIMER_VALUE)
+                self.assertEqual(response.status_code, 200)
+                self.assertEqual(response.json(), VBrainRequestHandler.answer)
+            except requests.exceptions.ReadTimeout:
+                self.fail('Request is taking too long to be answered')
+
+    def test_rest_GET_wrong_address(self):
+        baseurl = f'http://{httpserver_address[0]}:{httpserver_address[1]}'
+        url = baseurl + '/wrong'
+        try:
+            response = requests.get(url, timeout=TIMER_VALUE)
+            self.assertEqual(response.status_code, 404)
+        except requests.exceptions.ReadTimeout:
+            self.fail('Request is taking too long to be answered')
+
+
 if __name__ == '__main__':
     unittest.main()