test_bitrix_call.py

tests/test_bitrix_call.py

1. """bitrix_call against an httpx.MockTransport: success, Bitrix's in-body
2. errors, rate-limit retries, and that the webhook URL never escapes."""
4. import asyncio
5. import json
7. import httpx
8. import pytest
10. import server
12. run = asyncio.run
15. @pytest.fixture
16. def transport(monkeypatch):
17. sleeps: list[float] = []
19. async def no_sleep(seconds):
20. sleeps.append(seconds)
22. monkeypatch.setattr(server.asyncio, "sleep", no_sleep)
24. def install(handler):
25. monkeypatch.setattr(server, "_HTTP", httpx.AsyncClient(transport=httpx.MockTransport(handler)))
26. return sleeps
27. return install
30. def test_posts_json_to_the_method_url(transport):
31. seen = []
33. def handler(request):
34. seen.append((request.method, str(request.url), json.loads(request.content)))
35. return httpx.Response(200, json={"result": {"item": {"id": 1}}, "total": 1})
37. transport(handler)
38. assert run(server.bitrix_call("crm.item.get", {"id": 1})) == {"item": {"id": 1}}
39. assert run(server.bitrix_call("crm.item.get", {"id": 1}, envelope=True)) == {
40. "result": {"item": {"id": 1}}, "total": 1}
41. assert seen[0] == ("POST", f"{server.WEBHOOK_URL}/crm.item.get.json", {"id": 1})
42. run(server.bitrix_call("user.current"))
43. assert seen[-1][2] == {}
46. def test_error_in_body_raises_with_url_redacted(transport):
47. def handler(request):
48. return httpx.Response(400, json={"error": "NOT_FOUND",
49. "error_description": f"see {server.WEBHOOK_URL}/x"})
51. transport(handler)
52. with pytest.raises(server.BitrixError) as exc:
53. run(server.bitrix_call("crm.item.get", {"id": 1}))
54. assert str(exc.value) == "NOT_FOUND: see [webhook]/x"
57. @pytest.mark.parametrize("status, body, message", [
58. (500, b"<html>oops</html>", "HTTP 500 from m (no parseable body)"),
59. (200, b"not json", "Unparseable response from m"),
60. (403, b'{"result": null}', "HTTP 403 from m"),
61. ])
62. def test_unusable_responses(transport, status, body, message):
63. transport(lambda request: httpx.Response(status, content=body))
64. with pytest.raises(server.BitrixError) as exc:
65. run(server.bitrix_call("m"))
66. assert str(exc.value) == message
69. def test_rate_limit_is_retried(transport):
70. answers = [{"error": "QUERY_LIMIT_EXCEEDED"}, {"result": 7}]
71. sleeps = transport(lambda request: httpx.Response(200, json=answers.pop(0)))
72. assert run(server.bitrix_call("m")) == 7
73. assert sleeps == [1]
76. def test_rate_limit_gives_up_after_three_attempts(transport):
77. sleeps = transport(lambda request: httpx.Response(200, json={"error": "QUERY_LIMIT_EXCEEDED"}))
78. with pytest.raises(server.BitrixError) as exc:
79. run(server.bitrix_call("m"))
80. assert str(exc.value) == "QUERY_LIMIT_EXCEEDED after 3 retries"
81. assert sleeps == [1, 2, 3]
84. def test_transport_errors_never_leak_the_webhook(transport):
85. def handler(request):
86. raise httpx.ConnectError(f"cannot reach {request.url}", request=request)
88. transport(handler)
89. with pytest.raises(server.BitrixError) as exc:
90. run(server.bitrix_call("m"))
91. assert str(exc.value) == ("transport failure after 3 attempts calling m: "
92. "ConnectError: cannot reach [webhook]/m.json")
93. assert "example.invalid" not in str(exc.value)
96. def test_batch_encodes_commands(fake_bitrix):
97. bx = fake_bitrix({"batch": {"result": {"result": {"a": []}, "result_total": {"a": 3}}}})
98. out = run(server.bitrix_batch({"a": ("crm.item.list", {"filter": {"stageId": "NEW"}})}))
99. assert out == {"result": {"a": []}, "result_total": {"a": 3}}
100. assert bx.calls == [("batch", {"halt": 0, "cmd": {"a": "crm.item.list?filter%5BstageId%5D=NEW"}})]
101. assert run(server.bitrix_batch({})) == {}
102. with pytest.raises(server.BitrixError):
103. run(server.bitrix_batch({str(i): ("m", {}) for i in range(51)}))