Improved testing

.coveragerc

@@ -2,6 +2,7 @@
 branch = True
 source = ./labthings
 omit = .venv/*
+concurrency = greenlet
 
 [report]
 # Regexes for lines to exclude from consideration

.github/workflows/ci.yml

@@ -0,0 +1,52 @@
+name: CI
+
+on: [push]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v1
+        with:
+          fetch-depth: 1
+
+      - name: Set up Python 3.7
+        uses: actions/setup-python@v1
+        with:
+          python-version: 3.7
+
+      - name: Install Poetry
+        uses: dschep/install-poetry-action@v1.3
+
+      - name: Cache Poetry virtualenv
+        uses: actions/cache@v1
+        id: cache
+        with:
+          path: ~/.virtualenvs
+          key: poetry-${{ hashFiles('**/poetry.lock') }}
+          restore-keys: |
+            poetry-${{ hashFiles('**/poetry.lock') }}
+
+      - name: Set Poetry config
+        run: |
+          poetry config virtualenvs.in-project false
+          poetry config virtualenvs.path ~/.virtualenvs
+
+      - name: Install Dependencies
+        run: poetry install
+        if: steps.cache.outputs.cache-hit != 'true'
+
+      - name: Code Quality
+        run: poetry run black . --check
+
+      - name: Test with pytest
+        run: poetry run pytest --cov-report term-missing --cov-report=xml --cov=labthings ./tests
+
+      - name: Upload coverage to Codecov
+        uses: codecov/codecov-action@v1
+        with:
+          file: ./coverage.xml
+          flags: unittests
+          name: codecov-umbrella
+          fail_ci_if_error: true

.gitignore

@@ -50,6 +50,7 @@ coverage.xml
 *.py,cover
 .hypothesis/
 .pytest_cache/
+coverage_html_report/
 
 # Translations
 *.mo

labthings/core/lock.py

@@ -31,6 +31,9 @@ def __repr__(self):
             self._owner,
         )
 
+    def locked(self):
+        return self._block.locked()
+
     def acquire(self, blocking=True, timeout=None):
         """
         Acquire the mutex, blocking if *blocking* is true, for up to
@@ -68,22 +71,6 @@ def release(self):
     def __exit__(self, typ, value, tb):
         self.release()
 
-    # Internal methods used by condition variables
-
-    def _acquire_restore(self, count_owner):
-        count, owner = count_owner
-        self._block.acquire()
-        self._count = count
-        self._owner = owner
-
-    def _release_save(self):
-        count = self._count
-        self._count = 0
-        owner = self._owner
-        self._owner = None
-        self._block.release()
-        return (count, owner)
-
     def _is_owned(self):
         return self._owner is getcurrent()
 
@@ -109,22 +96,35 @@ def __init__(self, timeout=1, name=None):
     def locked(self):
         return self._lock.locked()
 
-    def acquire(self, blocking=True):
-        return self._lock.acquire(blocking, timeout=self.timeout)
+    def acquire(self, blocking=True, timeout=None, _strict=True):
+        if not timeout:
+            timeout = self.timeout
+        result = self._lock.acquire(blocking, timeout=timeout)
+        if _strict and not result:
+            raise LockError("ACQUIRE_ERROR", self)
+        else:
+            return result
 
     def __enter__(self):
-        result = self._lock.acquire(blocking=True, timeout=self.timeout)
-        if result:
-            return result
-        else:
-            raise LockError("ACQUIRE_ERROR", self)
+        return self.acquire(blocking=True, timeout=self.timeout)
 
     def __exit__(self, *args):
-        self._lock.release()
+        self.release()
 
     def release(self):
         self._lock.release()
 
+    @property
+    def _owner(self):
+        return self._lock._owner
+
+    @_owner.setter
+    def _owner(self, new_owner):
+        self._lock._owner = new_owner
+
+    def _is_owned(self):
+        return self._lock._is_owned()
+
 
 class CompositeLock:
     """
@@ -144,20 +144,53 @@ def __init__(self, locks, timeout=1):
         self.locks = locks
         self.timeout = timeout
 
-    def acquire(self, blocking=True):
-        return (lock.acquire(blocking=blocking) for lock in self.locks)
+    def acquire(self, blocking=True, timeout=None):
+        if not timeout:
+            timeout = self.timeout
+
+        lock_all = all(
+            [
+                lock.acquire(blocking=blocking, timeout=timeout, _strict=False)
+                for lock in self.locks
+            ]
+        )
 
-    def __enter__(self):
-        result = (lock.acquire(blocking=True) for lock in self.locks)
-        if all(result):
-            return result
-        else:
+        if not lock_all:
+            self._emergency_release()
             raise LockError("ACQUIRE_ERROR", self)
 
+        return True
+
+    def __enter__(self):
+        return self.acquire(blocking=True, timeout=self.timeout)
+
     def __exit__(self, *args):
-        for lock in self.locks:
-            lock.release()
+        return self.release()
 
     def release(self):
+        # If not all child locks are owner by caller
+        if not all([owner is getcurrent() for owner in self._owner]):
+            raise RuntimeError("cannot release un-acquired lock")
         for lock in self.locks:
-            lock.release()
+            if lock.locked():
+                lock.release()
+
+    def _emergency_release(self):
+        for lock in self.locks:
+            if lock.locked() and lock._is_owned():
+                lock.release()
+
+    def locked(self):
+        return any([lock.locked() for lock in self.locks])
+
+    @property
+    def _owner(self):
+        return [lock._owner for lock in self.locks]
+
+    @_owner.setter
+    def _owner(self, new_owner):
+        for lock in self.locks:
+            lock._owner = new_owner
+
+    def _is_owned(self):
+        return all([lock._is_owned() for lock in self.locks])

labthings/core/tasks/pool.py

@@ -1,11 +1,10 @@
-import threading
 import logging
 from functools import wraps
 from gevent import getcurrent
 
 from .thread import TaskThread
 
-from flask import copy_current_request_context
+from flask import copy_current_request_context, has_request_context
 
 
 class TaskMaster:
@@ -38,21 +37,26 @@ def states(self):
 
     def new(self, f, *args, **kwargs):
         # copy_current_request_context allows threads to access flask current_app
-        task = TaskThread(
-            target=copy_current_request_context(f), args=args, kwargs=kwargs
-        )
+        if has_request_context():
+            target = copy_current_request_context(f)
+        else:
+            target = f
+        task = TaskThread(target=target, args=args, kwargs=kwargs)
         self._tasks.append(task)
         return task
 
     def remove(self, task_id):
         for task in self._tasks:
-            if (task.id == task_id) and not task.isAlive():
-                del task
+            if (str(task.id) == str(task_id)) and task.dead:
+                self._tasks.remove(task)
 
     def cleanup(self):
-        for task in self._tasks:
-            if not task.isAlive():
-                del task
+        for i, task in enumerate(self._tasks):
+            if task.dead:
+                # Mark for delection
+                self._tasks[i] = None
+        # Remove items marked for deletion
+        self._tasks = [t for t in self._tasks if t]
 
 
 # Task management

labthings/core/utilities.py

@@ -55,10 +55,15 @@ def rupdate(destination_dict, update_dict):
     for k, v in update_dict.items():
         # Merge lists if they're present in both objects
         if isinstance(v, list):
+            # If key is missing from destination, create the list
             if k not in destination_dict:
                 destination_dict[k] = []
+            # If destination value is also a list, merge
             if isinstance(destination_dict[k], list):
                 destination_dict[k].extend(v)
+            # If destination exists but isn't a list, replace
+            else:
+                destination_dict[k] = v
         # Recursively merge dictionaries if the element is a dictionary
         elif isinstance(v, collections.abc.Mapping):
             if k not in destination_dict:

labthings/server/exceptions.py

@@ -1,4 +1,4 @@
-from flask import jsonify, escape
+from flask import escape
 from werkzeug.exceptions import default_exceptions
 from werkzeug.exceptions import HTTPException
 
@@ -31,7 +31,7 @@ def std_handler(self, error):
             or getattr(getattr(error, "__class__", None), "__name__", None)
             or None,
         }
-        return jsonify(response), status_code
+        return (response, status_code)
 
     def init_app(self, app):
         self.app = app

labthings/server/spec/utilities.py

@@ -6,7 +6,7 @@
 from ..fields import Field
 from marshmallow import Schema as BaseSchema
 
-from collections import Mapping
+from collections.abc import Mapping
 
 
 def update_spec(obj, spec: dict):

pyproject.toml

@@ -18,7 +18,7 @@ zeroconf = ">=0.24.5,<0.26.0"
 [tool.poetry.dev-dependencies]
 pytest = "^5.2"
 black = {version = "^19.10b0",allow-prereleases = true}
-coverage = "^5.0.4"
+pytest-cov = "^2.8.1"
 numpy = "^1.18.2"
 
 [build-system]

tests/test_core_exceptions.py

@@ -0,0 +1,26 @@
+from labthings.core import exceptions
+import pytest
+
+
+def test_lockerror_valid_code():
+    from threading import Lock
+
+    lock = Lock()
+
+    assert exceptions.LockError("ACQUIRE_ERROR", lock)
+    assert (
+        str(exceptions.LockError("ACQUIRE_ERROR", lock))
+        == f"ACQUIRE_ERROR: LOCK {lock}: Unable to acquire. Lock in use by another thread."
+    )
+
+
+def test_lockerror_invalid_code():
+    from threading import Lock
+
+    lock = Lock()
+
+    assert exceptions.LockError("INVALID_ERROR", lock)
+    assert (
+        str(exceptions.LockError("INVALID_ERROR", lock))
+        == f"INVALID_ERROR: LOCK {lock}: Unknown error."
+    )