FastLED 3.9.15
Loading...
Searching...
No Matches
test_server.py
Go to the documentation of this file.
1#!/usr/bin/env python3
2"""
3HTTP Test Server for Asio Client Example
4
5Mimics httpbin.org endpoints for local testing of fl::fetch API.
6
7Usage:
8 uv run python examples/Asio/Client/test_server.py
9"""
10
11import _thread
12import argparse
13import json
14import sys
15from http.server import BaseHTTPRequestHandler, HTTPServer
16from typing import Any
17
18from rich.console import Console
19
20
21console = Console()
22
23
24class HTTPBinHandler(BaseHTTPRequestHandler):
25 """HTTP request handler that mimics httpbin.org API."""
26
27 def log_message(self, format: str, *args: Any) -> None:
28 """Override to use rich console for logging."""
29 console.print(
30 f"[cyan]{self.address_string()}[/cyan] - {format % args}", style="dim"
31 )
32
33 def send_json_response(self, data: dict[str, Any], status_code: int = 200) -> None:
34 """Send JSON response with appropriate headers."""
35 json_data = json.dumps(data, indent=2)
36 self.send_response(status_code)
37 self.send_header("Content-Type", "application/json")
38 self.send_header("Content-Length", str(len(json_data)))
39 self.send_header("Access-Control-Allow-Origin", "*")
40 self.end_headers()
41 self.wfile.write(json_data.encode())
42
43 def send_text_response(self, text: str, status_code: int = 200) -> None:
44 """Send plain text response."""
45 self.send_response(status_code)
46 self.send_header("Content-Type", "text/html; charset=utf-8")
47 self.send_header("Content-Length", str(len(text)))
48 self.end_headers()
49 self.wfile.write(text.encode())
50
51 def do_GET(self) -> None:
52 """Handle GET requests."""
53 if self.path == "/":
54 # Root endpoint - HTML welcome page
55 html = """<!DOCTYPE html>
56<html>
57<head><title>FastLED Test Server</title></head>
58<body>
59<h1>FastLED HTTP Test Server</h1>
60<p>This server mimics httpbin.org endpoints for testing FastLED fetch API.</p>
61<h2>Available Endpoints:</h2>
62<ul>
63<li><a href="/json">/json</a> - Sample JSON slideshow data</li>
64<li><a href="/get">/get</a> - Echo request information</li>
65<li><a href="/ping">/ping</a> - Health check (returns "pong")</li>
66</ul>
67</body>
68</html>"""
69 self.send_text_response(html)
70
71 elif self.path == "/json":
72 # Mimic httpbin.org/json - slideshow data
73 data = {
74 "slideshow": {
75 "author": "FastLED Community",
76 "title": "FastLED Tutorial",
77 "slides": [
78 {
79 "title": "Introduction to FastLED",
80 "type": "tutorial",
81 },
82 {
83 "title": "LED Basics",
84 "type": "lesson",
85 },
86 {
87 "title": "HTTP Fetch API",
88 "type": "demo",
89 },
90 ],
91 }
92 }
93 console.print("[green]→ /json - Returning slideshow data[/green]")
94 self.send_json_response(data)
95
96 elif self.path.startswith("/get"):
97 # Mimic httpbin.org/get - echo request information
98 query_params: dict[str, str] = {}
99 if "?" in self.path:
100 query_string = self.path.split("?", 1)[1]
101 for param in query_string.split("&"):
102 if "=" in param:
103 key, value = param.split("=", 1)
104 query_params[key] = value
105
106 headers: dict[str, str] = {}
107 for header_name, header_value in self.headers.items():
108 headers[header_name] = header_value
109
110 data: dict[str, Any] = {
111 "args": query_params,
112 "headers": headers,
113 "origin": self.client_address[0],
114 "url": f"http://{self.headers.get('Host', 'localhost')}{self.path}",
115 }
116 console.print("[green]→ /get - Returning request info[/green]")
117 self.send_json_response(data)
118
119 elif self.path == "/ping":
120 # Health check endpoint
121 console.print("[green]→ /ping - Health check[/green]")
122 self.send_text_response("pong\n")
123
124 else:
125 # 404 Not Found
126 console.print(f"[red]→ {self.path} - Not Found[/red]")
127 self.send_json_response(
128 {"error": "Not Found", "path": self.path}, status_code=404
129 )
130
131 def do_POST(self) -> None:
132 """Handle POST requests."""
133 content_length = int(self.headers.get("Content-Length", 0))
134 post_data = self.rfile.read(content_length).decode("utf-8")
135
136 if self.path == "/post":
137 # Mimic httpbin.org/post - echo POST data
138 headers: dict[str, str] = {}
139 for header_name, header_value in self.headers.items():
140 headers[header_name] = header_value
141
142 json_data: Any = None
143 try:
144 if post_data:
145 json_data = json.loads(post_data)
146 else:
147 json_data = {}
148 except json.JSONDecodeError:
149 json_data = None
150
151 data: dict[str, Any] = {
152 "args": {},
153 "data": post_data,
154 "json": json_data,
155 "headers": headers,
156 "origin": self.client_address[0],
157 "url": f"http://{self.headers.get('Host', 'localhost')}{self.path}",
158 }
159 console.print("[green]→ POST /post - Returning echo data[/green]")
160 self.send_json_response(data)
161 else:
162 console.print(f"[red]→ POST {self.path} - Not Found[/red]")
163 self.send_json_response(
164 {"error": "Not Found", "path": self.path}, status_code=404
165 )
166
167
168def main() -> int:
169 """Main entry point."""
170 parser = argparse.ArgumentParser(description="HTTP test server for Client.ino")
171 parser.add_argument("--host", default="localhost", help="Server host")
172 parser.add_argument("--port", type=int, default=8081, help="Server port")
173 args = parser.parse_args()
174
175 server_address = (args.host, args.port)
176 httpd = HTTPServer(server_address, HTTPBinHandler)
177
178 console.print("\n[bold green]FastLED HTTP Test Server[/bold green]")
179 console.print(f"Listening on [cyan]http://{args.host}:{args.port}/[/cyan]\n")
180 console.print("[dim]Available endpoints:[/dim]")
181 console.print(" [cyan]GET /[/cyan] - Welcome page")
182 console.print(" [cyan]GET /json[/cyan] - Sample JSON slideshow data")
183 console.print(" [cyan]GET /get[/cyan] - Echo request information")
184 console.print(" [cyan]GET /ping[/cyan] - Health check")
185 console.print(" [cyan]POST /post[/cyan] - Echo POST data")
186 console.print("\n[yellow]Press Ctrl+C to stop[/yellow]\n")
187
188 httpd.serve_forever()
189 return 0
190
191
192if __name__ == "__main__":
193 try:
194 sys.exit(main())
195 except KeyboardInterrupt:
196 console.print("\n[yellow]Server stopped[/yellow]")
197 _thread.interrupt_main()
198 sys.exit(130)
test_server.HTTPBinHandler.do_POST
None do_POST(self)
Definition test_server.py:131
test_server.HTTPBinHandler.path
str path
Definition test_server.py:53
test_server.HTTPBinHandler.send_json_response
None send_json_response(self, dict[str, Any] data, int status_code=200)
Definition test_server.py:33
test_server.HTTPBinHandler.send_text_response
None send_text_response(self, str text, int status_code=200)
Definition test_server.py:43
test_server.HTTPBinHandler.log_message
None log_message(self, str format, *Any args)
Definition test_server.py:27
test_server.HTTPBinHandler.do_GET
None do_GET(self)
Definition test_server.py:51
test_server.HTTPBinHandler
Definition test_server.py:24
test_server.main
int main()
Definition test_server.py:168