archiver_test.archivers_test_set

This module lets you run a collection of tests against the data in multiple archiver clusters.

Module Contents

Classes

ArchiversTestSet	Represents data needed to run a compare test between two archiver clusters.
TestResult	Result of a test and method to pretty print as tables.

Functions

import_test_sets(→ list[ArchiversTestSet])	Loads a config file in json. Files are created via the [store_test_set] method.
truncate_repr(→ str)
check_equal(→ bool)
human_readable(→ list[str])
sets_check(→ list[TestResult])	Tests from a list of parameters:
event_rate_same(→ bool)	Calculates if the event rates are effectively the same.

Attributes

LOG
MAX_LEN_DISPLAY_VAL_STR
MAX_LEN_DISPLAY_ARCHIVE_EVENT_STR
ACCEPTABLE_EVENT_RATE_DIFF

archiver_test.archivers_test_set.LOG: logging.Logger

class archiver_test.archivers_test_set.ArchiversTestSet

Represents data needed to run a compare test between two archiver clusters.

Returns:

Storage of the configuration of a test.

Return type:

ArchiversTestSet

msg: str

pvs: list[str]

sop: datetime.datetime

eop: datetime.datetime

archiver_old: str

archiver_new: str

archiver_old_data: list[pathlib.Path] | None

async export(output_folder: pathlib.Path) → pathlib.Path

Stores the configuration for a test run, with the result.

Parameters:

output_folder – Path to store the config.

async store(datasets: dict[str, dict[str, archiver_test.pv_archive_events.PVArchiveEvents]], output_folder: pathlib.Path) → None

Store data from running a test for use in future tests, without redownloading the data.

Parameters:

• datasets – Data from running test_set

• output_folder – Path of folder to store the output data.

async load_old_data() → dict[str, archiver_test.pv_archive_events.PVArchiveEvents]

Loads old data

async check(output_folder: pathlib.Path | None) → TestResult

Tests a single ArchiverTestSet

Parameters:

output_folder – Path where to store the data from the run.

archiver_test.archivers_test_set.import_test_sets(config: pathlib.Path) → list[ArchiversTestSet]

Loads a config file in json. Files are created via the [store_test_set] method.

Parameters:

config – The file path of the config file.

archiver_test.archivers_test_set.MAX_LEN_DISPLAY_VAL_STR = 10

class archiver_test.archivers_test_set.TestResult

Bases: NamedTuple

Result of a test and method to pretty print as tables.

test_set: ArchiversTestSet

old_version: str

new_version: str

old_data_set: dict[str, archiver_test.pv_archive_events.PVArchiveEvents]

new_data_set: dict[str, archiver_test.pv_archive_events.PVArchiveEvents]

comparisons: dict[str, archiver_test.pv_archive_events.DataSetDiff]

get_reason(pv: str) → archiver_test.fail_reasons.Reason

Calculates a possible Reason why two archivers have different data.

Parameters:

pv – The pv with test failure.

check_protocol_difference(pv: str, old_details: dict[archiver_test.details.Detail, str], new_details: dict[archiver_test.details.Detail, str]) → bool

Checks if the reason for the test failure is inherent to the protocols being different. In this case, if the only thing different in the differences is the alarm status.

Parameters:

• pv (str) – pv name

• old_details (dict[Detail, str]) – old details

• new_details (dict[Detail, str]) – new details

Returns:

result of check

Return type:

bool

check_capacity_full(pv: str, old_details: dict[archiver_test.details.Detail, str]) → bool

If there are many lost events in the old

And the difference in number of events is not larger than 10% of number of the old archiver test period events.

Parameters:

• pv (str) – pv name

• old_details (dict[Detail, str]) – details from old archiver

Returns:

if true

Return type:

bool

check_not_archiving(old_details: dict[archiver_test.details.Detail, str], new_details: dict[archiver_test.details.Detail, str]) → bool

check_no_events(pv: str) → bool

check_types(pv: str) → bool

result_message(pv: str) → str

Returns a message saying pass or failure with a reason.

Parameters:

pv – PV to check.

print_pv_result(console: rich.console.Console, pv: str) → None

Prints a summary of a test to console for a single PV.

Parameters:

• console – Where to print the summary.

• pv – Which PV for the summary.

print_result() → None

Prints the test result as a colleciton of summary tables.

create_summary_table() → rich.table.Table

Creates a rich.Table summarizing the test results. With columns Number of events, Events different, Events Same, Mean difference and the test result.

Returns:

A table of summary test result information.

Return type:

Table

print_time_summary(console: rich.console.Console) → None

Print summary of the time period to the console.

Parameters:

console (Console) – console to print in

archiver_test.archivers_test_set.MAX_LEN_DISPLAY_ARCHIVE_EVENT_STR = 1000

archiver_test.archivers_test_set.truncate_repr(event: epicsarchiver.ArchiveEvent, amount: int = MAX_LEN_DISPLAY_ARCHIVE_EVENT_STR) → str

archiver_test.archivers_test_set.check_equal(pair: tuple[epicsarchiver.ArchiveEvent, epicsarchiver.ArchiveEvent]) → bool

archiver_test.archivers_test_set.human_readable(delta: dateutil.relativedelta.relativedelta) → list[str]

async archiver_test.archivers_test_set.sets_check(sets: list[ArchiversTestSet], output_folder: pathlib.Path | None) → list[TestResult]

Tests from a list of parameters:

Parameters:

• sets – list of test sets

• output_folder – Where to store any data.

Return dictionary of results keys of test_set.msg:

archiver_test.archivers_test_set.ACCEPTABLE_EVENT_RATE_DIFF = 0.1

archiver_test.archivers_test_set.event_rate_same(old: dict[archiver_test.details.Detail, str], new: dict[archiver_test.details.Detail, str]) → bool

Calculates if the event rates are effectively the same.

Parameters:

• old – Old Archiver details.

• new – New Archvier details

Returns:

Returns true or false if the same.