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()...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!