gahusb
64fbbb7958
Google이 비공식 trends endpoint 두 가지(/trends/.../rss + /trends/api/dailytrends) 모두 404로 폐기 (NAS에서 직접 호출 시 확정). 대안으로 YouTube Data API v3 mostPopular(regionCode=KR, 50개)로 source 교체: - source 이름: google_trends → youtube_trending - 키워드: 영상 제목 정제 (대괄호·이모지 제거, 60자 limit) - API 키: YOUTUBE_DATA_API_KEY (agent-office와 공유, .env 그대로 활용) - 키 미설정 시 graceful skip - docker-compose insta-lab에 환경변수 추가 - 테스트 9/9 pass (기존 6 + youtube 3 신규)
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
import os
|
|
import gc
|
|
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
|
|
gc.collect()
|
|
for ext in ("", "-wal", "-shm"):
|
|
try:
|
|
os.remove(path + ext)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def test_get_preferences_returns_defaults(client):
|
|
resp = client.get("/api/insta/preferences")
|
|
assert resp.status_code == 200
|
|
cats = {p["category"]: p["weight"] for p in resp.json()["categories"]}
|
|
assert cats == {"economy": 1.0, "psychology": 1.0, "celebrity": 1.0}
|
|
|
|
|
|
def test_put_preferences_upsert(client):
|
|
resp = client.put("/api/insta/preferences",
|
|
json={"categories": {"economy": 0.7, "psychology": 0.2, "tech": 0.5}})
|
|
assert resp.status_code == 200
|
|
cats = {p["category"]: p["weight"] for p in resp.json()["categories"]}
|
|
assert cats["economy"] == 0.7
|
|
assert cats["tech"] == 0.5
|
|
|
|
|
|
def test_list_trends_filter(client):
|
|
db_module.add_external_trend({"keyword": "A", "category": "economy",
|
|
"source": "naver_popular", "score": 1.0})
|
|
db_module.add_external_trend({"keyword": "B", "category": "celebrity",
|
|
"source": "google_trends", "score": 0.8})
|
|
resp = client.get("/api/insta/trends?source=naver_popular")
|
|
items = resp.json()["items"]
|
|
assert {it["keyword"] for it in items} == {"A"}
|
|
|
|
|
|
def test_collect_trends_kicks_background(client, monkeypatch):
|
|
from app import main, trend_collector
|
|
|
|
captured = {"called": False}
|
|
|
|
def fake_collect_all(cats):
|
|
captured["called"] = True
|
|
return {"naver_popular": 3, "youtube_trending": 2}
|
|
|
|
monkeypatch.setattr(trend_collector, "collect_all", fake_collect_all)
|
|
resp = client.post("/api/insta/trends/collect", json={})
|
|
assert resp.status_code == 200
|
|
task_id = resp.json()["task_id"]
|
|
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"
|
|
assert captured["called"] is True
|
|
|
|
|
|
def test_list_keywords_filters_by_source(client):
|
|
db_module.add_trending_keyword({"keyword": "M", "category": "economy",
|
|
"score": 0.4, "articles_count": 1, "source": "manual"})
|
|
db_module.add_external_trend({"keyword": "N", "category": "economy",
|
|
"source": "naver_popular", "score": 0.9})
|
|
resp = client.get("/api/insta/keywords?source=manual")
|
|
items = resp.json()["items"]
|
|
assert {it["keyword"] for it in items} == {"M"}
|