gahusb
d66a321982
YouTube 음악 파이프라인 Task 1 — 신규 5개 테이블과 헬퍼 함수 추가. 테이블: - video_pipelines: 파이프라인 단위 상태 머신 + 메타/리뷰 JSON - pipeline_jobs: 단계별 비동기 작업 상태/시간 - pipeline_feedback: 텔레그램 피드백 이력 - youtube_oauth_tokens: 채널 OAuth refresh/access 토큰 - youtube_setup: 단일 행 설정 (메타 템플릿/커버 프롬프트/리뷰 가중치/임계값/비주얼/공개정책) 헬퍼: - create_pipeline / get_pipeline / update_pipeline_state / list_pipelines - increment_feedback_count / record_feedback / get_feedback_history - create_pipeline_job / update_pipeline_job / list_pipeline_jobs - get_youtube_setup / update_youtube_setup - upsert_oauth_token / get_oauth_token / delete_oauth_token 테스트: - tests/test_pipeline_db.py: 7개 테스트 (생성/상태/피드백/잡/셋업/OAuth) - tests/conftest.py: freezegun 기반 freezer fixture 추가 - requirements.txt: freezegun>=1.4 추가 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
import os
|
|
import tempfile
|
|
import pytest
|
|
from app import db
|
|
|
|
|
|
@pytest.fixture
|
|
def fresh_db(monkeypatch, tmp_path):
|
|
db_path = tmp_path / "music.db"
|
|
monkeypatch.setattr(db, "DB_PATH", str(db_path))
|
|
db.init_db()
|
|
return db_path
|
|
|
|
|
|
def test_create_pipeline_inserts_row(fresh_db):
|
|
pid = db.create_pipeline(track_id=1)
|
|
row = db.get_pipeline(pid)
|
|
assert row["id"] == pid
|
|
assert row["state"] == "created"
|
|
assert row["track_id"] == 1
|
|
assert row["feedback_count_per_step"] == {}
|
|
|
|
|
|
def test_update_pipeline_state_records_started_at(fresh_db, freezer):
|
|
pid = db.create_pipeline(track_id=1)
|
|
freezer.move_to("2026-05-07T08:00:00")
|
|
db.update_pipeline_state(pid, "cover_pending")
|
|
row = db.get_pipeline(pid)
|
|
assert row["state"] == "cover_pending"
|
|
assert row["state_started_at"] == "2026-05-07T08:00:00"
|
|
|
|
|
|
def test_increment_feedback_count(fresh_db):
|
|
pid = db.create_pipeline(track_id=1)
|
|
db.increment_feedback_count(pid, "cover")
|
|
db.increment_feedback_count(pid, "cover")
|
|
row = db.get_pipeline(pid)
|
|
assert row["feedback_count_per_step"] == {"cover": 2}
|
|
|
|
|
|
def test_record_feedback(fresh_db):
|
|
pid = db.create_pipeline(track_id=1)
|
|
db.record_feedback(pid, "cover", "더 어둡게")
|
|
rows = db.get_feedback_history(pid)
|
|
assert len(rows) == 1
|
|
assert rows[0]["feedback_text"] == "더 어둡게"
|
|
|
|
|
|
def test_create_pipeline_job_lifecycle(fresh_db):
|
|
pid = db.create_pipeline(track_id=1)
|
|
job_id = db.create_pipeline_job(pid, "cover")
|
|
db.update_pipeline_job(job_id, status="running")
|
|
db.update_pipeline_job(job_id, status="succeeded", duration_ms=1234)
|
|
jobs = db.list_pipeline_jobs(pid)
|
|
assert jobs[0]["status"] == "succeeded"
|
|
assert jobs[0]["duration_ms"] == 1234
|
|
|
|
|
|
def test_youtube_setup_default_row_created_on_init(fresh_db):
|
|
setup = db.get_youtube_setup()
|
|
assert setup["review_threshold"] == 60
|
|
assert "metadata_template_json" in setup
|
|
|
|
|
|
def test_youtube_oauth_token_upsert(fresh_db):
|
|
db.upsert_oauth_token(
|
|
channel_id="UC123",
|
|
channel_title="My Channel",
|
|
avatar_url="https://...",
|
|
refresh_token="r1",
|
|
access_token="a1",
|
|
expires_at="2026-05-07T09:00:00",
|
|
)
|
|
tok = db.get_oauth_token()
|
|
assert tok["channel_id"] == "UC123"
|
|
assert tok["refresh_token"] == "r1"
|
|
db.upsert_oauth_token(
|
|
channel_id="UC123", channel_title="My Channel",
|
|
avatar_url=None, refresh_token="r2",
|
|
access_token="a2", expires_at="2026-05-07T10:00:00",
|
|
)
|
|
tok = db.get_oauth_token()
|
|
assert tok["refresh_token"] == "r2" # upsert
|
|
|
|
|
|
def test_update_pipeline_state_rejects_unknown_column(fresh_db):
|
|
pid = db.create_pipeline(track_id=1)
|
|
with pytest.raises(ValueError):
|
|
db.update_pipeline_state(pid, "cover_pending", evil_col="x; DROP TABLE")
|
|
|
|
|
|
def test_update_pipeline_job_rejects_unknown_column(fresh_db):
|
|
pid = db.create_pipeline(track_id=1)
|
|
job_id = db.create_pipeline_job(pid, "cover")
|
|
with pytest.raises(ValueError):
|
|
db.update_pipeline_job(job_id, evil_col="x")
|