e2e: Скилл для написания и запуска тестов rtp2httpd

--- name: e2e description: > Пишите, запускайте и отлаживайте сквозные тесты для rtp2httpd. ВСЕГДА используйте эту функцию, когда пользователь: (1) хочет написать новые сквозные/интеграционные тесты или добавить тестовые случаи в существующие файлы тестов, (2) запрашивает запуск тестов (например, "跑测试", "run tests", "run pytest", упоминает run-e2e.sh или uv), (3) нуждается в отладке неудачных или зависших тестов (таймауты, ошибки утверждений, ошибки импорта), (4) упоминает ЛЮБЫЙ файл в каталоге e2e/ (test_*.py, conftest.py, helpers/*, run-e2e.sh), (5) упоминает серверы-заглушки (MockRTSP*, MockHTTP*, MockFCC*, MockSTUN*), R2HProcess или тестовые фикстуры, (6) спрашивает об инфраструктуре тестирования (маркеры, фикстуры, область видимости, настройка многоадресной рассылки, выделение портов), (7) упоминает "端到端测试" (сквозное тестирование), "e2e test" (сквозной тест) или "integration test" (интеграционный тест) в контексте rtp2httpd. Эта функция содержит полное справочное руководство по API-помощникам, шаблонам тестирования и соглашениям. Без нее агент должен прочитать множество файлов, чтобы быстро узнать, что эта функция предоставляет. argument-hint: "[run|write|debug] [необязательный файл теста или ключевое слово]" --- # Сквозное тестирование rtp2httpd rtp2httpd — это C-демон, который проксирует потоки RTP многоадресной рассылки, RTSP и HTTP для HTTP-клиентов. Сквозные тесты основаны на Python (pytest) и находятся в каталоге `e2e/`. Они запускают реарный бинарный файл против серверов-заглушек и проверяют поведение в сети. ## Запуск тестов Все команды выполняются из корневого каталога проекта. Оболочка `run-e2e.sh` управляет вызовами uv/pytest: ```bash # Все тесты (параллельно, рекомендуется) ./scripts/run-e2e.sh # Последовательно (полезно для отладки) ./scripts/run-e2e.sh -p 1 # Один файл теста ./scripts/run-e2e.sh test_m3u.py # Фильтрация по ключевому слову ./scripts/run-e2e.sh -k "etag" # Фильтрация по маркеру ./scripts/run-e2e.sh -m "not multicast" # Остановка при первой неудаче ./scripts/run-e2e.sh -x # Сухой запуск (список тестов без запуска) ./scripts/run-e2e.sh --co ``` **Предварительное условие** — бинарный файл должен быть собран: ```bash cmake -B build -DCMAKE_BUILD_TYPE=Release -DENABLE_AGGRESSIVE_OPT=ON && cmake --build build -j$(getconf _NPROCESSORS_ONLN) ``` Зависимости Python управляются через uv (`pyproject.toml` в корневом каталоге проекта). ## Структура проекта ```text e2e/ ├── conftest.py # Общие фикстуры и маркеры ├── test_m3u.py # Тесты плейлистов M3U ├── test_multicast.py # Многоадресная потоковая передача RTP ├── test_rtsp_*.py # Прокси RTSP (транспорт, поиск, stun, прочее, content_base) ├── test_http_proxy*.py # HTTP-прокси (базовый, поиск, m3u_rewrite) ├── test_fcc.py # Быстрая смена каналов ├── test_config.py # Разбор конфигурации ├── test_auth.py # Аутентификация ├── test_error.py # Обработка ошибок ├── test_epg.py / test_pages.py / test_zerocopy.py └── helpers/ ├── __init__.py # Переэкспортирует все помощники (необходимо обновлять при добавлении новых) ├── constants.py # BINARY_PATH, LOOPBACK_IF, MCAST_ADDR, FIXTURES_DIR ├── ports.py # find_free_port(), wait_for_port() ├── http.py # http_get(), http_request(), stream_get() ├── rtp.py # make_rtp_packet(), MulticastSender ├── r2h_process.py # Оболочка R2HProcess ├── mock_rtsp.py # Варианты MockRTSPServer ├── mock_http.py # Варианты MockHTTPUpstream ├── mock_fcc.py # MockFCCServer └── mock_stun.py # MockSTUNServer ``` ## Написание тестов Перед написанием новых тестов прочитайте соответствующий существующий файл теста и помощники, чтобы соблюдать соглашения. ### Импорты Всегда импортируйте из `helpers` — он переэкспортирует все: ```python from helpers import ( R2HProcess, find_free_port, find_free_udp_port, http_get, http_request, stream_get, # добавьте другие по мере необходимости ) ``` ### Краткое описание API-помощников **Выделение портов** — никогда не используйте жестко закодированные порты: - `find_free_port()` — свободный TCP-порт - `find_free_udp_port()` — свободный UDP-порт - `find_free_udp_port_pair()` — четная/нечетная пара UDP для RTP/RTCP - `wait_for_port(port, host="127.0.0.1", timeout=5.0)` — блокирует, пока TCP-порт не начнет принимать соединения **HTTP-клиенты** — все возвращают `(status_code, headers_dict, body_bytes)`: - `http_get(host, port, path, timeout=5.0, headers=None)` - `http_request(host, port, method, path, timeout=5.0, headers=None, body=None)` - `stream_get(host, port, path, read_bytes=8192, timeout=10.0, headers=None)` — для потоковых ответов; читает до N байт, затем возвращает **Управление процессами**: - `R2HProcess(binary, port, extra_args=[], config_content=None)` - С `config_content`: записывает временную конфигурацию, передает `-c <path>` - Без конфигурации: передает `-C` (режим без конфигурации), используйте `extra_args` для параметров командной строки - `.start()` ждет, пока порт не начнет принимать соединения (таймаут 6 секунд) - `.stop()` завершает и очищает временную конфигурацию **Серверы-заглушки** — все имеют `.start()` / `.stop()` и `.port`: - `MockRTSPServer(port=0, sdp_control="*", content_base="auto", custom_sdp=None)` — переплетение TCP - `MockRTSPServerUDP()` — транспорт UDP - `MockRTSPServerSilent()` — принимает, но никогда не отвечает (тесты таймаутов) - `MockRTSPServerNoMedia()` — RTSP без данных в SDP - `MockRTSPServerNoTeardownResponse()` — игнорирует TEARDOWN - `MockHTTPUpstream(routes={path: {"status": N, "body": ..., "headers": {...}}})` — настраиваемый HTTP-сервер - `MockHTTPUpstreamSilent()` — принимает, но никогда не отвечает - `MockFCCServer()` — протоколы Telecom/Huawei FCC - `MockSTUNServer(port=0, mapped_port=0, mapped_ip="1.2.3.4", silent=False)` **RTP**: - `MulticastSender(addr=MCAST_ADDR, port=0, pps=200, ts_per_rtp=7, ...)` — отправляет многоадресный RTP на loopback - `make_rtp_packet(seq, timestamp, ssrc=0x12345678, payload_type=33, payload=None)` ### Общие фикстуры (conftest.py) - `r2h_binary` (сессия) — путь к бинарному файлу, пропускается, если отсутствует - `free_port` / `free_udp_port` (функция) — автоматически выделенные порты - `r2h_server` (функция) — предварительно запущенный R2HProcess с `-v 4 -m 100` - `multicast_sender` (функция) — запущенный MulticastSender - `mock_rtsp` / `mock_rtsp_udp` / `mock_rtsp_silent` / `mock_rtsp_no_media` / `mock_rtsp_no_teardown` - `mock_fcc` / `mock_http` / `mock_http_silent` ### Pytest Маркеры ```python @pytest.mark.multicast # требуется многоадресная рассылка на loopback @pytest.mark.rtsp # требуется сервер-заглушка RTSP @pytest.mark.fcc # требуется сервер-заглушка FCC @pytest.mark.http_proxy # требуется HTTP-прокси-заглушка @pytest.mark.slow # тесты, которые занимают больше времени ``` Применяйте ко всем тестам в файле с помощью `pytestmark = pytest.mark.multicast` на уровне модуля. ### Стратегия повторного использования R2HProcess Запуск rtp2httpd занимает время (запуск процесса + проверка готовности порта). Повторно используйте один и тот же экземпляр несколько раз, когда это возможно, чтобы значительно сократить время настройки теста. **Логика принятия решений:** 1. Могут ли все тесты в классе/модуле использовать одну и ту же конфигурацию и аргументы запуска? → Используйте фикстуру `scope="module"` или `scope="class"`. Это предпочтительный подход. 2. Тестам требуются разные конфигурации или аргументы? → Только в этом случае запускайте R2HProcess для каждого теста по требованию. ### Шаблоны тестирования **Шаблон 1 — Общая фикстура R2HProcess на уровне модуля (предпочтительно)** Стандартный подход. Запустите r2h один раз для всего файла, все тесты используют его: ```python @pytest.fixture(scope="module") def shared_r2h(r2h_binary): port = find_free_port() config = f""" [global] verbosity = 4 [bind] * {port} [services] #EXTM3U #EXTINF:-1,Channel One rtp://239.0.0.1:1234 """ r2h = R2HProcess(r2h_binary, port, config_content=config) r2h.start() yield r2h r2h.stop() class TestPlaylist: def test_playlist_served(self, shared_r2h): status, _, body = http_get("127.0.0.1", shared_r2h.port, "/playlist.m3u") assert status == 200 ``` Используйте `scope="class"`, когда разные классы в одном файле требуют разных конфигураций. **Шаблон 2 — R2HProcess для каждого теста (только при необходимости)** Когда конфигурация уникальна для одного теста и не может быть общим. Используйте `try/finally`: ```python def test_custom_max_clients(self, r2h_binary): port = find_free_port() config = f""" [global] verbosity = 4 maxclients = 1 [bind] * {port} """ r2h = R2HProcess(r2h_binary, port, config_content=config) try: r2h.start() status, _, _ = http_get("127.0.0.1", port, "/playlist.m3u") assert status == 200 finally: r2h.stop() ``` **Шаблон 3 — С серверами-заглушками (RTSP / HTTP / многоадресная рассылка)** Фикстура на уровне модуля, которая запускает как заглушку, так и r2h, возвращает оба, останавливает оба: ```python pytestmark = pytest.mark.rtsp @pytest.fixture(scope="module") def rtsp_env(r2h_binary): mock = MockRTSPServer() mock.start() port = find_free_port() config = f""" [global] verbosity = 4 [bind] * {port} [services] #EXTM3U #EXTINF:-1,RTSP Ch rtsp://127.0.0.1:{mock.port}/live """ r2h = R2HProcess(r2h_binary, port, config_content=config) r2h.start() yield r2h, mock r2h.stop() mock.stop() ``` Такой же шаблон работает для HTTP-прокси (с `MockHTTPUpstream`) и многоадресной рассылки (с `MulticastSender` и `extra_args=["-v", "4", "-m", "100", "-r", LOOPBACK_IF]` вместо `config_content`). ### Соглашения - **Повторное использование R2HProcess**: Используйте фикстуры `scope="module"`. Меньше запущенных процессов = более быстрые тесты. - **Именование файлов**: `test_<feature>.py` в каталоге `e2e/` - **Группировка классов**: Группируйте тесты по **функциональным подразделам** (`TestProxyRedirect`, `TestProxyStatusCodes`), а не по хронологии (`TestXxxNew`, `TestXxxMore`). Добавляйте новые тесты в существующий соответствующий класс. - **Именование тестов**: `test_<что>_<ожидаемое_поведение>` (например, `test_etag_present`, `test_if_none_match_304`) - **Выделение портов**: Всегда используйте `find_free_port()` / `find_free_udp_port()`, никогда не используйте жестко закодированные значения. - **Очистка**: Модульные/классовые фикстуры используют `yield` + `.stop()`. Экземпляры для каждого теста используют `try/finally`. - **Кодирование URL**: Используйте `%20` для пробелов в URL-адресах имен служб (например, `/Test%20Service`) - **Формат конфигурации**: INI-стиль с разделами `[global]`, `[bind]`, `[services]` - **Строки документации модуля**: Каждый файл теста начинается со строки документации, описывающей, что он охватывает. - **Параметризация**: Активно ищите похожие шаблоны тестов — если тесты отличаются только входными данными/ожидаемыми значениями, всегда используйте `@pytest.mark.parametrize` вместо копирования тестовых методов. ### Советы по отладке - Запустите один тест с подробным выводом: `./scripts/run-e2e.sh -p 1 -k "test_name" -x` - Проверьте, собран ли бинарный файл: `ls -la build/rtp2httpd` - Тест завис? Обычно тайм-аут `stream_get()` слишком короткий или отправитель многоадресной рассылки не достигает процесса.