"""banner-versions API 테스트. GET /api/banner-versions 와 POST /api/banner-versions/select 엔드포인트를 임시 HTTP 서버로 띄워 검증합니다. """ import json import os import socketserver import sys import threading import urllib.error import urllib.request from pathlib import Path import pytest _WORKSPACE = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(_WORKSPACE) not in sys.path: sys.path.insert(0, str(_WORKSPACE)) # --------------------------------------------------------------------------- # 테스트 데이터 # --------------------------------------------------------------------------- _BV_DATA = { "versions": { "v1456": { "task_id": "task-1456.1", "label": "버전 C (task-1456.1)", "description": "Cycle 2 수정", "available": True, "dq_scores": {"cell-1-test": 90, "cell-2-test": 85}, }, "v1460": { "task_id": "task-1460.1", "label": "피드백 수정 버전", "description": "피드백 기반 수정", "available": True, "dq_scores": {}, }, }, "cells": [ { "id": "cell-1-test", "label": "Cell 1", "brand": "test", "concept": "test-concept", }, { "id": "cell-2-test", "label": "Cell 2", "brand": "test", "concept": "test-concept2", }, ], "selections": {}, } _BANNER_FILES = [ "meta-feed-1080x1080.html", "meta-feed-1080x1080.png", "google-resp-1200x628.html", "google-resp-1200x628.png", ] # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def banner_test_env(tmp_path): """격리된 배너 테스트 환경을 구성합니다.""" # 1. banner-versions.json 생성 data_dir = tmp_path / "dashboard" / "data" data_dir.mkdir(parents=True) bv_file = data_dir / "banner-versions.json" bv_file.write_text( json.dumps(_BV_DATA, ensure_ascii=False, indent=2), encoding="utf-8" ) # 2. versions 디렉토리 + 더미 파일 생성 for version in ["v1456", "v1460"]: for cell in ["cell-1-test", "cell-2-test"]: cell_dir = ( tmp_path / "output" / "banners" / "versions" / version / cell ) cell_dir.mkdir(parents=True) for fname in _BANNER_FILES: (cell_dir / fname).write_text( f"dummy-{version}-{cell}-{fname}", encoding="utf-8" ) # 3. output/banners 디렉토리 (선택 복사 대상) (tmp_path / "output" / "banners").mkdir(parents=True, exist_ok=True) # 4. DataLoader가 reload_all() 시 참조하는 디렉토리 최소 생성 (tmp_path / "memory" / "logs").mkdir(parents=True, exist_ok=True) return tmp_path, bv_file @pytest.fixture def test_server(banner_test_env, monkeypatch): """배너 API 테스트용 임시 HTTP 서버를 띄웁니다.""" tmp_path, _ = banner_test_env import dashboard.server as server_mod # type: ignore[import-not-found] from dashboard.data_loader import DataLoader # type: ignore[import-not-found] from dashboard.server import DashboardHandler # type: ignore[import-not-found] # DataLoader를 tmp_path 기반으로 생성 loader = DataLoader(tmp_path) loader.reload_all() # DashboardHandler.__file__ 을 tmp 경로로 패치 # → Path(__file__).parent = tmp_path / "dashboard" # → banner-versions.json = tmp_path / "dashboard" / "data" / "banner-versions.json" monkeypatch.setattr( server_mod, "__file__", str(tmp_path / "dashboard" / "server.py") ) # routes_post.__file__ 도 tmp 경로로 패치 # → handle_post_banner_versions_select 내 Path(__file__).parent 가 tmp_path/dashboard 를 가리키게 함 import dashboard.routes_post as routes_post_mod monkeypatch.setattr( routes_post_mod, "__file__", str(tmp_path / "dashboard" / "routes_post.py") ) # routes_get.__file__ 도 tmp 경로로 패치 # → handle_get_banner_versions 내 Path(__file__).parent 가 tmp_path/dashboard 를 가리키게 함 import dashboard.routes_get as routes_get_mod monkeypatch.setattr( routes_get_mod, "__file__", str(tmp_path / "dashboard" / "routes_get.py") ) # 서브클래스로 data_loader와 dashboard_dir을 tmp 경로로 고정 class TestBannerHandler(DashboardHandler): def __init__(self, *args, **kwargs): # SimpleHTTPRequestHandler 생성자에게 directory를 직접 지정 # DashboardHandler.__init__ 의 self.dashboard_dir 을 우회 self.dashboard_dir = tmp_path / "dashboard" http_server_base = __import__("http.server", fromlist=["SimpleHTTPRequestHandler"]) http_server_base.SimpleHTTPRequestHandler.__init__( self, *args, directory=str(tmp_path), **kwargs ) TestBannerHandler.data_loader = loader # type: ignore[attr-defined] socketserver.ThreadingTCPServer.allow_reuse_address = True with socketserver.ThreadingTCPServer( ("127.0.0.1", 0), TestBannerHandler ) as httpd: actual_port = httpd.server_address[1] thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() yield f"http://127.0.0.1:{actual_port}", tmp_path httpd.shutdown() # --------------------------------------------------------------------------- # HTTP 유틸 # --------------------------------------------------------------------------- def _get(url: str): """GET 요청 → (status, body_dict)""" req = urllib.request.Request(url) with urllib.request.urlopen(req) as resp: return resp.status, json.loads(resp.read()) def _post(url: str, data: dict): """POST 요청 → (status, body_dict) — 4xx/5xx 도 파싱해서 반환""" body = json.dumps(data).encode() req = urllib.request.Request( url, data=body, headers={"Content-Type": "application/json"} ) try: with urllib.request.urlopen(req) as resp: return resp.status, json.loads(resp.read()) except urllib.error.HTTPError as e: return e.code, json.loads(e.read()) # --------------------------------------------------------------------------- # 테스트 # --------------------------------------------------------------------------- class TestBannerVersionsAPI: """Banner versions API 테스트.""" # ------------------------------------------------------------------ # 1. GET /api/banner-versions — 정상 응답 + 구조 검증 # ------------------------------------------------------------------ def test_get_banner_versions_success(self, test_server): """GET /api/banner-versions — 정상 응답 + 구조 검증.""" base_url, _ = test_server status, body = _get(f"{base_url}/api/banner-versions") assert status == 200, f"예상 200, 실제 {status}: {body}" # 최상위 키 존재 확인 assert "versions" in body, "응답에 'versions' 키 없음" assert "cells" in body, "응답에 'cells' 키 없음" assert "selections" in body, "응답에 'selections' 키 없음" # versions 구조 versions = body["versions"] assert isinstance(versions, list), "'versions' 는 리스트여야 함" assert len(versions) >= 2, f"버전이 최소 2개여야 함, 실제: {len(versions)}" version_ids = {v["id"] for v in versions} assert "v1456" in version_ids, "v1456 버전이 응답에 없음" assert "v1460" in version_ids, "v1460 버전이 응답에 없음" for v in versions: for key in ("id", "label", "description", "available", "cells"): assert key in v, f"version 항목에 '{key}' 키 없음: {v}" # cells 구조 cells = body["cells"] assert isinstance(cells, list), "'cells' 는 리스트여야 함" assert len(cells) >= 2, f"셀이 최소 2개여야 함, 실제: {len(cells)}" for c in cells: assert "id" in c, f"cell 항목에 'id' 키 없음: {c}" assert "name" in c, f"cell 항목에 'name' 키 없음: {c}" # ------------------------------------------------------------------ # 2a. POST /api/banner-versions/select — v1456 선택 성공 # ------------------------------------------------------------------ def test_post_select_v1456_success(self, test_server): """POST /api/banner-versions/select — v1456 선택 성공.""" base_url, tmp_path = test_server payload = {"cell_id": "cell-1-test", "version": "v1456"} status, body = _post(f"{base_url}/api/banner-versions/select", payload) assert status == 200, f"예상 200, 실제 {status}: {body}" assert body.get("status") == "ok" assert body.get("cell_id") == "cell-1-test" assert body.get("version") == "v1456" # 파일이 복사되었는지 확인 dst_dir = tmp_path / "output" / "banners" / "cell-1-test" for fname in _BANNER_FILES: assert (dst_dir / fname).exists(), f"복사된 파일 없음: {dst_dir / fname}" # ------------------------------------------------------------------ # 2b. POST /api/banner-versions/select — v1460 선택 성공 # ------------------------------------------------------------------ def test_post_select_v1460_success(self, test_server): """POST /api/banner-versions/select — v1460 선택 성공.""" base_url, tmp_path = test_server payload = {"cell_id": "cell-2-test", "version": "v1460"} status, body = _post(f"{base_url}/api/banner-versions/select", payload) assert status == 200, f"예상 200, 실제 {status}: {body}" assert body.get("status") == "ok" assert body.get("cell_id") == "cell-2-test" assert body.get("version") == "v1460" # 복사 확인 dst_dir = tmp_path / "output" / "banners" / "cell-2-test" for fname in _BANNER_FILES: assert (dst_dir / fname).exists(), f"복사된 파일 없음: {dst_dir / fname}" # ------------------------------------------------------------------ # 3. POST 후 GET — 선택 상태 유지 확인 # ------------------------------------------------------------------ def test_post_then_get_selection_persisted(self, test_server): """POST 후 GET — selections에 반영 확인.""" base_url, _ = test_server # 선택 post_status, post_body = _post( f"{base_url}/api/banner-versions/select", {"cell_id": "cell-1-test", "version": "v1460"}, ) assert post_status == 200, f"POST 실패: {post_body}" # 조회 get_status, get_body = _get(f"{base_url}/api/banner-versions") assert get_status == 200, f"GET 실패: {get_body}" selections = get_body.get("selections", {}) assert selections.get("cell-1-test") == "v1460", ( f"selections 에 선택 상태 반영 안 됨: {selections}" ) # ------------------------------------------------------------------ # 4. 존재하지 않는 버전 선택 시 400 에러 # ------------------------------------------------------------------ def test_post_select_invalid_version(self, test_server): """존재하지 않는 버전 선택 시 400 에러.""" base_url, _ = test_server status, body = _post( f"{base_url}/api/banner-versions/select", {"cell_id": "cell-1-test", "version": "v-nonexistent"}, ) assert status == 400, f"예상 400, 실제 {status}: {body}" assert "error" in body, f"응답에 'error' 키 없음: {body}" assert "Invalid version" in body["error"], ( f"에러 메시지에 'Invalid version' 없음: {body['error']}" ) # ------------------------------------------------------------------ # 5. 잘못된 cell_id 선택 시 400 에러 # ------------------------------------------------------------------ def test_post_select_invalid_cell_id(self, test_server): """잘못된 cell_id 선택 시 400 에러.""" base_url, _ = test_server status, body = _post( f"{base_url}/api/banner-versions/select", {"cell_id": "cell-invalid", "version": "v1456"}, ) assert status == 400, f"예상 400, 실제 {status}: {body}" assert "error" in body, f"응답에 'error' 키 없음: {body}" assert "Invalid cell_id" in body["error"], ( f"에러 메시지에 'Invalid cell_id' 없음: {body['error']}" )