talkscore-asr/tests/test_webhook.py

"""Тесты доставки результата на сторонний адрес."""
import hashlib
import hmac
import json

import pytest

from app.webhook import deliver, sign_payload, secret_fingerprint


class FakeResponse:
    def __init__(self, status):
        self.status_code = status


class TestSignature:
    def test_signature_matches_hmac(self):
        body = b'{"job_id":"x"}'
        expected = hmac.new(b"secret", body, hashlib.sha256).hexdigest()
        assert sign_payload(body, "secret") == expected

    def test_different_secret_gives_different_signature(self):
        body = b'{"job_id":"x"}'
        assert sign_payload(body, "a") != sign_payload(body, "b")

    def test_signature_changes_with_body(self):
        assert sign_payload(b"one", "s") != sign_payload(b"two", "s")


class TestDelivery:
    def test_successful_delivery_sends_once(self, monkeypatch):
        calls = []

        def fake_post(url, data=None, headers=None, timeout=None):
            calls.append((url, data, headers))
            return FakeResponse(200)

        monkeypatch.setattr("requests.post", fake_post)
        assert deliver("http://x/hook", {"job_id": "1"}, "s", delays=(0,)) is True
        assert len(calls) == 1

    def test_signature_header_present(self, monkeypatch):
        seen = {}

        def fake_post(url, data=None, headers=None, timeout=None):
            seen.update(headers)
            return FakeResponse(200)

        monkeypatch.setattr("requests.post", fake_post)
        deliver("http://x", {"job_id": "1"}, "секрет", delays=(0,))
        assert "X-Talkscore-Signature" in seen

    def test_no_signature_without_secret(self, monkeypatch):
        seen = {}
        monkeypatch.setattr("requests.post",
                            lambda url, data=None, headers=None, timeout=None:
                            (seen.update(headers), FakeResponse(200))[1])
        deliver("http://x", {"job_id": "1"}, "", delays=(0,))
        assert "X-Talkscore-Signature" not in seen

    def test_retries_on_server_error(self, monkeypatch):
        attempts = []
        monkeypatch.setattr("requests.post",
                            lambda url, data=None, headers=None, timeout=None:
                            (attempts.append(1), FakeResponse(500))[1])
        assert deliver("http://x", {"job_id": "1"}, "", delays=(0, 0, 0)) is False
        assert len(attempts) == 3

    def test_retries_on_network_failure(self, monkeypatch):
        attempts = []

        def boom(url, data=None, headers=None, timeout=None):
            attempts.append(1)
            raise OSError("сеть недоступна")

        monkeypatch.setattr("requests.post", boom)
        assert deliver("http://x", {"job_id": "1"}, "", delays=(0, 0)) is False
        assert len(attempts) == 2

    def test_stops_after_first_success(self, monkeypatch):
        attempts = []

        def flaky(url, data=None, headers=None, timeout=None):
            attempts.append(1)
            return FakeResponse(500 if len(attempts) == 1 else 200)

        monkeypatch.setattr("requests.post", flaky)
        assert deliver("http://x", {"job_id": "1"}, "", delays=(0, 0, 0)) is True
        assert len(attempts) == 2

    def test_payload_is_valid_json_utf8(self, monkeypatch):
        seen = {}
        monkeypatch.setattr("requests.post",
                            lambda url, data=None, headers=None, timeout=None:
                            (seen.update({"body": data}), FakeResponse(200))[1])
        deliver("http://x", {"text": "русский текст"}, "", delays=(0,))
        assert json.loads(seen["body"].decode("utf-8"))["text"] == "русский текст"


class TestDiagnostics:
    """Сверять настройки надо, не пересылая секрет: он осядет в переписке."""

    def test_fingerprint_is_stable(self):
        assert secret_fingerprint("odin-i-tot-zhe") == secret_fingerprint("odin-i-tot-zhe")

    def test_different_secrets_differ(self):
        assert secret_fingerprint("pervyy") != secret_fingerprint("vtoroy")

    def test_fingerprint_does_not_reveal_secret(self):
        secret = "ochen-dlinnyy-sekret-32-bayta-rovno"
        assert secret not in secret_fingerprint(secret)
        assert len(secret_fingerprint(secret)) == 8

    def test_empty_secret_gives_empty_fingerprint(self):
        assert secret_fingerprint("") == ""


class TestUnsignedDelivery:
    def test_no_header_without_secret(self, monkeypatch):
        """Пустой секрет = заголовка нет вовсе. Принимающая сторона видит
        не «неверную подпись», а её отсутствие - причины разные."""
        sent = {}

        class Response:
            status_code = 200
            text = ""

        def fake_post(url, data=None, headers=None, timeout=None):
            sent.update(headers or {})
            return Response()

        import app.webhook as wh
        monkeypatch.setattr(wh, "requests", type("R", (), {"post": staticmethod(fake_post)}),
                            raising=False)
        import sys
        monkeypatch.setitem(sys.modules, "requests",
                            type("R", (), {"post": staticmethod(fake_post)}))
        wh.deliver("http://example.com", {"job_id": "x"}, secret="", delays=(0,))
        assert "X-Talkscore-Signature" not in sent