feat(orchestrator): P3 wrapper input/encoding fix + P4 audit-only mode
P3 hotfix (2026-05-18 — verified during #46 retry attempt): - _run_with_tree_kill: encode input only when Popen is in binary mode. Previously force-encoded str→bytes even with encoding= set, breaking text-mode stdin pipes with: write() argument must be str, not bytes. - run_claude path was the only affected call site. - 3 new C7 regression tests (input+encoding / bytes+binary / auto-encode). - C3/C6 test fixtures hardened with DEVNULL stdio isolation. P4 audit-only mode (2026-05-19, prep for #50 integration audit): - _is_audit_issue: title-based detection for [INTEGRATION-AUDIT*], [AUDIT-ONLY], or "integration audit" phrase. - _audit_mode + --audit-only CLI flag: manual override regardless of title. - AUDIT_ONLY_NOTE injected into context pack across all stages/rounds. - Stage 3 (code-edit) YES gate: deterministic git status check. Changes touching src/**, templates/**, tests/** auto-reject Stage 3 YES and post a supplement-request comment. LLM-independent enforcement. - 26 new audit-mode tests (title detection, CLI override, forbidden prefix detection, allowed paths pass, Windows backslash normalization, quoted paths with spaces, git error fail-open, constants sanity). Total: 75/75 pytest pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
221
tests/orchestrator_unit/test_audit_mode.py
Normal file
221
tests/orchestrator_unit/test_audit_mode.py
Normal file
@@ -0,0 +1,221 @@
|
||||
"""P4 (2026-05-19) — audit-only mode verification.
|
||||
|
||||
Covers:
|
||||
- _is_audit_issue: title pattern detection (positive + negative)
|
||||
- _audit_mode: title-based + CLI override (AUDIT_ONLY_OVERRIDE)
|
||||
- _check_audit_only_violations: forbidden prefix detection via mocked git status
|
||||
- AUDIT_ONLY_NOTE injection into context pack (via build_context_pack contract)
|
||||
|
||||
Run: pytest -q tests/orchestrator_unit/test_audit_mode.py
|
||||
"""
|
||||
import sys
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).parent.parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
import orchestrator # noqa: E402
|
||||
from orchestrator import ( # noqa: E402
|
||||
_is_audit_issue,
|
||||
_audit_mode,
|
||||
_check_audit_only_violations,
|
||||
AUDIT_ONLY_FORBIDDEN_PREFIXES,
|
||||
AUDIT_ONLY_NOTE,
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
# _is_audit_issue — title detection
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestIsAuditIssue:
|
||||
def test_integration_audit_bracket(self):
|
||||
assert _is_audit_issue("[INTEGRATION-AUDIT-01] cumulative review") is True
|
||||
assert _is_audit_issue("[INTEGRATION-AUDIT-02] something") is True
|
||||
assert _is_audit_issue("[INTEGRATION-AUDIT] no number") is True
|
||||
|
||||
def test_audit_only_bracket(self):
|
||||
assert _is_audit_issue("[AUDIT-ONLY] doc consistency check") is True
|
||||
|
||||
def test_case_insensitive(self):
|
||||
assert _is_audit_issue("[integration-audit-03] foo") is True
|
||||
assert _is_audit_issue("[Audit-Only] bar") is True
|
||||
|
||||
def test_plain_integration_audit_phrase(self):
|
||||
assert _is_audit_issue("Quarterly integration audit for closed issues") is True
|
||||
assert _is_audit_issue("Integration Audit Q2") is True
|
||||
|
||||
def test_execution_issue_not_audit(self):
|
||||
"""execution sub-issue 가 audit 로 잘못 감지되면 안 됨."""
|
||||
assert _is_audit_issue("[IMP-15 실행-1] image_aspect_mismatch") is False
|
||||
assert _is_audit_issue("[IMP-15 exec-2] table overflow") is False
|
||||
|
||||
def test_unrelated_issues(self):
|
||||
assert _is_audit_issue("IMP-19 I4 zone 비중 분배") is False
|
||||
assert _is_audit_issue("Fix overflow bug") is False
|
||||
assert _is_audit_issue("docs(IMP-06): Stage 4 fix") is False
|
||||
|
||||
def test_empty_or_none(self):
|
||||
assert _is_audit_issue("") is False
|
||||
assert _is_audit_issue(None) is False
|
||||
|
||||
def test_audit_word_in_random_position_no_match(self):
|
||||
"""'audit' 가 단독으로 나오는 건 안 잡아야 함 — 'integration audit' 만."""
|
||||
assert _is_audit_issue("audit some code") is False
|
||||
assert _is_audit_issue("security audit") is False
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
# _audit_mode — combination with CLI override
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestAuditMode:
|
||||
def setup_method(self):
|
||||
# 각 테스트 전에 override 리셋.
|
||||
orchestrator.AUDIT_ONLY_OVERRIDE = False
|
||||
|
||||
def teardown_method(self):
|
||||
orchestrator.AUDIT_ONLY_OVERRIDE = False
|
||||
|
||||
def test_title_based_only(self):
|
||||
assert _audit_mode("[INTEGRATION-AUDIT-01] foo") is True
|
||||
assert _audit_mode("IMP-19 zone") is False
|
||||
|
||||
def test_cli_override_forces_audit(self):
|
||||
"""title 에 marker 없어도 CLI flag 가 audit mode 강제."""
|
||||
orchestrator.AUDIT_ONLY_OVERRIDE = True
|
||||
assert _audit_mode("IMP-19 zone") is True
|
||||
assert _audit_mode("any title") is True
|
||||
assert _audit_mode("") is True
|
||||
|
||||
def test_override_off_falls_back_to_title(self):
|
||||
orchestrator.AUDIT_ONLY_OVERRIDE = False
|
||||
assert _audit_mode("IMP-19 zone") is False
|
||||
assert _audit_mode("[INTEGRATION-AUDIT-01]") is True
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
# _check_audit_only_violations — git status parsing
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
|
||||
class _FakeCompleted:
|
||||
def __init__(self, stdout, returncode=0):
|
||||
self.stdout = stdout
|
||||
self.stderr = ""
|
||||
self.returncode = returncode
|
||||
|
||||
|
||||
class TestCheckAuditOnlyViolations:
|
||||
"""subprocess.run 을 monkeypatch 해서 다양한 git status 출력 시나리오 검증."""
|
||||
|
||||
def test_clean_tree(self, monkeypatch):
|
||||
def fake_run(*args, **kwargs):
|
||||
return _FakeCompleted(stdout="")
|
||||
monkeypatch.setattr(subprocess, "run", fake_run)
|
||||
assert _check_audit_only_violations() == []
|
||||
|
||||
def test_only_allowed_changes(self, monkeypatch):
|
||||
"""docs/architecture 변경만 있으면 violation 0."""
|
||||
stdout = (
|
||||
" M docs/architecture/INTEGRATION-AUDIT-01-REPORT.md\n"
|
||||
"?? docs/architecture/INTEGRATION-AUDIT-01-MATRIX.md\n"
|
||||
" M docs/architecture/PHASE-Z-IMPLEMENTATION-ISSUE-BACKLOG.md\n"
|
||||
)
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _FakeCompleted(stdout=stdout))
|
||||
assert _check_audit_only_violations() == []
|
||||
|
||||
def test_src_change_detected(self, monkeypatch):
|
||||
stdout = (
|
||||
" M src/phase_z2_pipeline.py\n"
|
||||
" M docs/architecture/INTEGRATION-AUDIT-01-REPORT.md\n"
|
||||
)
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _FakeCompleted(stdout=stdout))
|
||||
v = _check_audit_only_violations()
|
||||
assert v == ["src/phase_z2_pipeline.py"]
|
||||
|
||||
def test_templates_change_detected(self, monkeypatch):
|
||||
stdout = " M templates/phase_z2/families/something.html\n"
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _FakeCompleted(stdout=stdout))
|
||||
v = _check_audit_only_violations()
|
||||
assert v == ["templates/phase_z2/families/something.html"]
|
||||
|
||||
def test_tests_change_detected(self, monkeypatch):
|
||||
stdout = " M tests/phase_z2/test_overflow.py\n"
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _FakeCompleted(stdout=stdout))
|
||||
v = _check_audit_only_violations()
|
||||
assert v == ["tests/phase_z2/test_overflow.py"]
|
||||
|
||||
def test_multiple_violations(self, monkeypatch):
|
||||
stdout = (
|
||||
" M src/a.py\n"
|
||||
"?? src/b.py\n"
|
||||
" M templates/c.html\n"
|
||||
" M tests/d.py\n"
|
||||
" M docs/architecture/INTEGRATION-AUDIT-01-REPORT.md\n" # allowed
|
||||
" M data/runs/run123.json\n" # allowed (not in forbidden)
|
||||
)
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _FakeCompleted(stdout=stdout))
|
||||
v = _check_audit_only_violations()
|
||||
assert set(v) == {"src/a.py", "src/b.py", "templates/c.html", "tests/d.py"}
|
||||
|
||||
def test_renamed_file_destination_checked(self, monkeypatch):
|
||||
"""rename 의 경우 destination 만 검사."""
|
||||
stdout = "R docs/old.md -> src/new.py\n"
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _FakeCompleted(stdout=stdout))
|
||||
v = _check_audit_only_violations()
|
||||
assert v == ["src/new.py"]
|
||||
|
||||
def test_windows_backslash_path(self, monkeypatch):
|
||||
"""Windows backslash path 도 forward-slash 로 정규화돼서 매치."""
|
||||
stdout = " M src\\phase_z2_pipeline.py\n"
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _FakeCompleted(stdout=stdout))
|
||||
v = _check_audit_only_violations()
|
||||
assert v == ["src/phase_z2_pipeline.py"]
|
||||
|
||||
def test_quoted_path_with_spaces(self, monkeypatch):
|
||||
"""공백/특수문자 포함 path 는 quoted — quote strip 후 검사."""
|
||||
stdout = ' M "src/some file.py"\n'
|
||||
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: _FakeCompleted(stdout=stdout))
|
||||
v = _check_audit_only_violations()
|
||||
assert v == ["src/some file.py"]
|
||||
|
||||
def test_git_error_fails_open(self, monkeypatch):
|
||||
"""git 자체 실패 → 가드 false positive 안 만들고 빈 list 반환."""
|
||||
monkeypatch.setattr(subprocess, "run",
|
||||
lambda *a, **kw: _FakeCompleted(stdout="", returncode=128))
|
||||
assert _check_audit_only_violations() == []
|
||||
|
||||
def test_subprocess_exception_fails_open(self, monkeypatch):
|
||||
"""subprocess.run 자체가 raise 해도 가드 false positive X."""
|
||||
def boom(*a, **kw): raise RuntimeError("git missing")
|
||||
monkeypatch.setattr(subprocess, "run", boom)
|
||||
assert _check_audit_only_violations() == []
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
# AUDIT_ONLY_NOTE constants — sanity
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestAuditOnlyConstants:
|
||||
def test_note_mentions_forbidden_prefixes(self):
|
||||
for p in AUDIT_ONLY_FORBIDDEN_PREFIXES:
|
||||
assert p in AUDIT_ONLY_NOTE, f"AUDIT_ONLY_NOTE missing prefix mention: {p}"
|
||||
|
||||
def test_note_mentions_allowed_paths(self):
|
||||
assert "INTEGRATION-AUDIT-*.md" in AUDIT_ONLY_NOTE
|
||||
assert "PHASE-Z-IMPLEMENTATION-ISSUE-BACKLOG.md" in AUDIT_ONLY_NOTE
|
||||
|
||||
def test_note_states_no_code_edit(self):
|
||||
# "report" 또는 "NOT code" 표현 명시 확인 (LLM 가독성 가드).
|
||||
lower = AUDIT_ONLY_NOTE.lower()
|
||||
assert "audit report" in lower or "report writing" in lower
|
||||
assert "not code" in lower or "no production" in lower
|
||||
|
||||
def test_forbidden_prefixes_no_trailing_slash_issues(self):
|
||||
"""블랙리스트는 startswith 매치 — 'src' (slash 없음) 면 'srcfoo.py' 도 매칭돼서 false positive.
|
||||
모든 prefix 가 '/' 로 끝나야 함."""
|
||||
for p in AUDIT_ONLY_FORBIDDEN_PREFIXES:
|
||||
assert p.endswith("/"), f"prefix '{p}' must end with '/' to avoid false matches"
|
||||
@@ -252,6 +252,52 @@ class TestC6_OrphanGrandchildAfterNormalExit:
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
# C7: input + encoding path — run_claude 가 실제 사용하는 호출 모드.
|
||||
# 2026-05-18 production bug: str input + encoding="utf-8" 일 때
|
||||
# wrapper 가 input 을 강제로 bytes 인코딩 → Popen text mode pipe 에
|
||||
# bytes 쓰려다 TypeError: write() argument must be str, not bytes.
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
|
||||
class TestC7_InputEncodingPath:
|
||||
def test_str_input_with_encoding_utf8(self):
|
||||
"""run_claude 와 동일한 호출 모드 — input=str + encoding='utf-8'."""
|
||||
# stdin 에서 읽은 그대로 stdout 으로 echo. 한글 포함해서 encoding 검증.
|
||||
r = _run_with_tree_kill(
|
||||
[_py(), "-c", "import sys; sys.stdout.write(sys.stdin.read())"],
|
||||
input="hello 안녕\n",
|
||||
encoding="utf-8",
|
||||
timeout=10,
|
||||
)
|
||||
assert r.returncode == 0
|
||||
# encoding= 모드면 stdout 는 str 이어야 함.
|
||||
assert isinstance(r.stdout, str)
|
||||
assert "hello" in r.stdout
|
||||
assert "안녕" in r.stdout
|
||||
|
||||
def test_bytes_input_without_encoding(self):
|
||||
"""encoding 없으면 binary mode — input=bytes 그대로 통과."""
|
||||
r = _run_with_tree_kill(
|
||||
[_py(), "-c", "import sys; sys.stdout.buffer.write(sys.stdin.buffer.read())"],
|
||||
input=b"raw bytes",
|
||||
timeout=10,
|
||||
)
|
||||
assert r.returncode == 0
|
||||
assert isinstance(r.stdout, bytes)
|
||||
assert r.stdout == b"raw bytes"
|
||||
|
||||
def test_str_input_without_encoding_auto_encoded(self):
|
||||
"""input=str 인데 encoding 없으면 wrapper 가 자동 utf-8 인코딩."""
|
||||
r = _run_with_tree_kill(
|
||||
[_py(), "-c", "import sys; sys.stdout.buffer.write(sys.stdin.buffer.read())"],
|
||||
input="auto encode 한글",
|
||||
timeout=10,
|
||||
)
|
||||
assert r.returncode == 0
|
||||
assert isinstance(r.stdout, bytes)
|
||||
assert r.stdout.decode("utf-8") == "auto encode 한글"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
# Bonus: _SPAWNED discipline — 다중 호출 후 누적 안 됨
|
||||
# ─────────────────────────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user