111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""Unit tests for agent_team.state_store (§6.7)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from agent_team import state_store
|
|
from agent_team.state_store import (
|
|
IntegrityError,
|
|
atomic_write,
|
|
compute_content_hash,
|
|
read_checked,
|
|
write_checked,
|
|
)
|
|
|
|
|
|
def test_compute_content_hash_is_sha256_hex() -> None:
|
|
import hashlib
|
|
|
|
data = b"hello world"
|
|
assert compute_content_hash(data) == hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def test_compute_content_hash_distinguishes_inputs() -> None:
|
|
assert compute_content_hash(b"a") != compute_content_hash(b"b")
|
|
|
|
|
|
def test_atomic_write_creates_file_with_exact_bytes(tmp_path: Path) -> None:
|
|
target = tmp_path / "state.bin"
|
|
payload = b"\x00\x01binary\xff"
|
|
atomic_write(target, payload)
|
|
assert target.read_bytes() == payload
|
|
|
|
|
|
def test_atomic_write_creates_missing_parent_dirs(tmp_path: Path) -> None:
|
|
target = tmp_path / "nested" / "deep" / "state.bin"
|
|
atomic_write(target, b"x")
|
|
assert target.read_bytes() == b"x"
|
|
|
|
|
|
def test_atomic_write_overwrites_existing(tmp_path: Path) -> None:
|
|
target = tmp_path / "state.bin"
|
|
atomic_write(target, b"old")
|
|
atomic_write(target, b"new-and-longer")
|
|
assert target.read_bytes() == b"new-and-longer"
|
|
|
|
|
|
def test_atomic_write_leaves_no_temp_files(tmp_path: Path) -> None:
|
|
target = tmp_path / "state.bin"
|
|
atomic_write(target, b"data")
|
|
leftovers = [p for p in tmp_path.iterdir() if p.name != "state.bin"]
|
|
assert leftovers == []
|
|
|
|
|
|
def test_write_then_read_checked_roundtrip(tmp_path: Path) -> None:
|
|
target = tmp_path / "state.bin"
|
|
payload = json.dumps({"k": "v"}).encode()
|
|
write_checked(target, payload, schema_version=3)
|
|
assert read_checked(target, schema_version=3) == payload
|
|
|
|
|
|
def test_read_checked_schema_version_mismatch_raises(tmp_path: Path) -> None:
|
|
target = tmp_path / "state.bin"
|
|
write_checked(target, b"data", schema_version=1)
|
|
with pytest.raises(IntegrityError):
|
|
read_checked(target, schema_version=2)
|
|
|
|
|
|
def test_read_checked_content_corruption_raises(tmp_path: Path) -> None:
|
|
target = tmp_path / "state.bin"
|
|
write_checked(target, b"original", schema_version=1)
|
|
# Corrupt the payload without touching the sidecar -> hash mismatch.
|
|
target.write_bytes(b"tampered")
|
|
with pytest.raises(IntegrityError):
|
|
read_checked(target, schema_version=1)
|
|
|
|
|
|
def test_read_checked_missing_payload_raises(tmp_path: Path) -> None:
|
|
with pytest.raises(IntegrityError):
|
|
read_checked(tmp_path / "nope.bin", schema_version=1)
|
|
|
|
|
|
def test_read_checked_missing_sidecar_raises(tmp_path: Path) -> None:
|
|
target = tmp_path / "state.bin"
|
|
# Plain atomic_write writes payload but NOT the integrity sidecar.
|
|
atomic_write(target, b"data")
|
|
with pytest.raises(IntegrityError):
|
|
read_checked(target, schema_version=1)
|
|
|
|
|
|
def test_read_checked_garbled_sidecar_raises(tmp_path: Path) -> None:
|
|
target = tmp_path / "state.bin"
|
|
write_checked(target, b"data", schema_version=1)
|
|
meta_path = target.with_name(target.name + ".meta.json")
|
|
meta_path.write_bytes(b"not-json{{{")
|
|
with pytest.raises(IntegrityError):
|
|
read_checked(target, schema_version=1)
|
|
|
|
|
|
def test_module_exports_public_contract() -> None:
|
|
for name in (
|
|
"IntegrityError",
|
|
"atomic_write",
|
|
"compute_content_hash",
|
|
"read_checked",
|
|
):
|
|
assert name in state_store.__all__
|
|
assert hasattr(state_store, name)
|