360 lines
11 KiB
Python
360 lines
11 KiB
Python
"""
|
|
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")
|
|
SEED_CANDIDATES = ["/app/data/hackathon.json", "/app/hackathon.json"]
|
|
seed = next((p for p in SEED_CANDIDATES if os.path.exists(p)), None)
|
|
if seed is None:
|
|
raise SystemExit(f"seed 파일 없음: {SEED_CANDIDATES}")
|
|
shutil.copy(seed, 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
|
|
|
|
|
|
def t_empty_state_has_topics_and_stage():
|
|
from app import _empty_state
|
|
s = _empty_state()
|
|
assert s["settings"]["current_stage"] == "intro"
|
|
assert s["topics"] == {"categories": []}
|
|
|
|
|
|
def t_load_data_backfills_nested_settings():
|
|
"""기존 settings에 voting_open만 있을 때 current_stage 자동 보강."""
|
|
legacy = {
|
|
"people": [],
|
|
"settings": {"voting_open": True},
|
|
"titles": {},
|
|
"tie_breaks": {},
|
|
"votes": [],
|
|
}
|
|
save_data(legacy)
|
|
fresh = load_data()
|
|
assert fresh["settings"]["voting_open"] is True
|
|
assert fresh["settings"]["current_stage"] == "intro"
|
|
assert fresh["topics"] == {"categories": []}
|
|
|
|
|
|
def t_stage_helpers():
|
|
from app import get_stage, set_stage, can_accept_votes, load_data, save_data, _empty_state
|
|
save_data(_empty_state())
|
|
|
|
assert get_stage() == "intro"
|
|
assert can_accept_votes(load_data()) is False # intro → False
|
|
|
|
set_stage("vote")
|
|
d = load_data()
|
|
assert d["settings"]["current_stage"] == "vote"
|
|
assert d["settings"]["voting_open"] is True
|
|
assert can_accept_votes(d) is True
|
|
|
|
set_stage("topics")
|
|
d = load_data()
|
|
assert d["settings"]["current_stage"] == "topics"
|
|
assert d["settings"]["voting_open"] is True # 떠나도 voting_open 유지
|
|
assert can_accept_votes(d) is False # stage 다르면 False
|
|
|
|
# 명시적 마감 후 vote 다시 들어오면 voting_open 자동 True
|
|
d["settings"]["voting_open"] = False
|
|
save_data(d)
|
|
set_stage("vote")
|
|
assert can_accept_votes(load_data()) is True
|
|
|
|
# invalid stage → ValueError
|
|
try:
|
|
set_stage("invalid_stage")
|
|
raise AssertionError("expected ValueError")
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
def t_topics_helpers():
|
|
from app import get_topics, update_topics, load_data, save_data, _empty_state
|
|
save_data(_empty_state())
|
|
|
|
assert get_topics() == []
|
|
|
|
sample = [
|
|
{"id": "T1", "title": "테스트", "tagline": "tl", "tone": "tn",
|
|
"items": [f"item{i}" for i in range(10)]}
|
|
]
|
|
update_topics(sample)
|
|
assert get_topics() == sample
|
|
|
|
# atomic 갱신 확인
|
|
sample2 = [{"id": "T1", "title": "교체", "tagline": "tl", "tone": "tn",
|
|
"items": [f"x{i}" for i in range(10)]}]
|
|
update_topics(sample2)
|
|
assert get_topics() == sample2
|
|
|
|
|
|
def t_topics_seeded_after_assign():
|
|
from app import get_topics, _empty_state, save_data, load_data
|
|
# 빈 상태로 reset
|
|
save_data(_empty_state())
|
|
assert get_topics() == []
|
|
|
|
from assign_teams import ensure_topics_seeded
|
|
data = load_data()
|
|
ensure_topics_seeded(data)
|
|
save_data(data)
|
|
|
|
cats = get_topics()
|
|
assert len(cats) == 4
|
|
for c in cats:
|
|
assert len(c["items"]) == 10
|
|
assert c["id"] in ("T1", "T2", "T3", "T4")
|
|
assert c["title"]
|
|
assert c["tagline"]
|
|
assert c["tone"]
|
|
|
|
|
|
def t_make_qr_png():
|
|
from app import make_qr_png
|
|
png = make_qr_png("http://localhost:8501/?mode=vote")
|
|
# PNG signature 8 bytes
|
|
assert png[:8] == b"\x89PNG\r\n\x1a\n"
|
|
assert len(png) > 100
|
|
|
|
|
|
def t_compute_vote_url_priority():
|
|
import os as _os
|
|
from app import compute_vote_url, save_data, load_data, _empty_state
|
|
save_data(_empty_state())
|
|
|
|
# 1. settings.public_base_url 우선
|
|
d = load_data()
|
|
d["settings"]["public_base_url"] = "http://example.com:9000"
|
|
save_data(d)
|
|
assert compute_vote_url() == "http://example.com:9000/?mode=vote"
|
|
|
|
# 2. env fallback
|
|
d["settings"].pop("public_base_url")
|
|
save_data(d)
|
|
_os.environ["PUBLIC_BASE_URL"] = "http://env-host:7777"
|
|
assert compute_vote_url() == "http://env-host:7777/?mode=vote"
|
|
_os.environ.pop("PUBLIC_BASE_URL")
|
|
|
|
# 3. localhost / LAN fallback
|
|
url = compute_vote_url()
|
|
assert url.endswith("/?mode=vote")
|
|
assert url.startswith("http://")
|
|
|
|
|
|
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)
|
|
test("empty_state 신규 키", t_empty_state_has_topics_and_stage)
|
|
test("load_data nested 키 backfill", t_load_data_backfills_nested_settings)
|
|
test("stage 헬퍼", t_stage_helpers)
|
|
test("topics 헬퍼", t_topics_helpers)
|
|
test("topics 시드", t_topics_seeded_after_assign)
|
|
test("QR PNG 생성", t_make_qr_png)
|
|
test("vote URL 우선순위", t_compute_vote_url_priority)
|
|
|
|
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)
|