"""Юнит-тесты SynapseClient — httpx.MockTransport вместо сервера."""
import logging
from datetime import UTC, datetime
import httpx
import pytest
from gnexus_synapse import SynapseClient
from gnexus_synapse.exceptions import (
SynapseAuthError,
SynapseConfigError,
SynapseConnectionError,
SynapseError,
SynapseForbiddenError,
SynapseNotFoundError,
SynapseServerError,
SynapseValidationError,
)
BASE = "http://synapse.test"
def make_client(handler, **kw):
return SynapseClient("http://synapse.test", "syn_test",
http_client=httpx.Client(transport=httpx.MockTransport(handler)),
**kw)
def test_send_success_builds_request():
seen = {}
def handler(request: httpx.Request) -> httpx.Response:
seen["url"] = str(request.url)
seen["auth"] = request.headers.get("authorization")
seen["ua"] = request.headers.get("user-agent")
seen["body"] = json_loads(request.read())
return httpx.Response(202, json={"ok": True, "id": "e-1", "status": "queued",
"deduplicated": False})
event = make_client(handler).send("bugtrail", "test", "failed",
priority="high", payload={"user_id": "42"},
dedup_key="d1")
assert (event.id, event.status, event.deduplicated) == ("e-1", "queued", False)
assert seen["url"] == f"{BASE}/api/v1/events"
assert seen["auth"] == "Bearer syn_test"
assert seen["ua"].startswith("gnexus-synapse-py/")
assert seen["body"]["source"] == "bugtrail"
assert seen["body"]["priority"] == "high"
assert seen["body"]["dedup_key"] == "d1"
def test_send_maps_status_errors():
cases = {
401: (SynapseAuthError, "API-ключ неизвестен или отозван"),
403: (SynapseForbiddenError, "не совпадает с источником ключа"),
422: (SynapseValidationError, "Тип не зарегистрирован"),
500: (SynapseServerError, "boom"),
503: (SynapseServerError, "перегружен"),
418: (SynapseError, "teapot"),
}
def make_handler(code, detail):
return lambda request: httpx.Response(code, json={"detail": detail})
for code, (cls, detail) in cases.items():
client = make_client(make_handler(code, detail))
with pytest.raises(cls) as exc_info:
client.send("s", "x", "y")
assert exc_info.value.detail == detail
assert exc_info.value.status_code == code
def test_send_transport_error_wrapped():
def handler(request):
raise httpx.ConnectError("connection refused")
client = make_client(handler)
with pytest.raises(SynapseConnectionError) as exc_info:
client.send("s", "x", "y")
assert isinstance(exc_info.value.__cause__, httpx.ConnectError)
def test_local_validation_no_http():
called = {"n": 0}
def handler(request):
called["n"] += 1
return httpx.Response(202, json={"ok": True, "id": "e", "status": "queued"})
client = make_client(handler)
with pytest.raises(SynapseValidationError) as exc_info:
client.send("Wrong Source", "x", "y")
assert exc_info.value.status_code is None # локальная ошибка, не сервер
assert called["n"] == 0
def test_default_source_and_missing():
def handler(request):
assert json_loads(request.read())["source"] == "libtest"
return httpx.Response(202, json={"ok": True, "id": "e", "status": "queued"})
make_client(handler, default_source="libtest").send(subject="x", action="y")
with pytest.raises(SynapseConfigError, match="source"):
make_client(handler).send(subject="x", action="y")
with pytest.raises(SynapseConfigError, match="api_key"):
SynapseClient("http://s.test", api_key=None,
http_client=httpx.Client(transport=httpx.MockTransport(handler)),
default_source="s").send("s", "x", "y")
def test_emit_swallows_and_logs(caplog):
caplog.set_level(logging.WARNING, logger="gnexus.synapse")
handler = lambda request: httpx.Response(503, json={"detail": "нет воркеров"}) # noqa: E731
client = make_client(handler)
result = client.emit("s", "x", "y")
assert result is None
assert any("synapse emit failed" in r.message for r in caplog.records)
def test_batch_partial():
def handler(request):
body = json_loads(request.read())
assert len(body) == 2 # конверт без source локально отвергнут
# 2-й из посланных сервер забраковал (тип не зарегистрирован)
return httpx.Response(202, json={"results": [
{"ok": True, "id": "a", "status": "queued", "deduplicated": False},
{"ok": False, "index": 1, "detail": "Тип не зарегистрирован"},
]})
result = make_client(handler).send_batch([
{"source": "s", "subject": "x", "action": "y1"}, # сервер принял
{"subject": "x", "action": "y"}, # нет source → локальный reject
{"source": "s", "subject": "x", "action": "y2"}, # сервер отверг
])
assert result.all_accepted is False
assert [e.id for e in result.accepted] == ["a"]
assert [(r.index, r.detail) for r in result.rejected] == [
(1, "не задан source: передайте source= или default_source (env SYNAPSE_DEFAULT_SOURCE)"),
(2, "Тип не зарегистрирован"),
]
def test_batch_local_rejections_shift_index():
def handler(request):
body = json_loads(request.read())
assert len(body) == 1 # битый не посылается
return httpx.Response(202, json={"results": [
{"ok": True, "id": "a", "status": "queued", "deduplicated": False},
]})
result = make_client(handler).send_batch([
{"source": "s", "subject": "x", "action": "y"}, # ok
{"source": "BAD SOURCE", "subject": "x", "action": "y"}, # локальный reject
])
assert [e.id for e in result.accepted] == ["a"] # позиция 0 — оригинальная
r = result.rejected[0]
assert (r.index, "BAD SOURCE") == (r.index, "BAD SOURCE") and r.index == 1
def test_batch_empty_no_http():
def handler(request):
raise AssertionError("HTTP не нужен")
result = make_client(handler).send_batch([])
assert result.results == [] and result.all_accepted
def test_status_parses_and_404():
eid = "11111111-1111-1111-1111-111111111111"
other = "22222222-2222-2222-2222-222222222222"
status_body = {
"id": eid, "source": "bugtrail", "subject": "test", "action": "failed",
"priority": "high", "status": "delivered",
"created_at": datetime(2026, 10, 3, 12, 0, tzinfo=UTC).isoformat(),
"expires_at": None,
"deliveries": [
{"channel": "internal_log", "target": None, "status": "delivered",
"attempts": 1, "error": None, "rendered_message": "тест упал"},
{"channel": "telegram", "target": "qa-chat", "status": "retrying",
"attempts": 2, "error": "429", "rendered_message": None},
],
}
def handler(request):
if request.url.path.endswith(other):
return httpx.Response(404, json={"detail": "Событие не найдено"})
return httpx.Response(200, json=status_body)
client = make_client(handler)
st = client.status(eid)
assert st.status == "delivered"
assert st.created_at.tzinfo is not None
assert st.deliveries[1].status == "retrying" and st.deliveries[1].attempts == 2
with pytest.raises(SynapseNotFoundError):
client.status(other)
with pytest.raises(SynapseValidationError):
client.status("не-uuid")
def test_health_ready():
def handler(request):
if request.url.path == "/api/healthz":
return httpx.Response(200, json={"status": "ok", "service": "synapse"})
return httpx.Response(200, json={"ready": True, "components": {}})
client = make_client(handler)
assert client.health() == {"status": "ok", "service": "synapse"}
assert client.ready() == {"ready": True, "components": {}}
def test_env_fallback(monkeypatch):
monkeypatch.setenv("SYNAPSE_URL", "http://env-synapse.test")
monkeypatch.setenv("SYNAPSE_API_KEY", "syn_from_env")
monkeypatch.setenv("SYNAPSE_DEFAULT_SOURCE", "libtest")
seen = {}
def capture(request):
seen["url"] = str(request.url)
seen["auth"] = request.headers.get("authorization")
return httpx.Response(202, json={"ok": True, "id": "e", "status": "queued"})
client2 = SynapseClient(http_client=httpx.Client(transport=httpx.MockTransport(capture)))
client2.send(subject="x", action="y")
assert seen["url"] == "http://env-synapse.test/api/v1/events"
assert seen["auth"] == "Bearer syn_from_env"
monkeypatch.delenv("SYNAPSE_API_KEY")
with pytest.raises(SynapseConfigError, match="api_key"):
SynapseClient(http_client=httpx.Client(transport=httpx.MockTransport(capture)),
default_source="s").send("s", "x", "y")
# -- helpers -------------------------------------------------------------
def json_loads(raw: bytes) -> dict:
import json
return json.loads(raw)
