#!/usr/bin/env python3 """ Mock Bridge TPE — Serveur HTTP minimal pour simuler le bridge TPE. Utilisé pour les tests de l'extension Chrome sans TPE physique. Endpoints : GET /health → statut du serveur POST /payment → simule un paiement TPE (mode configurable) POST /cancel → simule une annulation TPE Configuration : 1. Variable d'environnement MOCK_MODE (prioritaire) 2. Fichier test/config.json → clé "mode" Modes supportés : "success", "refused", "timeout" Usage : cd test/ MOCK_MODE=success python mock-bridge.py # mode via env python mock-bridge.py # mode via config.json (défaut: success) python mock-bridge.py --port 9878 # port personnalisé """ import http.server import json import os import sys import time from datetime import datetime, timezone # ─── Configuration ──────────────────────────────────────────────────────────── def load_config(): """Charge la configuration : env var prioritaire, sinon config.json, sinon défaut 'success'.""" # 1. Variable d'environnement mode = os.environ.get("MOCK_MODE") if mode: return {"mode": mode, "port": int(os.environ.get("MOCK_PORT", 9877))} # 2. Fichier config.json (cherché dans le même dossier que ce script) config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json") if os.path.exists(config_path): try: with open(config_path, "r", encoding="utf-8") as f: cfg = json.load(f) return { "mode": cfg.get("mode", "success"), "port": int(cfg.get("port", 9877)), } except (json.JSONDecodeError, IOError) as e: print(f"[WARN] Erreur lecture config.json: {e}, fallback sur défauts") # 3. Défauts return {"mode": "success", "port": 9877} # ─── Logging ────────────────────────────────────────────────────────────────── def log_request(method, path, body=None, extra=""): """Log formaté d'une requête entrante.""" ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") line = f"[{ts}] {method} {path}" if body: line += f" | body={body[:200]}" # tronquer les corps longs if extra: line += f" | {extra}" print(line) # ─── Handler HTTP ───────────────────────────────────────────────────────────── class MockTPEHandler(http.server.BaseHTTPRequestHandler): """Handler HTTP pour le mock bridge TPE.""" # Rendu silencieux pour les logs HTTP par défaut de http.server # On log nous-mêmes dans chaque méthode. def _send_cors(self): """Envoie les headers CORS.""" self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization") def _send_json(self, status_code, data): """Envoie une réponse JSON avec CORS.""" self.send_response(status_code) self._send_cors() self.send_header("Content-Type", "application/json; charset=utf-8") body = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_header("Content-Length", len(body)) self.end_headers() self.wfile.write(body) def _read_body(self): """Lit le corps de la requête.""" content_length = int(self.headers.get("Content-Length", 0)) if content_length > 0: return self.rfile.read(content_length).decode("utf-8", errors="replace") return "" # ── OPTIONS (preflight CORS) ────────────────────────────────────────── def do_OPTIONS(self): """Répond aux requêtes OPTIONS (preflight CORS).""" log_request("OPTIONS", self.path) self.send_response(204) self._send_cors() self.end_headers() # ── GET /health ─────────────────────────────────────────────────────── def do_GET(self): """Route GET /health et /mode.""" if self.path == "/health": log_request("GET", self.path, extra="→ 200 OK") self._send_json(200, {"status": "ok"}) elif self.path == "/mode": log_request("GET", self.path, extra=f"→ mode={self.server.config['mode']}") self._send_json(200, {"mode": self.server.config["mode"]}) else: log_request("GET", self.path, extra="→ 404") self._send_json(404, {"error": "Not found"}) # ── POST /payment, /cancel, /mode ───────────────────────────────────── def do_POST(self): """Route les POST /payment, /cancel et /mode.""" body = self._read_body() if self.path == "/payment": self._handle_payment(body) elif self.path == "/cancel": self._handle_cancel(body) elif self.path == "/mode": self._handle_mode(body) else: log_request("POST", self.path, body, extra="→ 404") self._send_json(404, {"error": "Not found"}) # ── Logique /mode ───────────────────────────────────────────────────── def _handle_mode(self, body): """Change le mode du mock à chaud.""" try: data = json.loads(body) if body else {} except json.JSONDecodeError: log_request("POST", "/mode", body, extra="→ 400 JSON invalide") self._send_json(400, {"error": "JSON invalide"}) return new_mode = data.get("mode", "") if new_mode not in ("success", "refused", "timeout"): log_request("POST", "/mode", body, extra=f"→ 400 mode invalide: {new_mode}") self._send_json(400, { "error": f"Mode '{new_mode}' invalide. Modes: success, refused, timeout" }) return old = self.server.config["mode"] self.server.config["mode"] = new_mode log_request("POST", "/mode", body, extra=f"→ {old} → {new_mode}") self._send_json(200, {"mode": new_mode, "previous": old}) # ── Logique /payment ────────────────────────────────────────────────── def _handle_payment(self, body): """Traite une requete de paiement en mode interactif. Affiche un menu dans le terminal pour choisir la reponse.""" amount = "?" try: data = json.loads(body) if body else {} amount = data.get("amount", "?") except: pass log_request("POST", "/payment", body, extra=f"attente choix...") print(f"\n{'='*50}") print(f" NOUVELLE DEMANDE DE PAIEMENT : {amount} EUR") print(f" [s] Success [r] Refuse [t] Timeout [q] Quitter mock") print(f"{'='*50}") choice = "" while choice not in ("s", "r", "t", "q"): try: choice = input(" > ").strip().lower() except (EOFError, KeyboardInterrupt): choice = "q" if choice == "q": print(" Fermeture du mock...") self._send_json(200, {"success": False, "error": "Mock ferme"}) # Planter le serveur import os os._exit(0) if choice == "s": print(" -> SUCCES simule") self._send_json(200, { "success": True, "card_masked": "4970XXXX1234", "auth_code": "AUTO123", }) elif choice == "r": print(" -> REFUS simule") self._send_json(200, { "success": False, "error": "Carte refusee par la banque", }) elif choice == "t": print(" -> TIMEOUT simule (130s)...") sys.stdout.flush() time.sleep(130) self._send_json(504, { "success": False, "error": "Timeout TPE", }) # ── Logique /cancel ─────────────────────────────────────────────────── def _handle_cancel(self, body): """Traite une requête d'annulation.""" log_request("POST", "/cancel", body, extra="→ annulation OK") self._send_json(200, {"success": True}) # ── Suppression des logs par défaut ─────────────────────────────────── def log_message(self, format, *args): """Override : on log nous-mêmes, donc on supprime le log par défaut.""" pass # ─── Serveur ────────────────────────────────────────────────────────────────── def parse_args(): """Parse les arguments de ligne de commande rudimentaires.""" port = None args = sys.argv[1:] i = 0 while i < len(args): if args[i] == "--port" and i + 1 < len(args): port = int(args[i + 1]) i += 2 elif args[i] in ("-h", "--help"): print(__doc__) sys.exit(0) else: i += 1 return port def main(): """Point d'entrée principal.""" config = load_config() # Override port par ligne de commande cli_port = parse_args() if cli_port: config["port"] = cli_port port = config["port"] mode = config["mode"] print("=" * 55) print(" 🧪 Mock Bridge TPE — Serveur de test") print("=" * 55) print(f" Port : {port}") print(f" Mode : {mode}") print(f" URL : http://localhost:{port}") print("=" * 55) print(" Endpoints :") print(f" GET http://localhost:{port}/health") print(f" POST http://localhost:{port}/payment") print(f" POST http://localhost:{port}/cancel") print("=" * 55) print(" Ctrl+C pour arrêter\n") server = http.server.HTTPServer(("0.0.0.0", port), MockTPEHandler) server.config = config # attacher la config au serveur try: server.serve_forever() except KeyboardInterrupt: print("\n👋 Arrêt du mock bridge TPE.") server.server_close() if __name__ == "__main__": main()