videoDownloadTGbot/smoke_test.py

#!/usr/bin/env python3
"""Смоук-тест загрузчиков: гоняет реальные ссылки через запущенные сервисы.

Запускать после КАЖДОЙ правки, до коммита:

    python3 smoke_test.py

Ссылки подобраны не случайно — это реальные регрессии, которые уже ломались:
  * возрастное YouTube-видео ловит отсутствие рабочего JS-рантайма (n-challenge);
  * Instagram-ролик ловит устаревший экстрактор yt-dlp;
  * обычное видео проверяет, что базовый путь скачивания цел.

Зависимостей нет — только стандартная библиотека, чтобы запускалось на любом хосте.
Код возврата: 0 — все проверки прошли, 1 — есть падения.
"""
import json
import os
import sys
import time
import urllib.error
import urllib.request

YOUTUBE_URL = os.getenv('YOUTUBE_DOWNLOADER_URL', 'http://localhost:5557')
INSTAGRAM_URL = os.getenv('INSTAGRAM_DOWNLOADER_URL', 'http://localhost:5556')

YT_SMALL = 'https://www.youtube.com/watch?v=jNQXAC9IVRw'  # «Me at the zoo», 19 секунд
YT_AGE_RESTRICTED = 'https://youtu.be/UMyoCr2MnpM'        # требует cookies + решения n-challenge
IG_REEL = 'https://www.instagram.com/reel/Dak0fDTMv_i/'   # обычный публичный reel

MIN_VIDEO_BYTES = 100 * 1024  # меньше — значит пришло не видео, а заглушка/ошибка


def _post(url, payload=None, timeout=120):
    """POST с JSON-телом. Возвращает (http_code, body_bytes)."""
    data = json.dumps(payload or {}).encode()
    req = urllib.request.Request(
        url, data=data, headers={'Content-Type': 'application/json'}, method='POST'
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.status, resp.read()
    except urllib.error.HTTPError as e:
        return e.code, e.read()
    except Exception as e:
        return None, str(e).encode()


def check_cookies(service, base_url):
    """/cookies/check должен вернуть status=ok."""
    code, body = _post(f'{base_url}/cookies/check', timeout=180)
    if code != 200:
        return False, f'HTTP {code}'
    try:
        data = json.loads(body)
    except ValueError:
        return False, 'ответ не JSON'
    status = data.get('status', '?')
    if status == 'ok':
        return True, 'status=ok'
    return False, f"status={status}: {(data.get('detail') or '')[:120]}"


def check_formats(url):
    """/formats у YouTube — лёгкая проверка: cookies + JS-рантайм без скачивания."""
    code, body = _post(f'{YOUTUBE_URL}/formats', {'url': url}, timeout=180)
    if code != 200:
        detail = body.decode('utf-8', 'replace')[:150]
        return False, f'HTTP {code}: {detail}'
    try:
        formats = json.loads(body).get('formats', [])
    except ValueError:
        return False, 'ответ не JSON'
    if not formats:
        return False, 'форматы не найдены'
    return True, f'форматов: {len(formats)}'


def check_download(base_url, url):
    """/download/stream — полный путь до реального файла."""
    code, body = _post(f'{base_url}/download/stream', {'url': url}, timeout=600)
    if code != 200:
        detail = body.decode('utf-8', 'replace')[:150]
        return False, f'HTTP {code}: {detail}'
    size = len(body)
    if size < MIN_VIDEO_BYTES:
        return False, f'подозрительно мало: {size} байт'
    return True, f'{size / 1024:.0f} КБ'


CHECKS = [
    ('cookies YouTube',        lambda: check_cookies('youtube', YOUTUBE_URL)),
    ('cookies Instagram',      lambda: check_cookies('instagram', INSTAGRAM_URL)),
    ('YouTube возрастное',     lambda: check_formats(YT_AGE_RESTRICTED)),
    ('YouTube скачивание',     lambda: check_download(YOUTUBE_URL, YT_SMALL)),
    ('Instagram скачивание',   lambda: check_download(INSTAGRAM_URL, IG_REEL)),
]


def main():
    print(f'Смоук-тест: {YOUTUBE_URL} / {INSTAGRAM_URL}\n')

    results = []
    for name, fn in CHECKS:
        started = time.monotonic()
        try:
            ok, detail = fn()
        except Exception as e:  # проверка не должна ронять весь прогон
            ok, detail = False, f'исключение: {e}'
        results.append((name, ok, detail, time.monotonic() - started))

    width = max(len(name) for name, *_ in results)
    print(f'| {"Проверка".ljust(width)} | Итог | Время | Детали')
    print(f'|{"-" * (width + 2)}|------|-------|--------')
    for name, ok, detail, elapsed in results:
        mark = ' OK ' if ok else 'FAIL'
        print(f'| {name.ljust(width)} | {mark} | {elapsed:5.1f}с | {detail}')

    failed = [name for name, ok, *_ in results if not ok]
    print()
    if failed:
        print(f'ИТОГ: провалено {len(failed)} из {len(results)} — {", ".join(failed)}')
        return 1
    print(f'ИТОГ: все проверки прошли ({len(results)} из {len(results)})')
    return 0


if __name__ == '__main__':
    sys.exit(main())