gahusb
e1b1944f43
- insta-lab ranked_keywords: add dedup_window_days Query param (default 14, ge=1, le=90); pass to db.list_recent_issued_topics - service_proxy.insta_ranked: add dedup_window_days param (default 14); include in GET params - InstaAgent.on_schedule: read dedup_window_days from custom_config (default 14); pass to insta_ranked call - test_ranked_respects_dedup_window: verifies window param gates eligible flag correctly Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
79 lines
3.5 KiB
Python
79 lines
3.5 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from app import db, config, selection_judge
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(config, "DB_PATH", str(tmp_path / "insta.db"))
|
|
monkeypatch.setattr(db, "DB_PATH", str(tmp_path / "insta.db"))
|
|
monkeypatch.setattr(selection_judge, "judge_candidates", lambda c: {})
|
|
db.init_db()
|
|
from app.main import app
|
|
return TestClient(app)
|
|
|
|
|
|
def test_ranked_returns_sorted_eligible(client, monkeypatch):
|
|
db.add_trending_keyword({"keyword": "금리", "category": "economy", "score": 0.9})
|
|
r = client.get("/api/insta/keywords/ranked?threshold=0.0&limit=10")
|
|
assert r.status_code == 200
|
|
items = r.json()["items"]
|
|
assert len(items) >= 1
|
|
assert "final_score" in items[0] and "eligible" in items[0]
|
|
|
|
|
|
def test_decision_approve_publishes(client):
|
|
sid = db.add_card_slate({"keyword": "금리", "category": "economy"})
|
|
r = client.post(f"/api/insta/slates/{sid}/decision", json={"decision": "approved"})
|
|
assert r.status_code == 200
|
|
assert db.get_card_slate(sid)["status"] == "published"
|
|
|
|
|
|
def test_decision_reject(client):
|
|
sid = db.add_card_slate({"keyword": "환율", "category": "economy"})
|
|
r = client.post(f"/api/insta/slates/{sid}/decision", json={"decision": "rejected"})
|
|
assert r.status_code == 200
|
|
assert db.get_card_slate(sid)["status"] == "rejected"
|
|
|
|
|
|
def test_decision_invalid_400(client):
|
|
sid = db.add_card_slate({"keyword": "x", "category": "economy"})
|
|
r = client.post(f"/api/insta/slates/{sid}/decision", json={"decision": "maybe"})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_decision_unknown_slate_404(client):
|
|
r = client.post("/api/insta/slates/99999/decision", json={"decision": "approved"})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_ranked_respects_dedup_window(client):
|
|
"""dedup_window_days param이 list_recent_issued_topics window에 반영되는지 검증.
|
|
|
|
'금리' 키워드를 방금 approved(published) 상태로 기록한 뒤:
|
|
- dedup_window_days=30 → 방금 발행 = window 안 → eligible False
|
|
- dedup_window_days=1 → DB datetime이 정각 경계 직전이라도 여전히 1일 안이므로 eligible False
|
|
(확인: 반드시 eligible=False)
|
|
추가로 두 번째 키워드(word2)는 아직 발행 이력 없으므로 window 무관하게 eligible True.
|
|
"""
|
|
# 방금 발행된 키워드 등록 + 슬레이트 approved 처리
|
|
db.add_trending_keyword({"keyword": "금리", "category": "economy", "score": 0.9})
|
|
sid = db.add_card_slate({"keyword": "금리", "category": "economy"})
|
|
db.set_slate_decision(sid, "approved") # published_at = now
|
|
|
|
# 발행 이력 없는 키워드 추가
|
|
db.add_trending_keyword({"keyword": "환율", "category": "economy", "score": 0.8})
|
|
|
|
# window=30 → '금리'는 최근 발행이라 dedup 대상 → eligible False
|
|
r = client.get("/api/insta/keywords/ranked?threshold=0.0&limit=10&dedup_window_days=30")
|
|
assert r.status_code == 200
|
|
items = r.json()["items"]
|
|
keumni = next((i for i in items if i["keyword"] == "금리"), None)
|
|
assert keumni is not None, "'금리' 항목이 ranked 응답에 없음"
|
|
assert keumni["eligible"] is False, "dedup_window_days=30 내 발행 → eligible은 False여야 함"
|
|
|
|
# 발행 이력 없는 '환율'은 어떤 window에서도 eligible True
|
|
hwanul = next((i for i in items if i["keyword"] == "환율"), None)
|
|
assert hwanul is not None, "'환율' 항목이 ranked 응답에 없음"
|
|
assert hwanul["eligible"] is True, "발행 이력 없는 키워드는 eligible True여야 함"
|