hackerthon-vote/tests/e2e.py

"""
E2E 테스트 — JSON 단일 파일 기반.
실행: docker exec hackathon-vote python3 /tmp/e2e.py
"""
import os
import sys
import tempfile
import shutil

# 격리 데이터 파일 (실제 hackathon.json과 분리)
TEST_DATA = tempfile.mktemp(suffix=".json")
shutil.copy("/app/hackathon.json", TEST_DATA)
os.environ["DATA_PATH"] = TEST_DATA
os.environ["ADMIN_TOKEN"] = "test"

sys.path.insert(0, "/app")

from app import (  # noqa: E402
    load_data,
    save_data,
    insert_vote,
    list_votes,
    clear_votes,
    get_participants,
    get_teams,
    compute_winners,
    save_tie_break,
    clear_tie_break,
    set_voting_open,
    is_voting_open,
    set_title,
    get_titles,
    archive_results,
)


def reset_votes():
    """투표만 초기화 (people, settings 보존)."""
    data = load_data()
    data["votes"] = []
    data["tie_breaks"] = {}
    data["titles"] = {}
    data["settings"]["voting_open"] = True
    save_data(data)


def add_vote(name, emp, team, fun, polish, utility):
    insert_vote(name, emp, team, {
        "fun_team": fun,
        "polish_team": polish,
        "utility_team": utility,
    })


PASS = "✅"
FAIL = "❌"
results = []


def test(name, fn):
    try:
        fn()
        results.append((PASS, name))
        print(f"{PASS} {name}")
    except AssertionError as e:
        results.append((FAIL, f"{name}: {e}"))
        print(f"{FAIL} {name}: {e}")
    except Exception as e:
        results.append((FAIL, f"{name}: UNEXPECTED {type(e).__name__}: {e}"))
        print(f"{FAIL} {name}: UNEXPECTED {type(e).__name__}: {e}")


def t_load():
    parts = get_participants()
    teams = get_teams()
    assert len(parts) == 34, f"기대 34명, 실제 {len(parts)}"
    assert len(teams) == 7
    assert "한지승" in parts
    assert "김태현" not in parts


def t_voting_toggle():
    reset_votes()
    assert is_voting_open() is True
    set_voting_open(False)
    assert is_voting_open() is False
    set_voting_open(True)
    assert is_voting_open() is True


def t_simple_winners():
    reset_votes()
    for i in range(5):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀1", "팀2", "팀3")
    winners, _ = compute_winners()
    assert winners["fun_team"]["status"] == "ok"
    assert winners["fun_team"]["team"] == "팀1"
    assert winners["polish_team"]["team"] == "팀2"
    assert winners["utility_team"]["team"] == "팀3"


def t_priority_one_team_all_first():
    reset_votes()
    for i in range(5):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀1", "팀1", "팀1")
    for i in range(5, 8):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀2", "팀2", "팀1")
    for i in range(8, 10):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀3", "팀3", "팀2")
    winners, _ = compute_winners()
    assert winners["utility_team"]["team"] == "팀1"
    assert winners["polish_team"]["team"] == "팀2"
    assert winners["fun_team"]["team"] == "팀3"


def t_tie_pending():
    reset_votes()
    for i in range(3):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀5", "팀6", "팀1")
    for i in range(3, 6):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀5", "팀6", "팀2")
    winners, _ = compute_winners()
    assert winners["utility_team"]["status"] == "tie"
    assert set(winners["utility_team"]["tied"]) == {"팀1", "팀2"}


def t_tie_break_random():
    reset_votes()
    for i in range(3):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀5", "팀6", "팀1")
    for i in range(3, 6):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀5", "팀6", "팀2")
    save_tie_break("utility_team", "팀1", "random")
    winners, _ = compute_winners()
    assert winners["utility_team"]["status"] == "ok"
    assert winners["utility_team"]["team"] == "팀1"
    assert winners["utility_team"]["method"] == "random"
    clear_tie_break("utility_team")


def t_unique_voter():
    reset_votes()
    add_vote("홍길동", "E1", "팀1", "팀2", "팀3", "팀4")
    try:
        add_vote("홍길동", "E2", "팀1", "팀5", "팀6", "팀7")
        raise AssertionError("중복 INSERT 통과")
    except ValueError as e:
        assert str(e) == "DUPLICATE_VOTER"


def t_titles_persist():
    reset_votes()
    set_title("팀1", "Slack 자동 분류기")
    titles = get_titles()
    assert titles["팀1"] == "Slack 자동 분류기"


def t_archive_writes_file():
    reset_votes()
    for i in range(5):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀1", "팀2", "팀3")
    archive_dir = os.path.dirname(TEST_DATA) or "."
    before = set(f for f in os.listdir(archive_dir) if f.startswith("results_"))
    path = archive_results()
    assert path is not None, "archive None 반환"
    assert os.path.exists(path)
    after = set(f for f in os.listdir(archive_dir) if f.startswith("results_"))
    assert len(after) > len(before)
    os.unlink(path)


def t_archive_skip_pending():
    reset_votes()
    for i in range(3):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀5", "팀6", "팀1")
    for i in range(3, 6):
        add_vote(f"V{i}", f"E{i}", "팀7", "팀5", "팀6", "팀2")
    path = archive_results()
    assert path is None


def t_atomic_write():
    """save_data 후 load_data가 동일 데이터 반환."""
    reset_votes()
    data = load_data()
    data["titles"]["팀1"] = "test atomic"
    save_data(data)
    fresh = load_data()
    assert fresh["titles"]["팀1"] == "test atomic"


def t_vote_count():
    reset_votes()
    assert len(list_votes()) == 0
    add_vote("A", "1", "팀1", "팀2", "팀3", "팀4")
    assert len(list_votes()) == 1
    clear_votes()
    assert len(list_votes()) == 0


if __name__ == "__main__":
    print(f"# E2E (data={TEST_DATA})\n")
    test("hackathon.json 로드 (34명, 7팀)", t_load)
    test("투표 마감 토글", t_voting_toggle)
    test("단순 winner", t_simple_winners)
    test("priority 1팀 모든 분야 1위", t_priority_one_team_all_first)
    test("동률 status='tie'", t_tie_pending)
    test("tie_break 적용", t_tie_break_random)
    test("voter_name UNIQUE 강제", t_unique_voter)
    test("팀 제목 영속", t_titles_persist)
    test("archive 파일 생성", t_archive_writes_file)
    test("archive 동률 skip", t_archive_skip_pending)
    test("atomic write", t_atomic_write)
    test("clear_votes", t_vote_count)

    fails = sum(1 for r, _ in results if r == FAIL)
    print(f"\n# {len(results)} 중 통과 {len(results) - fails}, 실패 {fails}")
    os.unlink(TEST_DATA)
    sys.exit(0 if fails == 0 else 1)