asio_tcp_server_test.py 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647
  1. import re
  2. import os
  3. import socket
  4. import ttfw_idf
  5. @ttfw_idf.idf_example_test(env_tag="Example_WIFI")
  6. def test_examples_protocol_asio_tcp_server(env, extra_data):
  7. """
  8. steps: |
  9. 1. join AP
  10. 2. Start server
  11. 3. Test connects to server and sends a test message
  12. 4. Test evaluates received test message from server
  13. 5. Test evaluates received test message on server stdout
  14. """
  15. test_msg = b"echo message from client to server"
  16. dut1 = env.get_dut("tcp_echo_server", "examples/protocols/asio/tcp_echo_server", dut_class=ttfw_idf.ESP32DUT)
  17. # check and log bin size
  18. binary_file = os.path.join(dut1.app.binary_path, "asio_tcp_echo_server.bin")
  19. bin_size = os.path.getsize(binary_file)
  20. ttfw_idf.log_performance("asio_tcp_echo_server_bin_size", "{}KB".format(bin_size // 1024))
  21. ttfw_idf.check_performance("asio_tcp_echo_server_size", bin_size // 1024, dut1.TARGET)
  22. # 1. start test
  23. dut1.start_app()
  24. # 2. get the server IP address
  25. data = dut1.expect(re.compile(r" IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)"), timeout=30)
  26. # 3. create tcp client and connect to server
  27. cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  28. cli.settimeout(30)
  29. cli.connect((data[0], 2222))
  30. cli.send(test_msg)
  31. data = cli.recv(1024)
  32. # 4. check the message received back from the server
  33. if (data == test_msg):
  34. print("PASS: Received correct message")
  35. pass
  36. else:
  37. print("Failure!")
  38. raise ValueError('Wrong data received from asi tcp server: {} (expected:{})'.format(data, test_msg))
  39. # 5. check the client message appears also on server terminal
  40. dut1.expect(test_msg.decode())
  41. if __name__ == '__main__':
  42. test_examples_protocol_asio_tcp_server()