web-page-backend/backend/tests/test_purchase_manager.py

# 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()