🧪 instantiate tests

.github/workflows/python-package.yml

@@ -5,9 +5,9 @@ name: copley
 
 on:
   push:
-    branches: [ "master" ]
+    branches: [ "main" ]
   pull_request:
-    branches: [ "master" ]
+    branches: [ "main" ]
 
 jobs:
   test:
@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12"]
 
     steps:
     - uses: actions/checkout@8ade135a41bc03ea155e62e844d188df1ea18608 # v4
@@ -37,4 +37,4 @@ jobs:
          mypy copley
     - name: Pytest
       run: |
-        pytest
+        pytest tests

README.md

@@ -1,2 +1,2 @@
 # copley
-Python driver for Copley JV100i Tapped Density Tester
+Python ≥3.9 driver for Copley JV100i Tapped Density Tester

copley/__init__.py

@@ -17,13 +17,21 @@ def command_line(args: Any = None) -> None:
 
     parser = argparse.ArgumentParser(description="Read scale status.")
     parser.add_argument('address', help="The serial or IP address of the copley.")
+    parser.add_argument('-r', '--reset', action='store_true', help="Reset test on device.")
     args = parser.parse_args(args)
 
-    async def get() -> None:
-        async with TapDensity(address=args.address) as copley:
-            report = await copley.get_report()
-            print(json.dumps(report, indent=4))
-    asyncio.run(get())
+    if args.reset:
+        async def reset() -> None:
+            async with TapDensity(address=args.address) as copley:
+                await copley.reset()
+                print("OK")
+        asyncio.run(reset())
+    else:
+        async def get() -> None:
+            async with TapDensity(address=args.address) as copley:
+                report = await copley.get_report()
+                print(json.dumps(report, indent=4))
+        asyncio.run(get())
 
 
 if __name__ == '__main__':

copley/driver.py

@@ -37,6 +37,7 @@ class TapDensity:
     READ_SET_SPEED = "SS"
     READ_SERIAL_NUMBER = "SN"
     READ_FIRMWARE = "V"
+    RESET_TEST = "TRESET"
 
     def __init__(self, address, **kwargs):
         """Set up connection parameters, serial or IP address and port."""
@@ -66,6 +67,10 @@ async def get_report(self):
         response = await self.query(self.PRINT_REPORT)
         return self._parse(response)
 
+    async def reset(self):
+        """Reset the Copley."""
+        await self.hw._write(self.RESET_TEST)
+
     def _parse(self, response: str) -> dict:
         """Parse a tapped density report. Data output format is ASCII."""
         if response is None:
@@ -78,26 +83,23 @@ def _parse(self, response: str) -> dict:
                 elif "Calculation type" in line:
                     calc_type = split_line[1].strip()
                 elif "Set Speed" in line:
-                    try:
-                        set_speed = int(split_line[1].strip())
-                    except ValueError:
-                        set_speed = split_line[1].strip()
+                    set_speed = split_line[1].strip()
                 elif "Total Taps" in line:
                     total_taps = int(split_line[1].strip())
                 elif "Sample Weight, W" in line:
-                    sample_weight = float(split_line[1].strip())
+                    sample_weight = split_line[1].strip()
                 elif "Initial Volume" in line:
-                    init_volume = float(split_line[1].strip())
+                    init_volume = split_line[1].strip()
                 elif "Final Volume" in line:
-                    final_volume = float(split_line[1].strip())
+                    final_volume = split_line[1].strip()
                 elif "Bulk Density" in line:
-                    bulk_density = float(split_line[1].strip())
+                    bulk_density = split_line[1].strip()
                 elif "Tapped Density (g/mL)" in line:
-                    tapped_density = float(split_line[1].strip())
+                    tapped_density = split_line[1].strip()
                 elif "Hausner Ratio" in line:
-                    hausner_ratio = float(split_line[1].strip())
+                    hausner_ratio = split_line[1].strip()
                 elif "Compress. Index" in line:
-                    compress_index = float(split_line[1].strip())
+                    compress_index = split_line[1].strip()
             try:
                 return {
                     'serial_number': sn,
@@ -113,4 +115,4 @@ def _parse(self, response: str) -> dict:
                     'compress_index': compress_index
                 }
             except Exception as e:
-                return {f'Could not parse: {e}'}
+                raise e

copley/mock.py

@@ -0,0 +1,51 @@
+"""Contains mocks for driver objects for offline testing."""
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import MagicMock
+
+from .driver import TapDensity as RealTapDensity
+
+
+class AsyncClientMock(MagicMock):
+    """Magic mock that works with async methods."""
+
+    async def __call__(self, *args, **kwargs):
+        """Convert regular mocks into into an async coroutine."""
+        return super().__call__(*args, **kwargs)
+
+
+class TapDensity(RealTapDensity):
+    """Mocks the overhead stirrer driver for offline testing."""
+
+    def __init__(self, *args, **kwargs):
+        """Set up connection parameters with default port."""
+        super().__init__(*args, **kwargs)
+        self.hw = AsyncClientMock()
+
+    async def query(self, command):
+        """Return mock requests to query."""
+        if not self.lock:
+            self.lock = asyncio.Lock()
+        async with self.lock:  # lock releases on CancelledError
+            if command == self.PRINT_REPORT:
+                return 'hello world :)'
+
+    def _parse(self, response):
+        """Return mock requests to parsing."""
+        if response is None:
+            return {'on': False}
+        else:
+            return {
+                'serial_number': 12345,
+                'calc_type': "Fixed weight",
+                'set_speed': "300",
+                'total_taps': 1250,
+                'sample_weight': "2.77",
+                'init_volume': "170.0",
+                'final_volume': "155.0",
+                'bulk_density': "0.016",
+                'tapped_density': "0.018",
+                'hausner_ratio': "1.097",
+                'compress_index': "8.82"
+            }

copley/util.py

@@ -134,7 +134,7 @@ async def _readline(self):
     async def _readlines(self, num_lines):
         """Read lines."""
         await self._handle_connection()
-        lines = []
+        lines: list[str] = []
         while len(lines) < num_lines:
             try:
                 line = await self.connection['reader'].readuntil(self.eol)
@@ -222,3 +222,6 @@ def close(self):
 
     async def _handle_connection(self):
         self.open = True
+
+    async def _handle_communication(self):
+        pass

setup.cfg

@@ -1,14 +1,15 @@
-[flake8]
-max-complexity = 15
-max-line-length = 99
-docstring-convention = pep257
+[bdist_wheel]
 
+[pydocstyle]
 # no docstrings in __init__.py
-ignore = D104,D107,W503
+add_ignore = D104,D107
 
-[mypy]
-check_untyped_defs = True
-disallow_untyped_defs = True
+[tool:pytest]
+asyncio_mode = auto
+addopts = --cov=copley
+
+[tool:ruff]
+target-version = "py38"
 
-[mypy-tests.*]
-allow_untyped_defs = True
+[mypy]
+check_untyped_defs = true

tests/test_driver.py

@@ -0,0 +1,25 @@
+"""Test the copley driver responds with correct data."""
+from unittest import mock
+
+import pytest
+
+from copley import command_line
+from copley.mock import TapDensity
+
+ADDRESS = '192.168.10.18:23'
+
+
+@pytest.fixture
+def driver():
+    """Confirm the overhead stirrer correctly initializes."""
+    return TapDensity(ADDRESS)
+
+
+@mock.patch('copley.TapDensity', TapDensity)
+def test_driver_cli(capsys):
+    """Confirm the commandline interface works."""
+    command_line([ADDRESS])
+    captured = capsys.readouterr()
+    assert "bulk_density" in captured.out
+    assert "tapped_density" in captured.out
+    assert "null" not in captured.out