gendesign/ops/metrics/tg-relay/test_app.py

"""Тесты для tg-relay — ретранслятора Bot API продукта через Beget (#3471).

Гоняют реальный `ThreadingHTTPServer` на localhost (эфемерный порт), апстрим
`urllib.request.urlopen` подменяется моком — реальный api.telegram.org НЕ
дёргаем никогда.
"""

from __future__ import annotations

import http.client
import json
import logging
import threading
from unittest import mock

import pytest

import app as relay


def _fake_upstream_response(status: int = 200, body: bytes = b'{"ok": true, "result": []}'):
    resp = mock.MagicMock()
    resp.status = status
    resp.read.return_value = body
    resp.headers.get.return_value = "application/json"
    resp.__enter__.return_value = resp
    resp.__exit__.return_value = False
    return resp


@pytest.fixture()
def secret(monkeypatch):
    monkeypatch.setattr(relay, "RELAY_SECRET", "test-secret-value")
    return "test-secret-value"


@pytest.fixture()
def server(secret):
    httpd = relay.ThreadingHTTPServer(("127.0.0.1", 0), relay.Handler)
    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
    thread.start()
    try:
        yield httpd
    finally:
        httpd.shutdown()
        thread.join(timeout=5)


def _request(server, path, headers=None, method="GET", body=None):
    conn = http.client.HTTPConnection(*server.server_address, timeout=5)
    try:
        conn.request(method, path, body=body, headers=headers or {})
        resp = conn.getresponse()
        return resp.status, resp.read()
    finally:
        conn.close()


def test_redact_path_hides_token_from_method_path():
    assert relay.redact_path("/bot123456:ABC-DEF/sendMessage") == "/bot<REDACTED>/sendMessage"


def test_redact_path_hides_bare_token():
    assert relay.redact_path("/bot123456:ABC-DEF/") == "/bot<REDACTED>"
    assert relay.redact_path("/bot123456:ABC-DEF") == "/bot<REDACTED>"


def test_redact_path_leaves_non_bot_paths_untouched():
    assert relay.redact_path("/healthz") == "/healthz"


def test_missing_secret_rejected_without_touching_upstream(server):
    with mock.patch.object(relay.urllib.request, "urlopen") as mocked:
        status, _body = _request(server, "/bot123:TOK/getMe")
    assert status == 401
    mocked.assert_not_called()


def test_wrong_secret_rejected_without_touching_upstream(server):
    with mock.patch.object(relay.urllib.request, "urlopen") as mocked:
        status, _body = _request(server, "/bot123:TOK/getMe", headers={"X-Relay-Secret": "wrong"})
    assert status == 401
    mocked.assert_not_called()


def test_valid_secret_passes_method_path_and_body_unmodified(server, secret):
    with mock.patch.object(
        relay.urllib.request, "urlopen", return_value=_fake_upstream_response()
    ) as mocked:
        status, body = _request(
            server,
            "/bot123:TOK/sendMessage",
            headers={"X-Relay-Secret": secret, "Content-Type": "application/json"},
            method="POST",
            body=b'{"chat_id": 1, "text": "hi"}',
        )

    assert status == 200
    assert json.loads(body) == {"ok": True, "result": []}

    sent_request = mocked.call_args[0][0]
    assert sent_request.full_url == "https://api.telegram.org/bot123:TOK/sendMessage"
    assert sent_request.data == b'{"chat_id": 1, "text": "hi"}'
    assert sent_request.get_method() == "POST"


def test_get_updates_uses_upstream_timeout_wider_than_longest_client_poll(server, secret):
    """Клиент шлёт getUpdates(timeout=30) → httpx ждёт ответ 40с (30 + запас
    10с). Апстрим-таймаут ретранслятора обязан быть шире, иначе он обрубит
    long-poll раньше площадки и превратит штатный цикл в вечный network error."""
    with mock.patch.object(
        relay.urllib.request, "urlopen", return_value=_fake_upstream_response()
    ) as mocked:
        _request(
            server,
            "/bot123:TOK/getUpdates",
            headers={"X-Relay-Secret": secret},
            method="POST",
            body=b'{"offset": 1, "timeout": 30}',
        )

    _req, kwargs = mocked.call_args
    assert kwargs["timeout"] >= 40


def test_token_never_appears_in_logs(server, secret, caplog):
    token = "999888777:VerySecretTokenValue"
    with caplog.at_level(logging.INFO, logger="tg-relay"):
        with mock.patch.object(
            relay.urllib.request, "urlopen", return_value=_fake_upstream_response()
        ):
            _request(server, f"/bot{token}/getMe", headers={"X-Relay-Secret": secret})

    for record in caplog.records:
        assert token not in record.getMessage()


def test_unauthorized_attempt_does_not_leak_token_either(server, caplog):
    token = "999888777:VerySecretTokenValue"
    with caplog.at_level(logging.INFO, logger="tg-relay"):
        _request(server, f"/bot{token}/getMe")

    for record in caplog.records:
        assert token not in record.getMessage()