Vladimir BryzgalovandClaude Opus 5
e6e10dddd1
Опрос статуса заставлял принимающую сторону дёргать сервис каждые несколько секунд. Теперь при завершении задачи результат уходит POST-ом на заданный адрес, с подписью HMAC-SHA256 в заголовке и тремя попытками при неудаче. Адрес задаётся в настройках или параметром запроса. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
98 lines
3.6 KiB
Python
98 lines
3.6 KiB
Python
"""Тесты доставки результата на сторонний адрес."""
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from app.webhook import deliver, sign_payload
|
|
|
|
|
|
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"] == "русский текст"
|