81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from blackout_notifier.clients import EitaaClient, ExternalServiceError, SaapaClient
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, body: Any, *, status_code: int = 200) -> None:
|
|
self._body = body
|
|
self.status_code = status_code
|
|
self.ok = 200 <= status_code < 400
|
|
|
|
def json(self) -> Any:
|
|
return self._body
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, response: FakeResponse) -> None:
|
|
self.response = response
|
|
self.calls: list[tuple[str, str, dict[str, Any]]] = []
|
|
|
|
def post(self, url: str, **kwargs: Any) -> FakeResponse:
|
|
self.calls.append(("POST", url, kwargs))
|
|
return self.response
|
|
|
|
def get(self, url: str, **kwargs: Any) -> FakeResponse:
|
|
self.calls.append(("GET", url, kwargs))
|
|
return self.response
|
|
|
|
|
|
def _payload(number: int = 1) -> dict[str, Any]:
|
|
return {
|
|
"outage_number": number,
|
|
"outage_date": "1405/04/27",
|
|
"outage_start_time": "09:00",
|
|
"outage_stop_time": "11:00",
|
|
"outage_address": "آدرس",
|
|
"reason_outage": "مدیریت بار",
|
|
"is_planned": True,
|
|
}
|
|
|
|
|
|
def test_saapa_client_sends_expected_request() -> None:
|
|
session = FakeSession(FakeResponse({"data": [_payload()]}))
|
|
client = SaapaClient("secret", timeout=7, max_events_per_bill=25, session=session) # type: ignore[arg-type]
|
|
|
|
events = client.fetch("123", "1405/04/26", "1405/04/31")
|
|
|
|
assert len(events) == 1
|
|
_, _, kwargs = session.calls[0]
|
|
assert kwargs["headers"]["Authorization"] == "Bearer secret"
|
|
assert kwargs["json"]["bill_id"] == "123"
|
|
assert kwargs["timeout"] == 7
|
|
|
|
|
|
def test_saapa_rejects_oversized_or_duplicate_snapshot() -> None:
|
|
oversized = FakeSession(FakeResponse({"data": [_payload(1), _payload(2)]}))
|
|
client = SaapaClient("token", timeout=1, max_events_per_bill=1, session=oversized) # type: ignore[arg-type]
|
|
with pytest.raises(ExternalServiceError, match="limit is 1"):
|
|
client.fetch("123", "from", "to")
|
|
|
|
duplicate = FakeSession(FakeResponse({"data": [_payload(1), _payload(1)]}))
|
|
client = SaapaClient("token", timeout=1, max_events_per_bill=5, session=duplicate) # type: ignore[arg-type]
|
|
with pytest.raises(ExternalServiceError, match="duplicate outage_number"):
|
|
client.fetch("123", "from", "to")
|
|
|
|
|
|
def test_eitaa_requires_ok_response() -> None:
|
|
session = FakeSession(FakeResponse({"ok": False, "description": "denied"}))
|
|
client = EitaaClient("token", timeout=2, session=session) # type: ignore[arg-type]
|
|
|
|
with pytest.raises(ExternalServiceError, match="denied"):
|
|
client.send("chat", "message")
|
|
|
|
_, url, kwargs = session.calls[0]
|
|
assert url.endswith("/token/sendMessage")
|
|
assert kwargs["params"] == {"chat_id": "chat", "text": "message"}
|