Merge pull request #4 from MrClock8163/geocom-tests

GeoCom tests

Author: MrClock8163

.github/workflows/run-tests.yml

@@ -28,7 +28,7 @@ jobs:
       - name: Test with pytest
         run: |
           python -m pip install -r requirements.txt
-          python -m pytest -v --color=yes
+          python -m pytest
   
   linting:
     name: Linting

pyproject.toml

@@ -39,6 +39,6 @@ source = "vcs"
 version-file = "src/geocompy/_version.py"
 
 [tool.pytest.ini_options]
-addopts = ["-ra"]
-testpaths = ["tests", "src", "examples"]
+addopts = ["-ra", "-v", "--color=yes"]
+testpaths = ["tests"]
 required_plugins = "pytest-mock"

src/geocompy/tps1200p/__init__.py

@@ -166,10 +166,13 @@ def __init__(
         """Theodolite measurement and calculation subsystem."""
 
         for i in range(retry):
-            self._conn.send("\n")
-            response = self.com.nullproc()
-            if response.comcode and response.rpccode:
-                break
+            try:
+                self._conn.send("\n")
+                response = self.com.nullproc()
+                if response.comcode and response.rpccode:
+                    break
+            except Exception:
+                pass
 
             sleep(1)
         else:

src/geocompy/vivatps/__init__.py

@@ -180,10 +180,13 @@ def __init__(
         """Theodolite measurement and calculation subsystem."""
 
         for i in range(retry):
-            self._conn.send("\n")
-            response = self.com.nullproc()
-            if response.comcode and response.rpccode:
-                break
+            try:
+                self._conn.send("\n")
+                response = self.com.nullproc()
+                if response.comcode and response.rpccode:
+                    break
+            except Exception:
+                pass
 
             sleep(1)
         else:

tests/conftest.py

@@ -2,5 +2,6 @@
 
 pytest.register_assert_rewrite(
     "helpers",
+    "helpers_geocom",
     "helpers_gsionline"
 )

tests/helpers_geocom.py

@@ -0,0 +1,97 @@
+from typing import Callable, Any
+import re
+
+from geocompy import GeoComProtocol
+from geocompy.communication import Connection
+from geocompy.data import (
+    Byte
+)
+
+from helpers import faulty_parser
+
+
+class DummyGeoComConnection(Connection):
+    _RESP = re.compile(
+        r"^%R1P,"
+        r"(?P<comrc>\d+),"
+        r"(?P<tr>\d+):"
+        r"(?P<rc>\d+)"
+        r"(?:,(?P<params>.*))?$"
+    )
+
+    _CMD = re.compile(
+        r"^%R1Q,"
+        r"(?P<rpc>\d+):"
+        r"(?:(?P<params>.*))?$"
+    )
+
+    def send(self, message: str):
+        return
+
+    def exchange1(self, cmd: str) -> str:
+        if not self._CMD.match(cmd):
+            return "%R1P,0,0:2"
+
+        if cmd == "%R1Q,5008:":
+            return "%R1P,0,0:0,1996,'07','19','10','13','2f'"
+
+        return "%R1P,0,0:0"
+
+
+class GeoComTester:
+    @staticmethod
+    def test_parse_response(instrument: GeoComProtocol):
+        cmd = "%R1Q,5008:"
+        answer = "%R1P,0,0:0,1996,'07','19','10','13','2f'"
+        parsers: dict[str, Callable[[str], Any]] = {
+            "year": int,
+            "month": Byte.parse,
+            "day": Byte.parse,
+            "hour": Byte.parse,
+            "minute": Byte.parse,
+            "second": Byte.parse
+        }
+        response = instrument.parse_response(
+            cmd,
+            answer,
+            parsers
+        )
+        assert response.params["year"] == 1996
+
+        response = instrument.parse_response(
+            cmd,
+            "%R1P,1,0:",
+            parsers
+        )
+        assert len(response.params) == 0
+
+        parsers_faulty = parsers.copy()
+        parsers_faulty["year"] = faulty_parser
+        response = instrument.parse_response(
+            cmd,
+            answer,
+            parsers_faulty
+        )
+        assert len(response.params) == 0
+
+    @staticmethod
+    def test_request(instrument: GeoComProtocol):
+        response = instrument.request(
+            5008,
+            parsers={
+                "year": int,
+                "month": Byte.parse,
+                "day": Byte.parse,
+                "hour": Byte.parse,
+                "minute": Byte.parse,
+                "second": Byte.parse
+            }
+        )
+        assert response.params["year"] == 1996
+
+        response = instrument.request(
+            1,
+            (1, 2.0)
+        )
+        assert response.cmd == "%R1Q,1:1,2.0"
+        assert response.response == "%R1P,0,0:0"

tests/test_tps1200p.py

@@ -0,0 +1,31 @@
+import pytest
+
+from geocompy.communication import Connection
+from geocompy.tps1200p import TPS1200P
+
+from helpers_geocom import (
+    DummyGeoComConnection,
+    GeoComTester
+)
+
+
+@pytest.fixture
+def tps() -> TPS1200P:
+    return TPS1200P(DummyGeoComConnection())
+
+
+class TestTPS1200P:
+    def test_init(self):
+        conn_bad = Connection()
+        with pytest.raises(ConnectionError):
+            TPS1200P(conn_bad, retry=1)
+
+        conn_good = DummyGeoComConnection()
+        instrument = TPS1200P(conn_good)
+        assert instrument._precision == 15
+
+    def test_parse_response(self, tps: TPS1200P):
+        GeoComTester.test_parse_response(tps)
+
+    def test_request(self, tps: TPS1200P):
+        GeoComTester.test_request(tps)

tests/test_vivatps.py

@@ -0,0 +1,31 @@
+import pytest
+
+from geocompy.communication import Connection
+from geocompy.vivatps import VivaTPS
+
+from helpers_geocom import (
+    DummyGeoComConnection,
+    GeoComTester
+)
+
+
+@pytest.fixture
+def tps() -> VivaTPS:
+    return VivaTPS(DummyGeoComConnection())
+
+
+class TestVivaTPS:
+    def test_init(self):
+        conn_bad = Connection()
+        with pytest.raises(ConnectionError):
+            VivaTPS(conn_bad, retry=1)
+
+        conn_good = DummyGeoComConnection()
+        instrument = VivaTPS(conn_good)
+        assert instrument._precision == 15
+
+    def test_parse_response(self, tps: VivaTPS):
+        GeoComTester.test_parse_response(tps)
+
+    def test_request(self, tps: VivaTPS):
+        GeoComTester.test_request(tps)