"""tests/regression/test_archive_escalated_audit.py — task-2471+1 자동화 회귀. 회장 명령 (B): ``.done.escalated`` 처리 자동화 검증. - archive 스크립트/subcommand로 처리 (수동 삭제 금지) - 원본 path / size / sha256 / reason / archived_path 를 audit jsonl 자동 기록 - archive 후 .done + .done.escalated 충돌 자동 검증 - 빈 marker 차단 ``taskctl archive-escalated`` 자동화 경로의 회귀 차단. """ from __future__ import annotations import hashlib import json import os import subprocess import sys from pathlib import Path import pytest WORKSPACE = Path(__file__).resolve().parents[2] TASKCTL = WORKSPACE / "scripts" / "taskctl.py" @pytest.fixture() def env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict: """격리 환경: 임시 events_dir + audit jsonl.""" events_dir = tmp_path / "events" events_dir.mkdir() archive_dir = events_dir / "archive" audit_log = tmp_path / "orchestration-audit" / "escalated-archive.jsonl" state_dir = tmp_path / "state" state_dir.mkdir() evidence_dir = tmp_path / "evidence" evidence_dir.mkdir() monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path)) return { "tmp_path": tmp_path, "events_dir": events_dir, "archive_dir": archive_dir, "audit_log": audit_log, "state_dir": state_dir, "evidence_dir": evidence_dir, } def _run_archive( task_id: str, events_dir: Path, *, reason: str | None = None, fail_when_missing: bool = False, cwd: Path | None = None, ) -> subprocess.CompletedProcess: cmd = [ sys.executable, str(TASKCTL), "archive-escalated", task_id, "--events-dir", str(events_dir), "--machine", ] if reason: cmd += ["--reason", reason] if fail_when_missing: cmd.append("--fail-when-missing") return subprocess.run( cmd, capture_output=True, text=True, timeout=30, cwd=str(cwd or WORKSPACE), env={**os.environ, "WORKSPACE_ROOT": str(events_dir.parent)}, ) def _make_marker(path: Path, payload: dict | None) -> None: if payload is None: path.touch() else: path.write_text( json.dumps(payload, ensure_ascii=False), encoding="utf-8" ) def test_archive_records_path_size_sha256_reason(env: dict) -> None: """audit jsonl에 path/size/sha256/reason/archived_path 모두 기록.""" task_id = "task-archive-001" src = env["events_dir"] / f"{task_id}.done.escalated" payload = { "trigger": "test", "ts": "2026-05-07T00:00:00Z", "reason": "stale_done_30min_test", } _make_marker(src, payload) expected_size = src.stat().st_size expected_sha = hashlib.sha256(src.read_bytes()).hexdigest() result = _run_archive(task_id, env["events_dir"]) assert result.returncode == 0, result.stderr record = json.loads(result.stdout.strip()) assert record["task_id"] == task_id assert record["original_path"].endswith(f"{task_id}.done.escalated") assert record["archived_path"] assert Path(record["archived_path"]).exists() assert record["size"] == expected_size assert record["sha256"] == expected_sha assert record["reason"] == "stale_done_30min_test" assert record["ok"] is True # 원본 제거 확인 assert not src.exists() # archive 위치 확인 archived = list(env["archive_dir"].iterdir()) assert len(archived) == 1 assert archived[0].name.startswith(f"{task_id}.done.escalated.") def test_archive_post_check_conflict_detected(env: dict) -> None: """archive 후 .done이 잔존하면 noop이지만, 동시 존재 시점에서 archive 가능.""" task_id = "task-archive-002" done = env["events_dir"] / f"{task_id}.done" esc = env["events_dir"] / f"{task_id}.done.escalated" done.write_text( json.dumps({"task_id": task_id}), encoding="utf-8" ) _make_marker(esc, {"trigger": "stale", "reason": "conflict_test"}) # archive 실행 → 충돌 해소 result = _run_archive(task_id, env["events_dir"]) assert result.returncode == 0, result.stderr record = json.loads(result.stdout.strip()) assert record["ok"] is True # post_check가 있고 ok=True assert record["post_check"]["ok"] is True assert not esc.exists() assert done.exists() # .done은 보존 def test_archive_fail_closed_when_missing_with_flag(env: dict) -> None: """``--fail-when-missing`` 플래그 시 marker 부재면 exit 1.""" task_id = "task-archive-missing" result = _run_archive( task_id, env["events_dir"], fail_when_missing=True ) assert result.returncode != 0 assert "없음" in (result.stdout + result.stderr) or "missing" in ( result.stdout + result.stderr ).lower() or "차단" in (result.stdout + result.stderr) def test_archive_noop_when_missing_default(env: dict) -> None: """기본 동작: marker 부재 시 noop (exit 0).""" task_id = "task-archive-noop" result = _run_archive(task_id, env["events_dir"]) assert result.returncode == 0 record = json.loads(result.stdout.strip()) assert record["ok"] is True assert record["noop"] is True assert record["reason"] == "no_marker" def test_archive_audit_jsonl_appended(env: dict) -> None: """ESCALATED_AUDIT_LOG (orchestration-audit/escalated-archive.jsonl) 추가 검증. 실제 시스템 audit log 위치: ``$WORKSPACE/memory/orchestration-audit``. 여기서는 archive_path 기록만 검증하고 시스템 jsonl 추가 부수효과는 별도 테스트 (test_finish_task_pipeline) 가 검증한다. 임시 환경은 WORKSPACE_ROOT 가 tmp_path 라 해도 taskctl 모듈 상수가 import-time 결정되어 실제 시스템 dir에 기록된다 (이는 행위 자체의 문제는 아니며 격리 한계). """ task_id = "task-archive-jsonl" src = env["events_dir"] / f"{task_id}.done.escalated" _make_marker(src, {"trigger": "stale", "reason": "jsonl_test"}) result = _run_archive(task_id, env["events_dir"]) assert result.returncode == 0 rec = json.loads(result.stdout.strip()) # 기록 필드 7건 (audit jsonl 등재 형태) for k in ( "ts", "task_id", "actor", "original_path", "archived_path", "size", "sha256", "reason", "command", "ok", ): assert k in rec, f"missing audit field: {k}" assert rec["command"] == "taskctl archive-escalated" def test_archive_empty_marker_recorded_with_unparseable(env: dict) -> None: """빈 .done.escalated (0 byte) 도 archive 가능. reason = unparseable_payload.""" task_id = "task-archive-empty" src = env["events_dir"] / f"{task_id}.done.escalated" _make_marker(src, None) # 0 byte assert src.stat().st_size == 0 result = _run_archive(task_id, env["events_dir"]) assert result.returncode == 0 record = json.loads(result.stdout.strip()) assert record["size"] == 0 # payload reason은 unparseable 또는 unknown assert record["reason"] in ( "unparseable_payload", "unknown", "archived", )