"""테스트 전체에 적용되는 설정 — 프로덕션 DB/projects 디렉토리를 보호한다. ⚠️ CRITICAL — import 순서 (2026-04-27 hotfix, 데이터 손실 사고 직후 도입): 환경변수(`DATABASE_URL`, `PROJECTS_DIR`, `ENVIRONMENT`)는 모든 다른 import 보다 먼저 설정해야 한다. 그러지 않으면 다음 chain이 production 값을 캐시한다. from tests.conftest_pg import ... → from app.core.database import Base → from app.core.config import settings # ← settings = Settings() 인스턴스화 → engine = create_engine(settings.database_url, ...) # ← production engine 캐시 settings/engine은 module-level singleton이라 한번 인스턴스화되면 환경변수 재설정도 무시된다. 결과: autouse fixture가 `Base.metadata.drop_all(engine)`과 `shutil.rmtree(settings.projects_dir)`를 호출하면 production DB와 projects/ 폴더가 통째로 삭제됨. 따라서 이 파일에서는 stdlib import 후 즉시 환경변수를 설정하고, app 관련 import는 그 뒤로만. 새 import를 추가할 때는 반드시 환경변수 설정 라인 아래에 둘 것. ⚠️ DB 정책 (2026-04-27 hotfix): `feedback_postgresql_not_sqlite.md` — DB는 PostgreSQL. 테스트도 PG 사용. production(`theroad`)과 분리된 별도 PG DB(`theroad_test`)를 강제한다. DATABASE_URL_TEST 환경변수로 override 가능. """ # ── 1. stdlib만 우선 import (env 설정에 필요) ── import os import sys import tempfile from pathlib import Path # Machine-enforced 가드 (Claude PR #5 hotfix Issue 3): 미래 contributor가 실수로 # 이 위에 `from app.*` import를 추가하면 settings/engine이 production으로 캐시된다. # 그 시나리오를 fail-fast로 차단. _PRELOADED = {m for m in sys.modules if m.startswith("app.")} if _PRELOADED: raise RuntimeError( f"REFUSING test setup: app.* modules already imported before env block: " f"{sorted(_PRELOADED)}. tests/conftest.py 위에 app.* import를 추가하지 말 것 — " f"settings/engine이 production으로 캐시되어 production data wipe로 이어진다 " f"(2026-04-27 사고 회귀 가드)." ) # ── 2. env 강제 설정 — app/core/config·database import 보다 반드시 먼저 ── os.environ["ENVIRONMENT"] = "test" # ★★§2 통과 조건의 「`netprobe` 바깥 호출 **0**」을 막던 **유일한 한 건**. # litellm 이 **import 때** 요금표 JSON 을 GitHub 에서 가져온다 # (`litellm_core_utils/get_model_cost_map.py`). 유료 provider 호출은 아니지만 # 「바깥 호출 0」은 거짓이 된다 — 그래서 「거의 0」으로 넘기지 않고 닫는다. # litellm 이 스스로 주는 스위치다: 켜면 패키지에 든 표를 쓴다. os.environ.setdefault("LITELLM_LOCAL_MODEL_COST_MAP", "True") # 별도 PG test DB 강제. production `theroad`와 분리된 `theroad_test`를 default로 사용. # CI/local에서 DATABASE_URL_TEST를 override 가능. production DATABASE_URL과 같으면 fail-fast. _DEFAULT_TEST_DB_URL = "postgresql://theroad:theroad_dev_2026@localhost:5432/theroad_test" _test_db_url = os.environ.get("DATABASE_URL_TEST", _DEFAULT_TEST_DB_URL) # Sanity: test URL이 production env의 DATABASE_URL과 같으면 거부 (.env 캐시 방지). _prod_db_url = os.environ.get("DATABASE_URL", "") if _prod_db_url and _prod_db_url == _test_db_url: raise RuntimeError( f"REFUSING test setup: DATABASE_URL_TEST({_test_db_url!r}) == " f"production DATABASE_URL. 테스트는 분리된 DB여야 한다 (theroad vs theroad_test)." ) # production URL과 같은 database 이름(끝 segment)이 같으면도 거부 (실수 방어) if _prod_db_url and _prod_db_url.split("/")[-1] == _test_db_url.split("/")[-1]: raise RuntimeError( f"REFUSING test setup: DATABASE_URL_TEST and production DATABASE_URL share " f"the same database name ({_test_db_url.split('/')[-1]!r}). " f"분리된 별도 PG database name 필요." ) # Codex P1 fix: prod URL 덮어쓰기 전에 원본 보존. conftest_pg.py 의 # _assert_test_pg_url 이 이 원본과 비교 (덮어쓴 후의 DATABASE_URL 은 항상 # test URL 이라 false positive). DATABASE_URL_TEST 가 prod DSN 을 가리키는 # 케이스의 회귀 가드. if _prod_db_url: os.environ["_ORIGINAL_DATABASE_URL"] = _prod_db_url os.environ["DATABASE_URL"] = _test_db_url # PROJECTS_DIR은 임시 디렉토리로 강제 — 매 fixture rmtree가 production 폴더 안 건드리도록. _test_dir = tempfile.mkdtemp(prefix="theroad_test_") os.environ["PROJECTS_DIR"] = os.path.join(_test_dir, "test_projects") # 시험 기록을 프로덕션 감사 데이터에서 가른다 (2026-08-24). # 돈 가드(아래 2-b, netprobe)는 집 안 주소를 통과시킨다 — 요금이 없으니 옳다. # 그런데 자체 호스팅 Opik 이 바로 집 안 주소라, 시험이 프로덕션 프로젝트에 # **실제로 쓴다**. 실측: 프로덕션 trace 표집 4,200건 중 657건(15.6%)이 시험 # 기록이었다(prompt='xxxx'·'p'·'SAMPLE prompt'). 발견 경로는 # tests/services/image_capture/test_i2i_capture.py 가 진짜 record_provider_call # 을 부르는 것. # 막으면 자체 호스팅 Opik 이 죽으므로(2026-08-20 실측) **막지 않고 가른다**. # ★setdefault 가 아니라 강제 대입이다 — 기계 .env 의 값이 이기면 안 된다. TEST_OPIK_PROJECT = "theroad-scene-lab-test" os.environ["OPIK_PROJECT_NAME"] = TEST_OPIK_PROJECT # ── 2-b. 돈 가드 — 시험이 바깥으로 유료 호출을 못 내보내게 (2026-08-20) ── # # 위 2번은 2026-04-27 자료 삭제 사고 뒤 **데이터베이스와 프로젝트 디렉토리**를 # 시험용으로 강제하고 프로덕션과 같으면 기동을 거부한다. 자료는 그렇게 지키는데 # **돈은 안 지키고 있었다** — 제공자 열쇠가 그대로 살아 있고 기능 설정도 기계의 # `.env` 를 물어서, 서비스를 걷는 시험이 실제 유료 API 를 불렀다. # # 실측(2026-08-20, tests/netprobe.py 로 잰 값): `tests/` 한 바퀴에 **566건** # (OpenAI 560 · Gemini 6). 「열쇠 없이도 도는가」를 보려는 시험(`*_no_openai_key`) # 조차 실제로 호출을 내보내고 있었다. 파일마다 기능 설정을 끄는 방식으로는 못 # 막는다 — 프로젝트 영어 이름 생성(project_service)처럼 **끌 설정이 아예 없는** # 자리가 있기 때문이다. # # 그래서 자료 가드와 같은 모양으로 **기본 차단**을 둔다. 프로세스 안 호출 # (TestClient 등)과 집 안 주소(자체 호스팅 Opik)는 통과하고, 바깥으로 나가려 하면 # 그 자리에서 크게 실패한다 — 조용히 새지 않고 시험을 쓴 사람이 바로 안다. # 정말 바깥이 필요한 시험은 `@pytest.mark.allow_outbound` 를 붙인다. os.environ.setdefault("THEROAD_TEST_BLOCK_OUTBOUND", "1") # 막기만 하면 **느려진다** — 막힌 호출을 재시도 층이 여러 번 다시 보내고 그 # 사이마다 기다린다. 실측(tests/test_projects_api.py 12건): # 가드 꺼짐(종전) 68초 · 실제 유료 호출이 바깥으로 나감 # 가드 켜짐, 재시도 기본 298초 · 막히지만 4.4배 느려짐(재시도 폭풍) # 가드 켜짐, 재시도 0 14초 · 막히고 **5배 빨라짐** # 시험에서 재시도는 지킬 것이 없다 — 어차피 다시 보내도 막힌다. 재시도 자체를 # 확인하는 시험은 자기 값을 직접 덮어 쓰므로(예: test_grok_body_error_retry) # 이 기본값에 안 걸린다. os.environ.setdefault("LLM_MAX_RETRIES", "0") # ── 3. 이제 안전하게 다른 import 가능 ── import pytest # W3-1 Codex Critical: PostgreSQL lane fixture는 tests/conftest.py에서만 자동 발견. from tests.conftest_pg import pg_engine, pg_session # noqa: F401 (env 설정 후라야 안전) # D6 review M3: test DB schema ergonomics. # # Plan B5 결정: production 은 alembic-only ownership (`_migrations` block 미추가). # 그러나 test DB 는 alembic 미사용 — `init_db()` + `Base.metadata.create_all` # 패턴. fresh DB 에서는 새 D6 column (entity_canon.metadata_json) 자동 생성되지만, # 기존 stale test DB (이전 pytest 세션) 의 entity_canon 은 column 부재 → inspect- # only test 가 NoSuchColumn / ORM insert 가 NULL DEFAULT violation 으로 fail. # # session-scope autouse — 1회만 실행 (review iter2 I1): # 1) `Base.metadata.create_all(checkfirst=True)` — 누락 테이블 보장 (idempotent). # 2) idempotent ALTER 로 entity_canon.metadata_json 보강. # # `safe_drop_all` 을 쓰는 file (예: test_project_export_import.py) 은 자체 # `_setup_db()` 가 schema 보장하므로 전역 fixture 가 매번 보강할 필요 없음. # 동일 세션에서 그 file 이 inspect-only test 보다 먼저 돌면 cross-order 문제 # 가능 — 그 경우 inspect-only file 자체에 module-scope autouse 추가 (방어적). @pytest.fixture(scope="session", autouse=True) def _ensure_d6_test_db_schema(): """1회 — entity_canon metadata_json + image_asset 생성 영속화 컬럼 보장. create_all(checkfirst=True) 는 누락 *테이블* 만 만들고 기존 테이블에 *컬럼* 을 추가하지 않는다. stale `theroad_test` 의 기존 테이블에 신규 컬럼을 보강하기 위해 idempotent `ADD COLUMN IF NOT EXISTS` 를 친다(fresh DB 는 create_all 로 자동 포함). """ from sqlalchemy import text from sqlalchemy.exc import OperationalError, ProgrammingError from app.core.database import Base, engine as _engine # 모델 등록 — Base.metadata 에 모든 table class import 시점 보장. import app.models.catalog # noqa: F401 import app.models.project # noqa: F401 Base.metadata.create_all(_engine, checkfirst=True) # image_asset 생성 영속화 컬럼(Phase A) — stale test DB 재현성용 idempotent ALTER. _IMAGE_ASSET_GEN_COLS = ( "input_image_ids TEXT", "stage TEXT", "pipeline_role TEXT", "is_intermediate BOOLEAN NOT NULL DEFAULT false", "generation_call_id TEXT", "candidate_index INTEGER", "pipeline_metadata_json TEXT", # alembic 009 — true-intermediate 생애 메타. "disposition TEXT", "attempt_index INTEGER", ) # ── 여러 편 처리 (alembic 014) ──────────────────────────────────── # ★계약이 세 곳에 적혀 있다: alembic 014 · `database.py::_migrations` · # 여기. 한 곳만 고치면 그 판만 갈린다 — 셋을 같이 본다. _MULTI_EPISODE_COLS = ( ("entity_episode_link", "presence_status TEXT NOT NULL DEFAULT 'active'"), ("entity_episode_link", "episode_notes_json TEXT"), ("character_outlook", "episode_id TEXT"), ) with _engine.connect() as conn: try: conn.execute(text( "ALTER TABLE entity_canon ADD COLUMN IF NOT EXISTS " "metadata_json TEXT NOT NULL DEFAULT '{}'" )) for _coldef in _IMAGE_ASSET_GEN_COLS: conn.execute(text( f"ALTER TABLE image_asset ADD COLUMN IF NOT EXISTS {_coldef}" )) for _tbl, _coldef in _MULTI_EPISODE_COLS: conn.execute(text( f"ALTER TABLE {_tbl} ADD COLUMN IF NOT EXISTS {_coldef}" )) conn.commit() except (ProgrammingError, OperationalError) as exc: # create_all 직후 entity_canon 부재는 비정상 (model 등록 실패 / 권한 # 부재 / 잘못된 DB URL). silent skip 시 stale schema 가 다른 test 까지 # 전염되므로 raise — 운영자가 DB 상태 확인. raise RuntimeError( "_ensure_d6_test_db_schema: ALTER entity_canon failed even after " f"Base.metadata.create_all — investigate test DB state. Exception: {exc!r}" ) from exc yield # problems.md #6 review M2: prompt_loader 의 process-global drift cache 가 다른 # test 파일에서 emit 한 warning 으로 오염되지 않도록 매 test 시작 시 비운다. # fresh_drift_cache fixture (test_prompt_version_pack.py) 의 monkeypatch 와는 # ── 기계의 `.env` 를 물지 않게 (2026-08-25) ───────────────────────── # # 시험이 재야 하는 것은 「무엇이 계약인가」이지 「이 기계가 지금 어떻게 # 설정됐나」가 아니다. 그런데 운영 중인 기계의 `.env` 에는 실험용 플래그가 # 켜져 있고, 그것이 그대로 시험에 새어 들어온다. # # ★하루에 **네 번** 같은 자리에 빠졌다: # · `STILL_CINE_VERIFY_ENABLED=true` → 재사용 갈래가 판정을 사러 가며 # 「records 무쓰기」 단정이 깨졌다 (test_cine_provider_swap) # · 같은 플래그 → 스텁 client 를 넘겼는데도 관문이 **실제 VLM 을 불러** # netprobe 가 유료 시도 9건을 잡았다 (test_cine_transform_stage) # · `STILL_CINE_PROVIDER=reve` → 스텁이 꽂힌 자리를 안 타고 진짜 fal 로 # 나가려 했다 (test_still_cine_stage) # · `STILL_CINE_STAGE_DIRECTION_ENABLED=true` → 팩 selector 가 옮겨 가 # 지문 단정이 흔들렸다 # # 파일마다 fixture 를 붙이는 것은 같은 개입의 반복이라 여기 한 번만 둔다. # **켜는 것을 재는 시험은 자기 안에서 다시 켠다** — monkeypatch 가 이 # fixture 보다 나중에 걸리므로 그대로 이긴다. _ENV_PINNED_OFF = ( ("still_cine_verify_enabled", False), ("still_cine_stage_direction_enabled", False), ("still_cine_provider", "grok"), ) @pytest.fixture(autouse=True) def _pin_experimental_flags_off(): from app.core.config import settings saved = [] for key, value in _ENV_PINNED_OFF: if hasattr(settings, key): saved.append((key, getattr(settings, key))) object.__setattr__(settings, key, value) yield for key, value in saved: object.__setattr__(settings, key, value) # 별개로 — autouse 라 모든 test 가 cleanup 보장. @pytest.fixture(autouse=True) def _reset_prompt_loader_drift_cache(): try: from app.modules import prompt_loader prompt_loader._DRIFT_WARNED.clear() except Exception: pass yield # G4.6 Wave A3 RC-H — legacy SceneDetailStep test 한정 helper isolation. # # 본래 G4.6 Phase 4 commit 시 broader autouse 로 시작 — 회귀 방지 위해 # _build_entity_traits_block + validate_visible_entities_contract 를 모든 test # 에서 no-op 처리. 그러나 Codex iter 1 I2: "모든 SceneDetailStep integration # coverage 를 무력화" 지적. legacy mock-DB 테스트 (ctx 의 name_by_short_id # field 미설정 + EntityCanon row 0) 만 보호하고, 새 Phase 4 unit / integration # 은 실제 production hookup 을 타도록 path-based scope 축소. # # Phase 4 implementation B1 prebuild 후: helper 가 db 인자 0, ctx 의 prebuild # map 만 사용. Legacy fixture 가 ctx.name_by_short_id 를 설정 안 하면 (default # 빈 dict) visible character 의 missing fail-fast 발생 → legacy 회귀. # 따라서 본 isolation 은 legacy fixture 들에만 한정. _LEGACY_SCENE_DETAIL_FIXTURE_FILES = frozenset({ "test_scene_detail_analyze_one.py", "test_scene_detail_v3.py", "test_render_prompt_card_integration.py", "test_g3_2_close_skip_sentinel_drift.py", }) @pytest.fixture(autouse=True) def _g4_6_isolate_entity_traits_and_validator(request, monkeypatch): # path-based opt-in — legacy SceneDetailStep mock-DB fixture 영역만 patch. # 새 Phase 4 test (test_visible_entities_validator.py) + 새 integration 은 # 실제 production hookup 그대로 (Codex iter 1 I2). test_path = getattr(request.node, "path", None) or getattr( request.node, "fspath", None ) test_filename = Path(str(test_path)).name if test_path else "" if test_filename not in _LEGACY_SCENE_DETAIL_FIXTURE_FILES: yield return try: monkeypatch.setattr( "app.core.steps.detail_steps._build_entity_traits_block", lambda *args, **kwargs: "", ) monkeypatch.setattr( "app.core.visible_entities_validator.validate_visible_entities_contract", lambda *args, **kwargs: None, ) except (ImportError, AttributeError): # legacy file 인데 module 미로드 / attr 부재 — silent skip 한정. # 이전 generic `except Exception: pass` (Claude iter 1 M2) 보다 좁힘. pass yield # ── W3-F32: 경로/파일명 기반 auto-marking ── _CORE_PATH_SEGMENTS = ("/tests/core/", "/tests/services/", "/tests/pipeline/") _STARTUP_STEMS = frozenset({ "test_auth_api", "test_users_api", "test_pipeline_e2e", "test_applicability", "test_step_catalog", "test_manifest_fields", "test_settings_registry", "test_api_endpoint_decorator", "test_models", "test_progress", "test_activity_logger", "test_checkpoint_io", "test_security", "test_i18n", "test_moderation", "test_moderation_retry_strategies", "test_projects_api", "test_episodes_api", "test_entities_api", "test_exports_api", "test_provenance", "test_visual_world_rules", "test_episode_summary", "test_scene_summary", "test_project_export_import", }) _IMAGE_STEMS = frozenset({ "test_images_api", "test_image_validator", "test_fal_angle_edit", "test_gemini_i2i_angle", "test_variation_pipeline", "test_entity_dependency", "test_gpt_scene10_analysis", }) _LANE_MARKERS = frozenset({"core", "startup", "image", "full", "pg"}) def _classify(item) -> str: path = item.path p = path.as_posix() for seg in _CORE_PATH_SEGMENTS: if seg in p: return "core" stem = path.stem if stem in _STARTUP_STEMS: return "startup" if stem in _IMAGE_STEMS: return "image" return "full" def pytest_collection_modifyitems(config, items): for item in items: existing = {m.name for m in item.iter_markers()} if existing & _LANE_MARKERS: continue lane = _classify(item) item.add_marker(getattr(pytest.mark, lane)) # ── 돈 가드 (2026-08-20) — 위 「2-b」 절의 실행부 ───────────────────────── # 판별은 tests/netprobe.py 의 것을 **그대로 쓴다**. 두 곳에 따로 적으면 한쪽만 # 고쳐질 때 조용히 갈린다(이 세션에 실제로 그 부류의 결함이 있었다). def pytest_configure(config): config.addinivalue_line( "markers", "allow_outbound: 바깥 유료 호출을 허용한다 — 정말 필요할 때만") # Opik 프로젝트가 시험용인지 확인 — 아니면 시험 자체를 세운다 (2026-08-24). # ★fail-fast 다. 기록 실패를 non-fatal 로 삼키는 프로덕션 규약과 반대인데, # 여기서 조용히 넘어가면 프로덕션 감사 데이터가 오염되는 것을 아무도 모른다. # 값이 조용히 프로덕션으로 되돌아가는 것이 이 조치가 막으려는 바로 그 실패다. from app.core.config import settings if settings.opik_project_name != TEST_OPIK_PROJECT: raise RuntimeError( f"REFUSING test run: opik_project_name=" f"{settings.opik_project_name!r} — 시험이 프로덕션 Opik " f"프로젝트에 쓰려 한다. 위 OPIK_PROJECT_NAME 대입이 무력화됐다 " f"(app.* 가 env 블록보다 먼저 import 됐거나 settings 가 다른 " f"경로로 만들어졌다).") if os.environ.get("THEROAD_TEST_BLOCK_OUTBOUND") == "1": from tests import netprobe # ★붙임말로 **등록**한다. 그냥 함수만 부르면 netprobe 의 다른 훅 # (어느 시험이 냈는지 적는 pytest_runtest_protocol)이 안 붙어, # 출처가 전부 「수집·설정 단계」로 뭉개진다 — 총계만 남으면 못 고친다. if not config.pluginmanager.is_registered(netprobe): config.pluginmanager.register(netprobe, "netprobe")