espressif/esp-protocols
1
0
Fork 1
mirror of https://github.com/espressif/esp-protocols.git synced 2025-07-18 04:52:18 +02:00
Code Issues Packages Projects Releases Wiki Activity

style: format python files with isort and double-quote-string-fixer

Browse Source 
* Original commit: espressif/esp-idf@0146f258d7
This commit is contained in:
Fu Hanxi
2021-01-26 10:49:01 +08:00
committed by suren-gabrielyan-espressif
parent dec1d7dcf8
commit 49a00c2455
1 changed files with 28 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
examples/protocols/mdns/mdns_example_test.py
View File

@ -1,15 +1,15 @@
import re
import os import os
import re
import socket import socket
import time
import struct import struct
import subprocess
import time
from threading import Event, Thread
import dpkt import dpkt
import dpkt.dns import dpkt.dns
from threading import Thread, Event
import subprocess
from tiny_test_fw import DUT
import ttfw_idf import ttfw_idf
from tiny_test_fw import DUT
stop_mdns_server = Event() stop_mdns_server = Event()
esp_answered = Event() esp_answered = Event()
@ -18,7 +18,7 @@ esp_answered = Event()
def get_dns_query_for_esp(esp_host): def get_dns_query_for_esp(esp_host):
dns = dpkt.dns.DNS(b'\x00\x00\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x01') dns = dpkt.dns.DNS(b'\x00\x00\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x01')
dns.qd[0].name = esp_host + u'.local' dns.qd[0].name = esp_host + u'.local'
print("Created query for esp host: {} ".format(dns.__repr__())) print('Created query for esp host: {} '.format(dns.__repr__()))
return dns.pack() return dns.pack()
@ -32,26 +32,26 @@ def get_dns_answer_to_mdns(tester_host):
arr.name = tester_host arr.name = tester_host
arr.ip = socket.inet_aton('127.0.0.1') arr.ip = socket.inet_aton('127.0.0.1')
dns. an.append(arr) dns. an.append(arr)
print("Created answer to mdns query: {} ".format(dns.__repr__())) print('Created answer to mdns query: {} '.format(dns.__repr__()))
return dns.pack() return dns.pack()
def get_dns_answer_to_mdns_lwip(tester_host, id): def get_dns_answer_to_mdns_lwip(tester_host, id):
dns = dpkt.dns.DNS(b"\x5e\x39\x84\x00\x00\x01\x00\x01\x00\x00\x00\x00\x0a\x64\x61\x76\x69\x64" dns = dpkt.dns.DNS(b'\x5e\x39\x84\x00\x00\x01\x00\x01\x00\x00\x00\x00\x0a\x64\x61\x76\x69\x64'
b"\x2d\x63\x6f\x6d\x70\x05\x6c\x6f\x63\x61\x6c\x00\x00\x01\x00\x01\xc0\x0c" b'\x2d\x63\x6f\x6d\x70\x05\x6c\x6f\x63\x61\x6c\x00\x00\x01\x00\x01\xc0\x0c'
b"\x00\x01\x00\x01\x00\x00\x00\x0a\x00\x04\xc0\xa8\x0a\x6c") b'\x00\x01\x00\x01\x00\x00\x00\x0a\x00\x04\xc0\xa8\x0a\x6c')
dns.qd[0].name = tester_host dns.qd[0].name = tester_host
dns.an[0].name = tester_host dns.an[0].name = tester_host
dns.an[0].ip = socket.inet_aton('127.0.0.1') dns.an[0].ip = socket.inet_aton('127.0.0.1')
dns.an[0].rdata = socket.inet_aton('127.0.0.1') dns.an[0].rdata = socket.inet_aton('127.0.0.1')
dns.id = id dns.id = id
print("Created answer to mdns (lwip) query: {} ".format(dns.__repr__())) print('Created answer to mdns (lwip) query: {} '.format(dns.__repr__()))
return dns.pack() return dns.pack()
def mdns_server(esp_host): def mdns_server(esp_host):
global esp_answered global esp_answered
UDP_IP = "0.0.0.0" UDP_IP = '0.0.0.0'
UDP_PORT = 5353 UDP_PORT = 5353
MCAST_GRP = '224.0.0.251' MCAST_GRP = '224.0.0.251'
TESTER_NAME = u'tinytester.local' TESTER_NAME = u'tinytester.local'
@ -60,7 +60,7 @@ def mdns_server(esp_host):
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
sock.bind((UDP_IP,UDP_PORT)) sock.bind((UDP_IP,UDP_PORT))
mreq = struct.pack("4sl", socket.inet_aton(MCAST_GRP), socket.INADDR_ANY) mreq = struct.pack('4sl', socket.inet_aton(MCAST_GRP), socket.INADDR_ANY)
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
sock.settimeout(30) sock.settimeout(30)
while not stop_mdns_server.is_set(): while not stop_mdns_server.is_set():
@ -72,14 +72,14 @@ def mdns_server(esp_host):
dns = dpkt.dns.DNS(data) dns = dpkt.dns.DNS(data)
if len(dns.qd) > 0 and dns.qd[0].type == dpkt.dns.DNS_A: if len(dns.qd) > 0 and dns.qd[0].type == dpkt.dns.DNS_A:
if dns.qd[0].name == TESTER_NAME: if dns.qd[0].name == TESTER_NAME:
print("Received query: {} ".format(dns.__repr__())) print('Received query: {} '.format(dns.__repr__()))
sock.sendto(get_dns_answer_to_mdns(TESTER_NAME), (MCAST_GRP,UDP_PORT)) sock.sendto(get_dns_answer_to_mdns(TESTER_NAME), (MCAST_GRP,UDP_PORT))
elif dns.qd[0].name == TESTER_NAME_LWIP: elif dns.qd[0].name == TESTER_NAME_LWIP:
print("Received query: {} ".format(dns.__repr__())) print('Received query: {} '.format(dns.__repr__()))
sock.sendto(get_dns_answer_to_mdns_lwip(TESTER_NAME_LWIP, dns.id), addr) sock.sendto(get_dns_answer_to_mdns_lwip(TESTER_NAME_LWIP, dns.id), addr)
if len(dns.an) > 0 and dns.an[0].type == dpkt.dns.DNS_A: if len(dns.an) > 0 and dns.an[0].type == dpkt.dns.DNS_A:
if dns.an[0].name == esp_host + u'.local': if dns.an[0].name == esp_host + u'.local':
print("Received answer to esp32-mdns query: {}".format(dns.__repr__())) print('Received answer to esp32-mdns query: {}'.format(dns.__repr__()))
esp_answered.set() esp_answered.set()
except socket.timeout: except socket.timeout:
break break
@ -87,7 +87,7 @@ def mdns_server(esp_host):
continue continue
@ttfw_idf.idf_example_test(env_tag="Example_WIFI") @ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_mdns(env, extra_data): def test_examples_protocol_mdns(env, extra_data):
global stop_mdns_server global stop_mdns_server
""" """
@ -97,21 +97,21 @@ def test_examples_protocol_mdns(env, extra_data):
3. check the mdns name is accessible 3. check the mdns name is accessible
4. check DUT output if mdns advertized host is resolved 4. check DUT output if mdns advertized host is resolved
""" """
dut1 = env.get_dut("mdns-test", "examples/protocols/mdns", dut_class=ttfw_idf.ESP32DUT) dut1 = env.get_dut('mdns-test', 'examples/protocols/mdns', dut_class=ttfw_idf.ESP32DUT)
# check and log bin size # check and log bin size
binary_file = os.path.join(dut1.app.binary_path, "mdns-test.bin") binary_file = os.path.join(dut1.app.binary_path, 'mdns-test.bin')
bin_size = os.path.getsize(binary_file) bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance("mdns-test_bin_size", "{}KB".format(bin_size // 1024)) ttfw_idf.log_performance('mdns-test_bin_size', '{}KB'.format(bin_size // 1024))
# 1. start mdns application # 1. start mdns application
dut1.start_app() dut1.start_app()
# 2. get the dut host name (and IP address) # 2. get the dut host name (and IP address)
specific_host = dut1.expect(re.compile(r"mdns hostname set to: \[([^\]]+)\]"), timeout=30) specific_host = dut1.expect(re.compile(r'mdns hostname set to: \[([^\]]+)\]'), timeout=30)
specific_host = str(specific_host[0]) specific_host = str(specific_host[0])
thread1 = Thread(target=mdns_server, args=(specific_host,)) thread1 = Thread(target=mdns_server, args=(specific_host,))
thread1.start() thread1.start()
try: try:
ip_address = dut1.expect(re.compile(r" sta ip: ([^,]+),"), timeout=30)[0] ip_address = dut1.expect(re.compile(r' sta ip: ([^,]+),'), timeout=30)[0]
print("Connected to AP with IP: {}".format(ip_address)) print('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout: except DUT.ExpectTimeout:
stop_mdns_server.set() stop_mdns_server.set()
thread1.join() thread1.join()
@ -121,15 +121,15 @@ def test_examples_protocol_mdns(env, extra_data):
if not esp_answered.wait(timeout=30): if not esp_answered.wait(timeout=30):
raise ValueError('Test has failed: did not receive mdns answer within timeout') raise ValueError('Test has failed: did not receive mdns answer within timeout')
# 4. check DUT output if mdns advertized host is resolved # 4. check DUT output if mdns advertized host is resolved
dut1.expect(re.compile(r"mdns-test: Query A: tinytester.local resolved to: 127.0.0.1"), timeout=30) dut1.expect(re.compile(r'mdns-test: Query A: tinytester.local resolved to: 127.0.0.1'), timeout=30)
dut1.expect(re.compile(r"mdns-test: gethostbyname: tinytester-lwip.local resolved to: 127.0.0.1"), timeout=30) dut1.expect(re.compile(r'mdns-test: gethostbyname: tinytester-lwip.local resolved to: 127.0.0.1'), timeout=30)
dut1.expect(re.compile(r"mdns-test: getaddrinfo: tinytester-lwip.local resolved to: 127.0.0.1"), timeout=30) dut1.expect(re.compile(r'mdns-test: getaddrinfo: tinytester-lwip.local resolved to: 127.0.0.1'), timeout=30)
# 5. check the DUT answers to `dig` command # 5. check the DUT answers to `dig` command
dig_output = subprocess.check_output(['dig', '+short', '-p', '5353', '@224.0.0.251', dig_output = subprocess.check_output(['dig', '+short', '-p', '5353', '@224.0.0.251',
'{}.local'.format(specific_host)]) '{}.local'.format(specific_host)])
print('Resolving {} using "dig" succeeded with:\n{}'.format(specific_host, dig_output)) print('Resolving {} using "dig" succeeded with:\n{}'.format(specific_host, dig_output))
if not ip_address.encode('utf-8') in dig_output: if not ip_address.encode('utf-8') in dig_output:
raise ValueError("Test has failed: Incorrectly resolved DUT hostname using dig" raise ValueError('Test has failed: Incorrectly resolved DUT hostname using dig'
"Output should've contained DUT's IP address:{}".format(ip_address)) "Output should've contained DUT's IP address:{}".format(ip_address))
finally: finally:
stop_mdns_server.set() stop_mdns_server.set()