from typing import List
from atk.common.log import Logger
logging = Logger().get_logger()
class RunModesManager:
@classmethod
def op(cls, option, backend, run_modes: List[str]):
"""
操作所有run_modes
:params option: BaseRunMode的方法
:params backend: from atk.tasks.backends.backend import BaseBackend
:params run_modes: 被注册的run_mode名列表,List[str]
"""
from atk.tasks.run_modes import RunModesFactory
run_modes = run_modes or []
for run_mode in run_modes:
run_mode_ins = RunModesFactory.get_instance(run_mode)(backend=backend)
if hasattr(run_mode_ins, option):
method = getattr(run_mode_ins, option)
method()
else:
logging.warning(f"run_mode {run_mode} doesn't has method {option}")
logging.debug(
f"case id: {backend.case_config.id}, node: {backend.task_result.get_backend_name()}, "
f"run_modes {run_modes} {option} success."
)
@classmethod
def push(cls, backend, run_modes: List[str]):
"""
加载所有run_modes
"""
cls.op("push", backend, run_modes)
@classmethod
def pop(cls, backend, run_modes: List[str]):
"""
卸载所有run_modes
"""
cls.op("pop", backend, run_modes)