diff --git a/src/fixate/__main__.py b/src/fixate/__main__.py index fe06ff71..4baa5876 100644 --- a/src/fixate/__main__.py +++ b/src/fixate/__main__.py @@ -13,6 +13,7 @@ from fixate.core.ui import user_info_important, user_serial, user_ok from fixate.reporting import register_csv, unregister_csv from fixate.ui_cmdline import register_cmd_line, unregister_cmd_line +from fixate.core.common import TestScript import fixate.sequencer parser = ArgumentParser( @@ -335,7 +336,7 @@ def ui_run(self): return ReturnCodes.ERROR -def retrieve_test_data(test_suite, index): +def retrieve_test_data(test_suite, index) -> TestScript: """ Tries to retrieve test data from the loaded test_suite module :param test_suite: Imported module with tests available @@ -346,7 +347,7 @@ def retrieve_test_data(test_suite, index): data = test_suite.test_data except AttributeError: # Try legacy API - return test_suite.TEST_SEQUENCE + return test_suite.TEST_SCRIPT try: sequence = data[index] except KeyError as e: diff --git a/src/fixate/core/common.py b/src/fixate/core/common.py index aab698c3..3dfee39b 100644 --- a/src/fixate/core/common.py +++ b/src/fixate/core/common.py @@ -1,3 +1,5 @@ +from __future__ import annotations +import dataclasses import re import sys import threading @@ -7,6 +9,8 @@ import warnings from functools import wraps from collections import namedtuple +from typing import TypeVar, Generic, List, Optional, Union, Iterable + from fixate.core.exceptions import ParameterError, InvalidScalarQuantityError logger = logging.getLogger(__name__) @@ -326,16 +330,19 @@ def inner(*args, **kwargs): return inner +DmType = TypeVar("DmType") + + # The first line of the doc string will be reflected in the test logs. Please don't change. -class TestList: +class TestList(Generic[DmType]): """ Test List The TestList is a container for TestClasses and TestLists to set up a test hierarchy. They operate similar to a python list except that it has additional methods that can be overridden to provide additional functionality """ - def __init__(self, seq=None): - self.tests = [] + def __init__(self, seq: Optional[List[Union[TestClass[DmType], TestList[DmType]]]] = None): + self.tests: List[Union[TestClass[DmType], TestList[DmType]]] = [] if seq is None: seq = [] self.tests.extend(seq) @@ -352,56 +359,60 @@ def __init__(self, seq=None): self.test_desc = doc_string[0] self.test_desc_long = "\\n".join(doc_string[1:]) - def __getitem__(self, item): + def __getitem__(self, item) -> Union[TestClass[DmType], TestList[DmType]]: return self.tests.__getitem__(item) - def __contains__(self, item): + def __contains__(self, item) -> bool: return self.tests.__contains__(item) - def __setitem__(self, key, value): + def __setitem__(self, key: int, value: Union[TestClass[DmType], TestList[DmType]]) -> None: return self.tests.__setitem__(key, value) - def __delitem__(self, key): + def __delitem__(self, key: int) -> None: return self.tests.__delitem__(key) - def __len__(self): + def __len__(self) -> int: return self.tests.__len__() - def append(self, p_object): + def append(self, p_object: Union[TestClass[DmType], TestList[DmType]]) -> None: self.tests.append(p_object) - def extend(self, iterable): + def extend(self, iterable: Iterable[Union[TestClass[DmType], TestList[DmType]]]): self.tests.extend(iterable) - def insert(self, index, p_object): + def insert(self, index: int, p_object: Union[TestClass[DmType], TestList[DmType]]): self.tests.insert(index, p_object) - def index(self, value, start=None, stop=None): + def index(self, value: Union[TestClass[DmType], TestList[DmType]], start: int =None, stop: int=None): self.tests.index(value, start, stop) - def set_up(self): + def set_up(self, dm: DmType): """ Optionally override this to be called before the set_up of the included TestClass and/or TestList within this TestList """ + pass - def tear_down(self): + def tear_down(self, dm: DmType): """ Optionally override this to be called after the tear_down of the included TestClass's and/or TestList's within this TestList This will be called if the set_up has been called regardless of the success of the included TestClass's and/or TestList's """ + pass - def enter(self): + def enter(self, dm: DmType): """ This is called when being pushed onto the stack """ + pass - def exit(self): + def exit(self, dm: DmType): """ This is called when being popped from the stack """ + pass -class TestClass: +class TestClass(Generic[DmType]): """ This class is an abstract base class to implement tests. The first line of the docstring of the class that inherits this class will be recognised by logging and UI @@ -416,7 +427,6 @@ class TestClass: test_desc = None test_desc_long = None attempts = 1 - tests = [] retry_type = RT_PROMPT retry_exceptions = [BaseException] # Depreciated skip_exceptions = [] @@ -438,19 +448,28 @@ def __init__(self, skip=False): self.test_desc = doc_string[0] self.test_desc_long = "\\n".join(doc_string[1:]) - def set_up(self): + def set_up(self, dm: DmType): """ Optionally override this code that is executed before the test method is called """ + pass - def tear_down(self): + def tear_down(self, dm: DmType): """ Optionally override this code that is always executed at the end of the test whether it was successful or not """ + pass - def test(self): + def test(self, dm: DmType): """ This method should be overridden with the test code This is the test sequence code Use chk functions to set the pass fail criteria for the test """ + raise NotImplementedError + + +@dataclasses.dataclass +class TestScript(Generic[DmType]): + test_list: TestList[DmType] + dm_type: type(DmType) diff --git a/src/fixate/drivers/dmm/__init__.py b/src/fixate/drivers/dmm/__init__.py index fcf83cfa..556199ca 100644 --- a/src/fixate/drivers/dmm/__init__.py +++ b/src/fixate/drivers/dmm/__init__.py @@ -18,7 +18,7 @@ def open() -> DMM: - for DMM in (Fluke8846A, Keithley6500): + for driver_class in (Fluke8846A, Keithley6500): instrument = find_instrument_by_id(DMM.REGEX_ID) if instrument is not None: # We've found a configured instrument so try to open it @@ -30,7 +30,7 @@ def open() -> DMM: f"Unable to open DMM: {instrument.address}" ) from e # Instantiate driver with connected instrument - driver = DMM(resource) + driver = driver_class(resource) fixate.drivers.log_instrument_open(driver) return driver raise InstrumentNotFoundError diff --git a/src/fixate/sequencer.py b/src/fixate/sequencer.py index 9e4b3436..0c68bb2b 100644 --- a/src/fixate/sequencer.py +++ b/src/fixate/sequencer.py @@ -1,8 +1,11 @@ +import inspect import sys import time import re +from typing import Optional, Union, TypeVar, Generic + from pubsub import pub -from fixate.core.common import TestList, TestClass +from fixate.core.common import TestList, TestClass, TestScript from fixate.core.exceptions import SequenceAbort, CheckFail from fixate.core.ui import user_retry_abort_fail from fixate.core.checks import CheckResult @@ -94,7 +97,8 @@ def get_parent_level(level): class Sequencer: def __init__(self): - self.tests = TestList() + self.test_script: Optional[TestScript] = None + self._driver_manager = None self._status = "Idle" self.active_test = None self.ABORT = False @@ -159,9 +163,9 @@ def status(self, val): else: self._status = val - def load(self, val): - self.tests.append(val) - self.context.push(self.tests) + def load(self, test_script: TestScript): + self.context.push(test_script.test_list) + self.test_script = test_script self.end_status = "N/A" def count_tests(self): @@ -225,6 +229,32 @@ def run_sequence(self): top.current().exit() self.context.pop() + @property + def driver_manager(self): + if self._driver_manager is None: + self._driver_manager = self.test_script.dm_type() + return self._driver_manager + + def _cleanup_driver_manager(self): + """ + Attempt to call close on each instrument on the driver manager. + + We assume any non-private attribute on the driver manager (i.e. not starting with '_') + is potentially a driver to be closed. Iterate over all such items and if they + have a close method call it. + + Finally, set the driver_manager instance back to None, so that all drivers get + re-instantiated if they are needed again. + """ + drivers = [ + driver + for name, driver in inspect.getmembers(self._driver_manager) + if not name.startswith("_") + ] + for driver in drivers: + if hasattr(driver, "close"): + driver.close() + def run_once(self): """ Runs through the tests once as are pushed onto the context stack. @@ -244,7 +274,7 @@ def run_once(self): data=top.testlist, test_index=self.levels(), ) - top.testlist.exit() + top.testlist.exit(self.driver_manager) if self.context: self.context.top().index += 1 elif isinstance(top.current(), TestClass): @@ -261,7 +291,7 @@ def run_once(self): data=top.current(), test_index=self.levels(), ) - top.current().enter() + top.current().enter(self.driver_manager) self.context.push(top.current()) else: raise SequenceAbort("Unknown Test Item Type") @@ -271,6 +301,7 @@ def run_once(self): exception=sys.exc_info()[1], test_index=self.levels(), ) + self._cleanup_driver_manager() pub.sendMessage("Sequence_Abort", exception=e) self._handle_sequence_abort() return @@ -315,11 +346,11 @@ def run_test(self): # Run the test try: for index_context, current_level in enumerate(self.context): - current_level.current().set_up() - active_test.test() + current_level.current().set_up(self.driver_manager) + active_test.test(self.driver_manager) finally: for current_level in self.context[index_context::-1]: - current_level.current().tear_down() + current_level.current().tear_down(self.driver_manager) if not self.chk_fail: active_test_status = "PASS" self.tests_passed += 1 @@ -343,6 +374,8 @@ def run_test(self): exception=sys.exc_info()[1], test_index=self.levels(), ) + self._cleanup_driver_manager() + attempts = 0 active_test_status = "ERROR" if not self.retry_prompt(): @@ -358,9 +391,11 @@ def run_test(self): exception=sys.exc_info()[1], test_index=self.levels(), ) + self._cleanup_driver_manager() # Retry Logic pub.sendMessage("Test_Retry", data=active_test, test_index=self.levels()) + self._cleanup_driver_manager() pub.sendMessage( "Test_Complete", data=active_test, diff --git a/test-script.py b/test-script.py new file mode 100644 index 00000000..62f8c3d0 --- /dev/null +++ b/test-script.py @@ -0,0 +1,73 @@ +import dataclasses + +from fixate.core.checks import chk_true +from fixate.core.common import TestClass, TestList, TestScript +from fixate.core.ui import user_info +from fixate.drivers import dmm + + +class JigDmm: + """Just for my testing...""" + def voltage_dc(self, _range: float) -> None: + pass + + def measurement(self) -> float: + user_info("Do some measurement!!!") + return 0.0 + + def close(self): + user_info("Closing dummy dmm") + + +@dataclasses.dataclass +class Jig123DriverManager: + dmm: JigDmm = dataclasses.field(default_factory=JigDmm) + # This doesn't work right at the moment... not sure why yet. + # dmm: dmm.DMM = dataclasses.field(default_factory=dmm.open) + + +class Jig123TestList(TestList[Jig123DriverManager]): + pass + + +class Jig123TestClass(TestClass[Jig123DriverManager]): + pass + + +class FailTest(Jig123TestClass): + def set_up(self, dm: Jig123DriverManager): + dm.dmm.voltage_dc(_range=30) + + def test(self, dm: Jig123DriverManager): + v = dm.dmm.measurement() + user_info(f"voltage measured was {v}") + chk_true(False, "force a failure to see test cleanup") + +class PassTest(Jig123TestClass): + def test(self, dm: Jig123DriverManager): + user_info(f"voltage measured was {dm.dmm.measurement()}") + + + +# New TestScript class bundles a test list with a driver manager. In +# test_variants.py, we will define TestScript objects instead of top level +# test lists like we do now. I've used a dataclass for the driver manager, +# but it could be any class. It is also possible to define a small function +# to use as the `default_factor` say for a serial port, where we need to use +# findftdi to get the right port to open. + +# Note that this is a bit of a compromise and might end up being a problem. +# At the moment would be possible for different `drivers` on our standard +# driver manager to use each other. We can't (easily) control creation order +# here, so that could run into problems. The behaviour is also slightly +# different to what we have now were each driver get "automatically" opened +# when accessed for the first time. With this design the sequence would +# create the driver manger, calling the default factory of each driver. I'm +# not sure that is necessarily good or bad. But is subtly different to existing +# driver managers. +# + +TEST_SCRIPT = TestScript( + test_list=Jig123TestList([FailTest(), PassTest()]), + dm_type=Jig123DriverManager, +)