How to use enable_promisc method in autotest

Best Python code snippet using autotest_python

testing.py

Source:testing.py

...50	for line in lines[2:]:51		if 'lo' not in line.split(":")[0]:52			interfaces.append(line.split(":")[0])53	return54def enable_promisc(interface):55	#enable promiscuous listening mode on every interface:56	os.system(f'ifconfig {interface} promisc')57def disable_promisc(interface):58	#disable promiscuous listening mode on every interface:59	os.system(f'ifconfig {interface} -promisc')60def format_packet_data(packet):61	packet_data = {'timestamp' : packet.sniff_timestamp,}62	if 'ETH' in str(packet.layers):63		packet_data['eth'] = {	'src' : packet.eth.src,64								'dst' : packet.eth.dst,}65	if 'TCP' in str(packet.layers) or 'UDP' in str(packet.layers):66		packet_data['ip']  = {	'src' : packet.ip.src,67								'dst' : packet.ip.dst,}68	if 'TCP' in str(packet.layers):69		packet_data['tcp'] = {	'src' : packet.tcp.srcport,70								'dst' : packet.tcp.dstport,}71	if 'UDP' in str(packet.layers):72		packet_data['udp'] = {	'src' : packet.udp.srcport,73								'dst' : packet.udp.dstport,}74	if 'ARP' in str(packet.layers):75		packet_data['arp'] = {	'mac' : {	'src' : packet.arp.src_hw_mac,76											'dst' : packet.arp.dst_hw_mac,77										},78								'ip'  : {	'src' : packet.arp.src_proto_ipv4,79											'dst' : packet.arp.dst_proto_ipv4,80										},81								'code': packet.arp.opcode,}82	# packet.pretty_print()83	return json.dumps(packet_data)84def live_capture(interface=None):85	#start a live packet capture on the specified interface86	print(f'waiting for packets on {interface}')87	enable_promisc(interface)88	capture = pyshark.LiveCapture(interface=interface, only_summaries=False)89	for packet in capture.sniff_continuously():90		with open(f'./pcap_{interface}.log', 'a') as log:91			log.write(format_packet_data(packet) + '\n') 92	disable_promisc(interface)93	return94def create_captures():95	#create a seperate capture for each interface96	captures =[]97	find_interfaces()98	for interface in interfaces:99		p = multiprocessing.Process(target=live_capture, args=(interface,))100		captures.append(p)101		p.start()...

wscript

Source:wscript

1#!/usr/bin/env python2# encoding: utf-83import os4import time5APPNAME = 'csp-term'6VERSION = '2.0'7top	= '.'8out	= 'build'9# Scan modules in local lib dir10modules = ['lib/libcsp', 'lib/libparam', 'lib/libutil', 'lib/libgosh', 'lib/libftp', 'lib/liblog']11def options(ctx):12	ctx.load('gcc')13	if os.path.exists('../tools/eclipse.py'):14		ctx.load('eclipse', tooldir='../tools/')15	ctx.recurse(modules)16	ctx.load('eclipse')17	gr = ctx.add_option_group("CSP-Term options")18	gr.add_option('--with-adcs', action='store_true',help='Enable ADCS client module')19	20def configure(ctx):21	ctx.load('gcc')22	23	try:24		ctx.find_program("sphinx-build", var="SPHINX_BUILD")25	except ctx.errors.ConfigurationError:26		pass27	28	ctx.env.append_unique('INCLUDES_CSPTERM',['include', 'client', 'Prototypes/libex', \29                                                  'albertasat-gomspace/albertasat-on-board-computer/liba/Subsystem/include', \30                                                  'CObject/liba/Class', 'CObject/liba/util', 'pforth/csrc'])31	ctx.env.append_unique('FILES_CSPTERM', 'src/*.c')32	ctx.env.append_unique('LIBS_CSPTERM', ['pforth', 'm', 'cutil', 'cclass', 'rt', 'pthread', 'elf', 'ncurses'])33	ctx.env.append_unique('DEFINES_CSPTERM', ['AUTOMATION', 'BDEBUG', 'OUTPUT_LOG', 'OUTPUT_LOG_NAME="' + os.getcwd() + '/logs/output_log.txt"', \34                                                  'PFORTH_DIC_PATH="pforth/build/unix/pforth.dic"'])35	ctx.env.append_unique('LIBPATH_CSPTERM', [os.getcwd() + '/Prototypes/libex/debug', \36						  os.getcwd() + '/CObject/liba/Class/debug', \37						  os.getcwd() + '/CObject/liba/util/debug', \38                                                  os.getcwd() + '/pforth/build/unix'])39        ctx.env.append_unique('LINKFLAGS_CSPTERM', ['-O0'])40	ctx.env.append_unique('CFLAGS_CSPTERM', ['-O0'])41	42	# Check WAF can find the required libraries.43	# Options for CSP44	ctx.options.with_os = 'posix'45	ctx.options.enable_rdp = True46	ctx.options.enable_qos = True47	ctx.options.enable_crc32 = True48	ctx.options.enable_hmac = True49	ctx.options.enable_xtea = True50	ctx.options.enable_promisc = True51	ctx.options.enable_if_kiss = True52	ctx.options.enable_if_can = True53	ctx.options.enable_if_zmqhub = True54	ctx.options.disable_stlib = True55	ctx.options.with_rtable = 'cidr'56	ctx.options.with_driver_can = 'socketcan'57	ctx.options.with_driver_usart = 'linux'58	# Options for libftp59	ctx.options.enable_ftp_server = True60	ctx.options.enable_ftp_client = True61	ctx.options.enable_fat = False62	ctx.options.enable_uffs = False63	ctx.options.enable_newlib = True64	65	# Options for liblog66	ctx.options.enable_log_node = True67	# Options for libutil68	ctx.options.clock = 'linux'69	ctx.options.enable_driver_debug = True70	ctx.options.with_log = 'cdh'71	ctx.options.enable_vmem = True72	ctx.options.enable_lzo = True73	# Options for libadcs74	if ctx.options.with_adcs == True:75		ctx.define('WITH_ADCS', '1')76		ctx.options.enable_adcs_client = True77	78	# Options for libparam79	ctx.options.enable_param_local_client = True80	ctx.options.enable_param_csp_server = True81	ctx.options.enable_param_csp_client = True82	ctx.options.enable_param_cmd = True83	ctx.options.param_lock = 'none'84	ctx.define('CSPTERM_VERSION', VERSION)85	# Recurse and write config86	ctx.write_config_header('include/conf_cspterm.h', top=True, remove=True)87	ctx.recurse(modules, mandatory=False)	88def build(ctx):89	ctx(export_includes=ctx.env.INCLUDES_CSPTERM, name='include')90	ctx.recurse(modules, mandatory=False)91	ctx.program(92		source=ctx.path.ant_glob(ctx.env.FILES_CSPTERM), 93		stdlibpath = ['-LCObject/liba/Class/debug/', '-LCObject/liba/util/debug', '-Lpforth/build/unix'],94		defines = ctx.env.DEFINES_NANOMIND,95		target='csp-term', 96		use=['CSPTERM', 'csp', 'param', 'util', 'gosh', 'ftp', 'log', 'cclass', 'cutil', 'pforth'],97		lib=ctx.env.LIBS_CSPTERM + ctx.env.LIBS98		)99def doc(ctx):100	ctx(101		cwd    = ctx.path.abspath(),102		rule   = 'doxygen doc/breathe.doxygen 2>&1 > /dev/null || echo doxygen error',103		source = './doc/breathe.doxygen',104		target = './doxygen/xml/index.xml',105		)106	ctx(107		rule   = "${SPHINX_BUILD} -q -c ./doc -b html -D release="+VERSION+" -d build/doc/doctrees . build/doc/html",108		cwd    = ctx.path.abspath(),109		source = ctx.path.ant_glob('**/*.rst') + ['doxygen/xml/index.xml'],110		target = './doc/html/doc/index.html',111		)112	ctx(113		rule   = "${SPHINX_BUILD} -q -c ./doc -b latex -D release="+VERSION+" -d build/doc/doctrees . build/doc/latex",114		cwd    = ctx.path.abspath(),115		source = ctx.path.ant_glob('**/*.rst') + ['doxygen/xml/index.xml'],116		target = './doc/latex/csp-term.tex',117		)118	ctx(119		cwd    = ctx.path.abspath(),120		rule   = 'make -C build/doc/latex all-pdf 2>&1 > /dev/null || echo make error',121		source = './doc/latex/csp-term.tex',122		target = './doc/latex/csp-term.pdf',123		)124from waflib.Build import BuildContext125class Doc(BuildContext):126   cmd = 'doc'127   fun = 'doc'128   129def dist(ctx):130    if not ctx.path.find_node('build/doc/latex/csp-term.pdf'):131        ctx.fatal('You forgot to run \'waf doc\' first, we need to include the documentation in the output')132    git_rev = os.popen('git describe --always --dirty=-dirty 2> /dev/null || echo unknown').read().strip()133    os.system('cp build/doc/latex/csp-term.pdf doc/csp-term-'+VERSION+'.pdf')134    os.system('cp -r build/doc/html doc/html')...

test-net-dev.py

Source:test-net-dev.py

1#!/usr/bin/env python32import os3import shutil4import sys5import time6from pwn import *7DIRECTORY = os.environ.get('VMM_DIRECTORY', '/app')8DEBUG_CONSOLE_PATH = os.environ.get('VMM_DEBUG_CONSOLE_PATH', '/app/ooowsserial-debug.py')9DISK_IMAGE_PATH = os.environ.get('VMM_DISK_IMAGE_PATH', '/app/disk')10OS_BOOTED = b"OOO OS BOOTED"11PREPARE_NET = b"prepare_net"12SET_DRIVER_FEATURES = b"net_set_driver_features"13ENABLE_CHKSUM_TX = b"net_enable_chksum_tx"14DISABLE_CHKSUM_TX = b"net_disable_chksum_tx"15ENABLE_ETH_CRC_TX = b"net_enable_tx_eth"16DISABLE_ETH_CRC_TX = b"net_disable_tx_eth"17ENABLE_PROMISC = b"net_enable_promisc"18DISABLE_PROMISC = b"net_disable_promisc"19TX_NORMAL_PACKET = b"tx_pkt"20TX_IPV4_PACKET = b"tx_ipv4_pkt"21RX_PACKET = b"rx_pkt"22LISTEN_ONE_MSG_CMD = ["mosquitto_sub", "-C", "1", "-N", "-t", "VPC"]23SEND_ONE_MSG_CMD = ["mosquitto_pub", "-s", "-t", "VPC"]24NORMAL_PACKET = b"DSTMACSRCMACLN\x61TLIDFOTP11IPIPDTDTDTDTAAA"25NORMAL_PACKET_CRC = b"\x4d\x1a\x85\xa2"26IPV4_NORMAL_PACKET = b"DSTMACSRCMACLN\x45GTLIDFOTP__IPIPDTDTDTDTAA"27IPV4_NORMAL_PACKET_CRC = b"\x7b\xa2\x62\xfa"28IPV4_CHKSUM_PACKET = b"DSTMACSRCMACLN\x45GTLIDFOTP<tIPIPDTDTDTDTAA"29BROADCAST_PACKET =b"TAAG\xff\xff\xff\xff\xff\xffSRCMACBodyOfPacketYo"30NON_BROADCAST_PACKET =b"TAAG\x11\x11\x11\x11\x11\xffSRCMACBodyOfPacketYo"31# context.log_level = 'debug'32def check_pkt_tx(p, cmd, correct):33    # Try to receive a packet34    listen = process(LISTEN_ONE_MSG_CMD)35    p.sendline(cmd)36    received_packet = listen.readall()37    assert(len(received_packet) > 4)38    pkt_data = received_packet[4:]39    if len(pkt_data) != len(correct):40        print("PUBLIC: packet tx fail: incorrect size")41        sys.exit(-1)42    if pkt_data != correct:43        print("PUBLIC: packet tx fail: incorrect data received")44        sys.exit(-1)45    listen.kill()46def check_pkt_rx(p, cmd, pkt, expected):47    p.sendline(cmd)48    p.recvuntil(cmd)49    # give it time to call into the driver50    time.sleep(10)51    send = process(SEND_ONE_MSG_CMD)52    send.send(pkt)53    send.shutdown()54    hex_size_regex = b"([0-9a-bA-B]+)\n"55    result = p.recvregex(hex_size_regex)56    match = re.search(hex_size_regex, result)57    size_hex = match.group(1)58    size = int(size_hex, 16)59    if size != len(expected):60        print("PUBLIC: packet rx fail: incorrect size")61        sys.exit(-1)62    raw_pkt = p.recvn(size)63    if raw_pkt != expected:64        print("PUBLIC: packet rx fail: incorrect data received")65        sys.exit(-1)66    send.kill()67def main():68    os.chdir(DIRECTORY)69    # Check if we need to run mosquitto server70    mosquitto = None71    if not 'DO_NOT_START_MOSQUITTO' in os.environ:72        mosquitto = process(["/usr/sbin/mosquitto", "-c", "/etc/mosquitto/mosquitto.conf"])73    shutil.copyfile(DEBUG_CONSOLE_PATH, "./devices-bin/ooowsserial.py")74    p = process(["./vmm", "test", DISK_IMAGE_PATH, "1", "devices.config"])75    p.recvuntil(OS_BOOTED)76    # important, need to initialize the structures77    p.recvuntil(b'$')78    p.sendline(PREPARE_NET)79    p.recvuntil(b'$')80    p.sendline(SET_DRIVER_FEATURES)81    p.recvuntil(b'$')82    check_pkt_tx(p, TX_NORMAL_PACKET, NORMAL_PACKET)83    p.recvuntil(b'$')84    check_pkt_tx(p, TX_IPV4_PACKET, IPV4_NORMAL_PACKET)85    p.recvuntil(b'$')86    p.sendline(ENABLE_CHKSUM_TX)87    p.recvuntil(b'$')88    check_pkt_tx(p, TX_IPV4_PACKET, IPV4_CHKSUM_PACKET)89    # Verify that non IPv4 packets are not affected90    p.recvuntil(b'$')91    check_pkt_tx(p, TX_NORMAL_PACKET, NORMAL_PACKET)92    p.recvuntil(b'$')93    p.sendline(DISABLE_CHKSUM_TX)94    p.recvuntil(b'$')95    p.sendline(ENABLE_ETH_CRC_TX)96    p.recvuntil(b'$')97    check_pkt_tx(p, TX_NORMAL_PACKET, NORMAL_PACKET + NORMAL_PACKET_CRC)98    p.recvuntil(b'$')99    check_pkt_tx(p, TX_IPV4_PACKET, IPV4_NORMAL_PACKET + IPV4_NORMAL_PACKET_CRC)100    p.recvuntil(b'$')101    check_pkt_rx(p, RX_PACKET, BROADCAST_PACKET, BROADCAST_PACKET[4:])102    p.recvuntil(b'$')103    p.sendline(ENABLE_PROMISC)104    p.recvuntil(b'$')105    check_pkt_rx(p, RX_PACKET, NON_BROADCAST_PACKET, NON_BROADCAST_PACKET[4:])106    p.kill()107    if mosquitto:108        mosquitto.kill()109if __name__ == '__main__':...

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.