47 lines
1.9 KiB
Python
47 lines
1.9 KiB
Python
|
import inspect
|
||
|
|
import warnings
|
||
|
|
from typing import Callable
|
||
|
|
|
||
|
|
from UniTAP.dev.modules.opf.handlers.base import OpfHandlerBase, OPFDialogAnswer,\
|
||
|
|
WrongOPFFunctionSignature
|
||
|
|
from UniTAP.dev.modules.opf.parsers import OPFParametersParser
|
||
|
|
|
||
|
|
|
||
|
|
class OpfHandlerDefault(OpfHandlerBase):
|
||
|
|
"""
|
||
|
|
Class `OpfHandlerDefault` inherit functionality from `OpfHandlerBase` and allows overriding opf functions.
|
||
|
|
Function `assign_function_for_opf_id` allows doing it.
|
||
|
|
"""
|
||
|
|
def __init__(self):
|
||
|
|
super().__init__()
|
||
|
|
self.__user_callback_dict = {}
|
||
|
|
self.__parser = OPFParametersParser()
|
||
|
|
|
||
|
|
def handle(self, opf_id, reply_answer, *args) -> OPFDialogAnswer:
|
||
|
|
base_opf_handler = super()._is_opf_processed(opf_id)
|
||
|
|
user_opf_handler = self.__user_callback_dict.get(opf_id, None) is not None
|
||
|
|
if not user_opf_handler and not base_opf_handler:
|
||
|
|
warnings.warn(f"Test requested operator feedback dialog with id: {opf_id}, but user "
|
||
|
|
f"function was not provided. Test will be aborted.")
|
||
|
|
return reply_answer(OPFDialogAnswer.ABORT)
|
||
|
|
|
||
|
|
if user_opf_handler:
|
||
|
|
opf_params_for_user = self.__parser.parse(opf_id, *args)
|
||
|
|
return reply_answer(self.__user_callback_dict.get(opf_id, None)(*opf_params_for_user))
|
||
|
|
else:
|
||
|
|
return super().handle(opf_id, reply_answer, *args)
|
||
|
|
|
||
|
|
def assign_function_for_opf_id(self, opf_id: int, func: Callable):
|
||
|
|
"""
|
||
|
|
Assign new function for doing required actions from OPF.
|
||
|
|
|
||
|
|
Arguments:
|
||
|
|
opf_id (`int`) - id of OPF.
|
||
|
|
func (`Callable`) - object of function
|
||
|
|
"""
|
||
|
|
signature = inspect.getfullargspec(func)
|
||
|
|
if signature.annotations['return'] != OPFDialogAnswer:
|
||
|
|
raise WrongOPFFunctionSignature(f"Function MUST declare {OPFDialogAnswer} "
|
||
|
|
f"as return value")
|
||
|
|
self.__user_callback_dict[opf_id] = func
|