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

How to use all_off method in tavern

Best Python code snippet using tavern

tcpdump.py

Source:tcpdump.py

1# -*- coding: utf-8 -*-2'''3ââââââ âââ  ââââ  ââââ  ââ  ââ âââ  âââ ââââ4  ââ  ââ ââ ââ ââ ââ ââ ââ  ââ âââ ââââ ââ ââ5  ââ  ââ    ââââ  ââ ââ ââ  ââ ââ ââ ââ ââââ6  ââ  ââ ââ ââ    ââ ââ ââ  ââ ââ    ââ ââ7  ââ   âââ  ââ    ââââ   ââââ  ââ    ââ ââ8Jeff Thompson | 2014-15 | www.jeffreythompson.org9An online performance, documenting every server my computer10connects to over a period of one month.11'''12# REQUIRED IMPORTS13from functions import *14import os, socket, sys, pwd, glob, grp, time15from threading import Thread16from datetime import datetime17from subprocess import *18# SETUP VARIABLES19compare_stored_urls = 	True				# store unique (true) or all?20group_email_urls = 		True 				# combine 'perfora.net' urls?21col_width = 			10					# width of the "count" column22upload_interval = 		10 * 1000			# how often to re-upload index.php23skip_some_urls = 		True 				# ignore some URL/IPs?24log_filename = 			'AllServers_ASC.csv'25# some IP/URLs to ignore26skip_urls = [ '155.246.200.', 'jeff-thompson.home', 'jeff-thompsons-iphone', 'jeffs-ipad.local.' ]27# tlds that aren't really tlds28not_really_tlds = 		[ 'imap', 'imaps', 'ftp' ]29# COLORS30ALL_OFF = 				'\033[0m'31BOLD = 					'\033[1m'32FG_CYAN = 				'\033[36m'33REVERSE = 				'\033[7m'34# PRINT CONTROL35CLEAR_LINE = 			'\x1b[1A'36CURSOR_UP = 			'\x1b[2K'37# ETC38count = 				039prev_millis = 			millis()40# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -41# HI42os.system('cls' if os.name=='nt' else 'clear')43title = [ "______ ____    ____",44"/\\__  _/\\  _ `\\/\\  _ `\\",45"\\/_/\\ \\\\ \\ \\/\\_\\ \\ \\_\\ \\",46"   \\ \\ \\\\ \\ \\/_/\\ \\  __/",47"    \\ \\ \\\\ \\ \\_\\ \\ \\ \\/",48"     \\ \\_\\\\ \\____/\\ \\_\\",49"      \\/_/ \\/___/  \\/_/",50"____   __  __    _   _   ____",51"/\\  _ `\\/\\ \\/\\ \\  / \\_/ \\/\\  _ `\\",52"\\ \\ \\/\\ \\ \\ \\ \\ \\/\\      \\ \\ \\_\\ \\",53" \\ \\ \\ \\ \\ \\ \\ \\ \\ \\ \\__\\ \\ \\  __/",54"  \\ \\ \\_\\ \\ \\ \\_\\ \\ \\ \\_/\\ \\ \\ \\/",55"   \\ \\____/\\ \\_____\\ \\_\\\\ \\_\\ \\_\\",56"    \\/___/  \\/_____/\\/_/ \\/_/\\/_/" ]57for line in title:58	print FG_CYAN + BOLD + center_text(line) + ALL_OFF59print '\n' + FG_CYAN + center_text('A 24-Hour Performance Documenting Server Connections') + ALL_OFF60print '\n\n' + REVERSE + BOLD + center_text('SETUP') + ALL_OFF + '\n'61# LET US KNOW THE SETTINGS62print 'Combine email URLs:           ',63if group_email_urls:64	print FG_CYAN + BOLD + 'Yes' + ALL_OFF65else:66	print FG_CYAN + BOLD + 'No' + ALL_OFF67print 'Unique URLs only:             ',68if compare_stored_urls:69	print FG_CYAN + BOLD + 'Yes' + ALL_OFF70else:71	print FG_CYAN + BOLD + 'No' + ALL_OFF72# GET LOCAL DOMAIN73# ignore traffic from our own computer74print 'Ignoring local domain:',75local_domain = socket.gethostname()76local_domain = local_domain[:local_domain.rfind('.')]77local_domain = local_domain.lower()78print FG_CYAN + BOLD + '        ' + local_domain + ALL_OFF79# LOAD PREVIOUS URLS80# and prev count, start time, etc81print 'Gathering previous URLs...',82sys.stdout.flush()83previous_urls = set()84try:85	with open(log_filename) as f:86		for i, line in enumerate(f):87			if i == 0:88				continue89			data = line.split(',')90			if i == 1:91				start_time = data[1] + ' at ' + data[2]92			previous_urls.add(data[3].strip())93		count = int(data[0])94except:95	with open(log_filename, 'a') as f:96		f.write('count,date,time,url,subdomain,domain,tld,rest,ip,country,region,city,zip_code,lat,lon' + '\n')97	start_time = datetime.now().strftime('%B %d, %Y at %H:%M:%S')98print '\b\b\b\b:       ' + FG_CYAN + BOLD + 'Found ' + str(len(previous_urls)) + ' URLs' + ALL_OFF99# WHEN DID WE START?100# gathered from previous file or current time above101print 'Start time:                    ' + FG_CYAN + BOLD + start_time + ALL_OFF102# SET PERMISSIONS103# required to run tcpdump from Python104# or run manually first: sudo chown $USER:admin /dev/bp*105print 'Setting /dev/bp* permissions...',106sys.stdout.flush()107uid = pwd.getpwuid(os.getuid()).pw_name108uid = pwd.getpwnam(uid).pw_uid109gid = grp.getgrnam('admin').gr_gid110for f in glob.glob('/dev/bp*'):111	os.chown(f, uid, gid)112print '\b\b\b\b:  ' + FG_CYAN + BOLD + 'Set' + ALL_OFF113# RUN IT!114print '\n\n' + REVERSE + BOLD + center_text('GATHERING') + ALL_OFF + '\n'115print r_align('Count', col_width) + '   ' + 'Server'116try:117	while True:118		# RUN TCPDUMP119		# l = print output as single line120		# q = quiet121		# S = helps prevent "state accumulation" (ie memory leak)122		# p = promiscuous mode (should capture as much traffic as possible)123		p = Popen( [ 'sudo', 'tcpdump', '-lqSp' ], stdout=PIPE, stderr=STDOUT)124		125		# iterate results126		for row in iter(p.stdout.readline, 'b'):127			count += 1128			# parse URL from tcpdump129			url = parse_url(row, local_domain, group_email_urls, 130							compare_stored_urls, previous_urls)131			if url == None:132				continue133			# skip internal Stevens IPs, etc?134			# messy and nasty, I know...135			if skip_some_urls:136				skip = False137				for u in skip_urls:138					if u in url:139						skip = True140						break141				if skip:142					continue143			# all good? what'd we get?144			print FG_CYAN + r_align(str(count), col_width) + '   ' + BOLD + url + ALL_OFF145			# split URL into parts146			url_parts = split_url(url, not_really_tlds)147			# get location for address148			if '155.246.200.18.' in url:149				location = ('', '', '', '', '', '', '')		# ignore internal Stevens IP150			else:151				location = get_location(url)152			# log to file153			log_data(count, url, url_parts, location)154			# enough time passed? created sorted CSV files and upload155			if millis() > prev_millis + upload_interval:156				sort_files(log_filename)157				# upload all three files158				print '\n' + center_text('[ uploading... ]'),159				sys.stdout.flush()160				try:161					t = Thread(target=upload, args=())162					t.start()163					t.join()164					print CURSOR_UP + CLEAR_LINE + BOLD + center_text('[ uploading... DONE! ]') + ALL_OFF165				except:166					print CURSOR_UP + CLEAR_LINE + BOLD + center_text('[ uploading... ERROR, COULDN\'T UPLOAD! ]') + ALL_OFF167				prev_millis = millis()168				169except KeyboardInterrupt:170	print '\b\b   '			# a hack: no ^C on exit :)171	print '\n' + REVERSE + BOLD + center_text('CLOSING') + ALL_OFF + '\n'172	# final upload173	print 'Final upload:      ' + BOLD + FG_CYAN,174	sys.stdout.flush()175	try:176		sort_files(log_filename)177		t = Thread(target=upload, args=())178		t.start()179		t.join()180		print 'Done' + ALL_OFF181	except Exception, e:182		print 'Error uploading' + ALL_OFF183		# print str(e)184	# close it185	print 'Disconnecting FTP:  ' + BOLD + FG_CYAN + 'Closed' + ALL_OFF186	print 'Recorded URLS:      ' + BOLD + FG_CYAN + str(len(previous_urls)) + ' total' + ALL_OFF187	print 'Start time:         ' + FG_CYAN + BOLD + start_time + ALL_OFF188	print 'End time:           ' + FG_CYAN + BOLD + datetime.now().strftime('%B %d, %Y at %H:%M:%S') + ALL_OFF189	print FG_CYAN + '\n' + '[ bye bye ]' + ALL_OFF + '\n'190# all done!...

core.py

Source:core.py

...30        self.C = c31        self.D = d3233        self.init()34        self.all_off()3536    def init(self):37        """init3839        :return:40        """41        GPIO.setup(self.A, GPIO.OUT)42        GPIO.setup(self.B, GPIO.OUT)43        GPIO.setup(self.C, GPIO.OUT)44        GPIO.setup(self.D, GPIO.OUT)4546    def all_off(self):47        """all_off4849        :return:50        """51        GPIO.output(self.A, False)52        GPIO.output(self.B, False)53        GPIO.output(self.C, False)54        GPIO.output(self.D, False)5556    def __del__(self):57        """__del__5859        :return:60        """61        print('Cleaning GPIO...')62        GPIO.cleanup()6364    def steps_x4(self, steps=85):65        """steps_x46667        :return:68        """69        for step_index in range(steps):70            sleep(DELAY)71            self.all_off()72            GPIO.output(self.A, True)73            sleep(DELAY)74            self.all_off()75            GPIO.output(self.C, True)76            sleep(DELAY)77            self.all_off()78            GPIO.output(self.B, True)79            sleep(DELAY)80            self.all_off()81            GPIO.output(self.D, True)82            sleep(DELAY)8384    def steps_minus_x4(self, steps=85):85        """steps_minus_x48687        :return:88        """89        for step_index in range(steps):90            sleep(DELAY)91            self.all_off()92            GPIO.output(self.D, True)93            sleep(DELAY)94            self.all_off()95            GPIO.output(self.B, True)96            sleep(DELAY)97            self.all_off()98            GPIO.output(self.C, True)99            sleep(DELAY)100            self.all_off()101            GPIO.output(self.A, True)102            sleep(DELAY)103104    def rotate_180(self):105        """rotate_180106107        """108        self.steps_x4(85)109110    def rotate_20_steps(self):111        """rotate_20_steps112113        """114        self.steps_x4(5)
...

opcua_client.py

Source:opcua_client.py

1import logging2import time3from opcua import Client4class OPCUAClient:5    @staticmethod6    def illuminate_all(opcua_url, color, duration_sec):7        try:8            client = Client(opcua_url)9            client.connect()10            root_node = client.get_root_node()11            illuminate_all = root_node.get_child(['0:Objects', '3:art_net', '3:node_1_illuminate_all'])12            if illuminate_all:13                root_node.call_method(illuminate_all, color)14            else:15                logging.error('Could not find illuminate_all method on opcua server')16            time.sleep(duration_sec)17            all_off = root_node.get_child(['0:Objects', '3:art_net', '3:node_1_all_off'])18            if all_off:19                root_node.call_method(all_off)20            else:21                logging.error('Could not find all_off method on opcua server')22        except Exception as e:23            logging.error('Error while illuminate_all/all_off: {}'.format(e))24        finally:25            client.disconnect()26    @staticmethod27    def send_barcode(opcua_url, part_number, node):28        try:29            client = Client(opcua_url)30            client.connect()31            root_node = client.get_root_node()32            all_off = root_node.get_child(['0:Objects', '3:art_net', '3:node_1_all_off'])33            if all_off:34                root_node.call_method(all_off)35            else:36                logging.error('Could not find all_off method on opcua server')37            # Sleep, otherwise the led gets turned on and then off sometimes -> timing problem with upd packets?38            time.sleep(0.5)39            method_to_call = root_node.get_child(node)40            if method_to_call:41                res_call = root_node.call_method(method_to_call, part_number)42                return res_call43            else:44                return False45        except Exception as e:46            logging.ERROR("Error while send_barcode: {0}".format(e))47        finally:...

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.