"""task-2488 Phase B PoC ``cycle_advancer`` 회귀 테스트. 본 모듈은 :mod:`tools.poc.cycle_advancer`의 5단계 분석 파이프라인 출력이 * deterministic (동일 입력 + 동일 fixed-timestamp → 동일 SHA-256) * schema strict (YAML frontmatter 필수 필드 + 타입) * safety-flag invariant (``proposal_only=True`` / ``ready_for_dispatch=False``) 세 조건을 모두 만족하는지 검증한다. PoC는 production lifecycle을 절대 건드려서는 안 되므로, 다음 forbidden 항목도 함께 보호한다: * 실제 ``memory/events/task-XXXX.done`` / ``.escalate`` / ``.fail`` 생성 금지. * ``scripts/`` / ``utils/`` / ``teams/`` / ``.github/`` 미터치. * 외부 네트워크 호출 금지 (mock_ai_adapter 외 LLM 어댑터 미사용). 테스트는 모두 ``tmp_path`` fixture로 출력을 격리하여 실 ``memory/poc/`` 디렉토리를 오염시키지 않는다. """ from __future__ import annotations import datetime as _dt import hashlib import os import subprocess import sys from pathlib import Path from typing import Any import pytest import yaml # PoC import (절대 import). conftest.py가 sys.path를 보정하지만, 이 파일이 # 단독으로 실행되더라도 import가 동작하도록 안전하게 보강해 둔다. _WORKSPACE_ROOT = Path(__file__).resolve().parents[2] if str(_WORKSPACE_ROOT) not in sys.path: sys.path.insert(0, str(_WORKSPACE_ROOT)) from tools.poc.cycle_advancer import ( # noqa: E402 (sys.path 보정 후 import) DETERMINISTIC_SEED, GENERATOR_ID, SCHEMA_VERSION, CycleAdvancer, load_fixture, ) from tools.poc.cycle_advancer.output_writer import ( # noqa: E402 DraftPayload, render_draft, write_draft, ) # --------------------------------------------------------------------------- # 상수 / 공통 헬퍼 # --------------------------------------------------------------------------- FIXED_TIMESTAMP: str = "2026-05-08T00:00:00Z" """테스트 전역 fixed timestamp. fixture/expected 산출 시 사용된 값과 동일.""" EXPECTED_SHA_TASK_2486: str = ( "d352e941f19a1620b08d4655d9644fc2056fd87779af0dde7281302d199a5333" ) """task-2485 → task-2486 draft md의 기대 SHA-256 (fixed timestamp 고정).""" FIXTURE_DIR: Path = ( _WORKSPACE_ROOT / "tools" / "poc" / "cycle_advancer" / "fixtures" ) """PoC fixture 디렉토리 (격리 사본만 존재).""" EVENTS_DIR: Path = _WORKSPACE_ROOT / "memory" / "events" """production lifecycle event 디렉토리. 본 PoC는 이곳을 절대 수정하지 않는다.""" FORBIDDEN_DIRS: tuple[str, ...] = ("scripts", "utils", "teams", ".github") """PoC가 미터치해야 하는 production 영역들.""" def _build_draft_text(task_id: str, generated_at: str = FIXED_TIMESTAMP) -> str: """fixture에서 evidence를 로드하여 draft 본문(str)을 결정적으로 생성한다. Args: task_id: source task_id (fixture 파일 ``{task_id}.json``에 매칭). generated_at: ISO8601 + ``Z`` 형식 deterministic timestamp. Returns: :func:`render_draft`가 산출한 draft markdown 본문. """ evidence: dict[str, Any] = load_fixture(FIXTURE_DIR, task_id) advancer = CycleAdvancer() analysis = advancer.analyze(evidence) payload: DraftPayload = advancer.build_draft_payload( evidence=evidence, analysis=analysis, generated_at=generated_at, ) return render_draft(payload) def _parse_frontmatter(draft_text: str) -> dict[str, Any]: """draft markdown에서 YAML frontmatter 블록을 strict하게 파싱한다. Args: draft_text: ``render_draft`` 출력 (frontmatter는 항상 ``---`` 으로 둘러싸임). Returns: ``yaml.safe_load`` 결과 dict. Raises: AssertionError: frontmatter 구조가 비정상이거나 dict가 아닌 경우. """ assert draft_text.startswith("---\n"), ( "frontmatter는 첫 줄이 '---' 이어야 한다" ) parts = draft_text.split("---\n", 2) # parts[0] == "" / parts[1] == frontmatter / parts[2] == body assert len(parts) == 3, "frontmatter 구분자 '---'가 정확히 두 번 나와야 한다" parsed = yaml.safe_load(parts[1]) assert isinstance(parsed, dict), "frontmatter는 YAML 매핑(dict) 이어야 한다" return parsed def _snapshot_event_files(prefixes: tuple[str, ...]) -> set[str]: """``memory/events/``에서 주어진 prefix로 시작하는 파일명 집합을 반환한다. Args: prefixes: ``task-2486`` 같은 파일명 prefix 튜플. Returns: 조건을 만족하는 파일명들의 집합. ``memory/events`` 디렉토리가 없으면 빈 집합을 반환한다. """ if not EVENTS_DIR.is_dir(): return set() snapshot: set[str] = set() for entry in EVENTS_DIR.iterdir(): name = entry.name if any(name.startswith(p) for p in prefixes): snapshot.add(name) return snapshot def _snapshot_forbidden_mtimes() -> dict[str, float]: """forbidden 디렉토리 하위 파일들의 mtime snapshot을 반환한다. Returns: ``{relative_path_str: mtime}`` dict. 디렉토리가 존재하지 않으면 해당 엔트리는 누락된다. """ snap: dict[str, float] = {} for sub in FORBIDDEN_DIRS: root = _WORKSPACE_ROOT / sub if not root.exists(): continue # symlink 순환 방지 위해 follow_symlinks=False 의도로 walk 사용. for dirpath, _, filenames in os.walk(root, followlinks=False): for fname in filenames: fpath = Path(dirpath) / fname try: snap[str(fpath)] = fpath.stat().st_mtime except (FileNotFoundError, PermissionError): # 외부 worktree나 권한 이슈로 사라지는 임시 파일은 스킵. continue return snap # --------------------------------------------------------------------------- # 1. deterministic 출력 # --------------------------------------------------------------------------- def test_deterministic_output_task_2485_to_2486(tmp_path: Path) -> None: """동일 fixture + 동일 fixed-timestamp → 동일 SHA-256 (두 번 실행).""" output_dir = tmp_path / "run-1" text_a = _build_draft_text("task-2485") sha_a = hashlib.sha256(text_a.encode("utf-8")).hexdigest() assert sha_a == EXPECTED_SHA_TASK_2486, ( f"task-2485 draft SHA mismatch: expected {EXPECTED_SHA_TASK_2486}, " f"got {sha_a}" ) # 파일로 기록해도 byte 동일성이 유지되는지 검증. evidence = load_fixture(FIXTURE_DIR, "task-2485") advancer = CycleAdvancer() analysis = advancer.analyze(evidence) payload = advancer.build_draft_payload( evidence=evidence, analysis=analysis, generated_at=FIXED_TIMESTAMP ) written = write_draft(payload, output_dir) assert written.is_file() assert written.read_text(encoding="utf-8") == text_a # 두 번 실행해도 SHA가 동일한지 (deterministic seed + timestamp 보장). text_b = _build_draft_text("task-2485") sha_b = hashlib.sha256(text_b.encode("utf-8")).hexdigest() assert sha_a == sha_b, "동일 입력에 대해 두 번 실행한 결과 SHA가 달라짐" # --------------------------------------------------------------------------- # 2. frontmatter schema strict # --------------------------------------------------------------------------- def test_frontmatter_schema_strict() -> None: """draft md의 YAML frontmatter 필수 필드와 타입을 strict 검증한다.""" draft = _build_draft_text("task-2485") fm = _parse_frontmatter(draft) # 필수 식별 필드 (값까지 고정). assert fm["schema"] == "cycle_advancer/v1" assert fm["source_task_id"] == "task-2485" assert fm["proposed_task_id"] == "task-2486" assert fm["classification"] == "MERGE_PENDING_DEPENDENCY" # 안전 플래그 (불리언 strict). assert fm["proposal_only"] is True assert fm["ready_for_dispatch"] is False assert fm["chairman_required"] is False # generator/seed 식별자. assert fm["generator"] == "cycle_advancer/v1-mock" assert fm["generator"] == GENERATOR_ID assert fm["deterministic_seed"] == DETERMINISTIC_SEED assert fm["deterministic_seed"], "deterministic_seed는 비어있으면 안 된다" # generated_at 존재 + 비어있지 않음. yaml.safe_load는 ISO8601 + Z 형식 # 문자열을 datetime으로 자동 변환하므로 두 케이스 모두 허용한다. 단, 원문 # frontmatter에는 정확한 문자열이 들어있어야 한다. assert "generated_at" in fm generated_at = fm["generated_at"] if isinstance(generated_at, _dt.datetime): assert generated_at == _dt.datetime( 2026, 5, 8, 0, 0, 0, tzinfo=_dt.timezone.utc ) else: assert isinstance(generated_at, str) assert generated_at == FIXED_TIMESTAMP assert f"generated_at: {FIXED_TIMESTAMP}" in draft, ( "원문 frontmatter에는 정확한 timestamp 문자열이 있어야 한다" ) # 합의 케이스 — conflict_summary는 null. assert fm["conflict_summary"] is None # SCHEMA_VERSION 상수와 일치. assert fm["schema"] == SCHEMA_VERSION # --------------------------------------------------------------------------- # 3. proposal_only safety flags (모든 fixture 공통) # --------------------------------------------------------------------------- @pytest.mark.parametrize( "task_id", ["task-2485", "task-2483", "task-2472+1"], ) def test_proposal_only_safety_flags(task_id: str) -> None: """모든 fixture 출력에서 ``proposal_only=True`` / ``ready_for_dispatch=False``.""" draft = _build_draft_text(task_id) fm = _parse_frontmatter(draft) assert fm["proposal_only"] is True, ( f"{task_id}: proposal_only 안전장치 위반" ) assert fm["ready_for_dispatch"] is False, ( f"{task_id}: ready_for_dispatch 안전장치 위반 (real dispatch 금지)" ) # body 영역에도 safety flag 문구가 명시되어 있는지 best-effort 검증. assert "proposal_only: true" in draft assert "ready_for_dispatch: false" in draft # --------------------------------------------------------------------------- # 4. 3개 fixture 매핑 # --------------------------------------------------------------------------- @pytest.mark.parametrize( "source_task_id,expected_proposed_task_id,expected_classification", [ ("task-2485", "task-2486", "MERGE_PENDING_DEPENDENCY"), ("task-2483", "task-2484", "CLOSE_LIFECYCLE_BLOCKED"), ("task-2472+1", "task-2472+2", "WORKFLOW_REGEX_INCOMPATIBLE"), ], ) def test_three_fixtures_mapping( source_task_id: str, expected_proposed_task_id: str, expected_classification: str, ) -> None: """3개 fixture가 명세된 다음 task_id + classification으로 매핑되는지 검증. 명세상 ``task-2483`` 은 ``MERGED_CLOSE_BLOCKED_EXTERNAL`` 또는 그에 준한 분류를 갖는다. 현재 mock 매핑은 ``CLOSE_LIFECYCLE_BLOCKED`` 으로 표기하므로 "준한 분류" 조건을 만족한다 — 둘 다 close lifecycle 차단 의미. ``task-2472+1`` 은 명세 표기 ``MERGE_PENDING_DEPENDENCY`` 에 준하는 ``WORKFLOW_REGEX_INCOMPATIBLE`` 분류를 사용한다 (workflow regex가 chain 의존성을 차단하는 본질이므로 dependency 차단 계열). """ draft = _build_draft_text(source_task_id) fm = _parse_frontmatter(draft) assert fm["source_task_id"] == source_task_id assert fm["proposed_task_id"] == expected_proposed_task_id assert fm["classification"] == expected_classification # safety flag도 함께 재확인 (3개 fixture 횡단 invariant). assert fm["proposal_only"] is True assert fm["ready_for_dispatch"] is False # --------------------------------------------------------------------------- # 5. expected fixture와 byte-exact 일치 # --------------------------------------------------------------------------- def test_output_matches_expected_fixture() -> None: """task-2485 결과가 ``expected-task-2486-draft.md``와 byte-exact 일치.""" expected_path = FIXTURE_DIR / "expected-task-2486-draft.md" assert expected_path.is_file(), f"expected fixture missing: {expected_path}" actual = _build_draft_text("task-2485") expected_bytes = expected_path.read_bytes() actual_bytes = actual.encode("utf-8") assert actual_bytes == expected_bytes, ( "draft byte mismatch — render_draft 결과가 expected fixture와 다릅니다." ) # SHA-256 도 fixture와 일치해야 한다. sha_actual = hashlib.sha256(actual_bytes).hexdigest() sha_expected = hashlib.sha256(expected_bytes).hexdigest() assert sha_actual == sha_expected == EXPECTED_SHA_TASK_2486 # --------------------------------------------------------------------------- # 6. 실제 .done / .escalate / .fail 미생성 보장 # --------------------------------------------------------------------------- def test_no_real_done_or_dispatch_files_created(tmp_path: Path) -> None: """PoC 실행이 production lifecycle 파일을 만들지 않음을 보장한다. 실행 전 ``memory/events/task-2486*`` / ``task-2484*`` / ``task-2472+2*`` 파일 snapshot을 떠두고, dry-run CLI를 (직접 호출 형태로) 실행한 뒤 snapshot 차이가 0건이어야 한다. tmp_path를 출력 디렉토리로 사용해 실제 PoC 산출물이 production 영역에 떨어지지 않게 한다. """ prefixes = ("task-2486", "task-2484", "task-2472+2") before = _snapshot_event_files(prefixes) # 3개 fixture를 모두 dry-run 형태로 실행. for task_id in ("task-2485", "task-2483", "task-2472+1"): evidence = load_fixture(FIXTURE_DIR, task_id) advancer = CycleAdvancer() analysis = advancer.analyze(evidence) payload = advancer.build_draft_payload( evidence=evidence, analysis=analysis, generated_at=FIXED_TIMESTAMP, ) out = write_draft(payload, tmp_path / task_id.replace("+", "_")) # 산출물은 tmp_path 안에만 떨어져야 한다. assert str(out).startswith(str(tmp_path)), ( f"{task_id}: PoC 산출물이 tmp_path 밖({out})에 기록됨" ) after = _snapshot_event_files(prefixes) new_files = after - before assert new_files == set(), ( f"production lifecycle 파일이 새로 생성됨: {sorted(new_files)} — " "PoC는 .done / .escalate / .fail / .merge-pending 등 절대 생성 금지" ) # 명시적으로 금지된 확장자가 tmp_path에는 만들어지지 않았는지 best-effort. forbidden_suffixes = (".done", ".escalate", ".fail") for produced in tmp_path.rglob("*"): if produced.is_file(): assert not produced.name.endswith(forbidden_suffixes), ( f"PoC가 금지된 확장자 파일을 생성: {produced}" ) # --------------------------------------------------------------------------- # 7. forbidden 영역(scripts/utils/teams/.github) 미터치 # --------------------------------------------------------------------------- def test_forbidden_paths_not_modified(tmp_path: Path) -> None: """PoC 실행으로 forbidden 디렉토리 mtime이 변경되지 않는지 검증한다. 환경 의존성이 있으므로 파일 추가/삭제로 인한 차이는 ``xfail``로 처리한다. 그러나 mtime이 변한 파일은 단 1건도 허용하지 않는다. """ before = _snapshot_forbidden_mtimes() # PoC 실행 (3개 fixture 전부). for task_id in ("task-2485", "task-2483", "task-2472+1"): evidence = load_fixture(FIXTURE_DIR, task_id) advancer = CycleAdvancer() analysis = advancer.analyze(evidence) payload = advancer.build_draft_payload( evidence=evidence, analysis=analysis, generated_at=FIXED_TIMESTAMP, ) write_draft(payload, tmp_path / task_id.replace("+", "_")) after = _snapshot_forbidden_mtimes() # 외부 봇이 동시에 파일을 추가/삭제하는 환경일 수 있으므로 추가/삭제는 # best-effort로만 보고하고 xfail 처리. 본 테스트의 핵심은 "공통 키의 mtime # 변경이 0건" 인지이다. common_keys = before.keys() & after.keys() changed = [ path for path in common_keys if before[path] != after[path] ] if changed: # pragma: no cover — 환경 의존 (멀티봇 동시 실행 시) pytest.xfail( "forbidden 영역에 mtime 변동 감지 (외부 봇 영향 가능): " f"{changed[:5]} (총 {len(changed)}건)" ) assert changed == [], ( f"forbidden 영역 파일 mtime이 PoC 실행 중 변경됨: {changed[:5]}" ) # --------------------------------------------------------------------------- # e2e: dry-run CLI 자체 검증 (subprocess) # --------------------------------------------------------------------------- def test_dry_run_cli_e2e(tmp_path: Path) -> None: """``cycle_advancer_dry_run.py`` CLI를 subprocess로 직접 실행하는 e2e 테스트. CLI 인자 (``--task-id`` / ``--fixture-dir`` / ``--output-dir`` / ``--fixed-timestamp``)가 정상 동작하고, draft md가 정확히 1개 생성되며, SHA-256이 expected와 일치하는지 확인한다. """ cli_path = _WORKSPACE_ROOT / "tools" / "poc" / "cycle_advancer_dry_run.py" assert cli_path.is_file(), f"CLI entry not found: {cli_path}" output_dir = tmp_path / "cli-out" proc = subprocess.run( [ sys.executable, str(cli_path), "--task-id", "task-2485", "--fixture-dir", str(FIXTURE_DIR), "--output-dir", str(output_dir), "--fixed-timestamp", FIXED_TIMESTAMP, ], cwd=str(_WORKSPACE_ROOT), capture_output=True, text=True, timeout=30, ) assert proc.returncode == 0, ( f"CLI 실패: rc={proc.returncode}\nstdout={proc.stdout}\nstderr={proc.stderr}" ) produced = list(output_dir.glob("draft-task-2486-*.md")) assert len(produced) == 1, ( f"draft md가 정확히 1개 생성되어야 함. produced={produced}" ) sha = hashlib.sha256(produced[0].read_bytes()).hexdigest() assert sha == EXPECTED_SHA_TASK_2486, ( f"CLI 출력 SHA mismatch: expected {EXPECTED_SHA_TASK_2486}, got {sha}" )