101 lines
2.9 KiB
Python
Executable File
101 lines
2.9 KiB
Python
Executable File
from pathlib import Path
|
|
import logging
|
|
|
|
from caproto.sync.client import read
|
|
import pytest
|
|
|
|
from src.classes import read_config, IocNotRunning
|
|
|
|
|
|
def check_ioc_running():
|
|
config = read_config()
|
|
pvprefix = config['pvprefix']
|
|
|
|
"""
|
|
This function checks if the test IOC is already running.
|
|
"""
|
|
try:
|
|
read(f'{pvprefix}:IOCREADY', timeout=2)
|
|
|
|
# Reading the check recird was successfull -> We assume that the IOC
|
|
# is running
|
|
return
|
|
except TimeoutError:
|
|
# IOC startup failed in the given time -> Raise an error
|
|
path = Path.cwd() / 'ioc' / 'startioc'
|
|
raise IocNotRunning(f'Start the test IOC first by running {path}')
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def check_ioc_running_before_test():
|
|
check_ioc_running()
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
"""
|
|
This function adds the --stress flag to Pytest which can be given to
|
|
perform long-running stress tests.
|
|
"""
|
|
parser.addoption(
|
|
"--stresstest",
|
|
action="store_true",
|
|
default=False,
|
|
help="Run tests marked as stresstest"
|
|
)
|
|
parser.addoption(
|
|
"--log",
|
|
action="store",
|
|
default="INFO",
|
|
help="""
|
|
Set log level:
|
|
- DEBUG: Show all raw caput and caget commands
|
|
- INFO: Write high-level commands (e.g. move, stop, home etc.)
|
|
- WARNING: Not used
|
|
- ERROR: Serious error (e.g. motor could not be enabled / disabled)
|
|
- CRITICAL: Not used
|
|
All error levels higher than the defined one are forwarded to the Pytest
|
|
logger.
|
|
"""
|
|
)
|
|
|
|
|
|
def pytest_runtest_setup(item):
|
|
"""
|
|
This function provides a marker which can be used on test functions to
|
|
mark them as stress tests via `@pytest.mark.stresstest`.
|
|
"""
|
|
if 'stresstest' in item.keywords and not item.config.getoption("--stresstest"):
|
|
pytest.skip("Need --stress to run this test")
|
|
|
|
|
|
def parse_loglevel(value: str) -> int:
|
|
value = value.upper()
|
|
if value.isdigit():
|
|
return int(value)
|
|
|
|
if value in logging._nameToLevel: # DEBUG, INFO, …
|
|
return logging._nameToLevel[value]
|
|
|
|
raise ValueError(f"Unknown log level: {value}")
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def configure_el734_logging(request):
|
|
loglevel_str = request.config.getoption("--log")
|
|
level = parse_loglevel(loglevel_str)
|
|
|
|
logger = logging.getLogger("EL734")
|
|
logger.handlers.clear() # so pytest doesn't duplicate handlers
|
|
|
|
handler = logging.StreamHandler()
|
|
formatter = logging.Formatter("%(levelname)s [%(name)s]: %(message)s")
|
|
handler.setFormatter(formatter)
|
|
logger.addHandler(handler)
|
|
logger.setLevel(level)
|
|
|
|
# Set caproto logger level to INFO and disable propagation
|
|
caproto_logger = logging.getLogger("caproto")
|
|
caproto_logger.setLevel(logging.ERROR)
|
|
# Prevent caproto logs from propagatingto the root logger
|
|
caproto_logger.propagate = False
|