Mock TheSkyX

pocs/hardware.py

@@ -1,6 +1,6 @@
 """Information about hardware supported by Panoptes."""
 
-ALL_NAMES = sorted(['camera', 'dome', 'mount', 'night', 'weather'])
+ALL_NAMES = sorted(['camera', 'dome', 'mount', 'night', 'weather', 'theskyx'])
 
 
 def get_all_names(all_names=ALL_NAMES, without=list()):

pocs/tests/data/theskyx.json

@@ -0,0 +1,18 @@
+{
+    "localhost": {
+        "3040": {
+            "5dd3bdd5e788cb8b3880563a7633fc96": {
+                "request": "/* Java Script */",
+                "response": "75 6E 64 65 66 69 6E 65 64 7C 4E 6F 20 65 72 72 6F 72 2E 20 45 72 72 6F 72 20 3D 20 30 2E"
+            },
+            "95c72a49c488d59f60c022fcfecf4382": {
+                "request": "FOOBAR",
+                "response": "7C 55 6E 6B 6E 6F 77 6E 20 63 6F 6D 6D 61 6E 64 2E 20 45 72 72 6F 72 20 3D 20 33 30 33 2E"
+            },
+            "a6ceb538a4029b586782d371aac2b947": {
+                "request": "\n/* Java Script */\nvar Out;\nOut=Application.version\n",
+                "response": "31 30 2E 35 2E 30 7C 4E 6F 20 65 72 72 6F 72 2E 20 45 72 72 6F 72 20 3D 20 30 2E"
+            }
+        }
+    }
+}

pocs/tests/test_theskyx_utils.py

@@ -0,0 +1,72 @@
+import os
+import pytest
+
+from mocket import Mocket
+
+from pocs.utils import error
+from pocs.utils.theskyx import TheSkyX
+
+
+@pytest.fixture(scope="function")
+def skyx(request):
+    """Create TheSkyX class but don't connect.t
+
+    If running with a real connection TheSkyX then the Mokcet will
+    be disabled here.
+    """
+
+    # Use `--with-hardware thesky` on cli to run without mock
+    Mocket.enable('theskyx', '{}/pocs/tests/data'.format(os.getenv('POCS')))
+    if 'theskyx' in pytest.config.getoption('--with-hardware'):
+        Mocket.disable()
+
+    theskyx = TheSkyX(connect=False)
+
+    yield theskyx
+
+
+def test_default_connect():
+    """Test connection to TheSkyX
+
+    If not running with a real connection then use Mocket
+    """
+    # Use `--with-hardware thesky` on cli to run without mock
+    if 'theskyx' not in pytest.config.getoption('--with-hardware'):
+        Mocket.enable('theskyx', '{}/pocs/tests/data'.format(os.getenv('POCS')))
+
+    skyx = TheSkyX()
+    assert skyx.is_connected is True
+
+
+def test_no_connect_write(skyx):
+    with pytest.raises(error.BadConnection):
+        skyx.write('/* Java Script */')
+
+
+def test_no_connect_read(skyx):
+    with pytest.raises(error.BadConnection):
+        skyx.read()
+
+
+def test_write_bad_key(skyx):
+    skyx.connect()
+    skyx.write('FOOBAR')
+    with pytest.raises(error.TheSkyXKeyError):
+        skyx.read()
+
+
+def test_write_no_command(skyx):
+    skyx.connect()
+    skyx.write('/* Java Script */')
+    assert skyx.read() == 'undefined'
+
+
+def test_get_build(skyx):
+    js = '''
+/* Java Script */
+var Out;
+Out=Application.version
+'''
+    skyx.connect()
+    skyx.write(js)
+    assert skyx.read().startswith('10.5')

pocs/utils/error.py

@@ -72,6 +72,12 @@ class InvalidMountCommand(PanError):
     pass
 
 
+class BadConnection(PanError):
+
+    """ PanError raised when a connection is bad """
+    pass
+
+
 class BadSerialConnection(PanError):
 
     """ PanError raised when serial command is bad """
@@ -101,3 +107,13 @@ class SolveError(NotFound):
 
     """ Camera cannot be imported """
     pass
+
+
+class TheSkyXError(PanError):
+    """ Errors from TheSkyX """
+    pass
+
+
+class TheSkyXKeyError(TheSkyXError):
+    """ Errors from TheSkyX because bad key passed """
+    pass

pocs/utils/theskyx.py

@@ -1,6 +1,7 @@
 import socket
 
 from pocs import PanBase
+from pocs.utils import error
 
 
 class TheSkyX(PanBase):
@@ -41,22 +42,31 @@ def connect(self):
                 self.logger.info('Connected to TheSkyX via {}:{}'.format(self._host, self._port))
 
     def write(self, value):
-        assert isinstance(value, str)
-        self.socket.sendall(value.encode())
+        try:
+            assert isinstance(value, str)
+            self.socket.sendall(value.encode())
+        except AttributeError:
+            raise error.BadConnection("Not connected to TheSkyX")
 
     def read(self, timeout=5):
-        self.socket.settimeout(timeout)
-        response = None
-
         try:
-            response = self.socket.recv(4096).decode()
-            if '|' in response:
-                response, err = response.split('|')
-            if err is not None and 'No error' not in err:
-                self.logger.warning("Mount error: {}".format(err))
-            elif err is None:
-                self.logger.warning("Error status not returned")
-        except socket.timeout:
-            pass
-
-        return response
+            self.socket.settimeout(timeout)
+            response = None
+
+            try:
+                response = self.socket.recv(4096).decode()
+                if '|' in response:
+                    response, err = response.split('|')
+                if err is not None and 'No error' not in err:
+                    if 'Error = 303' in err:
+                        raise error.TheSkyXKeyError("Invalid TheSkyX key")
+
+                    raise error.TheSkyXError(err)
+                elif err is None:
+                    raise error.TheSkyXError("Error status not returned")
+            except socket.timeout:
+                pass
+
+            return response
+        except AttributeError:
+            raise error.BadConnection("Not connected to TheSkyX")

requirements.txt

@@ -19,4 +19,5 @@ codecov
 ffmpy
 google-cloud-storage
 dateparser
-coveralls
+coveralls
+mocket