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

How to use is_killed method in localstack

Best Python code snippet using localstack_python

poloniex.py

Source:poloniex.py

1# Import Built-Ins2import signal3import logging4import multiprocessing as mp5import time6# Import Third-Party7from autobahn.asyncio.wamp import ApplicationRunner, ApplicationSession8from asyncio import coroutine, get_event_loop9import requests10# Import Homebrew11from bitex.api.WSS.base import WSSAPI12# Init Logging Facilities13log = logging.getLogger(__name__)14class PoloniexSession(ApplicationSession):15    @coroutine16    def onJoin(self, *args, **kwargs):17        channel = self.config.extra['channel']18        def onTicker(*args, **kwargs):19            self.config.extra['queue'].put((channel, (args, kwargs, time.time())))20        if self.config.extra['is_killed'].is_set():21            raise KeyboardInterrupt()22        try:23            yield from self.subscribe(onTicker, self.config.extra['channel'])24        except Exception as e:25            raise26class PlnxEndpoint(mp.Process):27    def __init__(self, endpoint, q, **kwargs):28        super(PlnxEndpoint, self).__init__(name='%s Endpoint Process' %29                                                endpoint, **kwargs)30        self.endpoint = endpoint31        self.q = q32        self.is_killed = mp.Event()33    def run(self):34        self.runner = ApplicationRunner("wss://api.poloniex.com:443", 'realm1',35                                   extra={'channel': self.endpoint,36                                          'queue': self.q,37                                          'is_killed': self.is_killed})38        self.runner.run(PoloniexSession)39    def join(self, *args, **kwargs):40        self.is_killed.set()41        super(PlnxEndpoint, self).join(*args, **kwargs)42class PoloniexWSS(WSSAPI):43    def __init__(self, endpoints=None):44        super(PoloniexWSS, self).__init__(None, 'Poloniex')45        self.data_q = mp.Queue()46        self.connections = {}47        if endpoints:48            self.endpoints = endpoints49        else:50            r = requests.get('https://poloniex.com/public?command=returnTicker')51            self.endpoints = list(r.json().keys())52            self.endpoints.append('ticker')53        for endpoint in self.endpoints:54            self.connections[endpoint] = PlnxEndpoint(endpoint, self.data_q)55    def start(self):56        super(PoloniexWSS, self).start()57        for conn in self.connections:58            self.connections[conn].start()59    def stop(self):60        for conn in self.connections:61            self.connections[conn].join()62        super(PoloniexWSS, self).stop()63if __name__ == "__main__":64    wss = PoloniexWSS()65    wss.start()66    time.sleep(5)67    wss.stop()68    while not wss.data_q.empty():...

run-shape-fuzzer-tests.py

Source:run-shape-fuzzer-tests.py

1#!/usr/bin/env python2from __future__ import print_function, division, absolute_import3import sys, os, subprocess, tempfile, threading4def which(program):5	# https://stackoverflow.com/a/3770286	def is_exe(fpath):7		return os.path.isfile(fpath) and os.access(fpath, os.X_OK)8	fpath, _ = os.path.split(program)9	if fpath:10		if is_exe(program):11			return program12	else:13		for path in os.environ["PATH"].split(os.pathsep):14			exe_file = os.path.join(path, program)15			if is_exe(exe_file):16				return exe_file17	return None18def cmd(command):19	# https://stackoverflow.com/a/440840920	# https://stackoverflow.com/a/1001226221	with tempfile.TemporaryFile() as tempf:22		p = subprocess.Popen (command, stderr=tempf)23		is_killed = {'value': False}24		def timeout(p, is_killed):25			is_killed['value'] = True26			p.kill()27		timer = threading.Timer (2, timeout, [p, is_killed])28		try:29			timer.start()30			p.wait ()31			tempf.seek (0)32			text = tempf.read().decode ("utf-8").strip ()33			returncode = p.returncode34		finally:35			timer.cancel()36		if is_killed['value']:37			text = 'error: timeout, ' + text38			returncode = 139		return text, returncode40srcdir = os.environ.get ("srcdir", ".")41EXEEXT = os.environ.get ("EXEEXT", "")42top_builddir = os.environ.get ("top_builddir", ".")43hb_shape_fuzzer = os.path.join (top_builddir, "hb-shape-fuzzer" + EXEEXT)44if not os.path.exists (hb_shape_fuzzer):45	if len (sys.argv) == 1 or not os.path.exists (sys.argv[1]):46		print ("""Failed to find hb-shape-fuzzer binary automatically,47please provide it as the first argument to the tool""")48		sys.exit (1)49	hb_shape_fuzzer = sys.argv[1]50print ('hb_shape_fuzzer:', hb_shape_fuzzer)51fails = 052valgrind = None53if os.environ.get('RUN_VALGRIND', ''):54	valgrind = which ('valgrind')55parent_path = os.path.join (srcdir, "fonts")56for file in os.listdir (parent_path):57	path = os.path.join(parent_path, file)58	text, returncode = cmd ([hb_shape_fuzzer, path])59	if text.strip ():60		print (text)61	failed = False62	if returncode != 0 or 'error' in text:63		print ('failure on %s' % file)64		failed = True65	if valgrind:66		text, returncode = cmd ([valgrind, '--error-exitcode=1', hb_shape_fuzzer, path])67		if returncode:68			print (text)69			print ('failure on %s' % file)70			failed = True71	if failed:72		fails = fails + 173if fails:74	print ("%i shape fuzzer related tests failed." % fails)...

02. MuOnline v.2.py

Source:02. MuOnline v.2.py

1health = 1002bitcoins = 03rooms = input().split("|")4is_killed = False5for room in rooms:6    info_room = room.split()7    action, number = info_room[0], int(info_room[1])8    if action == "potion":9        difference = 100 - health10        if difference <= number:11            number = difference12        health += number13        print(f"You healed for {number} hp.")14        print(f"Current health: {health} hp.")15    elif action == "chest":16        print(f"You found {number} bitcoins.")17        bitcoins += number18    else:19        health -= number20        if health > 0:21            print(f"You slayed {action}.")22        else:23            print(f"You died! Killed by {action}.")24            print(f"Best room: {rooms.index(room)+1}")25            is_killed = True26            break27if not is_killed:...

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.