-
Notifications
You must be signed in to change notification settings - Fork 1
/
http_server_simple_test.py
172 lines (138 loc) · 6.26 KB
/
http_server_simple_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
#!/usr/bin/env python
#
# Copyright 2018 Espressif Systems (Shanghai) PTE LTD
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import division, print_function, unicode_literals
import os
import random
import re
import socket
import string
import threading
import time
from builtins import range
import ttfw_idf
from idf_http_server_test import client
from tiny_test_fw import Utility
class http_client_thread(threading.Thread):
def __init__(self, ip, port, delay):
threading.Thread.__init__(self)
self.ip = ip
self.port = port
self.delay = delay
self.exc = None
# Thread function used to open a socket and wait for specific amount of time before returning
def open_connection(self, ip, port, delay):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(delay)
s.connect((ip, port))
time.sleep(delay)
def run(self):
try:
self.open_connection(self.ip, self.port, self.delay)
except socket.timeout as e:
self.exc = e
def join(self, timeout=None):
threading.Thread.join(self)
if self.exc:
raise self.exc
# When running on local machine execute the following before running this script
# > make app bootloader
# > make print_flash_cmd | tail -n 1 > build/download.config
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_http_server_simple(env, extra_data):
# Acquire DUT
dut1 = env.get_dut('http_server', 'examples/protocols/http_server/simple', dut_class=ttfw_idf.ESP32DUT)
# Get binary file
binary_file = os.path.join(dut1.app.binary_path, 'simple.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('http_server_bin_size', '{}KB'.format(bin_size // 1024))
# Upload binary and start testing
Utility.console_log('Starting http_server simple test app')
dut1.start_app()
# Parse IP address of STA
Utility.console_log('Waiting to connect with AP')
got_ip = dut1.expect(re.compile(r'(?:[\s\S]*)IPv4 address: (\d+.\d+.\d+.\d+)'), timeout=30)[0]
got_port = dut1.expect(re.compile(r"(?:[\s\S]*)Starting server on port: '(\d+)'"), timeout=30)[0]
Utility.console_log('Got IP : ' + got_ip)
Utility.console_log('Got Port : ' + got_port)
# Expected Logs
dut1.expect('Registering URI handlers', timeout=30)
# Run test script
# If failed raise appropriate exception
Utility.console_log('Test /hello GET handler')
if not client.test_get_handler(got_ip, got_port):
raise RuntimeError
# Acquire host IP. Need a way to check it
dut1.expect(re.compile(r'(?:[\s\S]*)Found header => Host: (\d+.\d+.\d+.\d+)'), timeout=30)[0]
# Match additional headers sent in the request
dut1.expect('Found header => Test-Header-2: Test-Value-2', timeout=30)
dut1.expect('Found header => Test-Header-1: Test-Value-1', timeout=30)
dut1.expect('Found URL query parameter => query1=value1', timeout=30)
dut1.expect('Found URL query parameter => query3=value3', timeout=30)
dut1.expect('Found URL query parameter => query2=value2', timeout=30)
dut1.expect('Request headers lost', timeout=30)
Utility.console_log('Test /ctrl PUT handler and realtime handler de/registration')
if not client.test_put_handler(got_ip, got_port):
raise RuntimeError
dut1.expect('Unregistering /hello and /echo URIs', timeout=30)
dut1.expect('Registering /hello and /echo URIs', timeout=30)
# Generate random data of 10KB
random_data = ''.join(string.printable[random.randint(0,len(string.printable)) - 1] for _ in range(10 * 1024))
Utility.console_log('Test /echo POST handler with random data')
if not client.test_post_handler(got_ip, got_port, random_data):
raise RuntimeError
query = 'http://foobar'
Utility.console_log('Test /hello with custom query : ' + query)
if not client.test_custom_uri_query(got_ip, got_port, query):
raise RuntimeError
dut1.expect('Found URL query => ' + query, timeout=30)
query = 'abcd+1234%20xyz'
Utility.console_log('Test /hello with custom query : ' + query)
if not client.test_custom_uri_query(got_ip, got_port, query):
raise RuntimeError
dut1.expect('Found URL query => ' + query, timeout=30)
@ttfw_idf.idf_example_test(env_tag='Example_WIFI')
def test_examples_protocol_http_server_lru_purge_enable(env, extra_data):
# Acquire DUT
dut1 = env.get_dut('http_server', 'examples/protocols/http_server/simple', dut_class=ttfw_idf.ESP32DUT)
# Get binary file
binary_file = os.path.join(dut1.app.binary_path, 'simple.bin')
bin_size = os.path.getsize(binary_file)
ttfw_idf.log_performance('http_server_bin_size', '{}KB'.format(bin_size // 1024))
# Upload binary and start testing
Utility.console_log('Starting http_server simple test app')
dut1.start_app()
# Parse IP address of STA
Utility.console_log('Waiting to connect with AP')
got_ip = dut1.expect(re.compile(r'(?:[\s\S]*)IPv4 address: (\d+.\d+.\d+.\d+)'), timeout=30)[0]
got_port = dut1.expect(re.compile(r"(?:[\s\S]*)Starting server on port: '(\d+)'"), timeout=30)[0]
Utility.console_log('Got IP : ' + got_ip)
Utility.console_log('Got Port : ' + got_port)
# Expected Logs
dut1.expect('Registering URI handlers', timeout=30)
threads = []
# Open 20 sockets, one from each thread
for _ in range(20):
try:
thread = http_client_thread(got_ip, (int(got_port)), 20)
thread.start()
threads.append(thread)
except OSError as err:
Utility.console_log('Error: unable to start thread, ' + err)
for t in threads:
t.join()
if __name__ == '__main__':
test_examples_protocol_http_server_simple()
test_examples_protocol_http_server_lru_purge_enable()