#!/usr/bin/env python3
"""Command line interface for archiver test."""
# pyre-strict
import asyncio
import logging
import pathlib
from datetime import UTC, datetime, timedelta
from pathlib import Path
import click
from rich.logging import RichHandler
from archiver_test import archivers_test_set
[docs]
LOG: logging.Logger = logging.getLogger(__name__)
[docs]
def _handle_debug(
_ctx: click.core.Context | None,
_param: click.core.Option | click.core.Parameter | None,
debug: bool | int | str,
) -> bool | int | str:
"""Turn on DEBUG logs, if asked otherwise INFO default"""
format_msg = "%(message)s"
if debug:
logging.basicConfig(
level=logging.DEBUG,
format=format_msg,
datefmt="[%X]",
handlers=[RichHandler(rich_tracebacks=True)],
)
else:
logging.basicConfig(
level=logging.INFO,
format=format_msg,
datefmt="[%X]",
handlers=[RichHandler()],
)
return debug
[docs]
async def async_main(
test_sets: list[archivers_test_set.ArchiversTestSet], output_folder: Path | None
) -> int:
"""Async main, runs the tests and prints the outputs.
Args:
test_sets: ArchiverTestSets to run.
output_folder: Optional input to store the data from the tests.
"""
results = await archivers_test_set.sets_check(test_sets, output_folder)
for result in results:
result.print_result()
return 0
@click.group(context_settings={"help_option_names": ["-h", "--help"]})
@click.pass_context
[docs]
def cli(_ctx: click.core.Context) -> None:
"""Archiver Test checks the data from two archivers
to see the difference when querying for the same data.
Example test the PV DTL-010:EMR-TT-047:Temp for the last 20 seconds.
.. code-block:: console
archiver_test test -p DTL-010:EMR-TT-047:Temp -o archiver-linac-01.tn.esss.lu.se -n skybrewer.cslab.esss.lu.se -s 20 -t "sky archiver vs linac 01"
Importing from a configuration file:
.. code-block:: console
archiver_test load config.json
Outputting to a configuration file and to a json storage:
.. code-block:: console
archiver_test test --output-folder output
A copy of the downloaded data will be stored in
`output/store/archiver_host_name/pv_name/time_period.test_state.test_run_time.pickle`
and a copy of the config as `output/test_name-test_run_time.test_state.pickle`.
Loading existing data from another test run:
.. code-block:: console
archiver_test test --old_data output/store/archiver-linac-01.tn.esss.lu.se/DTL-010:EMR-GT-003:Pos/data.pickle
Output is presented as a list of a tables for each PV for each archiver. Including:
- Archival parameters
- Average mean
- Number of test events
- PASS or FAIL if the returned events are exactly the same.
- Possible reason why results could be different.
And a summary table with all with a calculated average mean difference.
""" # noqa: E501
@cli.command(context_settings={"help_option_names": ["-h", "--help"]})
@click.option(
"--pv",
"-p",
multiple=True,
default=["DTL-010:EMR-TT-047:Temp"],
type=str,
help="PV names to check.",
)
@click.option(
"--archiver_old",
"-o",
default="archiver-linac-01.tn.esss.lu.se",
help="Hostname of old archiver.",
)
@click.option(
"--archiver_new",
"-n",
default="sw-vm-11.cslab.esss.lu.se",
help="Hostname of new archiver",
)
@click.option(
"--seconds", "-s", default=60, help="Length of time period to check in seconds."
)
@click.option("--test_name", "-t", default="Test", help="Name of test.")
@click.option(
"--old_data",
multiple=True,
default=None,
type=click.Path(exists=True, file_okay=True, readable=True, path_type=Path),
)
@click.option(
"--debug",
is_flag=True,
callback=_handle_debug,
show_default=True,
help="Turn on debug logging",
)
@click.option(
"--output-folder",
type=click.Path(exists=True, dir_okay=True, writable=True, path_type=Path),
help="Output folder for data from test.",
)
@click.pass_context
[docs]
def test(
ctx: click.core.Context,
pvs: list[str],
archiver_old: str,
archiver_new: str,
seconds: int,
test_name: str,
output_folder: Path | None,
old_data: list[Path] | None,
debug: bool, # noqa: FBT001, ARG001
) -> None:
"""Runs a single test given input options"""
time_now = datetime.now(tz=UTC) - timedelta(seconds=30)
a_test_set = archivers_test_set.ArchiversTestSet(
test_name,
pvs,
time_now - timedelta(seconds=seconds),
time_now,
archiver_old,
archiver_new,
old_data,
)
LOG.info("TEST command running with config: {%s}", a_test_set)
ctx.exit(asyncio.run(async_main([a_test_set], output_folder)))
@cli.command(context_settings={"help_option_names": ["-h", "--help"]})
@click.pass_context
@click.argument(
"config",
required=True,
type=click.Path(exists=True, file_okay=True, readable=True, path_type=Path),
)
@click.option(
"--debug",
is_flag=True,
callback=_handle_debug,
show_default=True,
help="Turn on debug logging",
)
@click.option(
"--output-folder",
type=click.Path(exists=True, dir_okay=True, writable=True, path_type=Path),
help="Output folder for data from test.",
)
[docs]
def load(
ctx: click.core.Context,
config: pathlib.Path,
output_folder: pathlib.Path,
debug: bool, # noqa: ARG001, FBT001
) -> None:
"""Loads a file CONFIG of test sets in json.
Argument CONFIG is a json file specifying a list of tests to run.
"""
test_sets = archivers_test_set.import_test_sets(config)
LOG.info("LOAD command running with config: %s", test_sets)
ctx.exit(asyncio.run(async_main(test_sets, output_folder)))