258 lines
9.6 KiB
Python
258 lines
9.6 KiB
Python
"""Tests for us2kr.deliver.blogger (no network: the httpx client is a stub)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, date, datetime
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from us2kr.config import Settings
|
|
from us2kr.deliver import build_deliverers
|
|
from us2kr.deliver.base import DeliveryError
|
|
from us2kr.deliver.blogger import BloggerDeliverer, briefing_html, post_title
|
|
from us2kr.models import BriefingResult, MarketPayload, Report
|
|
|
|
TOKEN_OK = {"access_token": "at-1", "expires_in": 3600}
|
|
POST_OK = {"kind": "blogger#post", "id": "42", "url": "https://nybrief.blogspot.com/2026/09/post.html"}
|
|
|
|
|
|
def make_result(headline: str | None = "9월 첫 거래일 하락") -> BriefingResult:
|
|
payload = MarketPayload(
|
|
session_date=date(2026, 8, 31),
|
|
us_close_time=datetime(2026, 8, 31, 20, 0, tzinfo=UTC),
|
|
generated_at=datetime(2026, 9, 1, 6, 15, tzinfo=UTC),
|
|
krx_date=date(2026, 9, 1),
|
|
krx_open_today=True,
|
|
)
|
|
report = None
|
|
if headline is not None:
|
|
report = Report(
|
|
headline=headline,
|
|
market_overview="지수가 내렸습니다.",
|
|
key_drivers=["금리 상승"],
|
|
rates_fx_commodities="금리 상승.",
|
|
korea_implications=["반도체 부담"],
|
|
watchlist_today=["환율"],
|
|
risk_flags=["변동성"],
|
|
)
|
|
markdown = (
|
|
"# 🇺🇸→🇰🇷 미국 마감 브리핑 — 2026-09-01\n"
|
|
"미국 마감: 2026-09-01 05:00 KST\n\n"
|
|
"## 총평\n**하락 마감**\n\n"
|
|
"| 종목 | 종가 |\n|---|---|\n| S&P 500 | 7,686.42 |\n"
|
|
)
|
|
return BriefingResult(payload=payload, report=report, markdown=markdown, degraded=report is None)
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, status: int, body: dict[str, Any]):
|
|
self.status_code = status
|
|
self._body = body
|
|
self.text = json.dumps(body)
|
|
|
|
def json(self) -> dict[str, Any]:
|
|
return self._body
|
|
|
|
|
|
class FakeHttp:
|
|
"""Minimal stand-in for httpx.Client: scripted responses, recorded calls."""
|
|
|
|
def __init__(self, responses: list[Any]):
|
|
self._responses = list(responses)
|
|
self.calls: list[dict[str, Any]] = []
|
|
|
|
def request(self, method: str, url: str, **kwargs: Any) -> Any:
|
|
self.calls.append({"method": method, "url": url, **kwargs})
|
|
if not self._responses:
|
|
raise AssertionError("unexpected extra HTTP call")
|
|
item = self._responses.pop(0)
|
|
if isinstance(item, Exception):
|
|
raise item
|
|
return item
|
|
|
|
def post(self, url: str, **kwargs: Any) -> Any:
|
|
return self.request("POST", url, **kwargs)
|
|
|
|
|
|
def make_deliverer(responses: list[Any]) -> tuple[BloggerDeliverer, FakeHttp]:
|
|
client = FakeHttp(responses)
|
|
deliverer = BloggerDeliverer(
|
|
"cid", "csecret", "rtoken", "12345", labels=["미국증시"], client=client # type: ignore[arg-type]
|
|
)
|
|
return deliverer, client
|
|
|
|
|
|
# ------------------------------------------------------------------ html & title
|
|
def test_briefing_html_drops_h1_and_renders_tables():
|
|
html = briefing_html(make_result().markdown)
|
|
assert "미국 마감 브리핑 — 2026-09-01" not in html # H1 is the post title, not body
|
|
assert "<h2>총평</h2>" in html
|
|
assert "<strong>하락 마감</strong>" in html
|
|
# tables carry inline borders - Blogger themes style nothing themselves
|
|
assert '<table style="border-collapse:collapse' in html
|
|
assert ">7,686.42</td>" in html and 'border:1px solid #ccc' in html
|
|
assert "<table>" not in html and "<td>" not in html # every cell got the inline style
|
|
|
|
|
|
def test_post_title_with_and_without_headline():
|
|
assert post_title(make_result()) == "미국 마감 브리핑 2026-09-01 — 9월 첫 거래일 하락"
|
|
assert post_title(make_result(headline=None)) == "미국 마감 브리핑 2026-09-01 (미 2026-08-31 세션)"
|
|
|
|
|
|
# ------------------------------------------------------------------ delivery
|
|
EMPTY_LIST = {"kind": "blogger#postList"} # no items key on an empty blog
|
|
|
|
|
|
def test_deliver_refreshes_token_lists_then_inserts():
|
|
deliverer, client = make_deliverer(
|
|
[_Resp(200, TOKEN_OK), _Resp(200, EMPTY_LIST), _Resp(200, POST_OK)]
|
|
)
|
|
receipt = deliverer.deliver(make_result())
|
|
|
|
assert receipt == POST_OK["url"]
|
|
token_call, list_call, insert_call = client.calls
|
|
assert token_call["data"]["grant_type"] == "refresh_token"
|
|
assert list_call["method"] == "GET" and list_call["params"]["fetchBodies"] == "false"
|
|
assert insert_call["method"] == "POST"
|
|
assert "refresh_token" not in insert_call["url"] # secrets never in URLs
|
|
assert insert_call["url"].endswith("/blogs/12345/posts/")
|
|
assert insert_call["headers"]["Authorization"] == "Bearer at-1"
|
|
body = insert_call["json"]
|
|
assert body["title"].startswith("미국 마감 브리핑 2026-09-01")
|
|
assert body["labels"] == ["미국증시"] and "<table style=" in body["content"]
|
|
|
|
|
|
def test_deliver_updates_existing_post_for_same_day():
|
|
listing = {"items": [{"id": "42", "title": "미국 마감 브리핑 2026-09-01 — 이전 헤드라인"}]}
|
|
deliverer, client = make_deliverer(
|
|
[_Resp(200, TOKEN_OK), _Resp(200, listing), _Resp(200, POST_OK)]
|
|
)
|
|
receipt = deliverer.deliver(make_result())
|
|
assert receipt == POST_OK["url"]
|
|
update_call = client.calls[-1]
|
|
assert update_call["method"] == "PUT"
|
|
assert update_call["url"].endswith("/blogs/12345/posts/42")
|
|
|
|
|
|
def test_deliver_other_day_posts_do_not_match():
|
|
listing = {"items": [{"id": "41", "title": "미국 마감 브리핑 2026-08-29 — 다른 날"}]}
|
|
deliverer, client = make_deliverer(
|
|
[_Resp(200, TOKEN_OK), _Resp(200, listing), _Resp(200, POST_OK)]
|
|
)
|
|
deliverer.deliver(make_result())
|
|
assert client.calls[-1]["method"] == "POST" # inserted, not updated
|
|
|
|
|
|
def test_deliver_reuses_cached_token():
|
|
deliverer, client = make_deliverer(
|
|
[
|
|
_Resp(200, TOKEN_OK),
|
|
_Resp(200, EMPTY_LIST),
|
|
_Resp(200, POST_OK),
|
|
_Resp(200, EMPTY_LIST),
|
|
_Resp(200, POST_OK),
|
|
]
|
|
)
|
|
deliverer.deliver(make_result())
|
|
deliverer.deliver(make_result())
|
|
token_calls = [c for c in client.calls if "oauth2" in c["url"]]
|
|
assert len(token_calls) == 1
|
|
|
|
|
|
def test_deliver_retries_once_on_401_with_fresh_token():
|
|
deliverer, client = make_deliverer(
|
|
[
|
|
_Resp(200, TOKEN_OK),
|
|
_Resp(401, {"error": "invalid token"}), # list call rejected
|
|
_Resp(200, {"access_token": "at-2", "expires_in": 3600}),
|
|
_Resp(200, EMPTY_LIST),
|
|
_Resp(200, POST_OK),
|
|
]
|
|
)
|
|
receipt = deliverer.deliver(make_result())
|
|
assert receipt == POST_OK["url"]
|
|
assert client.calls[-1]["headers"]["Authorization"] == "Bearer at-2"
|
|
|
|
|
|
def test_deliver_lookup_failure_still_publishes(monkeypatch):
|
|
monkeypatch.setattr("us2kr.deliver.blogger.time.sleep", lambda _s: None)
|
|
deliverer, client = make_deliverer(
|
|
[_Resp(200, TOKEN_OK)] + [_Resp(503, {"error": "backend"})] * 3 + [_Resp(200, POST_OK)]
|
|
)
|
|
receipt = deliverer.deliver(make_result())
|
|
assert receipt == POST_OK["url"]
|
|
assert client.calls[-1]["method"] == "POST"
|
|
|
|
|
|
def test_deliver_retries_5xx_then_raises(monkeypatch):
|
|
monkeypatch.setattr("us2kr.deliver.blogger.time.sleep", lambda _s: None)
|
|
deliverer, _client = make_deliverer(
|
|
[_Resp(200, TOKEN_OK), _Resp(200, EMPTY_LIST)] + [_Resp(503, {"error": "backend"})] * 3
|
|
)
|
|
with pytest.raises(DeliveryError, match="HTTP 503"):
|
|
deliverer.deliver(make_result())
|
|
|
|
|
|
def test_deliver_does_not_retry_4xx():
|
|
deliverer, client = make_deliverer(
|
|
[_Resp(200, TOKEN_OK), _Resp(200, EMPTY_LIST), _Resp(403, {"error": "forbidden"})]
|
|
)
|
|
with pytest.raises(DeliveryError, match="HTTP 403"):
|
|
deliverer.deliver(make_result())
|
|
assert len(client.calls) == 3 # no extra attempts
|
|
|
|
|
|
def test_token_failure_raises_delivery_error():
|
|
deliverer, _ = make_deliverer([_Resp(400, {"error": "invalid_grant"})])
|
|
with pytest.raises(DeliveryError, match="token refresh failed"):
|
|
deliverer.deliver(make_result())
|
|
|
|
|
|
def test_network_error_wrapped(monkeypatch):
|
|
monkeypatch.setattr("us2kr.deliver.blogger.time.sleep", lambda _s: None)
|
|
deliverer, _ = make_deliverer(
|
|
[_Resp(200, TOKEN_OK), _Resp(200, EMPTY_LIST)] + [httpx.ConnectError("boom")] * 3
|
|
)
|
|
with pytest.raises(DeliveryError, match="ConnectError"):
|
|
deliverer.deliver(make_result())
|
|
|
|
|
|
def test_notify_error_never_raises():
|
|
deliverer, client = make_deliverer([])
|
|
deliverer.notify_error("subject", "detail") # logs only, no HTTP call
|
|
assert client.calls == []
|
|
|
|
|
|
def test_requires_all_credentials():
|
|
with pytest.raises(ValueError):
|
|
BloggerDeliverer("cid", "csecret", "", "12345")
|
|
|
|
|
|
# ------------------------------------------------------------------ wiring
|
|
def test_build_deliverers_includes_blogger_when_configured(tmp_path):
|
|
base = {
|
|
"US2KR_DATA_DIR": str(tmp_path),
|
|
"US2KR_REPORTS_DIR": str(tmp_path),
|
|
"US2KR_LOG_DIR": str(tmp_path),
|
|
}
|
|
plain = Settings(_env_file=None, **base)
|
|
assert [d.name for d in build_deliverers(plain)] == ["file"]
|
|
|
|
full = Settings(
|
|
_env_file=None,
|
|
**base,
|
|
BLOGGER_CLIENT_ID="cid",
|
|
BLOGGER_CLIENT_SECRET="cs",
|
|
BLOGGER_REFRESH_TOKEN="rt",
|
|
BLOGGER_BLOG_ID="1",
|
|
BLOGGER_LABELS="미국증시, 시황",
|
|
)
|
|
deliverers = build_deliverers(full)
|
|
assert [d.name for d in deliverers] == ["file", "blogger"]
|
|
blogger = deliverers[-1]
|
|
assert blogger.labels == ["미국증시", "시황"] # type: ignore[attr-defined]
|
|
assert blogger.blog_id == "1" # type: ignore[attr-defined]
|