gahusb
2a8635e9ed
- backend/ → lotto/ 디렉토리 이동 - docker-compose: lotto-backend→lotto, lotto-frontend→frontend - deploy scripts, nginx, agent-office config 네이밍 일괄 반영 - lotto/app/db.py에서 todos·blog_posts CREATE TABLE 제거 (personal로 이관 완료) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
310 lines
11 KiB
Python
310 lines
11 KiB
Python
# backend/tests/test_purchase_manager.py
|
|
import sys, os
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
|
|
# Also insert the backend root so that "backend.app" package is importable
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
|
|
|
|
import sqlite3
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
# ":memory:" 공유 커넥션 — 각 테스트에서 독립적으로 생성
|
|
def _make_mem_conn():
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def test_purchase_history_has_new_columns():
|
|
"""purchase_history 테이블에 신규 컬럼이 존재하는지 검증"""
|
|
import db
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
|
|
cols = {r["name"] for r in mem.execute("PRAGMA table_info(purchase_history)").fetchall()}
|
|
assert "numbers" in cols
|
|
assert "is_real" in cols
|
|
assert "source_strategy" in cols
|
|
assert "source_detail" in cols
|
|
assert "checked" in cols
|
|
assert "results" in cols
|
|
assert "total_prize" in cols
|
|
# 기존 컬럼도 유지
|
|
assert "draw_no" in cols
|
|
assert "amount" in cols
|
|
assert "sets" in cols
|
|
assert "prize" in cols
|
|
assert "note" in cols
|
|
mem.close()
|
|
|
|
|
|
def test_strategy_performance_table_exists():
|
|
"""strategy_performance 테이블이 생성되는지 검증"""
|
|
import db
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
|
|
cols = {r["name"] for r in mem.execute("PRAGMA table_info(strategy_performance)").fetchall()}
|
|
assert "strategy" in cols
|
|
assert "draw_no" in cols
|
|
assert "sets_count" in cols
|
|
assert "total_correct" in cols
|
|
assert "avg_score" in cols
|
|
mem.close()
|
|
|
|
|
|
def test_strategy_weights_table_exists():
|
|
"""strategy_weights 테이블이 생성되고 초기값이 있는지 검증"""
|
|
import db
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
|
|
rows = mem.execute("SELECT * FROM strategy_weights ORDER BY strategy").fetchall()
|
|
strategies = {r["strategy"] for r in rows}
|
|
assert strategies == {"combined", "simulation", "heatmap", "manual", "custom"}
|
|
# 가중치 합이 1.0
|
|
total_weight = sum(r["weight"] for r in rows)
|
|
assert abs(total_weight - 1.0) < 0.01
|
|
mem.close()
|
|
|
|
|
|
def test_add_purchase_with_numbers():
|
|
"""번호 포함 구매 등록"""
|
|
import db
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
result = db.add_purchase(
|
|
draw_no=1150,
|
|
amount=5000,
|
|
sets=5,
|
|
numbers=[[1, 2, 3, 4, 5, 6], [7, 8, 9, 10, 11, 12]],
|
|
is_real=False,
|
|
source_strategy="simulation",
|
|
source_detail={"run_id": 42},
|
|
)
|
|
assert result["draw_no"] == 1150
|
|
assert result["amount"] == 5000
|
|
assert result["is_real"] == 0
|
|
assert result["source_strategy"] == "simulation"
|
|
assert result["numbers"] == [[1, 2, 3, 4, 5, 6], [7, 8, 9, 10, 11, 12]]
|
|
assert result["source_detail"] == {"run_id": 42}
|
|
mem.close()
|
|
|
|
|
|
def test_get_purchases_filter_is_real():
|
|
"""is_real 필터 동작"""
|
|
import db
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
db.add_purchase(draw_no=1150, amount=5000, sets=5, is_real=True)
|
|
db.add_purchase(draw_no=1150, amount=1000, sets=1, is_real=False)
|
|
real_only = db.get_purchases(is_real=True)
|
|
virtual_only = db.get_purchases(is_real=False)
|
|
assert len(real_only) == 1
|
|
assert real_only[0]["is_real"] == 1
|
|
assert len(virtual_only) == 1
|
|
assert virtual_only[0]["is_real"] == 0
|
|
mem.close()
|
|
|
|
|
|
def test_get_purchase_stats_by_type():
|
|
"""실제/가상 분리 통계"""
|
|
import db
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
db.add_purchase(draw_no=1150, amount=5000, sets=5, is_real=True, source_strategy="manual")
|
|
db.add_purchase(draw_no=1150, amount=1000, sets=1, is_real=False, source_strategy="simulation")
|
|
stats = db.get_purchase_stats()
|
|
assert "total" in stats
|
|
assert "real" in stats
|
|
assert "virtual" in stats
|
|
assert "by_strategy" in stats
|
|
assert stats["total"]["sets"] == 6
|
|
assert stats["real"]["sets"] == 5
|
|
assert stats["virtual"]["sets"] == 1
|
|
assert "manual" in stats["by_strategy"]
|
|
assert "simulation" in stats["by_strategy"]
|
|
# 하위호환 필드
|
|
assert "total_records" in stats
|
|
assert stats["total_records"] == 2
|
|
mem.close()
|
|
|
|
|
|
def test_upsert_strategy_performance():
|
|
"""전략 성과 upsert"""
|
|
import db
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
# 최초 insert
|
|
db.upsert_strategy_performance(
|
|
strategy="simulation",
|
|
draw_no=1150,
|
|
sets_count=10,
|
|
total_correct=30,
|
|
max_correct=5,
|
|
prize_total=5000,
|
|
avg_score=3.0,
|
|
)
|
|
rows = db.get_strategy_performance(strategy="simulation")
|
|
assert len(rows) == 1
|
|
assert rows[0]["sets_count"] == 10
|
|
assert rows[0]["avg_score"] == 3.0
|
|
# upsert (동일 strategy+draw_no)
|
|
db.upsert_strategy_performance(
|
|
strategy="simulation",
|
|
draw_no=1150,
|
|
sets_count=20,
|
|
total_correct=60,
|
|
max_correct=6,
|
|
prize_total=10000,
|
|
avg_score=4.5,
|
|
)
|
|
rows = db.get_strategy_performance(strategy="simulation")
|
|
assert len(rows) == 1 # 중복 없이 1개
|
|
assert rows[0]["sets_count"] == 20
|
|
assert rows[0]["avg_score"] == 4.5
|
|
mem.close()
|
|
|
|
|
|
def test_update_strategy_weight():
|
|
"""전략 가중치 업데이트"""
|
|
import db
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
# 초기값 확인
|
|
weights_before = db.get_strategy_weights()
|
|
combined_before = next(w for w in weights_before if w["strategy"] == "combined")
|
|
original_weight = combined_before["weight"]
|
|
# 업데이트
|
|
db.update_strategy_weight(
|
|
strategy="combined",
|
|
weight=0.5,
|
|
ema_score=0.75,
|
|
total_sets=100,
|
|
total_hits_3plus=20,
|
|
)
|
|
weights_after = db.get_strategy_weights()
|
|
combined_after = next(w for w in weights_after if w["strategy"] == "combined")
|
|
assert combined_after["weight"] == 0.5
|
|
assert combined_after["ema_score"] == 0.75
|
|
assert combined_after["total_sets"] == 100
|
|
assert combined_after["total_hits_3plus"] == 20
|
|
mem.close()
|
|
|
|
|
|
# ── purchase_manager 테스트 ───────────────────────────────────────────────────
|
|
|
|
def _import_purchase_manager_with_mem(mem_conn):
|
|
"""purchase_manager를 메모리 DB에 연결된 상태로 임포트."""
|
|
import db
|
|
import importlib
|
|
# backend.app 패키지로 로드해 상대 임포트가 동작하게 함
|
|
import backend.app.purchase_manager as pm
|
|
return pm
|
|
|
|
|
|
def test_check_purchases_for_draw():
|
|
"""특정 회차 구매 건들의 결과 체크"""
|
|
import db
|
|
import backend.app.purchase_manager as pm
|
|
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
# 당첨번호 삽입: 1125회 [3,12,23,34,38,45] bonus=7
|
|
mem.execute(
|
|
"""INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(1125, "2024-12-01", 3, 12, 23, 34, 38, 45, 7),
|
|
)
|
|
mem.commit()
|
|
|
|
# 구매 등록: 1등 번호 세트 + 낙첨 세트
|
|
purchase = db.add_purchase(
|
|
draw_no=1125,
|
|
amount=2000,
|
|
sets=2,
|
|
numbers=[[3, 12, 23, 34, 38, 45], [1, 2, 3, 4, 5, 6]],
|
|
is_real=False,
|
|
source_strategy="simulation",
|
|
)
|
|
|
|
with patch("db._conn", return_value=mem), \
|
|
patch("backend.app.purchase_manager.get_draw", side_effect=lambda drw: db.get_draw(drw)), \
|
|
patch("backend.app.purchase_manager.get_purchases", side_effect=lambda **kw: db.get_purchases(**kw)), \
|
|
patch("backend.app.purchase_manager.update_purchase_results", side_effect=lambda *a, **kw: db.update_purchase_results(*a, **kw)), \
|
|
patch("backend.app.purchase_manager.upsert_strategy_performance", side_effect=lambda **kw: db.upsert_strategy_performance(**kw)):
|
|
count = pm.check_purchases_for_draw(1125)
|
|
|
|
assert count == 1
|
|
|
|
# 결과 확인
|
|
with patch("db._conn", return_value=mem):
|
|
checked = db.get_purchases(draw_no=1125, checked=True)
|
|
assert len(checked) == 1
|
|
results = checked[0]["results"]
|
|
assert results is not None
|
|
assert len(results) == 2
|
|
# 첫 번째 세트: 6개 일치 → 1등
|
|
assert results[0]["rank"] == 1
|
|
assert results[0]["correct"] == 6
|
|
# 두 번째 세트: 3 하나만 일치 → 낙첨(correct=1)
|
|
assert results[1]["rank"] == 0
|
|
assert results[1]["correct"] == 1
|
|
|
|
mem.close()
|
|
|
|
|
|
def test_check_purchases_updates_strategy_performance():
|
|
"""결과 체크 후 strategy_performance가 갱신되는지 검증"""
|
|
import db
|
|
import backend.app.purchase_manager as pm
|
|
|
|
mem = _make_mem_conn()
|
|
with patch("db._conn", return_value=mem):
|
|
db.init_db()
|
|
# 당첨번호 삽입: 1126회
|
|
mem.execute(
|
|
"""INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(1126, "2024-12-08", 1, 2, 3, 4, 5, 6, 7),
|
|
)
|
|
mem.commit()
|
|
|
|
db.add_purchase(
|
|
draw_no=1126,
|
|
amount=5000,
|
|
sets=5,
|
|
numbers=[[1, 2, 3, 4, 5, 6], [10, 20, 30, 40, 41, 42]],
|
|
is_real=False,
|
|
source_strategy="simulation",
|
|
)
|
|
|
|
with patch("db._conn", return_value=mem), \
|
|
patch("backend.app.purchase_manager.get_draw", side_effect=lambda drw: db.get_draw(drw)), \
|
|
patch("backend.app.purchase_manager.get_purchases", side_effect=lambda **kw: db.get_purchases(**kw)), \
|
|
patch("backend.app.purchase_manager.update_purchase_results", side_effect=lambda *a, **kw: db.update_purchase_results(*a, **kw)), \
|
|
patch("backend.app.purchase_manager.upsert_strategy_performance", side_effect=lambda **kw: db.upsert_strategy_performance(**kw)):
|
|
count = pm.check_purchases_for_draw(1126)
|
|
|
|
assert count == 1
|
|
|
|
with patch("db._conn", return_value=mem):
|
|
perf = db.get_strategy_performance(strategy="simulation")
|
|
|
|
assert len(perf) >= 1
|
|
entry = next((p for p in perf if p["draw_no"] == 1126), None)
|
|
assert entry is not None, "draw_no=1126 에 대한 strategy_performance 없음"
|
|
assert entry["strategy"] == "simulation"
|
|
assert entry["sets_count"] == 2 # 2개 세트
|
|
|
|
mem.close()
|