73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
"""cron_schdule.plugins.call_inbox_sync_api — httpx POST /email/sync, no live network."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
from cron_schdule import plugins
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, response, calls):
|
|
self._response=response
|
|
self.calls=calls
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *exc):
|
|
return None
|
|
|
|
async def post(self, url, **kwargs):
|
|
self.calls.append((url, kwargs))
|
|
return self._response
|
|
|
|
|
|
async def test_call_inbox_sync_api_posts_email_sync(monkeypatch):
|
|
monkeypatch.setenv("BACKEND_URL","http://backend-api:8000")
|
|
monkeypatch.setenv("CRON_INBOX_SYNC_TOKEN","cron-secret")
|
|
calls=[]
|
|
response=httpx.Response(
|
|
200,
|
|
json={"data":{"status":"queued","id":"run-1"},"total":1,"status_code":200},
|
|
request=httpx.Request("POST","http://backend-api:8000/email/sync"),
|
|
)
|
|
monkeypatch.setattr(
|
|
plugins.httpx,
|
|
"AsyncClient",
|
|
lambda *a, **k: _FakeClient(response, calls),
|
|
)
|
|
payload=await plugins.call_inbox_sync_api(top=50, skip=0, test_on=True)
|
|
assert payload["data"]["status"]=="queued"
|
|
assert calls[0][0]=="http://backend-api:8000/email/sync"
|
|
assert calls[0][1]["headers"]["Authorization"]=="Bearer cron-secret"
|
|
assert calls[0][1]["params"]["top"]==50
|
|
assert calls[0][1]["params"]["test_on"] is True
|
|
|
|
|
|
async def test_call_inbox_sync_api_requires_token(monkeypatch):
|
|
monkeypatch.setenv("BACKEND_URL","http://backend-api:8000")
|
|
monkeypatch.delenv("CRON_INBOX_SYNC_TOKEN", raising=False)
|
|
try:
|
|
await plugins.call_inbox_sync_api()
|
|
raise AssertionError("expected RuntimeError")
|
|
except RuntimeError as e:
|
|
assert "CRON_INBOX_SYNC_TOKEN" in str(e)
|
|
|
|
|
|
async def test_call_inbox_sync_api_raises_on_http_error(monkeypatch):
|
|
monkeypatch.setenv("BACKEND_URL","http://backend-api:8000")
|
|
monkeypatch.setenv("CRON_INBOX_SYNC_TOKEN","cron-secret")
|
|
request=httpx.Request("POST","http://backend-api:8000/email/sync")
|
|
response=httpx.Response(401, text="unauthorized", request=request)
|
|
monkeypatch.setattr(
|
|
plugins.httpx,
|
|
"AsyncClient",
|
|
lambda *a, **k: _FakeClient(response, []),
|
|
)
|
|
try:
|
|
await plugins.call_inbox_sync_api()
|
|
raise AssertionError("expected HTTPStatusError")
|
|
except httpx.HTTPStatusError as e:
|
|
assert e.response.status_code==401
|