KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use close_sync method in localstack
Best Python code snippet using localstack_python
virtual-device.py
Source:virtual-device.py 
...73 self.assertTrue(self.dev.is_open())74 def tearDown(self):75 if self._close_on_teardown:76 self.assertTrue(self.dev.is_open())77 self.dev.close_sync()78 self.assertFalse(self.dev.is_open())79 super().tearDown()80 def wait_timeout(self, interval):81 timeout_reached = False82 def on_timeout():83 nonlocal timeout_reached84 timeout_reached = True85 GLib.timeout_add(interval, on_timeout)86 while not timeout_reached:87 ctx.iteration(False)88 def send_command(self, command, *args):89 self.assertIn(command, ['INSERT', 'REMOVE', 'SCAN', 'ERROR', 'RETRY',90 'FINGER', 'UNPLUG', 'SLEEP', 'SET_ENROLL_STAGES', 'SET_SCAN_TYPE',91 'SET_CANCELLATION_ENABLED', 'SET_KEEP_ALIVE', 'IGNORED_COMMAND',92 'CONT'])93 with Connection(self.sockaddr) as con:94 params = ' '.join(str(p) for p in args)95 con.sendall('{} {}'.format(command, params).encode('utf-8'))96 while ctx.pending():97 ctx.iteration(False)98 def send_finger_report(self, has_finger, iterate=True):99 self.send_command('FINGER', 1 if has_finger else 0)100 if iterate:101 expected = (FPrint.FingerStatusFlags.PRESENT if has_finger102 else ~FPrint.FingerStatusFlags.PRESENT)103 while not (self.dev.get_finger_status() & expected):104 ctx.iteration(True)105 def send_error(self, error):106 self.assertIsInstance(error, FPrint.DeviceError)107 self.send_command('ERROR', int(error))108 def send_retry(self, retry):109 self.assertIsInstance(retry, FPrint.DeviceRetry)110 self.send_command('RETRY', int(retry))111 def send_auto(self, obj):112 if isinstance(obj, FPrint.DeviceError):113 self.send_error(obj)114 elif isinstance(obj, FPrint.DeviceRetry):115 self.send_retry(obj)116 elif isinstance(obj, FPrint.FingerStatusFlags):117 self.send_finger_report(obj & FPrint.FingerStatusFlags.PRESENT, iterate=False)118 elif isinstance(obj, FPrint.ScanType):119 self.send_command('SET_SCAN_TYPE', obj.value_nick)120 elif isinstance(obj, FPrint.Print) and obj.props.fpi_data:121 self.send_command('SCAN', obj.props.fpi_data.unpack())122 else:123 raise Exception('No known type found for {}'.format(obj))124 def set_keep_alive(self, value):125 self.send_command('SET_KEEP_ALIVE', 1 if value else 0)126 def send_sleep(self, interval):127 self.assertGreater(interval, 0)128 multiplier = 5 if 'UNDER_VALGRIND' in os.environ else 1129 self.send_command('SLEEP', interval * multiplier)130 def enroll_print(self, nick, finger, username='testuser', retry_scan=-1):131 self._enrolled = None132 def done_cb(dev, res):133 print("Enroll done")134 try:135 self._enrolled = dev.enroll_finish(res)136 except Exception as e:137 self._enrolled = e138 self._enroll_stage = -1139 def progress_cb(dev, stage, pnt, data, error):140 self._enroll_stage = stage141 self._enroll_progress_error = error142 self.assertLessEqual(retry_scan, self.dev.get_nr_enroll_stages())143 retries = 1144 should_retry = retry_scan > 0145 def enroll_in_progress():146 if self._enroll_stage < 0 and not self._enrolled:147 return True148 if isinstance(self._enrolled, Exception):149 raise(self._enrolled)150 nonlocal retries151 self.assertLessEqual(self._enroll_stage, self.dev.get_nr_enroll_stages())152 if should_retry and retries > retry_scan:153 self.assertEqual(self._enroll_stage, retries - 1)154 else:155 self.assertEqual(self._enroll_stage, retries)156 if retries == retry_scan + 1:157 self.assertIsNotNone(self._enroll_progress_error)158 self.assertEqual(self._enroll_progress_error.code, FPrint.DeviceRetry.TOO_SHORT)159 else:160 self.assertIsNone(self._enroll_progress_error)161 if self._enroll_stage < self.dev.get_nr_enroll_stages():162 self._enroll_stage = -1163 self.assertIsNone(self._enrolled)164 self.assertEqual(self.dev.get_finger_status(),165 FPrint.FingerStatusFlags.NEEDED)166 if retry_scan == retries:167 GLib.idle_add(self.send_auto, FPrint.DeviceRetry.TOO_SHORT)168 else:169 GLib.idle_add(self.send_command, 'SCAN', nick)170 retries += 1171 return not self._enrolled172 self.assertEqual(self.dev.get_finger_status(), FPrint.FingerStatusFlags.NONE)173 self.send_command('SCAN', nick)174 template = FPrint.Print.new(self.dev)175 template.set_finger(finger)176 template.set_username(username)177 self.dev.enroll(template, callback=done_cb, progress_cb=progress_cb)178 while enroll_in_progress():179 ctx.iteration(False)180 self.assertEqual(self._enroll_stage, retries if not should_retry else retries - 1)181 self.assertEqual(self._enroll_stage, self.dev.get_nr_enroll_stages())182 self.assertEqual(self.dev.get_finger_status(), FPrint.FingerStatusFlags.NONE)183 self.assertEqual(self._enrolled.get_device_stored(),184 self.dev.has_storage())185 self.assertEqual(self._enrolled.props.driver, self.dev.get_driver())186 self.assertEqual(self._enrolled.props.device_id, self.dev.get_device_id())187 self.assertEqual(self._enrolled.props.device_stored, self.dev.has_storage())188 self.assertEqual(self._enrolled.props.fpi_data.unpack(), nick)189 self.assertIsNone(self._enrolled.props.image)190 return self._enrolled191 def start_verify(self, p, identify=False):192 self._verify_match = None193 self._verify_fp = None194 self._verify_error = None195 self._verify_report_match = None196 self._verify_report_print = None197 self._verify_completed = False198 self._verify_reported = False199 self._cancellable = Gio.Cancellable()200 if identify:201 self.assertTrue(self.dev.supports_identify())202 def match_cb(dev, match, pnt, data, error):203 self._verify_reported = True204 self._verify_report_match = match205 self._verify_report_print = pnt206 self._verify_report_error = error207 def verify_cb(dev, res):208 try:209 self._verify_match, self._verify_fp = (210 dev.identify_finish(res) if identify else dev.verify_finish(res))211 except gi.repository.GLib.Error as e:212 self._verify_error = e213 self._verify_completed = True214 if identify:215 self.dev.identify(p if isinstance(p, list) else [p],216 cancellable=self._cancellable, match_cb=match_cb, callback=verify_cb)217 else:218 self.dev.verify(p, cancellable=self._cancellable, match_cb=match_cb,219 callback=verify_cb)220 def cancel_verify(self):221 self._cancellable.cancel()222 while not self._verify_completed:223 ctx.iteration(True)224 self.assertIsNone(self._verify_match)225 self.assertIsNotNone(self._verify_error)226 self.assertEqual(self.dev.get_finger_status(), FPrint.FingerStatusFlags.NONE)227 def complete_verify(self):228 while not self._verify_completed:229 ctx.iteration(True)230 if self._verify_error is not None:231 raise self._verify_error232 def check_verify(self, p, scan_nick, match, identify=False):233 if isinstance(scan_nick, str):234 self.send_command('SCAN', scan_nick)235 elif scan_nick is not None:236 self.send_auto(scan_nick)237 self.start_verify(p, identify)238 self.complete_verify()239 self.assertTrue(self._verify_reported)240 if not match:241 self.assertIsNone(self._verify_report_match)242 if identify:243 if match:244 self.assertIsNotNone(self._verify_report_match)245 self.assertIsNotNone(self._verify_match)246 else:247 if self._verify_fp:248 self.assertEqual(self._verify_fp.equal(p), match)249 if match:250 self.assertTrue(251 self._verify_fp.equal(self._verify_report_match))252 else:253 self.assertFalse(match)254 if isinstance(scan_nick, str) and not self.dev.has_storage():255 self.assertEqual(self._verify_fp.props.fpi_data.get_string(), scan_nick)256class VirtualDevice(VirtualDeviceBase):257 def test_device_properties(self):258 self.assertEqual(self.dev.get_driver(), 'virtual_device')259 self.assertEqual(self.dev.get_device_id(), '0')260 self.assertEqual(self.dev.get_name(), 'Virtual device for debugging')261 self.assertTrue(self.dev.is_open())262 self.assertEqual(self.dev.get_scan_type(), FPrint.ScanType.SWIPE)263 self.assertEqual(self.dev.get_nr_enroll_stages(), 5)264 self.assertFalse(self.dev.supports_identify())265 self.assertFalse(self.dev.supports_capture())266 self.assertFalse(self.dev.has_storage())267 self.assertEqual(self.dev.props.driver, self.dev.get_driver())268 self.assertEqual(self.dev.props.device_id, self.dev.get_device_id())269 self.assertEqual(self.dev.props.name, self.dev.get_name())270 self.assertEqual(self.dev.props.scan_type, self.dev.get_scan_type())271 self.assertEqual(self.dev.props.nr_enroll_stages, self.dev.get_nr_enroll_stages())272 self.assertEqual(self.dev.props.open, self.dev.is_open())273 def test_device_features(self):274 self.assertFalse(self.dev.has_feature(FPrint.DeviceFeature.CAPTURE))275 self.assertFalse(self.dev.has_feature(FPrint.DeviceFeature.IDENTIFY))276 self.assertTrue(self.dev.has_feature(FPrint.DeviceFeature.VERIFY))277 self.assertFalse(self.dev.has_feature(FPrint.DeviceFeature.DUPLICATES_CHECK))278 self.assertFalse(self.dev.has_feature(FPrint.DeviceFeature.STORAGE))279 self.assertFalse(self.dev.has_feature(FPrint.DeviceFeature.STORAGE_LIST))280 self.assertFalse(self.dev.has_feature(FPrint.DeviceFeature.STORAGE_DELETE))281 self.assertFalse(self.dev.has_feature(FPrint.DeviceFeature.STORAGE_CLEAR))282 def test_open_error(self):283 self._close_on_teardown = False284 self.send_command('IGNORED_COMMAND') # This will be consumed by close285 self.send_error(FPrint.DeviceError.PROTO) # This will be consumed by open286 with GLibErrorMessage('libfprint-virtual_device',287 GLib.LogLevelFlags.LEVEL_WARNING, 'Could not process command: *'):288 self.dev.close_sync()289 with self.assertRaises(GLib.Error) as error:290 self.dev.open_sync()291 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),292 FPrint.DeviceError.PROTO))293 def test_open_error_with_keep_alive(self):294 self._close_on_teardown = False295 self.set_keep_alive(True)296 self.dev.close_sync()297 self.send_error(FPrint.DeviceError.PROTO)298 with self.assertRaises(GLib.Error) as error:299 self.dev.open_sync()300 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),301 FPrint.DeviceError.PROTO))302 def test_delayed_open(self):303 self.send_command('IGNORED_COMMAND') # This will be consumed by close304 self.send_sleep(500) # This will be consumed by open305 with GLibErrorMessage('libfprint-virtual_device',306 GLib.LogLevelFlags.LEVEL_WARNING, 'Could not process command: *'):307 self.dev.close_sync()308 opened = False309 def on_opened(dev, res):310 nonlocal opened311 dev.open_finish(res)312 opened = True313 self.dev.open(callback=on_opened)314 self.wait_timeout(10)315 self.assertFalse(self.dev.is_open())316 self.wait_timeout(10)317 self.assertFalse(self.dev.is_open())318 while not opened:319 ctx.iteration(True)320 def test_delayed_open_with_keep_alive(self):321 self.set_keep_alive(True)322 self.dev.close_sync()323 opened = False324 def on_opened(dev, res):325 nonlocal opened326 dev.open_finish(res)327 opened = True328 self.send_sleep(500)329 self.dev.open(callback=on_opened)330 self.wait_timeout(10)331 self.assertFalse(self.dev.is_open())332 self.wait_timeout(10)333 self.assertFalse(self.dev.is_open())334 while not opened:335 ctx.iteration(True)336 def test_close_while_opening(self):337 self.set_keep_alive(True)338 self.dev.close_sync()339 opened = False340 def on_opened(dev, res):341 nonlocal opened342 dev.open_finish(res)343 opened = True344 self.send_sleep(500)345 self.dev.open(callback=on_opened)346 self.wait_timeout(10)347 self.assertFalse(self.dev.is_open())348 with self.assertRaises(GLib.Error) as error:349 self.dev.close_sync()350 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),351 FPrint.DeviceError.NOT_OPEN))352 while not opened:353 ctx.iteration(True)354 def test_enroll(self):355 matching = self.enroll_print('testprint', FPrint.Finger.LEFT_LITTLE)356 self.assertEqual(matching.get_username(), 'testuser')357 self.assertEqual(matching.get_finger(), FPrint.Finger.LEFT_LITTLE)358 def test_enroll_with_retry(self):359 matching = self.enroll_print('testprint', FPrint.Finger.LEFT_LITTLE, retry_scan=2)360 self.assertEqual(matching.get_username(), 'testuser')361 self.assertEqual(matching.get_finger(), FPrint.Finger.LEFT_LITTLE)362 def test_enroll_verify_match(self):363 matching = self.enroll_print('testprint', FPrint.Finger.LEFT_THUMB)364 self.check_verify(matching, 'testprint', match=True,365 identify=self.dev.supports_identify())366 def test_enroll_verify_no_match(self):367 matching = self.enroll_print('testprint', FPrint.Finger.LEFT_RING)368 self.check_verify(matching, 'not-testprint', match=False,369 identify=self.dev.supports_identify())370 def test_enroll_verify_error(self):371 matching = self.enroll_print('testprint', FPrint.Finger.LEFT_RING)372 with self.assertRaises(GLib.Error) as error:373 self.check_verify(matching, FPrint.DeviceError.GENERAL, match=False,374 identify=self.dev.supports_identify())375 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),376 FPrint.DeviceError.GENERAL))377 def test_enroll_verify_retry(self):378 with self.assertRaises(GLib.GError) as error:379 self.check_verify(FPrint.Print.new(self.dev),380 FPrint.DeviceRetry.TOO_SHORT, match=False)381 self.assertTrue(error.exception.matches(FPrint.DeviceRetry.quark(),382 FPrint.DeviceRetry.TOO_SHORT))383 def test_enroll_script_interactive(self):384 enrolled = None385 def done_cb(dev, res):386 nonlocal enrolled387 try:388 enrolled = dev.enroll_finish(res)389 except Exception as e:390 enrolled = e391 enroll_stage = 0392 enroll_progress_error = None393 def progress_cb(dev, stage, pnt, data, error):394 nonlocal enroll_stage, enroll_progress_error395 enroll_stage = stage396 enroll_progress_error = error397 def wait_for_next_stage(expected):398 nonlocal enroll_stage, enroll_progress_error399 enroll_progress_error = None400 next_stage = enroll_stage + 1401 while enroll_stage < next_stage and not enroll_progress_error:402 ctx.iteration(True)403 if isinstance(expected, FPrint.DeviceRetry):404 self.assertEqual(enroll_stage, next_stage - 1)405 self.assertEqual(enroll_progress_error.code, int(expected))406 else:407 self.assertEqual(enroll_stage, expected)408 self.assertIsNone(enroll_progress_error)409 self.assertIsNone(enrolled)410 self.send_sleep(50)411 self.send_command('SCAN', 'print-id')412 self.send_command('SCAN', 'print-id')413 self.send_auto(FPrint.DeviceRetry.TOO_SHORT)414 self.send_command('SCAN', 'print-id')415 self.send_sleep(50)416 self.send_command('SCAN', 'print-id')417 self.send_auto(FPrint.DeviceRetry.CENTER_FINGER)418 self.send_command('SCAN', 'another-id')419 self.send_command('SCAN', 'print-id')420 self.dev.enroll(FPrint.Print.new(self.dev), callback=done_cb,421 progress_cb=progress_cb)422 wait_for_next_stage(1)423 wait_for_next_stage(2)424 wait_for_next_stage(FPrint.DeviceRetry.TOO_SHORT)425 wait_for_next_stage(3)426 wait_for_next_stage(4)427 wait_for_next_stage(FPrint.DeviceRetry.CENTER_FINGER)428 wait_for_next_stage(FPrint.DeviceRetry.GENERAL)429 wait_for_next_stage(5)430 while not enrolled:431 ctx.iteration(True)432 self.assertEqual(enrolled.get_driver(), self.dev.get_driver())433 self.assertEqual(enrolled.props.fpi_data.unpack(), 'print-id')434 def test_enroll_script(self):435 self.send_command('SET_ENROLL_STAGES', 8)436 self.send_command('SCAN', 'print-id')437 self.send_command('SCAN', 'print-id')438 self.send_auto(FPrint.DeviceRetry.TOO_SHORT)439 self.send_command('SCAN', 'print-id')440 self.send_auto(FPrint.DeviceRetry.REMOVE_FINGER)441 self.send_command('SCAN', 'print-id')442 self.send_auto(FPrint.DeviceRetry.CENTER_FINGER)443 self.send_command('SCAN', 'print-id')444 self.send_sleep(10)445 self.send_sleep(20)446 self.send_auto(FPrint.DeviceRetry.GENERAL)447 self.send_auto(FPrint.DeviceRetry.REMOVE_FINGER)448 self.send_command('SCAN', 'print-id')449 self.send_command('SCAN', 'another-id')450 self.send_command('SCAN', 'print-id')451 self.send_command('SCAN', 'print-id')452 enrolled = self.dev.enroll_sync(FPrint.Print.new(self.dev))453 self.assertEqual(enrolled.get_driver(), self.dev.get_driver())454 self.assertEqual(enrolled.props.fpi_data.unpack(), 'print-id')455 return enrolled456 def test_enroll_verify_script(self):457 enrolled = self.test_enroll_script()458 self.send_auto(FPrint.DeviceRetry.CENTER_FINGER)459 with self.assertRaises(GLib.GError) as error:460 self.dev.verify_sync(enrolled)461 self.assertTrue(error.exception.matches(FPrint.DeviceRetry.quark(),462 FPrint.DeviceRetry.CENTER_FINGER))463 self.send_sleep(50)464 self.send_auto(FPrint.DeviceRetry.TOO_SHORT)465 with self.assertRaises(GLib.GError) as error:466 self.dev.verify_sync(enrolled)467 self.assertTrue(error.exception.matches(FPrint.DeviceRetry.quark(),468 FPrint.DeviceRetry.TOO_SHORT))469 self.send_command('SCAN', 'another-id')470 verify_match, verify_fp = self.dev.verify_sync(enrolled)471 self.assertFalse(verify_match)472 if self.dev.has_storage():473 self.assertIsNone(verify_fp)474 else:475 self.assertFalse(verify_fp.equal(enrolled))476 self.send_auto(enrolled)477 verify_match, verify_fp = self.dev.verify_sync(enrolled)478 self.assertTrue(verify_match)479 self.assertTrue(verify_fp.equal(enrolled))480 def test_finger_status(self):481 self.start_verify(FPrint.Print.new(self.dev),482 identify=self.dev.supports_identify())483 self.assertEqual(self.dev.get_finger_status(),484 FPrint.FingerStatusFlags.NEEDED)485 self.send_finger_report(True)486 self.assertEqual(self.dev.get_finger_status(),487 FPrint.FingerStatusFlags.NEEDED | FPrint.FingerStatusFlags.PRESENT)488 self.send_finger_report(False)489 self.assertEqual(self.dev.get_finger_status(), FPrint.FingerStatusFlags.NEEDED)490 self.cancel_verify()491 def test_finger_status_after_sleep(self):492 self.send_sleep(10)493 self.start_verify(FPrint.Print.new(self.dev),494 identify=self.dev.supports_identify())495 self.assertEqual(self.dev.get_finger_status(),496 FPrint.FingerStatusFlags.NONE)497 while self.dev.get_finger_status() != FPrint.FingerStatusFlags.NEEDED:498 ctx.iteration(True)499 self.assertEqual(self.dev.get_finger_status(),500 FPrint.FingerStatusFlags.NEEDED)501 self.send_finger_report(True)502 self.assertEqual(self.dev.get_finger_status(),503 FPrint.FingerStatusFlags.NEEDED | FPrint.FingerStatusFlags.PRESENT)504 self.send_finger_report(False)505 self.assertEqual(self.dev.get_finger_status(),506 FPrint.FingerStatusFlags.NEEDED)507 self.cancel_verify()508 def test_change_enroll_stages(self):509 notified_spec = None510 def on_stage_changed(dev, spec):511 nonlocal notified_spec512 notified_spec = spec513 self.dev.connect('notify::nr-enroll-stages', on_stage_changed)514 notified_spec = None515 self.send_command('SET_ENROLL_STAGES', 20)516 self.assertEqual(self.dev.get_nr_enroll_stages(), 20)517 self.assertEqual(notified_spec.name, 'nr-enroll-stages')518 notified_spec = None519 self.send_command('SET_ENROLL_STAGES', 1)520 self.assertEqual(self.dev.get_nr_enroll_stages(), 1)521 self.assertEqual(notified_spec.name, 'nr-enroll-stages')522 with GLibErrorMessage('libfprint-device',523 GLib.LogLevelFlags.LEVEL_CRITICAL, '*enroll_stages > 0*'):524 notified_spec = None525 self.send_command('SET_ENROLL_STAGES', 0)526 self.assertEqual(self.dev.get_nr_enroll_stages(), 1)527 self.assertIsNone(notified_spec)528 def test_quick_enroll(self):529 self.send_command('SET_ENROLL_STAGES', 1)530 self.assertEqual(self.dev.get_nr_enroll_stages(), 1)531 matching = self.enroll_print('testprint', FPrint.Finger.LEFT_LITTLE)532 self.assertEqual(matching.get_username(), 'testuser')533 self.assertEqual(matching.get_finger(), FPrint.Finger.LEFT_LITTLE)534 def test_change_scan_type(self):535 notified_spec = None536 def on_scan_type_changed(dev, spec):537 nonlocal notified_spec538 notified_spec = spec539 self.dev.connect('notify::scan-type', on_scan_type_changed)540 for scan_type in [FPrint.ScanType.PRESS, FPrint.ScanType.SWIPE]:541 notified_spec = None542 self.send_auto(scan_type)543 self.assertEqual(self.dev.get_scan_type(), scan_type)544 self.assertEqual(notified_spec.name, 'scan-type')545 with GLibErrorMessage('libfprint-virtual_device',546 GLib.LogLevelFlags.LEVEL_WARNING, '*Scan type*not found'):547 notified_spec = None548 self.send_command('SET_SCAN_TYPE', 'eye-contact')549 self.assertEqual(self.dev.get_scan_type(), FPrint.ScanType.SWIPE)550 self.assertIsNone(notified_spec)551 def test_device_unplug(self):552 self._close_on_teardown = False553 notified_spec = None554 def on_removed_notify(dev, spec):555 nonlocal notified_spec556 notified_spec = spec557 removed = False558 def on_removed(dev):559 nonlocal removed560 removed = True561 self.assertFalse(self.dev.props.removed)562 self.dev.connect('notify::removed', on_removed_notify)563 self.dev.connect('removed', on_removed)564 self.send_command('UNPLUG')565 self.assertEqual(notified_spec.name, 'removed')566 self.assertTrue(self.dev.props.removed)567 self.assertTrue(removed)568 with self.assertRaises(GLib.GError) as error:569 self.dev.close_sync()570 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),571 FPrint.DeviceError.REMOVED))572 def test_device_unplug_during_verify(self):573 self._close_on_teardown = False574 notified_spec = None575 def on_removed_notify(dev, spec):576 nonlocal notified_spec577 notified_spec = spec578 removed = False579 def on_removed(dev):580 nonlocal removed581 removed = True582 self.assertFalse(self.dev.props.removed)583 self.dev.connect('notify::removed', on_removed_notify)584 self.dev.connect('removed', on_removed)585 self.start_verify(FPrint.Print.new(self.dev),586 identify=self.dev.supports_identify())587 self.send_command('UNPLUG')588 self.assertEqual(notified_spec.name, 'removed')589 self.assertTrue(self.dev.props.removed)590 self.assertFalse(removed)591 with self.assertRaises(GLib.GError) as error:592 self.complete_verify()593 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),594 FPrint.DeviceError.REMOVED))595 self.assertTrue(removed)596 with self.assertRaises(GLib.GError) as error:597 self.dev.close_sync()598 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),599 FPrint.DeviceError.REMOVED))600 def test_device_sleep(self):601 self.send_sleep(1500)602 self.start_verify(FPrint.Print.new(self.dev),603 identify=self.dev.supports_identify())604 self.wait_timeout(300)605 self.assertFalse(self._verify_completed)606 self._cancellable.cancel()607 self.wait_timeout(200)608 self.assertTrue(self._verify_completed)609 self.cancel_verify()610 def test_device_sleep_on_cancellation(self):611 self.send_command('SET_CANCELLATION_ENABLED', int(False))612 self.send_sleep(1500)613 self.send_command('SCAN', 'foo-print')614 self.start_verify(FPrint.Print.new(self.dev),615 identify=self.dev.supports_identify())616 self.wait_timeout(300)617 self.assertFalse(self._verify_completed)618 self._cancellable.cancel()619 self.wait_timeout(300)620 self.assertFalse(self._verify_completed)621 self.cancel_verify()622 # Since we don't really cancel here, next command will be passed to release623 self._close_on_teardown = False624 with GLibErrorMessage('libfprint-virtual_device',625 GLib.LogLevelFlags.LEVEL_WARNING, 'Could not process command: SCAN *'):626 self.dev.close_sync()627 def test_device_sleep_before_completing_verify(self):628 enrolled = self.enroll_print('foo-print', FPrint.Finger.LEFT_RING)629 self.send_sleep(100)630 self.start_verify(enrolled, identify=self.dev.supports_identify())631 self.send_command('SCAN', 'bar-print')632 self.send_sleep(800)633 while not self._verify_reported:634 ctx.iteration(False)635 self.assertFalse(self._verify_completed)636 self.wait_timeout(10)637 self.assertFalse(self._verify_completed)638 self.complete_verify()639 self.assertTrue(self._verify_reported)640 def test_close_error(self):641 self._close_on_teardown = False642 close_res = None643 def on_closed(dev, res):644 nonlocal close_res645 try:646 close_res = dev.close_finish(res)647 except GLib.Error as e:648 close_res = e649 self.send_sleep(100)650 self.send_error(FPrint.DeviceError.BUSY)651 self.dev.close(callback=on_closed)652 self.wait_timeout(2)653 self.assertIsNone(close_res)654 while not close_res:655 ctx.iteration(True)656 self.assertEqual(close_res.code, int(FPrint.DeviceError.BUSY))657 def test_identify_unsupported(self):658 if self.dev.supports_identify():659 self.skipTest('Device supports identification')660 with self.assertRaises(GLib.Error) as error:661 self.dev.identify_sync([FPrint.Print.new(self.dev)])662 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),663 FPrint.DeviceError.NOT_SUPPORTED))664 def test_capture_unsupported(self):665 with self.assertRaises(GLib.Error) as error:666 self.dev.capture_sync(wait_for_finger=False)667 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),668 FPrint.DeviceError.NOT_SUPPORTED))669class VirtualDeviceClosed(VirtualDeviceBase):670 driver_name = 'virtual_device'671 def setUp(self):672 super().setUp()673 self._close_on_teardown = False674 self.dev.close_sync()675 self.assertFalse(self.dev.is_open())676 def test_close(self):677 with self.assertRaises(GLib.Error) as error:678 self.dev.close_sync()679 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),680 FPrint.DeviceError.NOT_OPEN))681 def test_enroll(self):682 with self.assertRaises(GLib.Error) as error:683 self.dev.enroll_sync(FPrint.Print.new(self.dev))684 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),685 FPrint.DeviceError.NOT_OPEN))686 def test_verify(self):687 with self.assertRaises(GLib.Error) as error:688 self.dev.verify_sync(FPrint.Print.new(self.dev))689 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),690 FPrint.DeviceError.NOT_OPEN))691 def test_identify(self):692 with self.assertRaises(GLib.Error) as error:693 self.dev.identify_sync([FPrint.Print.new(self.dev)])694 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),695 FPrint.DeviceError.NOT_OPEN))696 def test_capture(self):697 with self.assertRaises(GLib.Error) as error:698 self.dev.capture_sync(wait_for_finger=False)699 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),700 FPrint.DeviceError.NOT_OPEN))701 def test_delete_print(self):702 with self.assertRaises(GLib.Error) as error:703 self.dev.delete_print_sync(FPrint.Print.new(self.dev))704 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),705 FPrint.DeviceError.NOT_OPEN))706 def test_list_prints(self):707 with self.assertRaises(GLib.Error) as error:708 self.dev.list_prints_sync()709 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),710 FPrint.DeviceError.NOT_OPEN))711class VirtualDeviceBusyDeviceOperations(VirtualDeviceBase):712 driver_name = 'virtual_device'713 def setUp(self):714 super().setUp()715 self._close_on_teardown = False716 self.send_sleep(200)717 self.dev.close()718 def tearDown(self):719 while self.dev.is_open():720 ctx.iteration(True)721 super().tearDown()722 def test_open(self):723 self.set_keep_alive(True)724 while self.dev.is_open():725 ctx.iteration(True)726 self.assertFalse(self.dev.is_open())727 self.send_sleep(100)728 self.dev.open()729 with self.assertRaises(GLib.Error) as error:730 self.dev.open_sync()731 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),732 FPrint.DeviceError.BUSY))733 self.assertFalse(self.dev.is_open())734 while not self.dev.is_open():735 ctx.iteration(True)736 self.dev.close_sync()737 def test_close(self):738 with self.assertRaises(GLib.Error) as error:739 self.dev.close_sync()740 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),741 FPrint.DeviceError.BUSY))742 def test_enroll(self):743 with self.assertRaises(GLib.Error) as error:744 self.dev.enroll_sync(FPrint.Print.new(self.dev))745 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),746 FPrint.DeviceError.BUSY))747 def test_verify(self):748 with self.assertRaises(GLib.Error) as error:749 self.dev.verify_sync(FPrint.Print.new(self.dev))750 self.assertTrue(error.exception.matches(FPrint.DeviceError.quark(),751 FPrint.DeviceError.BUSY))752 def test_identify(self):753 with self.assertRaises(GLib.Error) as error:...
signaling.py
Source:signaling.py
...70 if self._javascript_callable:71 return IPython.display.JSON(result)72 return result73 74 def close_sync(self):75 loop = asyncio.get_event_loop()76 return loop.run_until_complete(self.close())77 78 def recv_nowait(self):79 try:80 return self._websocket.messages.popleft() # .get_nowait()81 #except (asyncio.queues.QueueEmpty, IndexError):82 except IndexError:83 pass84 85 async def receive(self):86 if self.__messages:87 message = self.__messages.pop(0)88 else:89 message = self.recv_nowait()90 if message:91 message = json.loads(message)["msg"]92 93 if message:94 logger.debug("< " + message)95 return object_from_string(message)96 97 def receive_sync(self):98 loop = asyncio.get_event_loop()99 message = loop.run_until_complete(self.receive())100 if message and self._javascript_callable:101 message = object_to_string(message)102 print('receive:', message)103 message = json.loads(message)104 message = IPython.display.JSON(message)105 return message106 107 async def send(self, obj):108 message = object_to_string(obj)109 logger.debug("> " + message)110 if self.__is_initiator:111 await self._http.post(self.__post_url, data=message)112 else:113 await self._websocket.send(json.dumps({"cmd": "send", "msg": message}))114 115 def send_sync(self, message):116 print('send:', message)117 if type(message) == str:118 message_json = json.loads(message)119 if 'candidate' in message_json:120 message_json['type'] = 'candidate'121 message_json["id"] = message_json["sdpMid"]122 message_json["label"] = message_json["sdpMLineIndex"]123 message = json.dumps(message_json) 124 message = object_from_string(message)125 loop = asyncio.get_event_loop()126 return loop.run_until_complete(self.send(message))127 128class ColabSignaling:129 def __init__(self, signaling_folder=None, webrtc_server=None, room=None, javacript_callable=False):130 if room is None:131 room = "".join([random.choice("0123456789") for x in range(10)])132 if webrtc_server is None and signaling_folder is None:133 raise ValueError('Either a WebRTC server or a signaling folder must be provided.')134 if webrtc_server is None:135 self._webrtc_server = FilesystemRTCServer(folder=signaling_folder)136 else:137 self._webrtc_server = webrtc_server138 139 self._room = room140 self._javascript_callable = javacript_callable141 if output and javacript_callable:142 output.register_callback(f'{room}.colab.signaling.connect', self.connect_sync)143 output.register_callback(f'{room}.colab.signaling.send', self.send_sync)144 output.register_callback(f'{room}.colab.signaling.receive', self.receive_sync)145 output.register_callback(f'{room}.colab.signaling.close', self.close_sync)146 @property147 def room(self):148 return self._room149 150 async def connect(self):151 data = self._webrtc_server.join(self._room)152 assert data["result"] == "SUCCESS"153 params = data["params"]154 self.__is_initiator = params["is_initiator"] == "true"155 self.__messages = params["messages"]156 self.__peer_id = params["peer_id"]157 158 logger.info(f"Room ID: {params['room_id']}")159 logger.info(f"Peer ID: {self.__peer_id}")160 return params161 def connect_sync(self):162 loop = asyncio.get_event_loop()163 result = loop.run_until_complete(self.connect())164 if self._javascript_callable:165 return IPython.display.JSON(result)166 return result167 168 async def close(self):169 if self._javascript_callable:170 return self.send_sync(BYE)171 else:172 await self.send(BYE)173 def close_sync(self):174 loop = asyncio.get_event_loop()175 return loop.run_until_complete(self.close())176 177 async def receive(self):178 message = self._webrtc_server.receive_message(self._room, self.__peer_id)179 # if self._javascript_callable:180 # print('ColabSignaling: sending message to Javascript peer:', message)181 # else:182 # print('ColabSignaling: sending message to Python peer:', message)183 if message and type(message) == str and not self._javascript_callable:184 message = object_from_string(message)185 return message186 def receive_sync(self):187 loop = asyncio.get_event_loop()...
virtual-sdcp.py
Source:virtual-sdcp.py
...56 template.props.enroll_date = date57 return self.dev.enroll_sync(template, None, progress_cb, None)58 def test_connect(self):59 self.dev.open_sync()60 self.dev.close_sync()61 def test_reconnect(self):62 # Ensure device was opened once before, this may be a reconnect if63 # it is the same process as another test.64 self.dev.open_sync()65 self.dev.close_sync()66 # Check that a reconnect happens on next open. To know about this, we67 # need to parse check log messages for that.68 success = [False]69 def log_func(domain, level, msg):70 print("log: '%s', '%s', '%s'" % (str(domain), str(level), msg))71 if msg == 'Reconnect succeeded':72 success[0] = True73 # Call default handler74 GLib.log_default_handler(domain, level, msg)75 handler_id = GLib.log_set_handler('libfprint-sdcp_device', GLib.LogLevelFlags.LEVEL_DEBUG, log_func)76 self.dev.open_sync()77 self.dev.close_sync()78 GLib.log_remove_handler('libfprint-sdcp_device', handler_id)79 assert success[0]80 def test_enroll(self):81 self.dev.open_sync()82 # Must return a print83 assert isinstance(self.enroll(), FPrint.Print)84 self.dev.close_sync()85 def test_verify(self):86 self.dev.open_sync()87 # Enroll a new print (will be the last), check that it verifies88 p = self.enroll()89 match, dev_print = self.dev.verify_sync(p)90 assert match91 # The returned "device" print is identical92 assert p.equal(dev_print)93 # We can do the same with it94 match, dev_print2 = self.dev.verify_sync(dev_print)95 assert match96 # Now, enroll a new print, causing the old one to not match anymore97 # (the device always claims to see the last enrolled print).98 self.enroll()99 match, dev_print = self.dev.verify_sync(p)100 assert match is False101 self.dev.close_sync()102if __name__ == '__main__':103 try:104 gi.require_version('FPrint', '2.0')105 from gi.repository import FPrint106 except Exception as e:107 print("Missing dependencies: %s" % str(e))108 sys.exit(77)109 # avoid writing to stderr...
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 localstack 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?
