142 lines
4.5 KiB
Python
142 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Мини-сервер на ПК для проверки HTTP-кнопок ButtonTask.
|
|
|
|
Запуск:
|
|
python tools/button-test-server.py
|
|
python tools/button-test-server.py --host 0.0.0.0 --port 8765
|
|
|
|
Примеры URL для кнопок (подставьте IP вашего ПК):
|
|
http://192.168.1.10:8765/ok
|
|
http://192.168.1.10:8765/fail
|
|
http://192.168.1.10:8765/btn?bid=01
|
|
http://192.168.1.10:8765/echo (тип: http_post, body: ping)
|
|
http://192.168.1.10:8765/slow?sec=8 (таймаут кнопки по умолчанию 7 с)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now().strftime("%H:%M:%S")
|
|
|
|
|
|
class ButtonTestHandler(BaseHTTPRequestHandler):
|
|
server_version = "ButtonTestHTTP/1.0"
|
|
|
|
def log_message(self, fmt: str, *args) -> None:
|
|
print(f"[{_now()}] {self.address_string()} {fmt % args}", flush=True)
|
|
|
|
def _send(self, code: int, body: str, content_type: str = "text/plain; charset=utf-8") -> None:
|
|
data = body.encode("utf-8")
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def _read_body(self) -> bytes:
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
return self.rfile.read(length) if length else b""
|
|
|
|
def do_GET(self) -> None:
|
|
path = urlparse(self.path).path
|
|
qs = parse_qs(urlparse(self.path).query)
|
|
|
|
if path in ("", "/"):
|
|
self._send(200, HELP_TEXT)
|
|
return
|
|
|
|
if path == "/ok":
|
|
self._send(200, "OK\n")
|
|
return
|
|
|
|
if path == "/fail":
|
|
self._send(500, "server error\n")
|
|
return
|
|
|
|
if path == "/btn":
|
|
bid = (qs.get("bid") or ["?"])[0]
|
|
payload = {"status": "ok", "bid": bid, "ts": time.time()}
|
|
self._send(200, json.dumps(payload, ensure_ascii=False), "application/json")
|
|
return
|
|
|
|
if path == "/slow":
|
|
sec = float((qs.get("sec") or ["3"])[0])
|
|
sec = max(0.0, min(sec, 60.0))
|
|
time.sleep(sec)
|
|
self._send(200, f"slept {sec:g}s\n")
|
|
return
|
|
|
|
if path == "/redirect":
|
|
self.send_response(302)
|
|
self.send_header("Location", "/ok")
|
|
self.end_headers()
|
|
return
|
|
|
|
self._send(404, f"not found: {path}\n")
|
|
|
|
def do_POST(self) -> None:
|
|
path = urlparse(self.path).path
|
|
raw = self._read_body()
|
|
text = raw.decode("utf-8", errors="replace")
|
|
|
|
if path == "/echo":
|
|
ct = self.headers.get("Content-Type", "text/plain")
|
|
out = (
|
|
f"method=POST\n"
|
|
f"content-type={ct}\n"
|
|
f"body={text}\n"
|
|
)
|
|
self._send(200, out)
|
|
return
|
|
|
|
if path == "/fail":
|
|
self._send(500, "post failed\n")
|
|
return
|
|
|
|
self._send(404, f"not found: {path}\n")
|
|
|
|
|
|
HELP_TEXT = """ButtonTask test server
|
|
|
|
GET /ok -> 200 (зелёная обводка)
|
|
GET /fail -> 500 (красная обводка)
|
|
GET /btn?bid=01 -> 200 JSON (как в примере config.json)
|
|
GET /slow?sec=3 -> задержка, проверка таймаута
|
|
GET /redirect -> 302 -> /ok
|
|
POST /echo -> эхо тела запроса
|
|
POST /fail -> 500
|
|
|
|
Все запросы пишутся в консоль.
|
|
"""
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Тестовый HTTP-сервер для кнопок ButtonTask")
|
|
parser.add_argument("--host", default="0.0.0.0", help="адрес (0.0.0.0 — доступ с сети)")
|
|
parser.add_argument("--port", type=int, default=8765, help="порт (по умолчанию 8765)")
|
|
args = parser.parse_args()
|
|
|
|
httpd = ThreadingHTTPServer((args.host, args.port), ButtonTestHandler)
|
|
print(f"ButtonTask test server: http://{args.host}:{args.port}/", flush=True)
|
|
print("Остановка: Ctrl+C", flush=True)
|
|
print(HELP_TEXT, flush=True)
|
|
try:
|
|
httpd.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\nstop", flush=True)
|
|
finally:
|
|
httpd.server_close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|