329 lines
11 KiB
Python
329 lines
11 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://<ip>:8765/ok
|
||
http://<ip>:8765/fail
|
||
|
||
Расширенная проверка ответа (responseCheck):
|
||
http://<ip>:8765/state?phase=ok
|
||
http://<ip>:8765/poll?pending=3
|
||
http://<ip>:8765/cleaning?result=pending
|
||
http://<ip>:8765/nested?status=ready
|
||
http://<ip>:8765/reset — сброс счётчиков poll
|
||
|
||
Latch/reset (plain-text OK, как у заказчика):
|
||
http://<ip>:8765/api/click/btn?bid=01*04*01 — вызов (trigger)
|
||
http://<ip>:8765/api/click/btn?bid=01*04*00 — сброс (reset)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import threading
|
||
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")
|
||
|
||
|
||
_poll_lock = threading.Lock()
|
||
_poll_counts: dict[str, int] = {}
|
||
_latch_lock = threading.Lock()
|
||
_latch_active: dict[str, bool] = {}
|
||
|
||
|
||
def _latch_key(handler: BaseHTTPRequestHandler, bid: str) -> str:
|
||
"""Group trigger/reset pairs by bid prefix (01*04*01 / 01*04*00 -> 01*04)."""
|
||
parts = bid.split("*")
|
||
if len(parts) >= 2:
|
||
return "*".join(parts[:-1])
|
||
return handler.client_address[0]
|
||
|
||
|
||
def _bid_action(bid: str) -> str:
|
||
parts = bid.split("*")
|
||
if parts and parts[-1] == "00":
|
||
return "reset"
|
||
if parts and parts[-1] == "01":
|
||
return "trigger"
|
||
return "unknown"
|
||
|
||
|
||
def _poll_key(handler: BaseHTTPRequestHandler, qs: dict) -> str:
|
||
custom = (qs.get("key") or [""])[0].strip()
|
||
if custom:
|
||
return custom
|
||
return handler.client_address[0]
|
||
|
||
|
||
def _next_poll_success(key: str, pending_n: int) -> tuple[str, int]:
|
||
with _poll_lock:
|
||
hit = _poll_counts.get(key, 0) + 1
|
||
_poll_counts[key] = hit
|
||
if hit <= pending_n:
|
||
return "pending", hit
|
||
_poll_counts[key] = 0
|
||
return "ok", hit
|
||
|
||
|
||
def _reset_poll(key: str | None = None) -> None:
|
||
with _poll_lock:
|
||
if key is None:
|
||
_poll_counts.clear()
|
||
else:
|
||
_poll_counts.pop(key, None)
|
||
|
||
|
||
class ButtonTestHandler(BaseHTTPRequestHandler):
|
||
server_version = "ButtonTestHTTP/1.1"
|
||
|
||
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.send_header("Access-Control-Allow-Origin", "*")
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def _send_json(self, code: int, payload: dict) -> None:
|
||
self._send(code, json.dumps(payload, ensure_ascii=False), "application/json; charset=utf-8")
|
||
|
||
def _read_body(self) -> bytes:
|
||
length = int(self.headers.get("Content-Length", 0))
|
||
return self.rfile.read(length) if length else b""
|
||
|
||
def _qs(self) -> dict:
|
||
return parse_qs(urlparse(self.path).query)
|
||
|
||
def _handle(self, path: str, qs: dict, method: str, body_text: str = "") -> bool:
|
||
if path in ("", "/"):
|
||
self._send(200, HELP_TEXT)
|
||
return True
|
||
|
||
if path == "/ok":
|
||
self._send(200, "OK\n")
|
||
return True
|
||
|
||
if path == "/fail":
|
||
self._send(500, "server error\n")
|
||
return True
|
||
|
||
if path == "/btn":
|
||
bid = (qs.get("bid") or ["?"])[0]
|
||
self._send_json(200, {"status": "ok", "bid": bid, "ts": time.time()})
|
||
return True
|
||
|
||
if path == "/api/click/btn":
|
||
bid = (qs.get("bid") or ["?"])[0]
|
||
action = _bid_action(bid)
|
||
key = _latch_key(self, bid)
|
||
note = ""
|
||
with _latch_lock:
|
||
if action == "trigger":
|
||
if _latch_active.get(key):
|
||
note = " (already active)"
|
||
else:
|
||
_latch_active[key] = True
|
||
elif action == "reset":
|
||
if _latch_active.get(key):
|
||
_latch_active[key] = False
|
||
note = " (cleared)"
|
||
else:
|
||
note = " (was idle)"
|
||
print(
|
||
f"[{_now()}] /api/click/btn bid={bid} action={action} key={key}{note}",
|
||
flush=True,
|
||
)
|
||
self._send(200, "OK")
|
||
return True
|
||
|
||
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 True
|
||
|
||
if path == "/redirect":
|
||
self.send_response(302)
|
||
self.send_header("Location", "/ok")
|
||
self.end_headers()
|
||
return True
|
||
|
||
if path == "/reset":
|
||
key = (qs.get("key") or [""])[0].strip() or None
|
||
_reset_poll(key)
|
||
msg = f"poll reset: {key or 'all'}"
|
||
self._send_json(200, {"success": "ok", "message": msg})
|
||
print(f"[{_now()}] {msg}", flush=True)
|
||
return True
|
||
|
||
if path == "/state":
|
||
phase = (qs.get("phase") or qs.get("success") or ["ok"])[0]
|
||
code = int((qs.get("code") or ["200"])[0])
|
||
self._send_json(code, {"success": phase, "ts": time.time(), "method": method})
|
||
return True
|
||
|
||
if path == "/poll":
|
||
pending_n = int((qs.get("pending") or ["3"])[0])
|
||
pending_n = max(0, min(pending_n, 50))
|
||
key = _poll_key(self, qs)
|
||
success, hit = _next_poll_success(key, pending_n)
|
||
self._send_json(200, {
|
||
"success": success,
|
||
"hit": hit,
|
||
"pending": pending_n,
|
||
"key": key,
|
||
"ts": time.time(),
|
||
})
|
||
return True
|
||
|
||
if path == "/cleaning":
|
||
result = (qs.get("result") or ["ok"])[0]
|
||
self._send_json(200, {
|
||
"success": result,
|
||
"service": "cleaning",
|
||
"message": {
|
||
"ok": "Вызов принят",
|
||
"pending": "Ожидание клининга",
|
||
"error": "Сервис недоступен",
|
||
}.get(result, result),
|
||
"ts": time.time(),
|
||
})
|
||
return True
|
||
|
||
if path == "/nested":
|
||
status = (qs.get("status") or ["ready"])[0]
|
||
self._send_json(200, {"data": {"status": status}, "ts": time.time()})
|
||
return True
|
||
|
||
if path == "/json":
|
||
# Произвольные поля: ?success=ok&code=200 или ?foo=bar&code=201
|
||
code = int((qs.get("code") or ["200"])[0])
|
||
payload = {k: v[0] for k, v in qs.items() if k != "code"}
|
||
if not payload:
|
||
payload = {"success": "ok"}
|
||
payload["ts"] = time.time()
|
||
self._send_json(code, payload)
|
||
return True
|
||
|
||
if path == "/badjson":
|
||
self._send(200, "{not-json", "application/json; charset=utf-8")
|
||
return True
|
||
|
||
if path == "/httpfail":
|
||
# HTTP 500, но тело с success=ok — для requireHttpSuccess
|
||
self._send_json(500, {"success": "ok", "note": "http 500"})
|
||
return True
|
||
|
||
if path == "/echo":
|
||
ct = self.headers.get("Content-Type", "text/plain")
|
||
if "json" in ct.lower() and body_text.strip():
|
||
try:
|
||
parsed = json.loads(body_text)
|
||
except json.JSONDecodeError:
|
||
parsed = {"raw": body_text}
|
||
self._send_json(200, {
|
||
"success": "ok",
|
||
"echo": parsed,
|
||
"method": method,
|
||
})
|
||
else:
|
||
out = f"method={method}\ncontent-type={ct}\nbody={body_text}\n"
|
||
self._send(200, out)
|
||
return True
|
||
|
||
return False
|
||
|
||
def do_GET(self) -> None:
|
||
path = urlparse(self.path).path
|
||
if not self._handle(path, self._qs(), "GET"):
|
||
self._send(404, f"not found: {path}\n")
|
||
|
||
def do_POST(self) -> None:
|
||
path = urlparse(self.path).path
|
||
body_text = self._read_body().decode("utf-8", errors="replace")
|
||
if not self._handle(path, self._qs(), "POST", body_text):
|
||
self._send(404, f"not found: {path}\n")
|
||
|
||
|
||
HELP_TEXT = """ButtonTask test server — сценарии проверки
|
||
|
||
=== Без responseCheck (только HTTP) ===
|
||
GET /ok -> 200
|
||
GET /fail -> 500
|
||
GET /slow?sec=3 -> задержка
|
||
GET /redirect -> 302 -> /ok
|
||
GET /httpfail -> 500 + JSON {"success":"ok"}
|
||
|
||
=== responseCheck: поле success ===
|
||
GET /state?phase=ok -> {"success":"ok"}
|
||
GET /state?phase=error -> {"success":"error"}
|
||
GET /state?phase=pending -> {"success":"pending"}
|
||
GET /state?phase=ok&code=500 -> HTTP 500 + JSON
|
||
|
||
=== Polling (pending -> ok) ===
|
||
GET /poll?pending=3 -> 3x pending, затем ok
|
||
GET /poll?pending=3&key=btn1 -> отдельный счётчик на кнопку
|
||
GET /reset -> сброс всех счётчиков
|
||
GET /reset?key=btn1 -> сброс одного
|
||
|
||
=== Latch/reset (bodyMatch: plain OK) ===
|
||
GET /api/click/btn?bid=01*04*01 -> trigger, тело "OK"
|
||
GET /api/click/btn?bid=01*04*00 -> reset, тело "OK"
|
||
(последний сегмент bid: 01=вызов, 00=сброс; в консоли видно состояние)
|
||
|
||
=== Прочее ===
|
||
GET /cleaning?result=ok|pending|error
|
||
GET /nested?status=ready -> {"data":{"status":"ready"}}
|
||
GET /json?success=ok&foo=bar
|
||
GET /badjson -> битый JSON при 200
|
||
POST /echo -> эхо (JSON in -> JSON out)
|
||
POST /state?phase=ok
|
||
|
||
Пример responseCheck для /poll?pending=3:
|
||
rules: success equals ok -> ok
|
||
success equals pending -> pending
|
||
success equals error -> error
|
||
poll: intervalMs=2000, maxAttempts=10
|
||
|
||
Все запросы пишутся в консоль.
|
||
"""
|
||
|
||
|
||
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())
|