From e7083fc77324948eb4b1334bfb9912926d3b6ec9 Mon Sep 17 00:00:00 2001 From: Clint Lawrence Date: Mon, 27 May 2024 17:07:35 +1000 Subject: [PATCH 1/3] initial ideas to share --- src/fixate/__main__.py | 5 +- src/fixate/core/common.py | 154 ++++++++++++++++------------- src/fixate/drivers/dmm/__init__.py | 4 +- src/fixate/sequencer.py | 55 +++++++++-- test-script.py | 72 ++++++++++++++ 5 files changed, 207 insertions(+), 83 deletions(-) create mode 100644 test-script.py 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 1e31bdc2..6fd41c97 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__) @@ -325,17 +329,74 @@ def inner(*args, **kwargs): return inner +DmType = TypeVar("DmType") +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 + as the name of the test with the remaining lines stored as self.test_desc_long which will show in the test logs + """ + + RT_ABORT = 1 # Abort the whole test sequence + RT_RETRY = 2 # Automatically retry up to "attempts" + RT_PROMPT = 3 # Prompt the user; Options are Abort the sequence, retry, or fail and continue + RT_FAIL = 4 # Automatically fail and move on + + test_desc = None + test_desc_long = None + attempts = 1 + retry_type = RT_PROMPT + retry_exceptions = [BaseException] # Depreciated + skip_exceptions = [] + abort_exceptions = [KeyboardInterrupt, AttributeError, NameError] + skip_on_fail = False + + def __init__(self, skip=False): + self.skip = skip + if not self.test_desc: + try: + doc_string = [ + line.strip() for line in self.__doc__.splitlines() if line.strip() + ] + except: + self.test_desc = self.__class__.__name__ + self.test_desc_long = "" + else: + if doc_string: + self.test_desc = doc_string[0] + self.test_desc_long = "\\n".join(doc_string[1:]) + + def set_up(self, dm: DmType): + """ + Optionally override this code that is executed before the test method is called + """ + pass + + 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, 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 + # 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,105 +413,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: - """ - 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 - as the name of the test with the remaining lines stored as self.test_desc_long which will show in the test logs - """ - - RT_ABORT = 1 # Abort the whole test sequence - RT_RETRY = 2 # Automatically retry up to "attempts" - RT_PROMPT = 3 # Prompt the user; Options are Abort the sequence, retry, or fail and continue - RT_FAIL = 4 # Automatically fail and move on - - test_desc = None - test_desc_long = None - attempts = 1 - tests = [] - retry_type = RT_PROMPT - retry_exceptions = [BaseException] # Depreciated - skip_exceptions = [] - abort_exceptions = [KeyboardInterrupt, AttributeError, NameError] - skip_on_fail = False - - def __init__(self, skip=False): - self.skip = skip - if not self.test_desc: - try: - doc_string = [ - line.strip() for line in self.__doc__.splitlines() if line.strip() - ] - except: - self.test_desc = self.__class__.__name__ - self.test_desc_long = "" - else: - if doc_string: - self.test_desc = doc_string[0] - self.test_desc_long = "\\n".join(doc_string[1:]) - - def set_up(self): - """ - Optionally override this code that is executed before the test method is called - """ - - def tear_down(self): - """ - Optionally override this code that is always executed at the end of the test whether it was successful or not - """ - - def test(self): - """ - 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 - """ +@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 e77c5b14..b7517d3c 100644 --- a/src/fixate/drivers/dmm/__init__.py +++ b/src/fixate/drivers/dmm/__init__.py @@ -19,7 +19,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: @@ -32,7 +32,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..24e648f1 --- /dev/null +++ b/test-script.py @@ -0,0 +1,72 @@ +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: + dummy: JigDmm = dataclasses.field(default_factory=JigDmm) + 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, +) From 565f681babcddf423c7b74b7c02fa7c0dcd3a458 Mon Sep 17 00:00:00 2001 From: Clint Lawrence Date: Mon, 27 May 2024 17:10:28 +1000 Subject: [PATCH 2/3] use the dummy so it works --- test-script.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/test-script.py b/test-script.py index 24e648f1..62f8c3d0 100644 --- a/test-script.py +++ b/test-script.py @@ -21,8 +21,9 @@ def close(self): @dataclasses.dataclass class Jig123DriverManager: - dummy: JigDmm = dataclasses.field(default_factory=JigDmm) - dmm: dmm.DMM = dataclasses.field(default_factory=dmm.open) + 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]): From ae96698743fb998b929a0048f2885bcc885138b6 Mon Sep 17 00:00:00 2001 From: Clint Lawrence Date: Tue, 28 May 2024 11:35:04 +1000 Subject: [PATCH 3/3] move TestClass back below TestList --- src/fixate/core/common.py | 113 +++++++++++++++++++------------------- 1 file changed, 58 insertions(+), 55 deletions(-) diff --git a/src/fixate/core/common.py b/src/fixate/core/common.py index 6fd41c97..1bfc5235 100644 --- a/src/fixate/core/common.py +++ b/src/fixate/core/common.py @@ -329,62 +329,8 @@ def inner(*args, **kwargs): return inner -DmType = TypeVar("DmType") -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 - as the name of the test with the remaining lines stored as self.test_desc_long which will show in the test logs - """ - RT_ABORT = 1 # Abort the whole test sequence - RT_RETRY = 2 # Automatically retry up to "attempts" - RT_PROMPT = 3 # Prompt the user; Options are Abort the sequence, retry, or fail and continue - RT_FAIL = 4 # Automatically fail and move on - - test_desc = None - test_desc_long = None - attempts = 1 - retry_type = RT_PROMPT - retry_exceptions = [BaseException] # Depreciated - skip_exceptions = [] - abort_exceptions = [KeyboardInterrupt, AttributeError, NameError] - skip_on_fail = False - - def __init__(self, skip=False): - self.skip = skip - if not self.test_desc: - try: - doc_string = [ - line.strip() for line in self.__doc__.splitlines() if line.strip() - ] - except: - self.test_desc = self.__class__.__name__ - self.test_desc_long = "" - else: - if doc_string: - self.test_desc = doc_string[0] - self.test_desc_long = "\\n".join(doc_string[1:]) - - def set_up(self, dm: DmType): - """ - Optionally override this code that is executed before the test method is called - """ - pass - - 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, 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 +DmType = TypeVar("DmType") # The first line of the doc string will be reflected in the test logs. Please don't change. @@ -466,6 +412,63 @@ def exit(self, dm: DmType): pass +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 + as the name of the test with the remaining lines stored as self.test_desc_long which will show in the test logs + """ + + RT_ABORT = 1 # Abort the whole test sequence + RT_RETRY = 2 # Automatically retry up to "attempts" + RT_PROMPT = 3 # Prompt the user; Options are Abort the sequence, retry, or fail and continue + RT_FAIL = 4 # Automatically fail and move on + + test_desc = None + test_desc_long = None + attempts = 1 + retry_type = RT_PROMPT + retry_exceptions = [BaseException] # Depreciated + skip_exceptions = [] + abort_exceptions = [KeyboardInterrupt, AttributeError, NameError] + skip_on_fail = False + + def __init__(self, skip=False): + self.skip = skip + if not self.test_desc: + try: + doc_string = [ + line.strip() for line in self.__doc__.splitlines() if line.strip() + ] + except: + self.test_desc = self.__class__.__name__ + self.test_desc_long = "" + else: + if doc_string: + self.test_desc = doc_string[0] + self.test_desc_long = "\\n".join(doc_string[1:]) + + def set_up(self, dm: DmType): + """ + Optionally override this code that is executed before the test method is called + """ + pass + + 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, 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]