from __future__ import annotations
|
|
from pathlib import Path
|
import tempfile
|
import unittest
|
|
from hibor_fast_collection.adb import RemoteFile
|
from hibor_fast_collection.cache import CacheWatcher
|
from hibor_fast_collection.models import Candidate, EvidenceState, PackageState, TerminalReceipt, TerminalStatus
|
from hibor_fast_collection.selection import FastScanner, ScanPage
|
from hibor_fast_collection.terminal import (
|
TERMINAL_KEYS_V003, TERMINAL_KEYS_V009, build_terminal, canonical_terminal_bytes,
|
legacy_v003_projection,
|
)
|
|
|
class FakeClock:
|
def __init__(self): self.value = 0.0
|
def __call__(self): return self.value
|
def sleep(self, amount): self.value += amount
|
|
|
class CacheSelectionTerminalTests(unittest.TestCase):
|
def test_cache_stability_and_ambiguity(self):
|
calls = [
|
(),
|
(RemoteFile("/c/a", "a", 10, "t1"),),
|
(RemoteFile("/c/a", "a", 10, "t1"),),
|
]
|
clock = FakeClock()
|
watcher = CacheWatcher(lambda: calls.pop(0) if calls else (), clock=clock, sleep=clock.sleep)
|
baseline = watcher.baseline()
|
match = watcher.wait_for_unique_stable(baseline, timeout_ms=5_000, poll_ms=100)
|
self.assertEqual(match.status, "UNIQUE_STABLE")
|
self.assertEqual(match.remote.name, "a")
|
|
def test_fast_scanner_early_stop(self):
|
pages = {
|
1: ScanPage((Candidate("a", "中国中免报告", None, None, score=100),), "1"),
|
2: ScanPage((Candidate("b", "中国中免深度", None, None, score=110),), "2"),
|
3: ScanPage((), "3"),
|
}
|
outcome = FastScanner(lambda n: pages[n]).scan(quantity=2, min_screens=3, normal_max_screens=5)
|
self.assertEqual((outcome.status, outcome.screens_scanned), ("ENOUGH_CONFIRMED", 3))
|
|
def test_terminal_51_and_legacy_43(self):
|
target = Path("t.json")
|
receipt = TerminalReceipt(target, None, False, False, EvidenceState.N, False, False, True,
|
None, None, None, PackageState.P00_INIT)
|
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": {}})
|
terminal = build_terminal(base, receipt, status=TerminalStatus.SUCCESS, stop_code=None)
|
self.assertEqual(tuple(terminal), TERMINAL_KEYS_V009)
|
self.assertEqual(len(terminal), 51)
|
legacy = legacy_v003_projection(terminal)
|
self.assertEqual((len(legacy), tuple(legacy)), (43, TERMINAL_KEYS_V003))
|
self.assertTrue(canonical_terminal_bytes(terminal).startswith(b'{"schema_version"'))
|
|
|
if __name__ == "__main__": unittest.main()
|
