-
Notifications
You must be signed in to change notification settings - Fork 70
Expand file tree
/
Copy pathtest_webserver.py
More file actions
105 lines (83 loc) · 3.43 KB
/
test_webserver.py
File metadata and controls
105 lines (83 loc) · 3.43 KB
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
"""
Unit tests for the MicroPythonOS webserver.
"""
import _thread
import sys
import time
import unittest
import deflate
import io
sys.path.insert(0, "../internal_filesystem/lib")
from mpos import TaskManager
from mpos.net.download_manager import DownloadManager
from mpos.webserver.webserver import WebServer
class TestWebServer(unittest.TestCase):
"""Test cases for WebServer."""
def tearDown(self):
"""Ensure the webserver is stopped after tests."""
if WebServer.is_started():
WebServer.stop()
TaskManager.stop()
def test_webserver_serves_webrepl_page(self):
"""Webserver should serve the WebREPL HTML page on root."""
def start_task_manager():
try:
TaskManager.enable()
TaskManager.start()
except KeyboardInterrupt:
print("TaskManager got KeyboardInterrupt, falling back to REPL shell...")
except Exception as exc:
print(f"TaskManager got exception: {exc}")
TaskManager.enable()
_thread.stack_size(TaskManager.good_stack_size())
_thread.start_new_thread(start_task_manager, ())
startup_timeout = 5.0
start_time = time.time()
while TaskManager.keep_running is not True and (time.time() - start_time) < startup_timeout:
time.sleep(0.05)
if TaskManager.keep_running is not True:
self.fail("TaskManager failed to start")
started = WebServer.start()
if not started:
self.fail("WebServer failed to start")
startup_wait = 3.0
startup_wait_start = time.time()
while (time.time() - startup_wait_start) < startup_wait:
if WebServer.is_started():
break
time.sleep(0.05)
response_state = {"data": None, "error": None, "done": False}
async def download_task():
response_bytes = None
last_error = None
url_attempts = ["http://localhost:7890/", "http://127.0.0.1:7890/"]
for url in url_attempts:
for _ in range(20):
try:
response_bytes = await DownloadManager.download_url(url)
break
except Exception as exc:
last_error = exc
await TaskManager.sleep(0.5)
if response_bytes is not None:
break
if response_bytes is None:
response_state["error"] = last_error or RuntimeError(
"WebServer did not respond before timeout"
)
else:
response_state["data"] = response_bytes
response_state["done"] = True
TaskManager.create_task(download_task())
timeout_seconds = 30.0
start_wait = time.time()
while not response_state["done"] and (time.time() - start_wait) < timeout_seconds:
time.sleep(0.1)
if response_state["data"] is None:
error = response_state["error"]
self.fail(f"WebServer response unavailable: {error}")
with deflate.DeflateIO(io.BytesIO(response_state["data"]), deflate.GZIP, 15) as gzip_stream:
response_text = gzip_stream.read().decode("utf-8", "replace")
self.assertIn("<title>MicroPythonOS WebREPL</title>", response_text)
WebServer.stop()
self.assertFalse(WebServer.is_started())