test(file_state): PR 10 — parity tests for evaluate_open/save (amend §D paired)

PYTHON_THINNING_PLAN §5 PR 10. Wave 1.5 amend §D paired parity test PR —
PR 11 (kind_codes 통합 + decision 매핑 lookup table 이관) 의 baseline.

새 테스트 26개 (총 33 = 7 기존 + 26 신규):

evaluate_open_file (9 시나리오):
- DIRECTORY/SYMLINK kind blocked, FILE_TOO_LARGE, size limit boundary,
  zero-byte allow toggle 양방향, NUL byte binary, high ASCII no NUL,
  binary_probe_bytes window 경계.

evaluate_save_file (17 시나리오):
- decision_code 0–5 전체 매트릭스.
- kind_codes 매트릭스: REGULAR_FILE/OTHER 동일 → OK,
  REGULAR→DIRECTORY/SYMLINK kind-specific 우선,
  REGULAR→OTHER 메타데이터 변경.
- size 단독 변경 / mtime 단독 변경 분리.
- baseline=None×candidate=None 경계 (baseline-unknown 우선).
- 사용자 보이는 message 5종 텍스트 핀 (amend A1: Python single source).

PR 11 land 후 본 33개가 *비트 동일하게* 통과해야 한다.

테스트: 33 그린 (parity 26 신규 + file_pipeline 기존 7).
plan: PR 10  표기.

boundary-claim:
  removes: []
  delete-count: 0
  ban-list: 'amend §D paired parity test PR — PR 11의 baseline'
  scenarios-added: 26

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-02 00:20:39 +09:00
parent c19aaaef1a
commit b47f7eba3b
2 changed files with 303 additions and 2 deletions

View File

@@ -12,8 +12,8 @@
> | PR 37 | ✅ | `2238b55` | `_rust_ffi.py` 1452 LOC → 6 모듈 패키지 (각 ≤400 LOC) |
> | PR 5.5 | ✅ | `c29e3f5` | diagnostics 청산은 *이미 일원화됨* — 인벤토리 정정 (no-op) |
> | PR 8 | ✅ | `32fc8ef` | `derive_venv_name` heuristic → `sessions_native::interpreter_probe` (~40 LOC) |
> | PR 9 | ✅ no-op | (TBD) | tree/list 잔여 호출자 0건 확인 — PR 2가 이미 일원화 완료 |
> | PR 10 | ⏭ pending | — | file_state parity tests (테스트-먼저, amend §D) |
> | PR 9 | ✅ no-op | `c19aaae` | tree/list 잔여 호출자 0건 확인 — PR 2가 이미 일원화 완료 |
> | PR 10 | ✅ | (TBD) | file_state parity tests +25 (총 33 시나리오, amend §D paired) |
> | PR 11 | ⏭ pending | — | file_state kind_codes 통합 + decision 매핑 lookup table (~120 LOC) |
> | PR 12 | ⏭ pending | — | eager_hydrate parity tests |
> | **PR 13a** | ⏭ Wave 2 게이트 | — | envelope 스펙 + ref impl + parity (PR-A 본체 가능) |

View File

@@ -0,0 +1,301 @@
"""Parity baseline for ``file_state.evaluate_open_file`` / ``evaluate_save_file``.
Wave 1.5 amend §D paired parity test PR — Python 본체의 *현재 동작*을
fixture로 핀해서 PR 11 (kind_codes 통합 + decision 매핑 lookup table 이관)
이 같은 결과를 반환하는지 보장한다.
기존 ``test_file_pipeline.py`` 7 시나리오를 보존하면서 +25 추가:
- open guard (size, kind, binary head, zero-byte allow toggle, edge sizes).
- save decision (각 decision_code 05 + kind_codes 4종 매트릭스 + boundary).
이관 PR(PR 11) 후에도 본 테스트는 *동일하게* 통과해야 한다.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from sessions.file_state import (
FileOpenGuardrails,
OpenFileRequest,
OpenOutcome,
ReloadChoice,
SaveConflictKind,
SaveFileRequest,
SaveOutcome,
UnsupportedOpenReason,
evaluate_open_file,
evaluate_save_file,
)
from sessions.remote import RemoteFileKind, RemoteFileMetadata
# ---------------------------------------------------------------------------
# evaluate_open_file — guard matrix
# ---------------------------------------------------------------------------
def _open_request(tmp_path: Path, **md_kwargs) -> OpenFileRequest:
md = RemoteFileMetadata(**{"mtime_ns": 1, "size_bytes": 4, **md_kwargs})
return OpenFileRequest(
remote_absolute_path="/r/w/a.txt",
local_cache_path=tmp_path / "a.txt",
remote_metadata=md,
)
def test_open_blocked_when_remote_is_directory(tmp_path: Path) -> None:
req = _open_request(tmp_path, kind=RemoteFileKind.DIRECTORY, size_bytes=4096)
res = evaluate_open_file(req, content_head=b"", guard_limits=FileOpenGuardrails())
assert res.outcome is OpenOutcome.BLOCKED_BY_POLICY
assert res.unsupported_reason is UnsupportedOpenReason.UNSUPPORTED_REMOTE_KIND
def test_open_blocked_when_remote_is_symlink(tmp_path: Path) -> None:
req = _open_request(tmp_path, kind=RemoteFileKind.SYMLINK, size_bytes=64)
res = evaluate_open_file(req, content_head=b"", guard_limits=FileOpenGuardrails())
assert res.outcome is OpenOutcome.BLOCKED_BY_POLICY
assert res.unsupported_reason is UnsupportedOpenReason.UNSUPPORTED_REMOTE_KIND
def test_open_blocked_when_size_exceeds_limit(tmp_path: Path) -> None:
guard = FileOpenGuardrails(max_open_bytes=128)
req = _open_request(tmp_path, size_bytes=1024)
res = evaluate_open_file(req, content_head=b"text", guard_limits=guard)
assert res.outcome is OpenOutcome.BLOCKED_BY_POLICY
assert res.unsupported_reason is UnsupportedOpenReason.FILE_TOO_LARGE
def test_open_ok_at_size_limit_boundary(tmp_path: Path) -> None:
guard = FileOpenGuardrails(max_open_bytes=8)
req = _open_request(tmp_path, size_bytes=8)
res = evaluate_open_file(req, content_head=b"abcdefgh", guard_limits=guard)
assert res.outcome is OpenOutcome.OK
def test_open_blocked_zero_byte_when_disallowed(tmp_path: Path) -> None:
guard = FileOpenGuardrails(allow_empty_files=False)
req = _open_request(tmp_path, size_bytes=0)
res = evaluate_open_file(req, content_head=b"", guard_limits=guard)
assert res.outcome is OpenOutcome.BLOCKED_BY_POLICY
assert res.unsupported_reason is UnsupportedOpenReason.ZERO_BYTE_READ_NOT_ALLOWED
def test_open_ok_zero_byte_when_allowed(tmp_path: Path) -> None:
guard = FileOpenGuardrails(allow_empty_files=True)
req = _open_request(tmp_path, size_bytes=0)
res = evaluate_open_file(req, content_head=b"", guard_limits=guard)
assert res.outcome is OpenOutcome.OK
def test_open_blocked_binary_with_nul_byte(tmp_path: Path) -> None:
req = _open_request(tmp_path, size_bytes=8)
res = evaluate_open_file(
req, content_head=b"good\x00data", guard_limits=FileOpenGuardrails()
)
assert res.outcome is OpenOutcome.BLOCKED_BINARY_HEURISTIC
def test_open_ok_with_high_ascii_no_nul(tmp_path: Path) -> None:
req = _open_request(tmp_path, size_bytes=8)
# 0x80 etc. without NUL — heuristic only flags NUL byte.
res = evaluate_open_file(
req, content_head=b"\x80\x81\x82text", guard_limits=FileOpenGuardrails()
)
assert res.outcome is OpenOutcome.OK
def test_open_binary_probe_window_respected(tmp_path: Path) -> None:
"""Bytes past ``binary_probe_bytes`` must not influence the heuristic."""
guard = FileOpenGuardrails(binary_probe_bytes=4)
req = _open_request(tmp_path, size_bytes=8)
res = evaluate_open_file(req, content_head=b"text\x00more", guard_limits=guard)
assert res.outcome is OpenOutcome.OK
# ---------------------------------------------------------------------------
# evaluate_save_file — kind_codes matrix + decision_code 0..5
# ---------------------------------------------------------------------------
def _save_request(tmp_path: Path, *, baseline=None, candidate=None) -> SaveFileRequest:
return SaveFileRequest(
remote_absolute_path="/r/w/f.py",
local_cache_path=tmp_path / "f.py",
baseline_remote_metadata=baseline,
candidate_remote_metadata=candidate,
)
@pytest.mark.parametrize(
"kind",
[
RemoteFileKind.REGULAR_FILE,
RemoteFileKind.OTHER,
],
)
def test_save_ok_when_metadata_matches_for_kind(
tmp_path: Path, kind: RemoteFileKind
) -> None:
meta = RemoteFileMetadata(mtime_ns=42, size_bytes=128, kind=kind)
res = evaluate_save_file(_save_request(tmp_path, baseline=meta, candidate=meta))
assert res.outcome is SaveOutcome.OK
def test_save_conflict_when_size_changed(tmp_path: Path) -> None:
baseline = RemoteFileMetadata(mtime_ns=1, size_bytes=10)
current = RemoteFileMetadata(mtime_ns=1, size_bytes=20)
res = evaluate_save_file(
_save_request(tmp_path, baseline=baseline, candidate=current)
)
assert res.conflict is not None
assert res.conflict.kind is SaveConflictKind.REMOTE_METADATA_CHANGED
def test_save_conflict_when_only_mtime_differs(tmp_path: Path) -> None:
baseline = RemoteFileMetadata(mtime_ns=1, size_bytes=10)
current = RemoteFileMetadata(mtime_ns=999, size_bytes=10)
res = evaluate_save_file(
_save_request(tmp_path, baseline=baseline, candidate=current)
)
assert res.conflict is not None
assert res.conflict.kind is SaveConflictKind.REMOTE_METADATA_CHANGED
assert (
res.conflict.reload_choice_hint is ReloadChoice.KEEP_LOCAL_AND_OVERWRITE_REMOTE
)
def test_save_conflict_when_kind_changed_to_other(tmp_path: Path) -> None:
baseline = RemoteFileMetadata(
mtime_ns=1, size_bytes=10, kind=RemoteFileKind.REGULAR_FILE
)
current = RemoteFileMetadata(mtime_ns=1, size_bytes=10, kind=RemoteFileKind.OTHER)
res = evaluate_save_file(
_save_request(tmp_path, baseline=baseline, candidate=current)
)
assert res.conflict is not None
assert res.conflict.kind is SaveConflictKind.REMOTE_METADATA_CHANGED
def test_save_conflict_when_path_became_symlink(tmp_path: Path) -> None:
baseline = RemoteFileMetadata(mtime_ns=1, size_bytes=10)
current = RemoteFileMetadata(mtime_ns=1, size_bytes=10, kind=RemoteFileKind.SYMLINK)
res = evaluate_save_file(
_save_request(tmp_path, baseline=baseline, candidate=current)
)
assert res.conflict is not None
assert res.conflict.kind is SaveConflictKind.REMOTE_PATH_IS_SYMLINK
assert res.conflict.reload_choice_hint is ReloadChoice.CANCEL
def test_save_conflict_baseline_unknown_with_candidate_present(
tmp_path: Path,
) -> None:
meta = RemoteFileMetadata(mtime_ns=1, size_bytes=1)
res = evaluate_save_file(_save_request(tmp_path, baseline=None, candidate=meta))
assert res.conflict is not None
assert res.conflict.kind is SaveConflictKind.BASELINE_UNKNOWN
assert res.conflict.reload_choice_hint is ReloadChoice.CANCEL
def test_save_conflict_baseline_unknown_when_both_none(tmp_path: Path) -> None:
"""No baseline takes precedence over remote-missing — see decision_code 1."""
res = evaluate_save_file(_save_request(tmp_path, baseline=None, candidate=None))
assert res.conflict is not None
assert res.conflict.kind is SaveConflictKind.BASELINE_UNKNOWN
def test_save_conflict_remote_missing_message_text(tmp_path: Path) -> None:
"""Pin user-visible message string — Python single-source-of-truth (amend A1)."""
baseline = RemoteFileMetadata(mtime_ns=1, size_bytes=10)
res = evaluate_save_file(_save_request(tmp_path, baseline=baseline, candidate=None))
assert res.conflict is not None
assert "disappeared" in res.conflict.message
assert res.conflict.kind is SaveConflictKind.REMOTE_FILE_MISSING
def test_save_conflict_directory_message_text(tmp_path: Path) -> None:
baseline = RemoteFileMetadata(mtime_ns=1, size_bytes=10)
current = RemoteFileMetadata(
mtime_ns=1, size_bytes=4096, kind=RemoteFileKind.DIRECTORY
)
res = evaluate_save_file(
_save_request(tmp_path, baseline=baseline, candidate=current)
)
assert res.conflict is not None
assert "directory" in res.conflict.message.lower()
def test_save_conflict_symlink_message_text(tmp_path: Path) -> None:
baseline = RemoteFileMetadata(mtime_ns=1, size_bytes=10)
current = RemoteFileMetadata(mtime_ns=1, size_bytes=10, kind=RemoteFileKind.SYMLINK)
res = evaluate_save_file(
_save_request(tmp_path, baseline=baseline, candidate=current)
)
assert res.conflict is not None
assert "symlink" in res.conflict.message.lower()
def test_save_conflict_metadata_changed_message_text(tmp_path: Path) -> None:
baseline = RemoteFileMetadata(mtime_ns=1, size_bytes=10)
current = RemoteFileMetadata(mtime_ns=2, size_bytes=10)
res = evaluate_save_file(
_save_request(tmp_path, baseline=baseline, candidate=current)
)
assert res.conflict is not None
assert "changed" in res.conflict.message.lower()
def test_save_conflict_baseline_unknown_message_text(tmp_path: Path) -> None:
meta = RemoteFileMetadata(mtime_ns=1, size_bytes=1)
res = evaluate_save_file(_save_request(tmp_path, baseline=None, candidate=meta))
assert res.conflict is not None
assert "metadata" in res.conflict.message.lower()
# ---------------------------------------------------------------------------
# kind_codes matrix — every (baseline_kind, candidate_kind) where same →OK,
# differ →METADATA_CHANGED, kind=DIRECTORY/SYMLINK on candidate trigger
# their own conflict variants regardless of size/mtime equality.
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"kind",
[
RemoteFileKind.REGULAR_FILE,
RemoteFileKind.OTHER,
],
)
def test_save_ok_for_same_kind_same_metadata(
tmp_path: Path, kind: RemoteFileKind
) -> None:
meta = RemoteFileMetadata(mtime_ns=7, size_bytes=42, kind=kind)
res = evaluate_save_file(_save_request(tmp_path, baseline=meta, candidate=meta))
assert res.outcome is SaveOutcome.OK
@pytest.mark.parametrize(
"candidate_kind, expected_kind",
[
(RemoteFileKind.DIRECTORY, SaveConflictKind.REMOTE_PATH_IS_DIRECTORY),
(RemoteFileKind.SYMLINK, SaveConflictKind.REMOTE_PATH_IS_SYMLINK),
],
)
def test_save_kind_changed_to_blocked_kind_overrides_metadata_match(
tmp_path: Path,
candidate_kind: RemoteFileKind,
expected_kind: SaveConflictKind,
) -> None:
"""Even with identical (mtime, size), changing kind to dir/symlink trips the
kind-specific conflict — Rust ``save_decision_code`` checks kind *before*
metadata equality."""
baseline = RemoteFileMetadata(
mtime_ns=1, size_bytes=10, kind=RemoteFileKind.REGULAR_FILE
)
candidate = RemoteFileMetadata(mtime_ns=1, size_bytes=10, kind=candidate_kind)
res = evaluate_save_file(
_save_request(tmp_path, baseline=baseline, candidate=candidate)
)
assert res.conflict is not None
assert res.conflict.kind is expected_kind