"""Тесты доставки результата на сторонний адрес.""" 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