epicsarchiver_mgmt.commands.validation ====================================== .. py:module:: epicsarchiver_mgmt.commands.validation .. autoapi-nested-parse:: Provide validation functions for the archiver management operations. Attributes ---------- .. autoapisummary:: epicsarchiver_mgmt.commands.validation.LOG epicsarchiver_mgmt.commands.validation.OPERATION_RESULT_STATUS epicsarchiver_mgmt.commands.validation.OPERATION_RESULT_OK epicsarchiver_mgmt.commands.validation.CONFIRMATION_PROMPT Exceptions ---------- .. autoapisummary:: epicsarchiver_mgmt.commands.validation.ValidOperationResultsError epicsarchiver_mgmt.commands.validation.ValidPVStatusError epicsarchiver_mgmt.commands.validation.RequestHTTPError epicsarchiver_mgmt.commands.validation.NotSamePVError epicsarchiver_mgmt.commands.validation.AlreadyNewProtocolError epicsarchiver_mgmt.commands.validation.DifferentArchiverClusterError Functions --------- .. autoapisummary:: epicsarchiver_mgmt.commands.validation.validate_operation_results epicsarchiver_mgmt.commands.validation.validate_pvs_status epicsarchiver_mgmt.commands.validation.validate_pvs_status_from_existing_info epicsarchiver_mgmt.commands.validation.validate_not_same epicsarchiver_mgmt.commands.validation.validate_current_protocol epicsarchiver_mgmt.commands.validation.validate_archiver_fqdns Module Contents --------------- .. py:data:: LOG :type: logging.Logger .. py:data:: OPERATION_RESULT_STATUS :value: 'status' .. py:data:: OPERATION_RESULT_OK :value: 'ok' .. py:data:: CONFIRMATION_PROMPT :value: 'Are you sure you want to proceed?' .. py:exception:: ValidOperationResultsError(action_results: dict[str, epicsarchiver_mgmt.archiver.mgmt.OperationResult | epicsarchiver_mgmt.archiver.mgmt.OperationResultList], operation_name: str) Bases: :py:obj:`epicsarchiver_mgmt.exceptions.BaseMgmtError` Exception for when the operation results are not valid. .. py:attribute:: action_results .. py:attribute:: operation_name .. py:function:: 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 Validate the results of an operation. :param pvs: The PVs that were acted on. :type pvs: Sequence[str] :param action_results: The results of the operation. :type action_results: list[OperationResult | OperationResultList] :param operation_name: The name of the operation. :type operation_name: str :param expected_operation_results: The expected statuses. Defaults to {OPERATION_RESULT_STATUS: [OPERATION_RESULT_OK]}. :type expected_operation_results: dict[str, list[str]] :raises ValidOperationResultsError: If the results are not valid. .. py:exception:: ValidPVStatusError(pv: str, archiving_status: epicsarchiver_mgmt.archiver.info.ArchivingStatus | None, expected_statuses: collections.abc.Sequence[epicsarchiver_mgmt.archiver.info.ArchivingStatus | None]) Bases: :py:obj:`epicsarchiver_mgmt.exceptions.BaseMgmtError` Exception for when a PV is not in the expected status. .. py:function:: 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 Validate the status of PVs. :param archiver_info: The archiver management server. :type archiver_info: ArchiverMgmtInfo :param pvs: The PVs to validate. :type pvs: Sequence[str] :param expected_statuses: The allowed statuses for the PVs. :type expected_statuses: Sequence[ArchivingStatus | None] :param existing_status_infos: The existing statuses of the PVs. Defaults to None. :type existing_status_infos: InfoResultList | None, optional .. py:function:: 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 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. .. py:exception:: RequestHTTPError(http_error: requests.HTTPError) Bases: :py:obj:`epicsarchiver_mgmt.exceptions.BaseMgmtError` Exception for when there is an HTTP error. .. py:attribute:: http_error .. py:exception:: NotSamePVError(pv: str) Bases: :py:obj:`epicsarchiver_mgmt.exceptions.BaseMgmtError` Exception for when the old and new PVs are the same. .. py:attribute:: pv .. py:function:: validate_not_same(pairs: collections.abc.Sequence[tuple[str, str]]) -> None Validate the rename operation. :param pairs: The pairs of PVs. :type pairs: Sequence[tuple[str, str]] :raises NotSamePVError: If the old and new PVs are the same. .. py:exception:: AlreadyNewProtocolError(pvs: set[str], protocol: epicsarchiver_mgmt.archiver.mgmt.EpicsProto) Bases: :py:obj:`epicsarchiver_mgmt.exceptions.BaseMgmtError` Exception for when a PV is already in the new protocol. .. py:attribute:: pvs .. py:attribute:: protocol .. py:function:: validate_current_protocol(archiver_info: epicsarchiver_mgmt.archiver.info.ArchiverMgmtInfo, pvs: collections.abc.Sequence[str], new_protocol: epicsarchiver_mgmt.archiver.mgmt.EpicsProto) -> None Validate the current protocol of the pvs is not the new protocol. :param archiver_info: The archiver management server. :type archiver_info: ArchiverMgmtInfo :param pvs: The PVs to validate. :type pvs: Sequence[str] :param new_protocol: The new protocol. :type new_protocol: EpicsProto :raises AlreadyNewProtocolError: If a PV is already in the new protocol. .. py:exception:: DifferentArchiverClusterError(archiver_fqdn: str) Bases: :py:obj:`epicsarchiver_mgmt.exceptions.BaseMgmtError` Exception for when the archiver FQDNs are not part of the same cluster. .. py:attribute:: archiver_fqdn .. py:function:: validate_archiver_fqdns(archiver_fqdns: collections.abc.Sequence[str]) -> None Validate the archiver FQDNs. :param archiver_fqdns: The archiver FQDNs to validate. :type archiver_fqdns: Sequence[str] :raises DifferentArchiverClusterError: If the archiver FQDNs are not part of the same cluster.