1# This example code is in the Public Domain (or CC0 licensed, at your option.)
2
3# Unless required by applicable law or agreed to in writing, this
4# software is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
5# CONDITIONS OF ANY KIND, either express or implied.
6
7# -*- coding: utf-8 -*-
8
9from __future__ import print_function, unicode_literals
10
11import os
12import re
13import socket
14import sys
15
16import ttfw_idf
17
18# -----------  Config  ----------
19PORT = 3333
20INTERFACE = 'eth0'
21# -------------------------------
22
23
24def udp_client(address, payload):
25    for res in socket.getaddrinfo(address, PORT, socket.AF_UNSPEC,
26                                  socket.SOCK_DGRAM, 0, socket.AI_PASSIVE):
27        family_addr, socktype, proto, canonname, addr = res
28    try:
29        sock = socket.socket(family_addr, socket.SOCK_DGRAM)
30        sock.settimeout(20.0)
31    except socket.error as msg:
32        print('Could not create socket')
33        print(os.strerror(msg.errno))
34        raise
35    try:
36        sock.sendto(payload.encode(), addr)
37        reply, addr = sock.recvfrom(128)
38        if not reply:
39            return
40        print('Reply[' + addr[0] + ':' + str(addr[1]) + '] - ' + str(reply))
41    except socket.timeout:
42        print('Socket operation timeout')
43        return str(None)
44    except socket.error as msg:
45        print('Error while sending or receiving data from the socket')
46        print(os.strerror(msg.errno))
47        sock.close()
48        raise
49    return reply.decode()
50
51
52@ttfw_idf.idf_example_test(env_tag='Example_WIFI_Protocols')
53def test_examples_protocol_socket_udpserver(env, extra_data):
54    MESSAGE = 'Data to ESP'
55    MAX_RETRIES = 3
56    """
57    steps:
58      1. join AP
59      2. have the board connect to the server
60      3. send and receive data
61    """
62    dut1 = env.get_dut('udp_server', 'examples/protocols/sockets/udp_server', dut_class=ttfw_idf.ESP32DUT)
63    # check and log bin size
64    binary_file = os.path.join(dut1.app.binary_path, 'udp_server.bin')
65    bin_size = os.path.getsize(binary_file)
66    ttfw_idf.log_performance('udp_server_bin_size', '{}KB'.format(bin_size // 1024))
67
68    # start test
69    dut1.start_app()
70
71    ipv4 = dut1.expect(re.compile(r' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), timeout=30)[0]
72    ipv6_r = r':'.join((r'[0-9a-fA-F]{4}',) * 8)    # expect all 8 octets from IPv6 (assumes it's printed in the long form)
73    ipv6 = dut1.expect(re.compile(r' IPv6 address: ({})'.format(ipv6_r)), timeout=30)[0]
74    print('Connected with IPv4={} and IPv6={}'.format(ipv4, ipv6))
75    dut1.expect(re.compile(r'Waiting for data'), timeout=10)
76
77    # test IPv4
78    for _ in range(MAX_RETRIES):
79        print('Testing UDP on IPv4...')
80        received = udp_client(ipv4, MESSAGE)
81        if received == MESSAGE:
82            print('OK')
83            break
84    else:
85        raise ValueError('IPv4: Did not receive UDP message after {} retries'.format(MAX_RETRIES))
86    dut1.expect(MESSAGE)
87
88    # test IPv6
89    for _ in range(MAX_RETRIES):
90        print('Testing UDP on IPv6...')
91        received = udp_client('{}%{}'.format(ipv6, INTERFACE), MESSAGE)
92        if received == MESSAGE:
93            print('OK')
94            break
95    else:
96        raise ValueError('IPv6: Did not receive UDP message after {} retries'.format(MAX_RETRIES))
97    dut1.expect(MESSAGE)
98
99
100if __name__ == '__main__':
101    if sys.argv[2:]:    # if two arguments provided:
102        # Usage: example_test.py <server_address> <message_to_send_to_server>
103        udp_client(sys.argv[1], sys.argv[2])
104    else:               # otherwise run standard example test as in the CI
105        test_examples_protocol_socket_udpserver()
106