100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
# music-lab/tests/test_market.py
|
|
|
|
def test_ingest_and_report(tmp_db):
|
|
from app.db import init_db
|
|
from app.market import ingest_trends, get_suggestions
|
|
init_db()
|
|
|
|
trends = [
|
|
{"source": "youtube", "country": "BR", "genre": "lo-fi", "keyword": "lofi study", "score": 0.9, "rank": 1, "metadata": {}},
|
|
{"source": "youtube", "country": "ID", "genre": "pop", "keyword": "pop hits", "score": 0.7, "rank": 2, "metadata": {}},
|
|
{"source": "billboard", "country": "US", "genre": "pop", "keyword": "top 40", "score": 0.8, "rank": 1, "metadata": {}},
|
|
]
|
|
report = ingest_trends(trends, "2026-05-01")
|
|
assert report["report_date"] == "2026-05-01"
|
|
assert len(report["top_genres"]) >= 2
|
|
# pop이 lo-fi보다 score 높아야 함 (2건)
|
|
genres_by_score = [g["genre"] for g in report["top_genres"]]
|
|
assert genres_by_score[0] == "pop"
|
|
|
|
suggestions = get_suggestions(limit=3)
|
|
assert len(suggestions) >= 1
|
|
assert "suno_prompt" in suggestions[0]
|
|
|
|
|
|
def test_ingest_idempotent(tmp_db):
|
|
"""같은 날 두 번 ingest해도 report가 upsert 돼야 함."""
|
|
from app.db import init_db, get_trend_reports
|
|
from app.market import ingest_trends
|
|
init_db()
|
|
|
|
trends = [{"source": "youtube", "country": "BR", "genre": "lo-fi",
|
|
"keyword": "chill", "score": 0.8, "rank": 1, "metadata": {}}]
|
|
ingest_trends(trends, "2026-05-01")
|
|
ingest_trends(trends, "2026-05-01") # 두 번째
|
|
|
|
reports = get_trend_reports()
|
|
assert len([r for r in reports if r["report_date"] == "2026-05-01"]) == 1
|
|
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
def test_market_endpoints_empty(tmp_db):
|
|
"""Empty DB: /report/latest returns 404, /suggest returns []."""
|
|
from app.db import init_db
|
|
init_db()
|
|
from app.main import app
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/api/music/market/report/latest")
|
|
assert resp.status_code == 404
|
|
|
|
resp = client.get("/api/music/market/suggest")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["suggestions"] == []
|
|
|
|
|
|
def test_market_ingest_endpoint(tmp_db):
|
|
"""POST /ingest returns ok, GET /report/latest returns report, GET /trends returns data."""
|
|
from app.db import init_db
|
|
init_db()
|
|
from app.main import app
|
|
client = TestClient(app)
|
|
|
|
payload = {
|
|
"trends": [
|
|
{"source": "youtube", "country": "BR", "genre": "lo-fi",
|
|
"keyword": "lofi", "score": 0.9, "rank": 1, "metadata": {}},
|
|
],
|
|
"report_date": "2026-05-01",
|
|
}
|
|
resp = client.post("/api/music/market/ingest", json=payload)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["ok"] is True
|
|
assert data["trends_saved"] == 1
|
|
assert data["report_date"] == "2026-05-01"
|
|
|
|
resp = client.get("/api/music/market/report/latest")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["report_date"] == "2026-05-01"
|
|
|
|
resp = client.get("/api/music/market/trends")
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["trends"]) == 1
|
|
|
|
|
|
def test_ingest_empty_trends(tmp_db):
|
|
"""Empty trends list ingests without error and returns a well-formed report."""
|
|
from app.db import init_db
|
|
from app.market import ingest_trends
|
|
init_db()
|
|
|
|
report = ingest_trends([], "2026-05-02")
|
|
assert report["report_date"] == "2026-05-02"
|
|
assert report["top_genres"] == []
|
|
assert report["recommended_styles"] == []
|
|
assert isinstance(report["insights"], str)
|
|
assert len(report["insights"]) > 0
|