How to use _check_error_codes method in lisa

Best Python code snippet using lisa_python

state.py

Source:state.py

...136    )137    @classmethod138    def new(cls) -> '_RetryOrCatch':139        return cls()140    def _check_error_codes(self):141        if len(self.error_equals) == 0:142            raise exc.StateValidationError(143                f"{C.ErrorEquals!r} has to be a NON EMPTY list!"144            )145        for error_code in self.error_equals:146            if not ErrorCodeEnum.contains(error_code):147                raise exc.StateValidationError(148                    f"{error_code!r} is not a valid Error Code!"149                )150    def _add_error(self, error_code: str) -> '_RetryOrCatch':151        if error_code not in self.error_equals:152            self.error_equals.append(error_code)153        return self154    def if_all_error(self) -> '_RetryOrCatch':155        return self._add_error(ErrorCodeEnum.AllError.value)156    def if_heartbeat_timeout_error(self) -> '_RetryOrCatch':157        return self._add_error(ErrorCodeEnum.HeartbeatTimeoutError.value)158    def if_timeout_error(self) -> '_RetryOrCatch':159        return self._add_error(ErrorCodeEnum.TimeoutError.value)160    def if_task_failed_error(self) -> '_RetryOrCatch':161        return self._add_error(ErrorCodeEnum.TaskFailedError.value)162    def if_permissions_error(self) -> '_RetryOrCatch':163        return self._add_error(ErrorCodeEnum.PermissionsError.value)164    def if_result_path_match_failure_error(self) -> '_RetryOrCatch':165        return self._add_error(ErrorCodeEnum.ResultPathMatchFailureError.value)166    def if_parameter_path_failure_error(self) -> '_RetryOrCatch':167        return self._add_error(ErrorCodeEnum.ParameterPathFailureError.value)168    def if_branch_failed_error(self) -> '_RetryOrCatch':169        return self._add_error(ErrorCodeEnum.BranchFailedError.value)170    def if_no_choice_matched_error(self) -> '_RetryOrCatch':171        return self._add_error(ErrorCodeEnum.NoChoiceMatchedError.value)172    def if_intrinsic_failure_error(self) -> '_RetryOrCatch':173        return self._add_error(ErrorCodeEnum.IntrinsicFailureError.value)174    def if_data_limit_exceeded_error(self) -> '_RetryOrCatch':175        return self._add_error(ErrorCodeEnum.DataLimitExceededError.value)176    def if_lambda_unknown_error(self) -> '_RetryOrCatch':177        return self._add_error(ErrorCodeEnum.LambdaUnknownError.value)178    def if_lambda_service_error(self) -> '_RetryOrCatch':179        return self._add_error(ErrorCodeEnum.LambdaServiceError.value)180    def if_lambda_aws_error(self) -> '_RetryOrCatch':181        return self._add_error(ErrorCodeEnum.LambdaAWSError.value)182    def if_lambda_sdk_client_error(self) -> '_RetryOrCatch':183        return self._add_error(ErrorCodeEnum.LambdaSdkClientError.value)184    def if_lambda_too_many_requests_error(self) -> '_RetryOrCatch':185        return self._add_error(ErrorCodeEnum.LambdaTooManyRequestsError.value)186    def _serialize(self) -> dict:187        data = self.to_dict()188        data = self._to_alias(data)189        return data190@attr.s191class Retry(_RetryOrCatch):192    """193    Reference:194    - https://docs.aws.amazon.com/step-functions/latest/dg/concepts-error-handling.html#error-handling-retrying-after-an-error195    """196    interval_seconds: T.Optional[int] = attr.ib(197        default=None, metadata={C.ALIAS: C.IntervalSeconds},198    )199    backoff_rate: T.Optional[T.Union[float, int]] = attr.ib(200        default=None, metadata={C.ALIAS: C.BackoffRate},201    )202    max_attempts: T.Optional[int] = attr.ib(203        default=None, metadata={C.ALIAS: C.MaxAttempts},204    )205    def with_interval_seconds(self, sec: int) -> 'Retry':206        self.interval_seconds = sec207        return self208    def with_back_off_rate(self, rate: T.Union[float, int]) -> 'Retry':209        self.backoff_rate = rate210        return self211    def with_max_attempts(self, attempts: int) -> 'Retry':212        self.max_attempts = attempts213        return self214    def _pre_serialize_validation(self):215        self._check_error_codes()216@attr.s217class Catch(_RetryOrCatch):218    """219    Reference:220    - https://docs.aws.amazon.com/step-functions/latest/dg/concepts-error-handling.html#error-handling-fallback-states221    """222    next: T.Optional[str] = attr.ib(223        default=None, metadata={C.ALIAS: C.Next},224    )225    result_path: T.Optional[str] = attr.ib(226        default=None, metadata={C.ALIAS: C.ResultPath},227    )228    def next_then(self, state: 'StateType'):229        self.next = state.id230        return self231    def with_result_path(self, result_path: str):232        """233        A path that determines what input is sent to the state specified234        in the Next field.235        """236        self.result_path = result_path237        return self238    def _check_next(self):239        if self.next is None:240            raise exc.ValidationError(f"{C.Catch}.{C.Next} is not defined!")241    def _check_result(self):242        if self.result_path is not None:243            if not is_json_path(self.result_path):244                raise exc.ValidationError(245                    f"{C.Catch}.{C.ResultPath} = {self.result_path!r} "246                    f"is not a valid JSON path!"247                )248    def _pre_serialize_validation(self):249        self._check_error_codes()250        self._check_next()251@attr.s252class _HasRetryCatch(State):253    retry: T.List['Retry'] = attr.ib(254        factory=list, metadata={C.ALIAS: C.Retry},255    )256    catch: T.List['Catch'] = attr.ib(257        factory=list, metadata={C.ALIAS: C.Catch},258    )259    def _serialize_retry_catch_fields(self, data: dict) -> dict:260        if self.retry:261            data[C.Retry] = [262                retry.serialize()263                for retry in self.retry...

service.py

Source:service.py

...59            cmd_result.assert_exit_code()60    def restart_service(self, name: str, ignore_exit_code: int = 0) -> None:61        cmd_result = self.run(f"{name} restart", shell=True, sudo=True, force_run=True)62        # optionally ignore exit code if it matches our expected non-zero value63        _check_error_codes(cmd_result, ignore_exit_code)64class Systemctl(Tool):65    @property66    def command(self) -> str:67        return "systemctl"68    @property69    def can_install(self) -> bool:70        return False71    def stop_service(self, name: str) -> None:72        if self._check_service_running(name):73            cmd_result = self.run(f"stop {name}", shell=True, sudo=True, force_run=True)74            cmd_result.assert_exit_code()75    def restart_service(self, name: str, ignore_exit_code: int = 0) -> None:76        cmd_result = self.run(f"restart {name}", shell=True, sudo=True, force_run=True)77        _check_error_codes(cmd_result, ignore_exit_code)78    def enable_service(self, name: str) -> None:79        cmd_result = self.run(f"enable {name}", shell=True, sudo=True, force_run=True)80        cmd_result.assert_exit_code()81    def hibernate(self) -> None:82        self.run_async("hibernate", sudo=True, force_run=True)83    def _check_exists(self) -> bool:84        return True85    def _check_service_exists(self, name: str) -> bool:86        cmd_result = self.run(87            f"--full --no-pager status {name}", shell=True, sudo=True, force_run=True88        )89        if (90            "could not be found" in cmd_result.stdout91            or "not-found" in cmd_result.stdout92        ):93            return False94        return True95    def _check_service_running(self, name: str) -> bool:96        cmd_result = self.run(97            f"--full --no-pager status {name}", shell=True, sudo=True, force_run=True98        )99        return (100            "could not be found" not in cmd_result.stdout101            or "not-found" in cmd_result.stdout102        ) and 0 == cmd_result.exit_code103def _check_error_codes(cmd_result: ExecutableResult, error_code: int = 0) -> None:...

test_retry_and_catch.py

Source:test_retry_and_catch.py

...9)10class TestRetryAndCatch:11    def test_validation(self):12        with pytest.raises(exc.StateValidationError):13            Retry.new()._check_error_codes()14        with pytest.raises(exc.StateValidationError):15            retry = Retry.new()16            retry.error_equals = ["InvalidErrorCode"]17            retry._check_error_codes()18        Retry.new().if_all_error()._check_error_codes()19    def test_add_error(self):20        retry = Retry.new()21        for attribute in _RetryOrCatch.__dict__:22            if attribute.startswith("if_"):23                getattr(retry, attribute)()24    def test_serialize(self):25        # Retry26        retry = (27            Retry.new()28            .with_interval_seconds(10)29            .with_back_off_rate(2.0)30            .with_max_attempts(3)31            .if_lambda_service_error()32            .if_lambda_aws_error()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.