from __future__ import annotations
|
|
from pathlib import Path
|
import sys
|
import tempfile
|
import unittest
|
|
from hibor_fast_collection.models import EvidenceState, PackageState, TerminalReceipt, TerminalStatus
|
from hibor_fast_collection.budget import Budget
|
from hibor_fast_collection.process import ProcessSupervisor
|
from hibor_fast_collection.terminal import TERMINAL_KEYS_V003, TerminalWriter, build_terminal
|
|
|
def terminal_value(target: Path):
|
base = {key: None for key in TERMINAL_KEYS_V003}
|
base.update({
|
"project_id": "project-info", "task_id": "T", "handoff_id": "H",
|
"items": [], "prohibited_action_attestation": {},
|
})
|
intended = TerminalReceipt(target, target, True, True, EvidenceState.V, True, False,
|
True, 2, "0" * 64, None, PackageState.P10_CLOSED)
|
return build_terminal(base, intended, status=TerminalStatus.SUCCESS, stop_code=None)
|
|
|
class ProcessTerminalWriterTests(unittest.TestCase):
|
def test_process_success_and_stream_separation(self):
|
result = ProcessSupervisor().run(
|
(sys.executable, "-c", "import sys;sys.stdout.buffer.write(b'OUT');sys.stderr.buffer.write(b'ERR')"),
|
timeout_ms=5_000,
|
)
|
self.assertTrue(result.started and result.exited)
|
self.assertEqual((result.exit_code, result.stdout_bytes, result.stderr_bytes), (0, b"OUT", b"ERR"))
|
self.assertIsNone(result.error_code)
|
|
def test_process_timeout_is_bounded_and_terminal(self):
|
result = ProcessSupervisor(kill_wait_ms=3_000).run(
|
(sys.executable, "-c", "import time;time.sleep(5)"), timeout_ms=100,
|
)
|
self.assertTrue(result.started)
|
self.assertTrue(result.timed_out)
|
self.assertTrue(result.terminate_issued)
|
self.assertTrue(result.exited or result.liveness_unknown)
|
self.assertIsNotNone(result.error_code)
|
|
def test_terminal_exact_replay_valid_and_drift_invalid(self):
|
with tempfile.TemporaryDirectory() as temp:
|
target = Path(temp) / "terminal.json"
|
value = terminal_value(target)
|
writer = TerminalWriter()
|
first = writer.persist(target, value, source_state=PackageState.P09_TIMING)
|
second = writer.persist(target, value, source_state=PackageState.P09_TIMING)
|
self.assertEqual((first.presence, second.presence), (EvidenceState.V, EvidenceState.V))
|
target.write_bytes(target.read_bytes() + b"\n")
|
third = writer.persist(target, value, source_state=PackageState.P09_TIMING)
|
self.assertEqual(third.presence, EvidenceState.I)
|
self.assertFalse(third.persisted)
|
|
def test_terminal_spent_close_budget_returns_no_attempt_receipt(self):
|
class Clock:
|
value = 0.0
|
def __call__(self): return self.value
|
with tempfile.TemporaryDirectory() as temp:
|
target = Path(temp) / "terminal.json"
|
value = terminal_value(target)
|
clock = Clock()
|
budget = Budget(100, 10, clock=clock)
|
clock.value = 0.101
|
receipt = TerminalWriter().persist(
|
target, value, source_state=PackageState.P09_TIMING,
|
checkpoint=lambda: budget.checkpoint("close", "terminal", batch_close=True),
|
)
|
self.assertEqual((receipt.presence, receipt.persist_attempted,
|
receipt.persisted, receipt.error_code),
|
(EvidenceState.N, False, False, None))
|
self.assertFalse(target.exists())
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
