100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
"""
|
|
Tests for core.http_client status handling.
|
|
|
|
Regression guard: the client used to return a truthy body for ANY status, so a
|
|
4xx/5xx read as success. Now only 2xx/3xx yield a result; 4xx returns None with
|
|
no retry; 5xx retries then returns None.
|
|
|
|
Run:
|
|
python -m unittest discover -s tests
|
|
"""
|
|
import unittest
|
|
from unittest.mock import patch, AsyncMock
|
|
|
|
import core.http_client as hc
|
|
|
|
|
|
class FakeResp:
|
|
def __init__(self, status, json_data=None, content_type="application/json", text=""):
|
|
self.status = status
|
|
self._json = json_data
|
|
self.content_type = content_type
|
|
self._text = text
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
async def json(self):
|
|
return self._json
|
|
|
|
async def text(self):
|
|
return self._text
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, responses):
|
|
self._responses = list(responses)
|
|
self.closed = False
|
|
self.calls = []
|
|
|
|
def _make(self, method, url, **kwargs):
|
|
self.calls.append((method, url))
|
|
return self._responses.pop(0)
|
|
|
|
def get(self, url, **kwargs):
|
|
return self._make("get", url, **kwargs)
|
|
|
|
def post(self, url, **kwargs):
|
|
return self._make("post", url, **kwargs)
|
|
|
|
def patch(self, url, **kwargs):
|
|
return self._make("patch", url, **kwargs)
|
|
|
|
|
|
class TestHttpStatus(unittest.IsolatedAsyncioTestCase):
|
|
def _install(self, responses):
|
|
session = FakeSession(responses)
|
|
hc._session = session
|
|
return session
|
|
|
|
async def asyncTearDown(self):
|
|
hc._session = None
|
|
|
|
async def test_2xx_json_returned(self):
|
|
self._install([FakeResp(200, json_data={"ok": True})])
|
|
result = await hc.api_post("http://x/act")
|
|
self.assertEqual(result, {"ok": True})
|
|
|
|
async def test_4xx_returns_none_no_retry(self):
|
|
session = self._install([FakeResp(404, text="not found")])
|
|
result = await hc.api_post("http://x/missing")
|
|
self.assertIsNone(result) # rejection is not success
|
|
self.assertEqual(len(session.calls), 1) # 4xx is not retried
|
|
|
|
async def test_5xx_retries_then_none(self):
|
|
session = self._install([FakeResp(500, text="boom")] * 3)
|
|
with patch.object(hc.asyncio, "sleep", new=AsyncMock()):
|
|
result = await hc.api_post("http://x/act")
|
|
self.assertIsNone(result)
|
|
self.assertEqual(len(session.calls), 3) # retried up to max
|
|
|
|
async def test_5xx_then_2xx_recovers(self):
|
|
session = self._install([FakeResp(503, text="try later"),
|
|
FakeResp(200, json_data={"ok": 1})])
|
|
with patch.object(hc.asyncio, "sleep", new=AsyncMock()):
|
|
result = await hc.api_post("http://x/act")
|
|
self.assertEqual(result, {"ok": 1})
|
|
self.assertEqual(len(session.calls), 2)
|
|
|
|
async def test_2xx_non_json_returns_status(self):
|
|
self._install([FakeResp(204, content_type="text/plain")])
|
|
result = await hc.api_post("http://x/act")
|
|
self.assertEqual(result, {"status_code": 204})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|