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

Merge pull request #116 from gabsuren/examples/mdns_examples_integration

Browse Source 
mdns: Example tests integration
This commit is contained in:
david-cermak
2022-09-15 11:41:58 +02:00
committed by GitHub
parent cc4d33d871 a045c1c885
commit b8d1e58778
3 changed files with 130 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
components/mdns/examples/pytest.ini Normal file
View File

@ -0,0 +1,24 @@
[pytest]
# only the files with prefix `pytest_` would be recognized as pytest test scripts.
python_files = pytest_*.py
addopts =
-s
--embedded-services esp,idf
-W ignore::_pytest.warning_types.PytestExperimentalApiWarning
--tb short
# ignore DeprecationWarning
filterwarnings =
ignore:Call to deprecated create function (.*)\(\):DeprecationWarning
# log related
log_cli = True
log_cli_level = INFO
log_cli_format = %(asctime)s %(levelname)s %(message)s
log_cli_date_format = %Y-%m-%d %H:%M:%S
log_file = test.log
log_file_level = INFO
log_file_format = %(asctime)s %(levelname)s %(message)s
log_file_date_format = %Y-%m-%d %H:%M:%S

69
components/mdns/examples/mdns_example_test.py → components/mdns/examples/pytest_mdns.py
View File

@ -5,19 +5,19 @@ import socket
import struct
import subprocess
import time
import pytest
from threading import Event, Thread
import dpkt
import dpkt.dns
import ttfw_idf
from tiny_test_fw import DUT
from tiny_test_fw.Utility import console_log
from pytest_embedded import Dut
import subprocess
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.qd[0].name = esp_host + u'.local'
console_log('Created query for esp host: {} '.format(dns.__repr__()))
print('Created query for esp host: {} '.format(dns.__repr__()))
return dns.pack()
@ -31,7 +31,7 @@ def get_dns_answer_to_mdns(tester_host):
arr.name = tester_host
arr.ip = socket.inet_aton('127.0.0.1')
dns.an.append(arr)
console_log('Created answer to mdns query: {} '.format(dns.__repr__()))
print('Created answer to mdns query: {} '.format(dns.__repr__()))
return dns.pack()
@ -80,18 +80,18 @@ def mdns_server(esp_host, events):
dns = dpkt.dns.DNS(data)
if len(dns.qd) > 0 and dns.qd[0].type == dpkt.dns.DNS_A:
if dns.qd[0].name == TESTER_NAME:
console_log('Received query: {} '.format(dns.__repr__()))
print('Received query: {} '.format(dns.__repr__()))
sock.sendto(get_dns_answer_to_mdns(TESTER_NAME), (MCAST_GRP, UDP_PORT))
elif dns.qd[0].name == TESTER_NAME_LWIP:
console_log('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)
if len(dns.an) > 0 and dns.an[0].type == dpkt.dns.DNS_A:
console_log('Received answer from {}'.format(dns.an[0].name))
print('Received answer from {}'.format(dns.an[0].name))
if dns.an[0].name == esp_host + u'.local':
console_log('Received answer to esp32-mdns query: {}'.format(dns.__repr__()))
print('Received answer to esp32-mdns query: {}'.format(dns.__repr__()))
events['esp_answered'].set()
if dns.an[0].name == esp_host + u'-delegated.local':
console_log('Received answer to esp32-mdns-delegate query: {}'.format(dns.__repr__()))
print('Received answer to esp32-mdns-delegate query: {}'.format(dns.__repr__()))
events['esp_delegated_answered'].set()
except socket.timeout:
break
@ -99,7 +99,7 @@ def mdns_server(esp_host, events):
continue
def test_examples_protocol_mdns(env, config):
def test_examples_protocol_mdns(dut):
"""
steps: |
1. obtain IP address + init mdns example
@ -107,60 +107,31 @@ def test_examples_protocol_mdns(env, config):
3. check the mdns name is accessible
4. check DUT output if mdns advertized host is resolved
"""
dut1 = env.get_dut('mdns-test', 'examples/protocols/mdns', dut_class=ttfw_idf.ESP32DUT, app_config_name=config)
# check and log bin size
binary_file = os.path.join(dut1.app.binary_path, 'mdns_test.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('mdns-test_bin_size', '{}KB'.format(bin_size // 1024))
# 1. start mdns application
dut1.start_app()
# 2. get the dut host name (and IP address)
specific_host = dut1.expect(re.compile(r'mdns hostname set to: \[([^\]]+)\]'), timeout=30)[0]
specific_host = dut.expect(re.compile(b'mdns hostname set to: \[(.*?)\]')).group(1).decode()
mdns_server_events = {'stop': Event(), 'esp_answered': Event(), 'esp_delegated_answered': Event()}
mdns_responder = Thread(target=mdns_server, args=(str(specific_host), mdns_server_events))
ip_address = dut.expect(re.compile(b'IPv4 address:([a-zA-Z0-9]*).*')).group(1).decode()
print('Connected to AP with IP: {}'.format(ip_address))
try:
ip_address = dut1.expect(re.compile(r' eth ip: ([^,]+),'), timeout=30)[0]
console_log('Connected to AP with IP: {}'.format(ip_address))
except DUT.ExpectTimeout:
raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP')
try:
# 3. check the mdns name is accessible
# 3. check the mdns name is accessible.
mdns_responder.start()
if not mdns_server_events['esp_answered'].wait(timeout=30):
raise ValueError('Test has failed: did not receive mdns answer within timeout')
if not mdns_server_events['esp_delegated_answered'].wait(timeout=30):
raise ValueError('Test has failed: did not receive mdns answer for delegated host within timeout')
# 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: 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)
dut.expect(re.compile(b'mdns-test: Query A: tinytester.local resolved to: 127.0.0.1'), timeout=30)
dut.expect(re.compile(b'mdns-test: gethostbyname: tinytester-lwip.local resolved to: 127.0.0.1'), timeout=30)
dut.expect(re.compile(b'mdns-test: getaddrinfo: tinytester-lwip.local resolved to: 127.0.0.1'), timeout=30)
# 5. check the DUT answers to `dig` command
dig_output = subprocess.check_output(['dig', '+short', '-p', '5353', '@224.0.0.251',
'{}.local'.format(specific_host)])
console_log('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:
raise ValueError('Test has failed: Incorrectly resolved DUT hostname using dig'
"Output should've contained DUT's IP address:{}".format(ip_address))
finally:
mdns_server_events['stop'].set()
mdns_responder.join()
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1')
def test_examples_protocol_mdns_default(env, _):
test_examples_protocol_mdns(env, 'eth_def')
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1')
def test_examples_protocol_mdns_socket(env, _):
test_examples_protocol_mdns(env, 'eth_socket')
@ttfw_idf.idf_example_test(env_tag='Example_EthKitV1')
def test_examples_protocol_mdns_custom_netif(env, _):
test_examples_protocol_mdns(env, 'eth_custom_netif')
if __name__ == '__main__':
test_examples_protocol_mdns_default()