Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run lisa automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
