How to use wait_for_close method in lisa

Best Python code snippet using lisa_python

118

Source:118

...41        while not self.has_opened and time.clock() < timeout:42            self.process_events()43        if not self.has_opened:44            self.fail('Timeout while waiting for the tray to open')45    def wait_for_close(self):46        timeout = time.clock() + 347        while not self.has_closed and time.clock() < timeout:48            self.process_events()49        if not self.has_opened:50            self.fail('Timeout while waiting for the tray to close')51class GeneralChangeTestsBaseClass(ChangeBaseClass):52    def test_change(self):53        result = self.vm.qmp('change', device='drive0', target=new_img,54                                       arg=iotests.imgfmt)55        self.assert_qmp(result, 'return', {})56        self.wait_for_open()57        self.wait_for_close()58        result = self.vm.qmp('query-block')59        self.assert_qmp(result, 'return[0]/tray_open', False)60        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)61    def test_blockdev_change_medium(self):62        result = self.vm.qmp('blockdev-change-medium', device='drive0',63                                                       filename=new_img,64                                                       format=iotests.imgfmt)65        self.assert_qmp(result, 'return', {})66        self.wait_for_open()67        self.wait_for_close()68        result = self.vm.qmp('query-block')69        self.assert_qmp(result, 'return[0]/tray_open', False)70        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)71    def test_eject(self):72        result = self.vm.qmp('eject', device='drive0', force=True)73        self.assert_qmp(result, 'return', {})74        self.wait_for_open()75        result = self.vm.qmp('query-block')76        self.assert_qmp(result, 'return[0]/tray_open', True)77        self.assert_qmp_absent(result, 'return[0]/inserted')78    def test_tray_eject_change(self):79        result = self.vm.qmp('eject', device='drive0', force=True)80        self.assert_qmp(result, 'return', {})81        self.wait_for_open()82        result = self.vm.qmp('query-block')83        self.assert_qmp(result, 'return[0]/tray_open', True)84        self.assert_qmp_absent(result, 'return[0]/inserted')85        result = self.vm.qmp('blockdev-change-medium', device='drive0',86                                                       filename=new_img,87                                                       format=iotests.imgfmt)88        self.assert_qmp(result, 'return', {})89        self.wait_for_close()90        result = self.vm.qmp('query-block')91        self.assert_qmp(result, 'return[0]/tray_open', False)92        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)93    def test_tray_open_close(self):94        result = self.vm.qmp('blockdev-open-tray', device='drive0', force=True)95        self.assert_qmp(result, 'return', {})96        self.wait_for_open()97        result = self.vm.qmp('query-block')98        self.assert_qmp(result, 'return[0]/tray_open', True)99        if self.was_empty == True:100            self.assert_qmp_absent(result, 'return[0]/inserted')101        else:102            self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)103        result = self.vm.qmp('blockdev-close-tray', device='drive0')104        self.assert_qmp(result, 'return', {})105        if self.has_real_tray or not self.was_empty:106            self.wait_for_close()107        result = self.vm.qmp('query-block')108        if self.has_real_tray or not self.was_empty:109            self.assert_qmp(result, 'return[0]/tray_open', False)110        else:111            self.assert_qmp(result, 'return[0]/tray_open', True)112        if self.was_empty == True:113            self.assert_qmp_absent(result, 'return[0]/inserted')114        else:115            self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)116    def test_tray_eject_close(self):117        result = self.vm.qmp('eject', device='drive0', force=True)118        self.assert_qmp(result, 'return', {})119        self.wait_for_open()120        result = self.vm.qmp('query-block')121        self.assert_qmp(result, 'return[0]/tray_open', True)122        self.assert_qmp_absent(result, 'return[0]/inserted')123        result = self.vm.qmp('blockdev-close-tray', device='drive0')124        self.assert_qmp(result, 'return', {})125        if self.has_real_tray:126            self.wait_for_close()127        result = self.vm.qmp('query-block')128        if self.has_real_tray:129            self.assert_qmp(result, 'return[0]/tray_open', False)130        else:131            self.assert_qmp(result, 'return[0]/tray_open', True)132        self.assert_qmp_absent(result, 'return[0]/inserted')133    def test_tray_open_change(self):134        result = self.vm.qmp('blockdev-open-tray', device='drive0', force=True)135        self.assert_qmp(result, 'return', {})136        self.wait_for_open()137        result = self.vm.qmp('query-block')138        self.assert_qmp(result, 'return[0]/tray_open', True)139        if self.was_empty == True:140            self.assert_qmp_absent(result, 'return[0]/inserted')141        else:142            self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)143        result = self.vm.qmp('blockdev-change-medium', device='drive0',144                                                       filename=new_img,145                                                       format=iotests.imgfmt)146        self.assert_qmp(result, 'return', {})147        self.wait_for_close()148        result = self.vm.qmp('query-block')149        self.assert_qmp(result, 'return[0]/tray_open', False)150        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)151    def test_cycle(self):152        result = self.vm.qmp('blockdev-add',153                             options={'node-name': 'new',154                                      'driver': iotests.imgfmt,155                                      'file': {'filename': new_img,156                                               'driver': 'file'}})157        self.assert_qmp(result, 'return', {})158        result = self.vm.qmp('blockdev-open-tray', device='drive0', force=True)159        self.assert_qmp(result, 'return', {})160        self.wait_for_open()161        result = self.vm.qmp('query-block')162        self.assert_qmp(result, 'return[0]/tray_open', True)163        if self.was_empty == True:164            self.assert_qmp_absent(result, 'return[0]/inserted')165        else:166            self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)167        result = self.vm.qmp('x-blockdev-remove-medium', device='drive0')168        self.assert_qmp(result, 'return', {})169        result = self.vm.qmp('query-block')170        self.assert_qmp(result, 'return[0]/tray_open', True)171        self.assert_qmp_absent(result, 'return[0]/inserted')172        result = self.vm.qmp('x-blockdev-insert-medium', device='drive0',173                                                       node_name='new')174        self.assert_qmp(result, 'return', {})175        result = self.vm.qmp('query-block')176        self.assert_qmp(result, 'return[0]/tray_open', True)177        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)178        result = self.vm.qmp('blockdev-close-tray', device='drive0')179        self.assert_qmp(result, 'return', {})180        self.wait_for_close()181        result = self.vm.qmp('query-block')182        self.assert_qmp(result, 'return[0]/tray_open', False)183        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)184    def test_close_on_closed(self):185        result = self.vm.qmp('blockdev-close-tray', device='drive0')186        # Should be a no-op187        self.assert_qmp(result, 'return', {})188        self.assertEquals(self.vm.get_qmp_events(wait=False), [])189    def test_remove_on_closed(self):190        if self.has_opened:191            # Empty floppy drive192            return193        result = self.vm.qmp('x-blockdev-remove-medium', device='drive0')194        self.assert_qmp(result, 'error/class', 'GenericError')195    def test_insert_on_closed(self):196        if self.has_opened:197            # Empty floppy drive198            return199        result = self.vm.qmp('blockdev-add',200                             options={'node-name': 'new',201                                      'driver': iotests.imgfmt,202                                      'file': {'filename': new_img,203                                               'driver': 'file'}})204        self.assert_qmp(result, 'return', {})205        result = self.vm.qmp('x-blockdev-insert-medium', device='drive0',206                                                       node_name='new')207        self.assert_qmp(result, 'error/class', 'GenericError')208class TestInitiallyFilled(GeneralChangeTestsBaseClass):209    was_empty = False210    def setUp(self, media, interface):211        qemu_img('create', '-f', iotests.imgfmt, old_img, '1440k')212        qemu_img('create', '-f', iotests.imgfmt, new_img, '1440k')213        self.vm = iotests.VM().add_drive(old_img, 'media=%s' % media, interface)214        self.vm.launch()215    def tearDown(self):216        self.vm.shutdown()217        os.remove(old_img)218        os.remove(new_img)219    def test_insert_on_filled(self):220        result = self.vm.qmp('blockdev-add',221                             options={'node-name': 'new',222                                      'driver': iotests.imgfmt,223                                      'file': {'filename': new_img,224                                               'driver': 'file'}})225        self.assert_qmp(result, 'return', {})226        result = self.vm.qmp('blockdev-open-tray', device='drive0')227        self.assert_qmp(result, 'return', {})228        self.wait_for_open()229        result = self.vm.qmp('x-blockdev-insert-medium', device='drive0',230                                                       node_name='new')231        self.assert_qmp(result, 'error/class', 'GenericError')232class TestInitiallyEmpty(GeneralChangeTestsBaseClass):233    was_empty = True234    def setUp(self, media, interface):235        qemu_img('create', '-f', iotests.imgfmt, new_img, '1440k')236        self.vm = iotests.VM().add_drive(None, 'media=%s' % media, interface)237        self.vm.launch()238    def tearDown(self):239        self.vm.shutdown()240        os.remove(new_img)241    def test_remove_on_empty(self):242        result = self.vm.qmp('blockdev-open-tray', device='drive0')243        self.assert_qmp(result, 'return', {})244        self.wait_for_open()245        result = self.vm.qmp('x-blockdev-remove-medium', device='drive0')246        # Should be a no-op247        self.assert_qmp(result, 'return', {})248class TestCDInitiallyFilled(TestInitiallyFilled):249    TestInitiallyFilled = TestInitiallyFilled250    has_real_tray = True251    def setUp(self):252        self.TestInitiallyFilled.setUp(self, 'cdrom', 'ide')253class TestCDInitiallyEmpty(TestInitiallyEmpty):254    TestInitiallyEmpty = TestInitiallyEmpty255    has_real_tray = True256    def setUp(self):257        self.TestInitiallyEmpty.setUp(self, 'cdrom', 'ide')258class TestFloppyInitiallyFilled(TestInitiallyFilled):259    TestInitiallyFilled = TestInitiallyFilled260    has_real_tray = False261    def setUp(self):262        self.TestInitiallyFilled.setUp(self, 'disk', 'floppy')263class TestFloppyInitiallyEmpty(TestInitiallyEmpty):264    TestInitiallyEmpty = TestInitiallyEmpty265    has_real_tray = False266    def setUp(self):267        self.TestInitiallyEmpty.setUp(self, 'disk', 'floppy')268        # FDDs not having a real tray and there not being a medium inside the269        # tray at startup means the tray will be considered open270        self.has_opened = True271class TestChangeReadOnly(ChangeBaseClass):272    def setUp(self):273        qemu_img('create', '-f', iotests.imgfmt, old_img, '1440k')274        qemu_img('create', '-f', iotests.imgfmt, new_img, '1440k')275        self.vm = iotests.VM()276    def tearDown(self):277        self.vm.shutdown()278        os.chmod(old_img, 0666)279        os.chmod(new_img, 0666)280        os.remove(old_img)281        os.remove(new_img)282    def test_ro_ro_retain(self):283        os.chmod(old_img, 0444)284        os.chmod(new_img, 0444)285        self.vm.add_drive(old_img, 'media=disk,read-only=on', 'floppy')286        self.vm.launch()287        result = self.vm.qmp('query-block')288        self.assert_qmp(result, 'return[0]/tray_open', False)289        self.assert_qmp(result, 'return[0]/inserted/ro', True)290        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)291        result = self.vm.qmp('blockdev-change-medium', device='drive0',292                                                       filename=new_img,293                                                       format=iotests.imgfmt,294                                                       read_only_mode='retain')295        self.assert_qmp(result, 'return', {})296        self.wait_for_open()297        self.wait_for_close()298        result = self.vm.qmp('query-block')299        self.assert_qmp(result, 'return[0]/tray_open', False)300        self.assert_qmp(result, 'return[0]/inserted/ro', True)301        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)302    def test_ro_rw_retain(self):303        os.chmod(old_img, 0444)304        self.vm.add_drive(old_img, 'media=disk,read-only=on', 'floppy')305        self.vm.launch()306        result = self.vm.qmp('query-block')307        self.assert_qmp(result, 'return[0]/tray_open', False)308        self.assert_qmp(result, 'return[0]/inserted/ro', True)309        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)310        result = self.vm.qmp('blockdev-change-medium', device='drive0',311                                                       filename=new_img,312                                                       format=iotests.imgfmt,313                                                       read_only_mode='retain')314        self.assert_qmp(result, 'return', {})315        self.wait_for_open()316        self.wait_for_close()317        result = self.vm.qmp('query-block')318        self.assert_qmp(result, 'return[0]/tray_open', False)319        self.assert_qmp(result, 'return[0]/inserted/ro', True)320        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)321    def test_rw_ro_retain(self):322        os.chmod(new_img, 0444)323        self.vm.add_drive(old_img, 'media=disk', 'floppy')324        self.vm.launch()325        result = self.vm.qmp('query-block')326        self.assert_qmp(result, 'return[0]/tray_open', False)327        self.assert_qmp(result, 'return[0]/inserted/ro', False)328        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)329        result = self.vm.qmp('blockdev-change-medium', device='drive0',330                                                       filename=new_img,331                                                       format=iotests.imgfmt,332                                                       read_only_mode='retain')333        self.assert_qmp(result, 'error/class', 'GenericError')334        self.assertEquals(self.vm.get_qmp_events(wait=False), [])335        result = self.vm.qmp('query-block')336        self.assert_qmp(result, 'return[0]/tray_open', False)337        self.assert_qmp(result, 'return[0]/inserted/ro', False)338        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)339    def test_ro_rw(self):340        os.chmod(old_img, 0444)341        self.vm.add_drive(old_img, 'media=disk,read-only=on', 'floppy')342        self.vm.launch()343        result = self.vm.qmp('query-block')344        self.assert_qmp(result, 'return[0]/tray_open', False)345        self.assert_qmp(result, 'return[0]/inserted/ro', True)346        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)347        result = self.vm.qmp('blockdev-change-medium',348                             device='drive0',349                             filename=new_img,350                             format=iotests.imgfmt,351                             read_only_mode='read-write')352        self.assert_qmp(result, 'return', {})353        self.wait_for_open()354        self.wait_for_close()355        result = self.vm.qmp('query-block')356        self.assert_qmp(result, 'return[0]/tray_open', False)357        self.assert_qmp(result, 'return[0]/inserted/ro', False)358        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)359    def test_rw_ro(self):360        os.chmod(new_img, 0444)361        self.vm.add_drive(old_img, 'media=disk', 'floppy')362        self.vm.launch()363        result = self.vm.qmp('query-block')364        self.assert_qmp(result, 'return[0]/tray_open', False)365        self.assert_qmp(result, 'return[0]/inserted/ro', False)366        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)367        result = self.vm.qmp('blockdev-change-medium',368                             device='drive0',369                             filename=new_img,370                             format=iotests.imgfmt,371                             read_only_mode='read-only')372        self.assert_qmp(result, 'return', {})373        self.wait_for_open()374        self.wait_for_close()375        result = self.vm.qmp('query-block')376        self.assert_qmp(result, 'return[0]/tray_open', False)377        self.assert_qmp(result, 'return[0]/inserted/ro', True)378        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)379    def test_make_rw_ro(self):380        self.vm.add_drive(old_img, 'media=disk', 'floppy')381        self.vm.launch()382        result = self.vm.qmp('query-block')383        self.assert_qmp(result, 'return[0]/tray_open', False)384        self.assert_qmp(result, 'return[0]/inserted/ro', False)385        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)386        result = self.vm.qmp('blockdev-change-medium',387                             device='drive0',388                             filename=new_img,389                             format=iotests.imgfmt,390                             read_only_mode='read-only')391        self.assert_qmp(result, 'return', {})392        self.wait_for_open()393        self.wait_for_close()394        result = self.vm.qmp('query-block')395        self.assert_qmp(result, 'return[0]/tray_open', False)396        self.assert_qmp(result, 'return[0]/inserted/ro', True)397        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)398    def test_make_ro_rw(self):399        os.chmod(new_img, 0444)400        self.vm.add_drive(old_img, 'media=disk', 'floppy')401        self.vm.launch()402        result = self.vm.qmp('query-block')403        self.assert_qmp(result, 'return[0]/tray_open', False)404        self.assert_qmp(result, 'return[0]/inserted/ro', False)405        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)406        result = self.vm.qmp('blockdev-change-medium',407                             device='drive0',408                             filename=new_img,409                             format=iotests.imgfmt,410                             read_only_mode='read-write')411        self.assert_qmp(result, 'error/class', 'GenericError')412        self.assertEquals(self.vm.get_qmp_events(wait=False), [])413        result = self.vm.qmp('query-block')414        self.assert_qmp(result, 'return[0]/tray_open', False)415        self.assert_qmp(result, 'return[0]/inserted/ro', False)416        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)417    def test_make_rw_ro_by_retain(self):418        os.chmod(old_img, 0444)419        self.vm.add_drive(old_img, 'media=disk,read-only=on', 'floppy')420        self.vm.launch()421        result = self.vm.qmp('query-block')422        self.assert_qmp(result, 'return[0]/tray_open', False)423        self.assert_qmp(result, 'return[0]/inserted/ro', True)424        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)425        result = self.vm.qmp('blockdev-change-medium', device='drive0',426                                                       filename=new_img,427                                                       format=iotests.imgfmt,428                                                       read_only_mode='retain')429        self.assert_qmp(result, 'return', {})430        self.wait_for_open()431        self.wait_for_close()432        result = self.vm.qmp('query-block')433        self.assert_qmp(result, 'return[0]/tray_open', False)434        self.assert_qmp(result, 'return[0]/inserted/ro', True)435        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)436    def test_make_ro_rw_by_retain(self):437        os.chmod(new_img, 0444)438        self.vm.add_drive(old_img, 'media=disk', 'floppy')439        self.vm.launch()440        result = self.vm.qmp('query-block')441        self.assert_qmp(result, 'return[0]/tray_open', False)442        self.assert_qmp(result, 'return[0]/inserted/ro', False)443        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)444        result = self.vm.qmp('blockdev-change-medium', device='drive0',445                                                       filename=new_img,446                                                       format=iotests.imgfmt,447                                                       read_only_mode='retain')448        self.assert_qmp(result, 'error/class', 'GenericError')449        self.assertEquals(self.vm.get_qmp_events(wait=False), [])450        result = self.vm.qmp('query-block')451        self.assert_qmp(result, 'return[0]/tray_open', False)452        self.assert_qmp(result, 'return[0]/inserted/ro', False)453        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)454    def test_rw_ro_cycle(self):455        os.chmod(new_img, 0444)456        self.vm.add_drive(old_img, 'media=disk', 'floppy')457        self.vm.launch()458        result = self.vm.qmp('query-block')459        self.assert_qmp(result, 'return[0]/tray_open', False)460        self.assert_qmp(result, 'return[0]/inserted/ro', False)461        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)462        result = self.vm.qmp('blockdev-add',463                             options={'node-name': 'new',464                                      'driver': iotests.imgfmt,465                                      'read-only': True,466                                      'file': {'filename': new_img,467                                               'driver': 'file'}})468        self.assert_qmp(result, 'return', {})469        result = self.vm.qmp('blockdev-open-tray', device='drive0', force=True)470        self.assert_qmp(result, 'return', {})471        self.wait_for_open()472        result = self.vm.qmp('query-block')473        self.assert_qmp(result, 'return[0]/tray_open', True)474        self.assert_qmp(result, 'return[0]/inserted/ro', False)475        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)476        result = self.vm.qmp('x-blockdev-remove-medium', device='drive0')477        self.assert_qmp(result, 'return', {})478        result = self.vm.qmp('query-block')479        self.assert_qmp(result, 'return[0]/tray_open', True)480        self.assert_qmp_absent(result, 'return[0]/inserted')481        result = self.vm.qmp('x-blockdev-insert-medium', device='drive0',482                                                       node_name='new')483        self.assert_qmp(result, 'return', {})484        result = self.vm.qmp('query-block')485        self.assert_qmp(result, 'return[0]/tray_open', True)486        self.assert_qmp(result, 'return[0]/inserted/ro', True)487        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)488        result = self.vm.qmp('blockdev-close-tray', device='drive0')489        self.assert_qmp(result, 'return', {})490        self.wait_for_close()491        result = self.vm.qmp('query-block')492        self.assert_qmp(result, 'return[0]/tray_open', False)493        self.assert_qmp(result, 'return[0]/inserted/ro', True)494        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)495GeneralChangeTestsBaseClass = None496TestInitiallyFilled = None497TestInitiallyEmpty = None498class TestBlockJobsAfterCycle(ChangeBaseClass):499    def setUp(self):500        qemu_img('create', '-f', iotests.imgfmt, old_img, '1M')501        self.vm = iotests.VM()502        self.vm.launch()503        result = self.vm.qmp('blockdev-add',504                             options={'id': 'drive0',...

test_rpc_server.py

Source:test_rpc_server.py

1from __future__ import print_function2import gevent3import logbook4from unittest import TestCase5from infi.rpc import (Client, AutoTimeoutClient, Server, ServiceBase, ZeroRPCClientTransport, ZeroRPCServerTransport,6                      rpc_call, rpc_deferred)7from infi.rpc.errors import TimeoutExpired8from infi.pyutils.contexts import contextmanager9from logbook import Logger10logger = Logger('test')11class FooService(ServiceBase):12    @rpc_call13    def no_delay_call(self):14        return 115    @rpc_call16    def delay_call(self):17        gevent.sleep(1)18        return 219    @rpc_deferred20    def no_delay_deferred_call(self):21        return 322    @rpc_deferred23    def delay_deferred_call(self):24        gevent.sleep(1)25        return 426    @rpc_call27    def stuck_call(self):28        from time import sleep29        sleep(1)30        return 531@contextmanager32def server_context(service, **kwargs):33    server_transport = ZeroRPCServerTransport.create_tcp(8192)34    server = Server(transport=server_transport, service=service, **kwargs)35    server.bind()36    try:37        yield server38    finally:39        if server:40            server.unbind()41def create_client():42    return Client(ZeroRPCClientTransport.create_tcp(8192))43class ServerRPCTestCase(TestCase):44    def test_no_delay_call(self):45        with server_context(FooService(), max_response_time=1):46            client = create_client()47            result = client.no_delay_call(async_rpc=True)48            self.assertTrue(result.is_done())49            self.assertEqual(1, result.get_result())50    def test_delay_call__above_max_response_time(self):51        with server_context(FooService(), max_response_time=0.5):52            client = create_client()53            result = client.delay_call(async_rpc=True)54            self.assertFalse(result.is_done())55            self.assertEqual(2, result.get_result())56    def test_delay_call__below_max_response_time(self):57        with server_context(FooService(), max_response_time=2):58            client = create_client()59            result = client.delay_call(async_rpc=True)60            self.assertTrue(result.is_done())61            self.assertEqual(2, result.get_result())62    def test_no_delay_deferred_call(self):63        with server_context(FooService(), max_response_time=1):64            client = create_client()65            result = client.no_delay_deferred_call(async_rpc=True)66            self.assertFalse(result.is_done())67            self.assertEqual(3, result.get_result())68    def test_delay_deferred_call__below_max_response_time(self):69        with server_context(FooService(), max_response_time=2):70            client = create_client()71            result = client.delay_deferred_call(async_rpc=True)72            self.assertFalse(result.is_done())73            self.assertEqual(4, result.get_result())74    def test_delay_deferred_call__above_max_response_time(self):75        with server_context(FooService(), max_response_time=0.5):76            client = create_client()77            result = client.delay_deferred_call(async_rpc=True)78            self.assertFalse(result.is_done())79            self.assertEqual(4, result.get_result())80    def test_auto_timeout_client__timeout_implicilty_found(self):81        with server_context(FooService(), max_response_time=0.5):82            client = AutoTimeoutClient(ZeroRPCClientTransport.create_tcp(8192))83            client.no_delay_call()84            self.assertEqual(0.5, client.get_server_max_response_time())85    def test_auto_timeout_client__timeout_explicitly_found(self):86        with server_context(FooService(), max_response_time=0.5):87            client = AutoTimeoutClient(ZeroRPCClientTransport.create_tcp(8192))88            self.assertEqual(0.5, client.get_server_max_response_time())89    def test_auto_timeout_client__short_timeout_on_stuck_server(self):90        import time91        from threading import Event92        wait_for_start = Event()93        wait_for_close = Event()94        def thread_server(wait_for_start, wait_for_close):95            try:96                print(("starting server, hub: {}".format(gevent.hub.get_hub())))97                with logbook.NullHandler().applicationbound():98                    with server_context(FooService(), max_response_time=0.1):99                        print("server started.")100                        wait_for_start.set()101                        while not wait_for_close.is_set():102                            gevent.sleep(0.1)103            except:104                import traceback105                traceback.print_exc()106        from gevent.threadpool import ThreadPool107        t = ThreadPool(1)108        t.size = 1109        t.spawn(thread_server, wait_for_start, wait_for_close)110        try:111            print(("starting client, hub: {}".format(gevent.hub.get_hub())))112            client = AutoTimeoutClient(ZeroRPCClientTransport.create_tcp(8192), timeout_calc_func=lambda n: n * 2)113            wait_for_start.wait()114            print("client started.")115            t1 = time.time()116            self.assertRaises(TimeoutExpired, client.stuck_call)117            t2 = time.time()118            # This test should always pass although we're dealing with timing and non-deterministic measures since119            # stuck_call() is stuck for an entire second while we're comparing time to 0.2 (almost an order of a120            # magnitude)121            self.assertAlmostEqual(0.2, t2 - t1, delta=0.2)122        finally:123            wait_for_close.set()...

runner.py

Source:runner.py

...12    Returns:13        A :class:`pwnlib.tubes.process.process` tube to interact with the process.14    Example:15        >>> p = run_assembly('mov ebx, 3; mov eax, SYS_exit; int 0x80;')16        >>> p.wait_for_close()17        >>> p.poll()18        319        >>> p = run_assembly('mov r0, #12; mov r7, #1; svc #0', arch='arm')20        >>> p.wait_for_close()21        >>> p.poll()22        1223    """24    return ELF.from_assembly(assembly).process()25@LocalContext26def run_shellcode(bytes, **kw):27    """Given assembled machine code bytes, execute them.28    Example:29        >>> bytes = asm('mov ebx, 3; mov eax, SYS_exit; int 0x80;')30        >>> p = run_shellcode(bytes)31        >>> p.wait_for_close()32        >>> p.poll()33        334        >>> bytes = asm('mov r0, #12; mov r7, #1; svc #0', arch='arm')35        >>> p = run_shellcode(bytes, arch='arm')36        >>> p.wait_for_close()37        >>> p.poll()38        1239    """40    return ELF.from_bytes(bytes, **kw).process()41@LocalContext42def run_assembly_exitcode(assembly):43    """44    Given an assembly listing, assemble and execute it, and wait for45    the process to die.46    Returns:47        The exit code of the process.48    Example:49        >>> run_assembly_exitcode('mov ebx, 3; mov eax, SYS_exit; int 0x80;')50        351    """52    p = run_assembly(assembly)53    p.wait_for_close()54    return p.poll()55@LocalContext56def run_shellcode_exitcode(bytes):57    """58    Given assembled machine code bytes, execute them, and wait for59    the process to die.60    Returns:61        The exit code of the process.62    Example:63        >>> bytes = asm('mov ebx, 3; mov eax, SYS_exit; int 0x80;')64        >>> run_shellcode_exitcode(bytes)65        366    """67    p = run_shellcode(bytes)68    p.wait_for_close()...

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.