From b276794c0adc0756de16aea5a65b951efa09d4cd Mon Sep 17 00:00:00 2001 From: Yuriy Yuriev Date: Fri, 22 May 2026 19:01:44 +0700 Subject: [PATCH] add test --- managers/proxy_manager.py | 4 +- test_webhook.py | 182 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 185 insertions(+), 1 deletion(-) create mode 100644 test_webhook.py diff --git a/managers/proxy_manager.py b/managers/proxy_manager.py index 12bb9e5..27be1ef 100644 --- a/managers/proxy_manager.py +++ b/managers/proxy_manager.py @@ -221,8 +221,9 @@ class ProxyManager: if not explicit_type: self._needs_detection.add(proxy.id) + random.shuffle(self._proxies) logger.info(f"Loaded {len(loaded)} proxies from {self.proxy_file}") - + # Статистика по типам type_counts = defaultdict(int) for p in loaded: @@ -374,6 +375,7 @@ class ProxyManager: if all_used: logger.info("All proxies used once, resetting used list") self._used_ids.clear() + random.shuffle(self._proxies) logger.debug(f"No proxies available (attempt {attempt})") diff --git a/test_webhook.py b/test_webhook.py new file mode 100644 index 0000000..6881e81 --- /dev/null +++ b/test_webhook.py @@ -0,0 +1,182 @@ +""" +Тест webhook-сервера Heleket. + +Использование: + # Отправить тестовый webhook локально (симуляция от Heleket) + python test_webhook.py --local + + # Попросить Heleket повторно отправить webhook по uuid платежа + python test_webhook.py --resend --uuid + + # Попросить Heleket повторно отправить по order_id + python test_webhook.py --resend --order-id +""" + +import asyncio +import sys +import argparse +import base64 +import hashlib +import json +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) + +import aiohttp + +from config.settings import settings + + +def _make_sign_request(body_bytes: bytes, api_key: str) -> str: + """Подпись исходящего запроса к Heleket API.""" + b64 = base64.b64encode(body_bytes).decode("utf-8") + return hashlib.md5((b64 + api_key).encode("utf-8")).hexdigest() + + +def _make_webhook_sign(payload: dict, api_key: str) -> str: + """Подпись входящего webhook (как Heleket её считает).""" + data = {k: v for k, v in payload.items() if k != "sign"} + data_json = json.dumps(data, separators=(",", ":"), ensure_ascii=False) + data_json = data_json.replace("/", "\\/") # PHP json_encode escapes / + b64 = base64.b64encode(data_json.encode("utf-8")).decode("utf-8") + return hashlib.md5((b64 + api_key).encode("utf-8")).hexdigest() + + +async def test_local(port: int, api_key: str): + """Отправляет тестовый webhook на локальный сервер с правильной подписью.""" + url = f"http://127.0.0.1:{port}/heleket/webhook" + + payload = { + "type": "payment", + "uuid": "test-uuid-0000-0000-0000-000000000001", + "order_id": "test_order_123", + "amount": "100.00", + "payment_amount": "100.00", + "payment_amount_usd": "1.10", + "merchant_amount": "98.00", + "commission": "2.00", + "is_final": True, + "status": "paid", + "currency": "RUB", + "payer_currency": "USDT", + "network": "tron", + "from": "TTestWalletAddress123", + "txid": "test_txid_abc123", + } + payload["sign"] = _make_webhook_sign(payload, api_key) + + print(f"\n→ POST {url}") + print(f" payload: {json.dumps(payload, ensure_ascii=False, indent=2)}") + + try: + async with aiohttp.ClientSession() as session: + async with session.post( + url, + json=payload, + timeout=aiohttp.ClientTimeout(total=10), + ) as resp: + body = await resp.text() + print(f"\n← HTTP {resp.status}: {body}") + if resp.status == 200: + print("✅ Webhook принят успешно") + elif resp.status == 400: + print("❌ Ошибка подписи или неверный формат") + elif resp.status == 403: + print("❌ Заблокирован по IP (HELEKET_WHITELIST_IP=true)") + print(" Для локального теста отключи: HELEKET_WHITELIST_IP=false") + except aiohttp.ClientConnectorError: + print(f"❌ Не могу подключиться к {url}") + print(" Убедись что бот запущен и webhook-сервер работает") + except Exception as e: + print(f"❌ Ошибка: {e}") + + +async def resend_webhook(uuid: str = None, order_id: str = None): + """Просит Heleket повторно отправить webhook на наш сервер.""" + merchant_uuid = settings.HELEKET_MERCHANT_UUID + api_key = settings.HELEKET_API_KEY + + if not merchant_uuid or not api_key: + print("❌ HELEKET_MERCHANT_UUID или HELEKET_API_KEY не заданы в .env") + return + + if not uuid and not order_id: + print("❌ Укажи --uuid или --order-id") + return + + payload = {} + if uuid: + payload["uuid"] = uuid + if order_id: + payload["order_id"] = order_id + + body = json.dumps(payload, separators=(",", ":")).encode("utf-8") + sign = _make_sign_request(body, api_key) + + headers = { + "merchant": merchant_uuid, + "sign": sign, + "Content-Type": "application/json", + } + + print(f"\n→ POST https://api.heleket.com/v1/payment/resend") + print(f" payload: {payload}") + + try: + async with aiohttp.ClientSession() as session: + async with session.post( + "https://api.heleket.com/v1/payment/resend", + data=body, + headers=headers, + timeout=aiohttp.ClientTimeout(total=15), + ) as resp: + data = await resp.json() + print(f"\n← HTTP {resp.status}: {json.dumps(data, ensure_ascii=False, indent=2)}") + + if data.get("state") == 0: + print("✅ Heleket повторно отправит webhook на твой сервер") + print(f" URL: {settings.HELEKET_WEBHOOK_URL}") + else: + msg = data.get("message", "unknown error") + print(f"❌ Ошибка: {msg}") + if "Notification not found" in str(msg): + print(" Платёж создан без url_callback — webhook не настроен") + elif "Too much resend" in str(msg): + print(" Превышен лимит 10 повторных отправок") + except Exception as e: + print(f"❌ Ошибка запроса: {e}") + + +def main(): + parser = argparse.ArgumentParser(description="Тест webhook Heleket") + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument("--local", action="store_true", + help="Отправить тестовый webhook на локальный сервер") + group.add_argument("--resend", action="store_true", + help="Попросить Heleket повторно отправить webhook") + + parser.add_argument("--uuid", help="UUID платежа (для --resend)") + parser.add_argument("--order-id", help="Order ID платежа (для --resend)") + parser.add_argument("--port", type=int, default=settings.HELEKET_WEBHOOK_PORT, + help=f"Порт webhook-сервера (по умолчанию: {settings.HELEKET_WEBHOOK_PORT})") + + args = parser.parse_args() + + api_key = settings.HELEKET_API_KEY + if not api_key: + print("❌ HELEKET_API_KEY не задан в .env") + sys.exit(1) + + if args.local: + print("🧪 Тест локального webhook-сервера") + if settings.HELEKET_WHITELIST_IP: + print("⚠️ HELEKET_WHITELIST_IP=true — локальный тест будет заблокирован по IP") + print(" Временно поставь HELEKET_WHITELIST_IP=false для теста") + asyncio.run(test_local(args.port, api_key)) + else: + print("📤 Запрос повторной отправки webhook через Heleket API") + asyncio.run(resend_webhook(args.uuid, args.order_id)) + + +if __name__ == "__main__": + main()