epicsarchiver_mgmt.commands.validation

Provide validation functions for the archiver management operations.

Attributes

`LOG`
`OPERATION_RESULT_STATUS`
`OPERATION_RESULT_OK`
`CONFIRMATION_PROMPT`

Exceptions

`ValidOperationResultsError`	Exception for when the operation results are not valid.
`ValidPVStatusError`	Exception for when a PV is not in the expected status.
`RequestHTTPError`	Exception for when there is an HTTP error.
`NotSamePVError`	Exception for when the old and new PVs are the same.
`AlreadyNewProtocolError`	Exception for when a PV is already in the new protocol.
`DifferentArchiverClusterError`	Exception for when the archiver FQDNs are not part of the same cluster.

Functions

`validate_operation_results`(→ None)	Validate the results of an operation.
`validate_pvs_status`(→ None)	Validate the status of PVs.
`validate_pvs_status_from_existing_info`(→ None)	Validate the status of PVs from existing information.
`validate_not_same`(→ None)	Validate the rename operation.
`validate_current_protocol`(→ None)	Validate the current protocol of the pvs is not the new protocol.
`validate_archiver_fqdns`(→ None)	Validate the archiver FQDNs.

Module Contents

epicsarchiver_mgmt.commands.validation.LOG: logging.Logger[source]

epicsarchiver_mgmt.commands.validation.OPERATION_RESULT_STATUS = 'status'[source]

epicsarchiver_mgmt.commands.validation.OPERATION_RESULT_OK = 'ok'[source]

epicsarchiver_mgmt.commands.validation.CONFIRMATION_PROMPT = 'Are you sure you want to proceed?'[source]

exception epicsarchiver_mgmt.commands.validation.ValidOperationResultsError(action_results: dict[str, epicsarchiver_mgmt.archiver.mgmt.OperationResult | epicsarchiver_mgmt.archiver.mgmt.OperationResultList], operation_name: str)[source]

Bases: epicsarchiver_mgmt.exceptions.BaseMgmtError

Exception for when the operation results are not valid.

action_results[source]

operation_name[source]

epicsarchiver_mgmt.commands.validation.validate_operation_results(pvs: collections.abc.Sequence[str], action_results: list[epicsarchiver_mgmt.archiver.mgmt.OperationResult] | epicsarchiver_mgmt.archiver.mgmt.OperationResultList, operation_name: str, *, expected_operation_results: dict[str, list[str]] | None = None) → None[source]

Validate the results of an operation.

Parameters:

pvs (Sequence[str]) – The PVs that were acted on.
action_results (list[OperationResult | OperationResultList]) – The results of the operation.
operation_name (str) – The name of the operation.
expected_operation_results (dict[str, list[str]]) – The expected statuses. Defaults to {OPERATION_RESULT_STATUS: [OPERATION_RESULT_OK]}.

Raises:

ValidOperationResultsError – If the results are not valid.

exception epicsarchiver_mgmt.commands.validation.ValidPVStatusError(pv: str, archiving_status: epicsarchiver_mgmt.archiver.info.ArchivingStatus | None, expected_statuses: collections.abc.Sequence[epicsarchiver_mgmt.archiver.info.ArchivingStatus | None])[source]

Bases: epicsarchiver_mgmt.exceptions.BaseMgmtError

Exception for when a PV is not in the expected status.

epicsarchiver_mgmt.commands.validation.validate_pvs_status(archiver_info: epicsarchiver_mgmt.archiver.info.ArchiverMgmtInfo, pvs: collections.abc.Sequence[str], expected_statuses: collections.abc.Sequence[epicsarchiver_mgmt.archiver.info.ArchivingStatus | None], existing_status_infos: epicsarchiver_mgmt.archiver.info.InfoResultList | None = None) → None[source]

Validate the status of PVs.

Parameters:

archiver_info (ArchiverMgmtInfo) – The archiver management server.
pvs (Sequence[str]) – The PVs to validate.
expected_statuses (Sequence[ArchivingStatus | None]) – The allowed statuses for the PVs.
existing_status_infos (InfoResultList | None, optional) – The existing statuses of the PVs. Defaults to None.

epicsarchiver_mgmt.commands.validation.validate_pvs_status_from_existing_info(pvs: collections.abc.Sequence[str], expected_statuses: collections.abc.Sequence[epicsarchiver_mgmt.archiver.info.ArchivingStatus | None], existing_status_infos: epicsarchiver_mgmt.archiver.info.InfoResultList) → None[source]

Validate the status of PVs from existing information.

This is used when the status information is already available and does not need to be fetched from the archiver. This is used to avoid making multiple requests to the archiver.

Args:.: pvs (Sequence[str]): The PVs to validate. expected_statuses (Sequence[ArchivingStatus | None]): The allowed statuses for the PVs. existing_status_infos (InfoResultList): The existing statuses of the PVs.

Raises:: ValidPVStatusError – If a PV is not in the expected status.

exception epicsarchiver_mgmt.commands.validation.RequestHTTPError(http_error: requests.HTTPError)[source]

Bases: epicsarchiver_mgmt.exceptions.BaseMgmtError

Exception for when there is an HTTP error.

http_error[source]

exception epicsarchiver_mgmt.commands.validation.NotSamePVError(pv: str)[source]

Bases: epicsarchiver_mgmt.exceptions.BaseMgmtError

Exception for when the old and new PVs are the same.

pv[source]

epicsarchiver_mgmt.commands.validation.validate_not_same(pairs: collections.abc.Sequence[tuple[str, str]]) → None[source]

Validate the rename operation.

Parameters:: pairs (Sequence[tuple[str, str]]) – The pairs of PVs.
Raises:: NotSamePVError – If the old and new PVs are the same.

exception epicsarchiver_mgmt.commands.validation.AlreadyNewProtocolError(pvs: set[str], protocol: epicsarchiver_mgmt.archiver.mgmt.EpicsProto)[source]

Bases: epicsarchiver_mgmt.exceptions.BaseMgmtError

Exception for when a PV is already in the new protocol.

pvs[source]

protocol[source]

epicsarchiver_mgmt.commands.validation.validate_current_protocol(archiver_info: epicsarchiver_mgmt.archiver.info.ArchiverMgmtInfo, pvs: collections.abc.Sequence[str], new_protocol: epicsarchiver_mgmt.archiver.mgmt.EpicsProto) → None[source]

Validate the current protocol of the pvs is not the new protocol.

Parameters:

archiver_info (ArchiverMgmtInfo) – The archiver management server.
pvs (Sequence[str]) – The PVs to validate.
new_protocol (EpicsProto) – The new protocol.

Raises:

AlreadyNewProtocolError – If a PV is already in the new protocol.

exception epicsarchiver_mgmt.commands.validation.DifferentArchiverClusterError(archiver_fqdn: str)[source]

Bases: epicsarchiver_mgmt.exceptions.BaseMgmtError

Exception for when the archiver FQDNs are not part of the same cluster.

archiver_fqdn[source]

epicsarchiver_mgmt.commands.validation.validate_archiver_fqdns(archiver_fqdns: collections.abc.Sequence[str]) → None[source]

Validate the archiver FQDNs.

Parameters:: archiver_fqdns (Sequence[str]) – The archiver FQDNs to validate.
Raises:: DifferentArchiverClusterError – If the archiver FQDNs are not part of the same cluster.