(self, readiness_check_type: ReadinessCheck, bash_command: str,
expected_keyword: str, max_attempts: int, interval_waiting_time: int,
adb_action: str = None)
| 190 | self.deploy() |
| 191 | |
| 192 | def check_adb_command(self, readiness_check_type: ReadinessCheck, bash_command: str, |
| 193 | expected_keyword: str, max_attempts: int, interval_waiting_time: int, |
| 194 | adb_action: str = None) -> None: |
| 195 | success = False |
| 196 | for _ in range(1, max_attempts): |
| 197 | if success: |
| 198 | break |
| 199 | else: |
| 200 | try: |
| 201 | output = subprocess.check_output( |
| 202 | bash_command.split()).decode(UTF8) |
| 203 | if expected_keyword in str(output).lower(): |
| 204 | if readiness_check_type is self.ReadinessCheck.POP_UP_WINDOW: |
| 205 | subprocess.check_call(adb_action, shell=True) |
| 206 | else: |
| 207 | self.logger.info( |
| 208 | f"{self.device_type} is {readiness_check_type.value}!") |
| 209 | success = True |
| 210 | else: |
| 211 | self.logger.info(f"[attempt: {_}] {self.device_type} is not {readiness_check_type.value}! " |
| 212 | f"will check again in {interval_waiting_time} seconds") |
| 213 | time.sleep(interval_waiting_time) |
| 214 | except subprocess.CalledProcessError: |
| 215 | self.logger.warning("command cannot be executed! will continue...") |
| 216 | time.sleep(2) |
| 217 | continue |
| 218 | else: |
| 219 | if readiness_check_type is self.ReadinessCheck.POP_UP_WINDOW: |
| 220 | self.logger.info(f"Pop up windows '{expected_keyword}' is not found!") |
| 221 | else: |
| 222 | raise RuntimeError( |
| 223 | f"{readiness_check_type.value} is checked {_} times!") |
| 224 | |
| 225 | def wait_until_ready(self) -> None: |
| 226 | super().wait_until_ready() |
no outgoing calls