gahusb
b3982c8f72
- Idempotent ALTER TABLE adds source column (default 'manual') + idx_tk_source index - New account_preferences table seeded with economy/psychology/celebrity at weight=1.0 - add_trending_keyword now accepts optional source param - New helpers: add_external_trend, list_trends, get_preferences, upsert_preferences - test_db updated: six→seven tables; test_preferences_crud.py (7 new tests, all pass) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
import os
|
|
import gc
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
from app import db as db_module
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_db(monkeypatch):
|
|
fd, path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
monkeypatch.setattr(db_module, "DB_PATH", path)
|
|
db_module.init_db()
|
|
yield path
|
|
gc.collect()
|
|
for ext in ("", "-wal", "-shm"):
|
|
try:
|
|
os.remove(path + ext)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def test_init_db_creates_account_preferences(tmp_db):
|
|
with db_module._conn() as conn:
|
|
rows = conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
|
|
names = {r[0] for r in rows}
|
|
assert "account_preferences" in names
|
|
|
|
|
|
def test_init_db_seeds_default_weights(tmp_db):
|
|
prefs = db_module.get_preferences()
|
|
cats = {p["category"]: p["weight"] for p in prefs}
|
|
assert cats["economy"] == pytest.approx(1.0)
|
|
assert cats["psychology"] == pytest.approx(1.0)
|
|
assert cats["celebrity"] == pytest.approx(1.0)
|
|
|
|
|
|
def test_upsert_preferences_replaces_weights(tmp_db):
|
|
db_module.upsert_preferences({"economy": 0.6, "psychology": 0.3, "celebrity": 0.1, "tech": 0.5})
|
|
prefs = {p["category"]: p["weight"] for p in db_module.get_preferences()}
|
|
assert prefs["economy"] == pytest.approx(0.6)
|
|
assert prefs["tech"] == pytest.approx(0.5)
|
|
assert "celebrity" in prefs and prefs["celebrity"] == pytest.approx(0.1)
|
|
|
|
|
|
def test_trending_keywords_source_column_exists(tmp_db):
|
|
with db_module._conn() as conn:
|
|
cols = [r[1] for r in conn.execute("PRAGMA table_info(trending_keywords)").fetchall()]
|
|
assert "source" in cols
|
|
|
|
|
|
def test_add_trending_keyword_default_source(tmp_db):
|
|
kid = db_module.add_trending_keyword({
|
|
"keyword": "K", "category": "economy", "score": 0.5, "articles_count": 3,
|
|
})
|
|
with db_module._conn() as conn:
|
|
row = conn.execute("SELECT source FROM trending_keywords WHERE id=?", (kid,)).fetchone()
|
|
assert row[0] == "manual"
|
|
|
|
|
|
def test_add_external_trend_stores_source(tmp_db):
|
|
tid = db_module.add_external_trend({
|
|
"keyword": "급등주", "category": "economy", "source": "naver_popular", "score": 0.9,
|
|
})
|
|
rows = db_module.list_trends(source="naver_popular")
|
|
assert any(r["id"] == tid and r["keyword"] == "급등주" for r in rows)
|
|
|
|
|
|
def test_list_trends_filters_by_source_and_category(tmp_db):
|
|
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": 1.0})
|
|
only_naver = db_module.list_trends(source="naver_popular")
|
|
assert {r["keyword"] for r in only_naver} == {"A"}
|
|
only_celeb_google = db_module.list_trends(source="google_trends", category="celebrity")
|
|
assert {r["keyword"] for r in only_celeb_google} == {"B"}
|