TestMu Conference is Back! Future of Testing + AI LambdaTest G2 Announcement

How to use test_watcher method in uiautomator

Best Python code snippet using uiautomator

testcase.py

Source:testcase.py

1##########################################################################2# This file is part of WTFramework.3#4#    WTFramework is free software: you can redistribute it and/or modify5#    it under the terms of the GNU General Public License as published by6#    the Free Software Foundation, either version 3 of the License, or7#    (at your option) any later version.8#9#    WTFramework is distributed in the hope that it will be useful,10#    but WITHOUT ANY WARRANTY; without even the implied warranty of11#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the12#    GNU General Public License for more details.13#14#    You should have received a copy of the GNU General Public License15#    along with WTFramework.  If not, see <http://www.gnu.org/licenses/>.16##########################################################################17import sys18from unittest.case import _UnexpectedSuccess, SkipTest19import warnings20from six import u21import unittest222class WatchedTestCase(unittest2.TestCase):23    '''24    This test case extends the unittest.TestCase to add support for 25    registering TestWatchers for listening on TestEvents.26    '''27    def __init__(self, *args, **kwargs):28        self.__wtf_test_watchers__ = []29        super(WatchedTestCase, self).__init__(*args, **kwargs)30    # '_' prefix is added to hide it form nosetest31    def _register_watcher(self, watcher, position=-1):32        """33        Register a test watcher.34        Args:35            watcher: A test watcher to register.36        Kwargs:37            position: position in execution queue to insert this watcher.38        """39        self.__wtf_test_watchers__.insert(position, watcher)40    # '_' prefix is added to hide it form nosetest41    def _unregister_watcher(self, watcher):42        """"43        Unregister a test watcher.44        Args:45            watcher : Reference to TestWatcher to unregister.46        """47        self.__wtf_test_watchers__.remove(watcher)48    def get_log(self):49        """50        Get a log of events fired.51        Returns:52            list - list of string names of events fired.53        """54        log = []55        for watcher in self.__wtf_test_watchers__:56            log = watcher.get_log() + log57        return log58    def run(self, result=None):59        """60        Overriding the run() method to insert calls to our TestWatcher call-backs.61        Most of this method is a copy of the unittest.TestCase.run() method source.62        Kwargs:63            result: TestResult object.64        """65        orig_result = result66        if result is None:67            result = self.defaultTestResult()68            startTestRun = getattr(result, 'startTestRun', None)69            if startTestRun is not None:70                startTestRun()71        # Track if clean up was run, so we can run clean up if setup failed.72        did_tear_down_execute = False73        self._resultForDoCleanups = result74        result.startTest(self)75        testMethod = getattr(self, self._testMethodName)76        if (getattr(self.__class__, "__unittest_skip__", False) or77                getattr(testMethod, "__unittest_skip__", False)):78            # If the class or method was skipped.79            try:80                skip_why = (getattr(self.__class__, '__unittest_skip_why__', '')81                            or getattr(testMethod, '__unittest_skip_why__', ''))82                self._addSkip(result, skip_why)83            finally:84                result.stopTest(self)85            return86        try:87            success = False88            try:89                # Run our test watcher actions.90                for test_watcher in self.__wtf_test_watchers__:91                    test_watcher.before_setup(self, result)92                # Run test setup.93                self.setUp()94            except SkipTest as e:95                self._addSkip(result, str(e))96            except Exception:97                result.addError(self, sys.exc_info())98            else:99                try:100                    # Run our test watcher actions.101                    for test_watcher in self.__wtf_test_watchers__:102                        test_watcher.before_test(self, result)103                    # Run our test104                    testMethod()105                    # Run our test watcher post test actions.106                    for test_watcher in self.__wtf_test_watchers__:107                        test_watcher.on_test_pass(self, result)108                except self.failureException as e:109                    result.addFailure(self, sys.exc_info())110                    # Run our test watcher on fail actions.111                    for test_watcher in self.__wtf_test_watchers__:112                        test_watcher.on_test_failure(self, result, e)113                except _UnexpectedSuccess:114                    addUnexpectedSuccess = getattr(115                        result, 'addUnexpectedSuccess', None)116                    if addUnexpectedSuccess is not None:117                        addUnexpectedSuccess(self)118                    else:119                        warnings.warn(u("Use of a TestResult without an addUnexpectedSuccess method is deprecated"),120                                      DeprecationWarning)121                        result.addFailure(self, sys.exc_info())122                except SkipTest as e:123                    self._addSkip(result, str(e))124                except Exception as e:125                    result.addError(self, sys.exc_info())126                    # Run our test watcher on error actions.127                    for test_watcher in self.__wtf_test_watchers__:128                        test_watcher.on_test_error(self, result, e)129                else:130                    success = True131                try:132                    did_tear_down_execute = True133                    # Run our test watcher after test actions.134                    for test_watcher in self.__wtf_test_watchers__:135                        test_watcher.after_test(self, result)136                    # Do tear down.137                    self.tearDown()138                except Exception:139                    result.addError(self, sys.exc_info())140                    success = False141                finally:  # Run our test watcher actions for after tear down..142                    for test_watcher in self.__wtf_test_watchers__:143                        test_watcher.after_teardown(self, result)144            cleanUpSuccess = self.doCleanups()145            success = success and cleanUpSuccess146            if success:147                result.addSuccess(self)148        finally:149            # Execute tear down if it did not get executed.150            if not did_tear_down_execute:151                # Run our test watcher after test actions.152                try:153                    for test_watcher in self.__wtf_test_watchers__:154                        test_watcher.after_test(self, result)155                    self.tearDown()156                except:157                    # do nothing, test case would already failed and failure is158                    # already handled.159                    pass160                finally:  # Run our test watcher actions for after tear down..161                    for test_watcher in self.__wtf_test_watchers__:162                        test_watcher.after_teardown(self, result)163            # Remove test watchers.  For some strange reason these apply to all test164            # cases, not just the currently running one.  So we remove them165            # here.166            self.__wtf_test_watchers__ = []167            result.stopTest(self)168            if orig_result is None:169                stopTestRun = getattr(result, 'stopTestRun', None)170                if stopTestRun is not None:...

sample_multiprocess_script.py

Source:sample_multiprocess_script.py

1# Copyright (c) 2014, Eventbrite and Contributors2# All rights reserved.3# Redistribution and use in source and binary forms, with or without4# modification, are permitted provided that the following conditions are5# met:6# Redistributions of source code must retain the above copyright notice,7# this list of conditions and the following disclaimer.8# Redistributions in binary form must reproduce the above copyright9# notice, this list of conditions and the following disclaimer in the10# documentation and/or other materials provided with the distribution.11# Neither the name of Eventbrite nor the names of its contributors may12# be used to endorse or promote products derived from this software13# without specific prior written permission.14# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS15# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT16# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR17# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT18# HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,19# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT20# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,21# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY22# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT23# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE24# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.25import multiprocessing26import subprocess27import sys28import threading29def main():30    num_processes = int(multiprocessing.cpu_count() * 2.5)31    tests = []32    for i in range(num_processes):33        test_command = TEST_CMD_TEMPLATE % (34            i,35            num_processes,36        )37        tests.append(TestWatcher(test_command))38    returncode = 039    for test_watcher in tests:40        test_watcher.join()41        if test_watcher.returncode > 0:42            returncode += test_watcher.returncode43        for line in test_watcher.stderr.splitlines():44            if not (45                line.endswith(' ... ok') or46                '... SKIP' in line47            ):48                sys.stderr.write(line + '\n')49    return returncode50class TestWatcher(threading.Thread):51    def __init__(self, command):52        super(TestWatcher, self).__init__()53        self.command = command54        self.stdout = ''55        self.stderr = ''56        self.start()57        self.returncode = 058    def run(self):59        p = subprocess.Popen(60            self.command,61            shell=True,62            stdout=subprocess.PIPE,63            stderr=subprocess.PIPE,64        )65        self.stdout, self.stderr = p.communicate()66        self.returncode = p.returncode67if __name__ == '__main__':...

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.