import asyncio
import httpx
from devplacepy.services import news as news_mod
from devplacepy.services.news import NewsService, _extract_grade, _get_ai_key, _get_article_images
from devplacepy.database import get_table
from devplacepy.utils import generate_uid
API_URL = "http://news.test/api"
AI_URL = "http://ai.test/v1/chat"
LINK_HIGH = "http://news.test/high"
LINK_LOW = "http://news.test/low"
class FakeResp:
def __init__(self, json_data=None, text="", status=200):
self._json = json_data
self.text = text
self.status_code = status
def raise_for_status(self):
if self.status_code >= 400:
raise httpx.HTTPError(f"status {self.status_code}")
def json(self):
return self._json
class FakeClient:
def __init__(self, articles):
self.articles = articles
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
return False
async def get(self, url, timeout=None):
if url == API_URL:
return FakeResp(json_data={"articles": self.articles})
return FakeResp(text='<img src="http://img.test/a.png">')
async def post(self, url, json=None, headers=None, timeout=None):
prompt = json["messages"][0]["content"]
if "FailArticle" in prompt:
return FakeResp(status=500, text="err")
if "EmptyArticle" in prompt:
return FakeResp(json_data={"choices": [{"message": {"content": ""}}]})
if "BadArticle" in prompt:
return FakeResp(json_data={"choices": [{"message": {"content": "no number"}}]})
grade = "9" if "HighArticle" in prompt else "3"
return FakeResp(json_data={"choices": [{"message": {"content": grade}}]})
class FailingApiClient(FakeClient):
async def get(self, url, timeout=None):
if url == API_URL:
raise httpx.HTTPError("api down")
return FakeResp(text="")
def _settings_stub(threshold="7"):
def fake_get_setting(key, default=None):
return {
"news_api_url": API_URL,
"news_ai_url": AI_URL,
"news_ai_model": "test-model",
"news_grade_threshold": threshold,
"news_ai_key": "",
}.get(key, default)
return fake_get_setting
def test_extract_grade_parsing():
assert _extract_grade("8") == 8
assert _extract_grade("Grade: 9") == 9
assert _extract_grade("Score is 7/10") == 7
assert _extract_grade("0") is None
assert _extract_grade("11") is None
assert _extract_grade("not a number") is None
def test_get_ai_key_env_precedence(monkeypatch):
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
monkeypatch.setenv("NEWS_AI_KEY", "primary")
assert _get_ai_key() == "primary"
def test_get_ai_key_setting_fallback(monkeypatch):
monkeypatch.delenv("NEWS_AI_KEY", raising=False)
monkeypatch.setattr(news_mod, "get_setting", lambda key, default=None: "from-setting" if key == "news_ai_key" else default)
assert _get_ai_key() == "from-setting"
def test_get_ai_key_openrouter_fallback(monkeypatch):
monkeypatch.delenv("NEWS_AI_KEY", raising=False)
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
monkeypatch.setenv("OPENROUTER_API_KEY", "router-key")
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
assert _get_ai_key() == "router-key"
def test_get_ai_key_openai_fallback(monkeypatch):
monkeypatch.delenv("NEWS_AI_KEY", raising=False)
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
monkeypatch.setenv("OPENAI_API_KEY", "openai-key")
assert _get_ai_key() == "openai-key"
def test_get_ai_key_empty_when_unset(monkeypatch):
monkeypatch.delenv("NEWS_AI_KEY", raising=False)
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
assert _get_ai_key() == ""
def test_grade_article_empty_content_returns_none(local_db, monkeypatch):
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
grade = asyncio.run(NewsService()._grade_article(
{"title": "EmptyArticle", "description": "d", "content": "c"}, AI_URL, "m", FakeClient([])))
assert grade is None
def test_grade_article_unparseable_returns_none(local_db, monkeypatch):
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
grade = asyncio.run(NewsService()._grade_article(
{"title": "BadArticle", "description": "d", "content": "c"}, AI_URL, "m", FakeClient([])))
assert grade is None
def test_run_once_handles_api_failure(local_db, monkeypatch):
monkeypatch.setattr(news_mod, "get_setting", _settings_stub())
monkeypatch.setattr(news_mod.httpx, "AsyncClient", lambda *a, **k: FailingApiClient([]))
asyncio.run(NewsService().run_once())
def test_run_once_updates_existing_news_row(local_db, monkeypatch):
external_id = f"news-{generate_uid()}"
existing_uid = generate_uid()
get_table("news").insert({
"uid": existing_uid, "external_id": external_id, "slug": "",
"title": "Old Title", "status": "draft", "grade": 0, "synced_at": "2020-01-01",
})
articles = [{"guid": external_id, "title": "HighArticle", "description": "d", "content": "c",
"link": "", "feed_name": "Feed", "author": "A", "published": "2026-01-01"}]
monkeypatch.setattr(news_mod, "get_setting", _settings_stub(threshold="7"))
monkeypatch.setattr(news_mod.httpx, "AsyncClient", lambda *a, **k: FakeClient(articles))
asyncio.run(NewsService().run_once())
row = get_table("news").find_one(uid=existing_uid)
assert row["title"] == "HighArticle"
assert row["status"] == "published"
assert get_table("news").count(external_id=external_id) == 1
def test_get_article_images_filters_and_dedupes():
html = (
'<img src="http://img.test/a.png">'
"<img src='http://img.test/a.png'>"
'<img src="http://img.test/b.svg">'
'<img src="/relative.png">'
)
client = FakeClient([])
async def fake_get(url, timeout=None):
return FakeResp(text=html)
client.get = fake_get
images = asyncio.run(_get_article_images("http://news.test/page", client))
assert [img["url"] for img in images] == ["http://img.test/a.png"]
def test_get_article_images_network_error_returns_empty():
client = FakeClient([])
async def failing_get(url, timeout=None):
raise httpx.HTTPError("down")
client.get = failing_get
assert asyncio.run(_get_article_images("http://news.test/page", client)) == []
def test_run_once_publishes_grades_and_is_idempotent(local_db, monkeypatch):
g_high, g_low, g_fail = (f"news-{generate_uid()}" for _ in range(3))
articles = [
{"guid": g_high, "title": "HighArticle", "description": "d", "content": "c",
"link": LINK_HIGH, "feed_name": "Feed", "author": "A", "published": "2026-01-01"},
{"guid": g_low, "title": "LowArticle", "description": "d", "content": "c",
"link": LINK_LOW, "feed_name": "Feed", "author": "A", "published": "2026-01-01"},
{"guid": g_fail, "title": "FailArticle", "description": "d", "content": "c",
"link": "", "feed_name": "Feed", "author": "A", "published": "2026-01-01"},
{"guid": "", "title": "NoGuid", "description": "d", "content": "c",
"link": "", "feed_name": "Feed", "author": "A", "published": "2026-01-01"},
]
monkeypatch.setattr(news_mod, "get_setting", _settings_stub(threshold="7"))
monkeypatch.setattr(news_mod.httpx, "AsyncClient", lambda *a, **k: FakeClient(articles))
asyncio.run(NewsService().run_once())
news = get_table("news")
sync = get_table("news_sync")
assert news.find_one(external_id=g_high)["status"] == "published"
assert news.find_one(external_id=g_high)["grade"] == 9
assert news.find_one(external_id=g_low)["status"] == "draft"
assert news.find_one(external_id=g_low)["grade"] == 3
fail_row = news.find_one(external_id=g_fail)
assert fail_row["status"] == "draft"
assert fail_row["grade"] == 0
assert sync.find_one(external_id=g_fail)["status"] == "grading_failed"
assert sync.find_one(external_id=g_high)["status"] == "graded"
asyncio.run(NewsService().run_once())
assert news.count(external_id=g_high) == 1
assert news.count(external_id=g_fail) == 1
Reference in New Issue View Git Blame Copy Permalink