feat(insta-lab): main.py FastAPI endpoints + BackgroundTasks
13 REST endpoints covering health, status, news, keywords, slates, tasks, and prompt templates. All background functions are async def so FastAPI awaits them without asyncio.run conflicts. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
91
insta-lab/tests/test_main.py
Normal file
91
insta-lab/tests/test_main.py
Normal file
@@ -0,0 +1,91 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app import db as db_module
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(monkeypatch):
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
monkeypatch.setattr(db_module, "DB_PATH", path)
|
||||
db_module.init_db()
|
||||
from app import main
|
||||
monkeypatch.setattr(main, "DB_PATH", path)
|
||||
with TestClient(main.app) as c:
|
||||
yield c
|
||||
import gc
|
||||
gc.collect()
|
||||
for ext in ("", "-wal", "-shm"):
|
||||
try:
|
||||
os.remove(path + ext)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def test_health(client):
|
||||
resp = client.get("/health")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["ok"] is True
|
||||
|
||||
|
||||
def test_status_endpoint(client):
|
||||
resp = client.get("/api/insta/status")
|
||||
assert resp.status_code == 200
|
||||
j = resp.json()
|
||||
assert "naver_api" in j and "anthropic_api" in j
|
||||
|
||||
|
||||
def test_news_articles_listing(client):
|
||||
db_module.add_news_article({
|
||||
"category": "economy", "title": "T1", "link": "https://x/1", "summary": "S",
|
||||
})
|
||||
resp = client.get("/api/insta/news/articles?category=economy&days=7")
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()["items"]) == 1
|
||||
|
||||
|
||||
def test_keywords_listing(client):
|
||||
db_module.add_trending_keyword({
|
||||
"keyword": "K", "category": "economy", "score": 0.5, "articles_count": 3,
|
||||
})
|
||||
resp = client.get("/api/insta/keywords?category=economy")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["items"][0]["keyword"] == "K"
|
||||
|
||||
|
||||
def test_create_slate_kicks_background_task(client, monkeypatch):
|
||||
from app import main, card_writer, card_renderer
|
||||
|
||||
def fake_write(keyword, category, articles=None):
|
||||
return db_module.add_card_slate({
|
||||
"keyword": keyword, "category": category, "status": "draft",
|
||||
"cover_copy": {"headline": "H", "body": "B", "accent_color": "#000"},
|
||||
"body_copies": [{"headline": f"h{i}", "body": f"b{i}"} for i in range(8)],
|
||||
"cta_copy": {"headline": "C", "body": "B", "cta": "F"},
|
||||
})
|
||||
|
||||
async def fake_render(slate_id, template="default/card.html.j2"):
|
||||
for i in range(1, 11):
|
||||
db_module.add_card_asset(slate_id, i, f"/tmp/{slate_id}_{i}.png", "h")
|
||||
return [f"/tmp/{slate_id}_{i}.png" for i in range(1, 11)]
|
||||
|
||||
monkeypatch.setattr(card_writer, "write_slate", fake_write)
|
||||
monkeypatch.setattr(card_renderer, "render_slate", fake_render)
|
||||
|
||||
resp = client.post("/api/insta/slates", json={"keyword": "K", "category": "economy"})
|
||||
assert resp.status_code == 200
|
||||
task_id = resp.json()["task_id"]
|
||||
# poll task
|
||||
for _ in range(20):
|
||||
st = client.get(f"/api/insta/tasks/{task_id}").json()
|
||||
if st["status"] in ("succeeded", "failed"):
|
||||
break
|
||||
assert st["status"] == "succeeded"
|
||||
slate_id = st["result_id"]
|
||||
detail = client.get(f"/api/insta/slates/{slate_id}").json()
|
||||
assert detail["status"] == "rendered"
|
||||
assert len(detail["assets"]) == 10
|
||||
Reference in New Issue
Block a user