86 lines
3.4 KiB
Python
86 lines
3.4 KiB
Python
import json
|
|
import unittest
|
|
from email.message import Message
|
|
from urllib.error import HTTPError
|
|
from unittest.mock import patch
|
|
|
|
from ai_daily_report.clients import FetchTextError, BlogApiClient, OpenAICompatibleClient, fetch_text
|
|
|
|
|
|
class FakeResponse:
|
|
status = 200
|
|
|
|
def __init__(self, body):
|
|
self.body = body
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def read(self):
|
|
return self.body
|
|
|
|
|
|
class ClientTests(unittest.TestCase):
|
|
def test_fetch_text_decodes_response(self):
|
|
with patch("urllib.request.urlopen", return_value=FakeResponse("ok".encode("utf-8"))):
|
|
self.assertEqual(fetch_text("https://example.com", 1), "ok")
|
|
|
|
def test_fetch_text_retries_transient_http_errors(self):
|
|
responses = [
|
|
HTTPError("https://example.com", 503, "Service Unavailable", {}, None),
|
|
FakeResponse("ok".encode("utf-8")),
|
|
]
|
|
with patch("urllib.request.urlopen", side_effect=responses) as urlopen:
|
|
self.assertEqual(fetch_text("https://example.com", 1, retries=1, backoff_seconds=0), "ok")
|
|
|
|
self.assertEqual(urlopen.call_count, 2)
|
|
|
|
def test_fetch_text_does_not_retry_404_and_classifies_error(self):
|
|
with patch(
|
|
"urllib.request.urlopen",
|
|
side_effect=HTTPError("https://example.com", 404, "Not Found", {}, None),
|
|
) as urlopen:
|
|
with self.assertRaises(FetchTextError) as context:
|
|
fetch_text("https://example.com", 1, retries=2, backoff_seconds=0)
|
|
|
|
self.assertEqual(urlopen.call_count, 1)
|
|
self.assertEqual(context.exception.error_type, "http_404")
|
|
self.assertEqual(context.exception.http_status, 404)
|
|
|
|
def test_openai_compatible_client_returns_message_content(self):
|
|
body = json.dumps({"choices": [{"message": {"content": "hello"}}]}).encode("utf-8")
|
|
with patch("urllib.request.urlopen", return_value=FakeResponse(body)):
|
|
client = OpenAICompatibleClient(api_key="key", base_url="https://llm.example/v1", model="model")
|
|
self.assertEqual(client.chat("prompt"), "hello")
|
|
|
|
def test_blog_api_client_create_and_publish(self):
|
|
responses = [
|
|
FakeResponse(json.dumps({"slug": "ai-2026-06-04"}).encode("utf-8")),
|
|
FakeResponse(json.dumps({"ok": True}).encode("utf-8")),
|
|
]
|
|
with patch("urllib.request.urlopen", side_effect=responses):
|
|
client = BlogApiClient(base_url="https://blog.example", token="token")
|
|
self.assertEqual(client.create_post({"title": "t"})["slug"], "ai-2026-06-04")
|
|
client.publish_post("ai-2026-06-04")
|
|
|
|
def test_blog_api_client_slug_lookup_falls_back_to_query_endpoint(self):
|
|
responses = [
|
|
HTTPError("https://blog.example/api/service/posts/ai-2026-06-10", 404, "Not Found", Message(), None),
|
|
FakeResponse(json.dumps({"items": [{"slug": "ai-2026-06-10", "content": "body"}]}).encode("utf-8")),
|
|
]
|
|
with patch("urllib.request.urlopen", side_effect=responses) as urlopen:
|
|
client = BlogApiClient(base_url="https://blog.example", token="token")
|
|
post = client.get_post_by_slug("ai-2026-06-10")
|
|
|
|
self.assertIsNotNone(post)
|
|
assert post is not None
|
|
self.assertEqual(post["slug"], "ai-2026-06-10")
|
|
self.assertEqual(urlopen.call_count, 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|