KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _execute_with_timeout method in robotframework-androidlibrary

Best Python code snippet using robotframework-androidlibrary_python

__init__.py

Source:__init__.py

...82            args.append('-http-proxy')83            args.append(http_proxy)84        logging.debug("$> %s", ' '.join(args))85        self._emulator_proc = subprocess.Popen(args)86        rc, output, errput = self._execute_with_timeout([self._adb, 'wait-for-device'], max_timeout=80, max_attempts=1)87        if rc != 0 and retries > 0:88                self.stop_emulator()89                logging.warn("adb did not respond, retry starting %s " % retries)90                self.start_emulator(avd_name, no_window, language, country, save_snapshot, retries - 1)91    def stop_emulator(self):92        '''93        Halts a previously started Android Emulator.94        '''95        if not hasattr(self, '_emulator_proc'):96            logging.warn("Could not stop Android Emulator: It was not started.")97            return98        self._emulator_proc.terminate()99        self._emulator_proc.kill()100        self._emulator_proc.wait()101        self._emulator_proc = None102    def _execute_with_timeout(self, cmd, max_attempts=3, max_timeout=120):103        logging.debug("$> %s # with timeout %ds", ' '.join(cmd), max_timeout)104        attempt = 0105        while attempt < max_attempts:106            attempt = attempt + 1107            out = tempfile.NamedTemporaryFile(delete=False)108            err = tempfile.NamedTemporaryFile(delete=False)109            p = killableprocess.Popen(cmd, stdout=out, stderr=err)110            p.wait(max_timeout)111            out.flush()112            out.close()113            err.flush()114            err.close()115            # -9 and 127 are returned by killableprocess when a timeout happens116            if  p.returncode == -9 or p.returncode == 127:117                logging.warn("Executing %s failed executing in less then %d seconds and was killed, attempt number %d of %d" % (118                    ' '.join(cmd), max_timeout, attempt, max_attempts))119                continue120        try:121            outfile = open(out.name, 'r')122            errfile = open(err.name, 'r')123            return p.returncode, outfile.read(), errfile.read()124        finally:125            outfile.close()126            os.unlink(out.name)127            errfile.close()128            os.unlink(errfile.name)129    def _wait_for_package_manager(self):130        attempts = 0131        max_attempts = 3132        while attempts < max_attempts:133            rc, output, errput = self._execute_with_timeout([134                self._adb, "wait-for-device", "shell", "pm", "path", "android"],135                max_timeout=60, max_attempts=3)136            assert rc == 0, "Waiting for package manager failed: %d, %r, %r" % (rc, output, errput)137            if not 'Could not access the Package Manager.' in output:138                return139        raise AssertionError(output)140    def uninstall_application(self, package_name):141        self._wait_for_package_manager()142        rc, output, errput = self._execute_with_timeout([self._adb, "uninstall", package_name])143        assert rc == 0, "Uninstalling application failed: %d, %r" % (rc, output)144        assert output is not None145        logging.debug(output)146        assert 'Error' not in output, output147    def install_application(self, apk_file):148        '''149        Installs the given Android application package file (APK) on the emulator along with the test server.150        For instrumentation (and thus all remote keywords to work) both .apk151        files must be signed with the same key.152        `apk_file` Path the the application to install153        '''154        self._wait_for_package_manager()155        rc, output, errput = self._execute_with_timeout([self._adb, "install", "-r", apk_file], max_timeout=240)156        logging.debug(output)157        assert rc == 0, "Installing application failed: %d, %r" % (rc, output)158        assert output is not None159        assert 'Error' not in output, output160    def wait_for_device(self, timeout=120):161        '''162        Wait for the device to become available163        '''164        rc, output, errput = self._execute_with_timeout([self._adb, 'wait-for-device'], max_timeout=timeout / 3, max_attempts=3)165        assert rc == 0, "wait for device application failed: %d, %r" % (rc, output)166    def send_key(self, key_code):167        '''168        Send key event with the given key code. See http://developer.android.com/reference/android/view/KeyEvent.html for a list of available key codes.169        `key_code` The key code to send170        '''171        rc, output, errput = self._execute_with_timeout([self._adb, 'shell', 'input', 'keyevent', '%d' % key_code], max_attempts=1)172        assert rc == 0173    def press_back_button(self):174        '''175        Presses the back button.176        '''177        response = self._perform_action("go_back")178        assert response["success"] is True, "Could not press back button:: %s" % (response["message"])179    def press_menu_button(self):180        '''181        Press the menu button ("KEYCODE_MENU"), same as '| Send Key | 82 |'182        '''183        self.send_key(82)184    def set_device_endpoint(self, host='localhost', port=34777):185        """*DEPRECATED* Use 'Set Device Url' instead.186        Set the device endpoint where the application is started.187        If not set the endpoint defaults to 'localhost:34777'.188        `host` the endpoint's host189        `port` the endpoint's port190        """191        self.set_device_url('http://%s:%d' % (host, int(port)))192    def set_device_url(self, url='http://localhost:34777/'):193        """194        Set the device url where the application is started.195        `url` the base url to use for all requests196        """197        parsed_url = urlparse(url)198        self._port = parsed_url.port199        self._hostname = parsed_url.hostname200        self._url = url201    def start_testserver(self, package_name):202        '''203        *DEPRECATED* Use 'Start TestServer with apk' instead.204        Does not work with calabash-android >= 0.3.0205        Start the remote test server inside the Android Application.206        `package_name` fully qualified name of the application to test207        '''208        if not self._url:209            self.set_device_url()210        assert self._hostname == 'localhost', (211            "Device Url was set to %s, but should be set to localhost with the "212            "'Set Device Url' keyword to use a local testserver"213        )214        rc, output, errput = self._execute_with_timeout([215            self._adb,216            "wait-for-device",217            "forward",218            "tcp:%d" % self._port,219            "tcp:7102"220        ])221        args = [222            self._adb,223            "wait-for-device",224            "shell",225            "am",226            "instrument",227            "-e",228            "class",229            "sh.calaba.instrumentationbackend.InstrumentationBackend",230            "%s.test/sh.calaba.instrumentationbackend.CalabashInstrumentationTestRunner" % package_name,231        ]232        logging.debug("$> %s", ' '.join(args))233        self._testserver_proc = subprocess.Popen(args)234    def start_testserver_with_apk(self, apk):235        '''236        Works only with calabash-android >= 0.3.0237        Start the remote test server238        `apk` path to the apk to controll239        '''240        if not self._url:241            self.set_device_url()242        assert self._hostname == 'localhost', (243            "Device Url was set to %s, but should be set to localhost with the "244            "'Set Device Url' keyword to use a local testserver"245        )246        rc, output, errput = self._execute_with_timeout([247            self._adb,248            "wait-for-device",249            "forward",250            "tcp:%d" % self._port,251            "tcp:7102"252        ])253        package_name, main_activity = self._main_activity_from_apk(apk)254        if '.' not in main_activity or main_activity[0] == '.':255            main_activity = "%s.%s" % (package_name, main_activity.lstrip('.'))256        args = [257            self._adb,258            "shell",259            "am",260            "instrument",261            "-w",262            "-e",263            "target_package",264            package_name,265            "-e",266            "main_activity",267            main_activity,268            "-e",269            "test_server_port",270            str(7102),271            "-e",272            "class",273            "sh.calaba.instrumentationbackend.InstrumentationBackend",274            "%s.test/sh.calaba.instrumentationbackend.CalabashInstrumentationTestRunner" % package_name,275        ]276        self._testserver_proc = subprocess.Popen(args)277    def _main_activity_from_apk(self, apk):278        '''279        Returns the package_name and the Main-Action280        from a given apk281        '''282        rc, output, errput = self._execute_with_timeout([self._calabash_bin_path, "extract-manifest", apk])283        xmldoc = minidom.parseString(output)284        manifest = xmldoc.getElementsByTagName("manifest")285        assert len(manifest) > 0, "No <manifest> tag found in manifest file"286        manifest = manifest[0]287        package = manifest.getAttribute("package")288        assert package is not None, "Could not find package name in apk: %s manifest: %s" % (apk, output)289        for node in xmldoc.getElementsByTagName("action"):290            if node.getAttribute("android:name") == "android.intent.action.MAIN":291                return package, node.parentNode.parentNode.getAttribute("android:name")292        return package, None293    def stop_testserver(self):294        '''295        Halts a previously started Android Emulator.296        '''...

command.py

Source:command.py

...180            return self._exit_code(rc)181        if isinstance(self._exit_code, Iterable):182            return rc in self._exit_code183        return rc == self._exit_code184    def _execute_with_timeout(self, cmdline: List[AnyStr], result: Result) -> bool:185        try:186            if self._timeout:187                logging.info("... with timeout of %d seconds", self._timeout, extra=self.extra)188            rc = Command._popen(cmdline, result, needs_shell=self._needs_shell, encoding=self._encoding,189                                timeout=self._timeout, extra=self.extra)190            result.success = self._is_success(rc)191            if result.success:192                logging.warning("%s succeeded with exit code %d", cmdline[0], rc, extra=self.extra)193            else:194                logging.error("%s failed with exit code %d", cmdline[0], rc, extra=self.extra)195        except TimeoutError:196            logging.error("Execution timed out!", extra=self.extra)197        return result.success198    def _execute_with_rest(self, cmdline: List[AnyStr], result: Result) -> bool:199        if self._rest_period:200            logging.info("... resting for %d seconds", self._rest_period, extra=self.extra)201            sleep(self._rest_period)202        return self._execute_with_timeout(cmdline, result)203    def _execute_with_retry(self, cmdline: List[AnyStr], result: Result) -> bool:204        tries = self._retry205        for i in range(0, tries):206            logging.log(logging.DEBUG if not i else logging.WARN, "... with retry %d/%d", i+1, tries, extra=self.extra)207            success = self._execute_with_rest(cmdline, result)208            if success:209                break210        return success211    def _execute_with_lock(self, cmdline: List[AnyStr], result: Result) -> bool:212        @contextmanager213        def lock(cmd):214            _lock = None215            if self._exclusive:216                lock_path = Path(gettempdir(), f"{Path(cmd).name}.lock")...

instance_test.py

Source:instance_test.py

...55        print('Solution:')56        print(self.solution)57        print()58    def execute_optimization(self):59        self._execute_with_timeout(self._execute_optimization())60    def execute_solver(self):61        self._execute_with_timeout(self._execute_solver())62    def _execute_with_timeout(self, fun_to_execute):63        p = Process(target=fun_to_execute, name=self.test_name)64        p.start()65        p.join(timeout=self.timeout_minutes * 60 * 1000)66        p.terminate()67    def _execute_optimization(self):68        t_start = time.time()69        independent = IndependentOptimization(self.instance)70        independent_solution = independent.execute()71        evolutionary = EvolutionaryOptimization(independent_solution, self.instance)72        self.solution = evolutionary.execute()73        self.time = time.time() - t_start74        self._calculate_score()75    def _execute_solver(self):76        t_start = time.time()...

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.