from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
from pathlib import Path
|
import tempfile
|
import unittest
|
from unittest.mock import patch
|
|
import hibor_fast_collection.workflow as workflow_module
|
from hibor_fast_collection.adb import RemoteFile
|
from hibor_fast_collection.budget import Budget
|
from hibor_fast_collection.cache import CacheMatch
|
from hibor_fast_collection.models import Candidate, ContractError, TaskSpec
|
from hibor_fast_collection.quota import AppQuotaObservation, QuotaLedger
|
from hibor_fast_collection.selection import ScanOutcome
|
from hibor_fast_collection.workflow import HiborWorkflow, _identity
|
|
from helpers import task_spec
|
|
|
PDF = b"%PDF-1.4\n1 0 obj<<>>endobj\ntrailer<<>>\n%%EOF\n"
|
|
|
class FakeAdb:
|
def __init__(self):
|
self.triggered = False
|
self.remote = RemoteFile(
|
"/sdcard/Android/data/cn.com.hibor/files/myfile/report.pdf",
|
"report.pdf", len(PDF), "2026-07-29 10:00",
|
)
|
|
def preflight(self):
|
return object()
|
|
def screenshot(self, target: Path):
|
target.parent.mkdir(parents=True, exist_ok=True)
|
target.write_bytes(b"\x89PNG\r\n\x1a\nFAKE")
|
|
def list_cache(self):
|
return (self.remote,) if self.triggered else ()
|
|
def remote_sha256(self, remote_path: str):
|
self.assert_path(remote_path)
|
return hashlib.sha256(PDF).hexdigest()
|
|
def pull(self, remote_path: str, local_staging: Path, *, timeout_ms: int):
|
self.assert_path(remote_path)
|
local_staging.parent.mkdir(parents=True, exist_ok=True)
|
local_staging.write_bytes(PDF)
|
return object()
|
|
def assert_path(self, value: str):
|
if value != self.remote.path:
|
raise AssertionError(value)
|
|
|
class FakeUi:
|
def __init__(self, adb: FakeAdb):
|
self.adb = adb
|
self.candidate = Candidate("CANDIDATE-001", "中国中免深度研究", None, None, score=120,
|
bounds=(1, 1, 2, 2))
|
|
def search(self, query: str):
|
self.query = query
|
|
def scan(self, **kwargs):
|
return ScanOutcome("ENOUGH_CONFIRMED", (self.candidate,), 3, 1, 0)
|
|
def open_detail(self, candidate: Candidate):
|
return Candidate(candidate.candidate_id, candidate.title, "测试证券", "2026-07-29",
|
("分析师甲",), 1, candidate.score, candidate.bounds)
|
|
def trigger_current_detail(self):
|
self.adb.triggered = True
|
self.trigger_calls = getattr(self, "trigger_calls", 0) + 1
|
|
|
class WorkflowFakeE2ETests(unittest.TestCase):
|
def test_one_report_complete_without_external_process(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="collect-one")
|
raw["destination"] = str(root / "archive")
|
spec = TaskSpec.from_mapping(raw)
|
adb = FakeAdb()
|
result = HiborWorkflow(spec).collect(adb=adb, ui=FakeUi(adb), execute=True,
|
synthetic=True)
|
self.assertEqual((result["status"], result["exit_code"]), ("SUCCESS", 0))
|
self.assertEqual((result["triggered"], result["succeeded"]), (1, 1))
|
self.assertEqual(len(result["items"]), 1)
|
self.assertEqual(result["items"][0]["sha256"], hashlib.sha256(PDF).hexdigest())
|
self.assertTrue(Path(result["manifest_path"]).is_file())
|
self.assertTrue(Path(result["delivery_path"]).is_file())
|
self.assertTrue(Path(result["timing_path"]).is_file())
|
self.assertTrue(Path(result["terminal_path"]).is_file())
|
parsed = json.loads(Path(result["terminal_path"]).read_text(encoding="utf-8"))
|
self.assertEqual(parsed["status"], "SUCCESS")
|
ledger = spec.quota_path.read_text(encoding="utf-8")
|
self.assertIn("BASELINE_ESTIMATE", ledger)
|
self.assertIn("CONSUME_CONFIRMED", ledger)
|
self.assertIn("ARTIFACT_SUCCESS", ledger)
|
|
def test_quota_exhaustion_stops_before_trigger_and_writes_failure_manifest(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="collect-one")
|
raw["destination"] = str(root / "archive")
|
spec = TaskSpec.from_mapping(raw)
|
QuotaLedger(spec.quota_path).initialize(
|
task_id=spec.task_id, requester_role=spec.requester,
|
handoff_id=spec.handoff_id, known_floor=25, evidence_ref="TEST-FLOOR-25",
|
)
|
adb = FakeAdb()
|
result = HiborWorkflow(spec).collect(adb=adb, ui=FakeUi(adb), execute=True,
|
synthetic=True)
|
self.assertEqual(result["status"], "PARTIAL_QUOTA_STOP")
|
self.assertEqual((result["triggered"], adb.triggered), (0, False))
|
self.assertEqual(result["items"][0]["stop_code"], "QUOTA_EXHAUSTED")
|
self.assertTrue(Path(result["manifest_path"]).is_file())
|
|
def test_cache_timeout_marks_consumption_uncertain(self):
|
class ImmediateTimeout:
|
def __init__(self, list_files): self.list_files = list_files
|
def baseline(self): return {}
|
def wait_for_unique_stable(self, baseline, *, timeout_ms):
|
return CacheMatch("NONE_TIMEOUT", None, 1, ())
|
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="collect-one")
|
raw["destination"] = str(root / "archive")
|
spec = TaskSpec.from_mapping(raw)
|
adb = FakeAdb()
|
result = HiborWorkflow(spec).collect(
|
adb=adb, ui=FakeUi(adb), execute=True, synthetic=True,
|
watcher_factory=ImmediateTimeout,
|
)
|
self.assertEqual((result["status"], result["exit_code"]), ("STATE_UNCERTAIN", 27))
|
self.assertEqual((result["triggered"], result["quota_uncertain"]), (0, 1))
|
self.assertEqual(result["items"][0]["quota_state"], "UNCERTAIN")
|
self.assertTrue(Path(result["manifest_path"]).is_file())
|
|
def test_confirmed_quota_survives_post_confirm_validation_failure(self):
|
class BadRemoteHashAdb(FakeAdb):
|
def remote_sha256(self, remote_path: str):
|
self.assert_path(remote_path)
|
return "0" * 64
|
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="collect-one")
|
raw["destination"] = str(root / "archive")
|
spec = TaskSpec.from_mapping(raw)
|
adb = BadRemoteHashAdb()
|
result = HiborWorkflow(spec).collect(
|
adb=adb, ui=FakeUi(adb), execute=True, synthetic=True,
|
)
|
item = result["items"][0]
|
self.assertEqual((item["quota_state"], item["stop_code"]),
|
("CONFIRMED", "HASH_MISMATCH"))
|
self.assertIsNotNone(item["quota_terminal_event_id"])
|
self.assertIsNotNone(item["quota_artifact_event_id"])
|
ledger = spec.quota_path.read_text(encoding="utf-8")
|
self.assertIn("CONSUME_CONFIRMED", ledger)
|
self.assertIn("ARTIFACT_DUPLICATE_OR_FAILED", ledger)
|
self.assertNotIn(",RELEASE,", ledger)
|
with Path(result["manifest_path"]).open("r", encoding="utf-8", newline="") as stream:
|
row = next(csv.DictReader(stream))
|
self.assertEqual(row["quota_terminal_event_id"], item["quota_terminal_event_id"])
|
self.assertEqual(row["quota_artifact_event_id"], item["quota_artifact_event_id"])
|
self.assertEqual((row["bytes"], row["sha256"], row["relative_path"]),
|
("", "", ""))
|
|
def test_artifact_success_facts_survive_item_close_deadline(self):
|
class Clock:
|
value = 0.0
|
def __call__(self): return self.value
|
|
clock = Clock()
|
|
class AdvanceAfterArtifactQuota(QuotaLedger):
|
def artifact(self, **kwargs):
|
event = super().artifact(**kwargs)
|
clock.value = 600.001
|
return event
|
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="batch", quantity=2)
|
raw["destination"] = str(root / "archive")
|
spec = TaskSpec.from_mapping(raw)
|
budget = Budget(spec.total_budget_ms, spec.close_reserve_ms,
|
batch_increment_ms=spec.batch_increment_budget_ms,
|
clock=clock)
|
adb = FakeAdb()
|
with patch.object(workflow_module, "QuotaLedger", AdvanceAfterArtifactQuota):
|
result = HiborWorkflow(spec, budget=budget).collect(
|
adb=adb, ui=FakeUi(adb), execute=True, synthetic=True,
|
)
|
item = result["items"][0]
|
expected_sha = hashlib.sha256(PDF).hexdigest()
|
self.assertEqual((result["status"], item["stop_code"], item["quota_state"]),
|
("TIME_BUDGET_STOP", "DEADLINE_EXPIRED", "CONFIRMED"), result)
|
self.assertEqual((item["bytes"], item["sha256"]), (len(PDF), expected_sha))
|
self.assertTrue(Path(item["final_path"]).is_file())
|
self.assertIsNotNone(item["quota_artifact_event_id"])
|
ledger = spec.quota_path.read_text(encoding="utf-8")
|
self.assertEqual(ledger.count("ARTIFACT_SUCCESS"), 1)
|
self.assertNotIn("ARTIFACT_DUPLICATE_OR_FAILED", ledger)
|
with Path(result["manifest_path"]).open("r", encoding="utf-8", newline="") as stream:
|
row = next(csv.DictReader(stream))
|
self.assertEqual((row["download_status"], row["status"], row["stop_code"]),
|
("SUCCESS", "SUCCESS", "DEADLINE_EXPIRED"))
|
self.assertEqual((int(row["bytes"]), row["sha256"], row["relative_path"]),
|
(len(PDF), expected_sha, item["final_path"]))
|
self.assertEqual(row["quota_artifact_event_id"], item["quota_artifact_event_id"])
|
|
def test_execution_requires_explicit_gate(self):
|
with tempfile.TemporaryDirectory() as temp:
|
spec = TaskSpec.from_mapping(task_spec(Path(temp), mode="collect-one"))
|
with self.assertRaisesRegex(Exception, "explicit True"):
|
HiborWorkflow(spec).collect(adb=object(), ui=object())
|
|
def test_hard_filter_rejects_excluded_title(self):
|
with tempfile.TemporaryDirectory() as temp:
|
raw = task_spec(Path(temp), mode="collect-one")
|
raw["exclude"] = ["禁止词"]
|
spec = TaskSpec.from_mapping(raw)
|
candidate = Candidate("C", "中国中免禁止词报告", "测试证券", "2026-07-29",
|
("分析师甲",), 8, 120, (1, 1, 2, 2))
|
with self.assertRaises(ContractError):
|
HiborWorkflow(spec)._enforce_hard_filters(candidate)
|
|
def test_completed_request_replay_never_taps_or_mutates_ledger(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="collect-one")
|
raw["destination"] = str(root / "archive")
|
spec = TaskSpec.from_mapping(raw)
|
adb = FakeAdb()
|
ui = FakeUi(adb)
|
first = HiborWorkflow(spec).collect(adb=adb, ui=ui, execute=True, synthetic=True)
|
ledger_before = spec.quota_path.read_bytes()
|
second_adb = FakeAdb()
|
second_ui = FakeUi(second_adb)
|
second = HiborWorkflow(spec).collect(adb=second_adb, ui=second_ui,
|
execute=True, synthetic=True)
|
self.assertEqual((first["run_id"], second["run_id"]),
|
(second["run_id"], second["run_id"]))
|
self.assertFalse(second_adb.triggered)
|
self.assertFalse(hasattr(second_ui, "trigger_calls"))
|
self.assertEqual(spec.quota_path.read_bytes(), ledger_before)
|
|
def test_project_pure_reuse_passes_at_safe_zero_without_adb_or_ledger_write(self):
|
class NoAdb:
|
def set_timeout_provider(self, provider): self.provider = provider
|
def preflight(self): raise AssertionError("pure project reuse must not use ADB")
|
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
first_raw = task_spec(root, mode="collect-one")
|
first_raw["destination"] = str(root / "archive")
|
first_spec = TaskSpec.from_mapping(first_raw)
|
adb = FakeAdb()
|
first = HiborWorkflow(first_spec).collect(adb=adb, ui=FakeUi(adb),
|
execute=True, synthetic=True)
|
quota_date = QuotaLedger(first_spec.quota_path).snapshot().quota_date
|
ledger = QuotaLedger(first_spec.quota_path, quota_date=quota_date)
|
observation = AppQuotaObservation.build(
|
device_serial="emulator-5554", quota_date=ledger.quota_date,
|
captured_at_utc="2026-07-29T02:00:00Z", visible_remaining=0,
|
ui_snapshot_fingerprint="a" * 64,
|
)
|
ledger.observe_app_remaining(task_id="OBS", requester_role="R", handoff_id="H",
|
observation=observation)
|
before = first_spec.quota_path.read_bytes()
|
second_raw = task_spec(root, mode="collect-one")
|
second_raw["task_id"] = "TASK-TEST-REUSE"
|
second_raw["handoff_id"] = "HANDOFF-TEST-REUSE"
|
second_raw["performance_slot_id"] = "SLOT-REUSE"
|
second_raw["destination"] = str(root / "archive")
|
second_spec = TaskSpec.from_mapping(second_raw)
|
result = HiborWorkflow(second_spec).collect(
|
adb=NoAdb(), ui=object(), execute=True, synthetic=True,
|
)
|
self.assertEqual((result["status"], result["triggered"], result["succeeded"]),
|
("SUCCESS", 0, 1))
|
self.assertTrue(result["items"][0]["reused_without_new_trigger"])
|
self.assertEqual(result["quota_safe_available"], 0)
|
self.assertEqual(second_spec.quota_path.read_bytes(), before)
|
|
def test_manifest_conflict_is_not_exposed_as_verified_path(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="collect-one")
|
raw["destination"] = str(root / "archive")
|
spec = TaskSpec.from_mapping(raw)
|
workflow = HiborWorkflow(spec)
|
conflict = spec.output_path / workflow._run_id() / "manifest.csv"
|
conflict.parent.mkdir(parents=True)
|
conflict.write_bytes(b"CONFLICT")
|
adb = FakeAdb()
|
result = workflow.collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True)
|
self.assertNotEqual(result["status"], "SUCCESS")
|
self.assertIsNone(result["manifest_path"])
|
persisted = json.loads(Path(result["terminal_path"]).read_text(encoding="utf-8"))
|
self.assertIsNone(persisted["manifest_path"])
|
|
def test_delivery_and_timing_conflicts_preserve_only_verified_prefix(self):
|
for conflict_name, expected in (("delivery.md", (True, False, False)),
|
("timing.json", (True, True, False))):
|
with self.subTest(conflict_name=conflict_name), tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="collect-one")
|
raw["task_id"] = f"TASK-{conflict_name}"
|
raw["destination"] = str(root / "archive")
|
spec = TaskSpec.from_mapping(raw)
|
workflow = HiborWorkflow(spec)
|
run_root = spec.output_path / workflow._run_id()
|
path = run_root / conflict_name
|
path.parent.mkdir(parents=True)
|
path.write_bytes(b"CONFLICT")
|
adb = FakeAdb()
|
result = workflow.collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True)
|
actual = (result["manifest_path"] is not None,
|
result["delivery_path"] is not None,
|
result["timing_path"] is not None)
|
self.assertEqual(actual, expected)
|
self.assertNotEqual(result["status"], "SUCCESS")
|
|
def test_existing_invalid_terminal_stops_before_adb_and_trigger(self):
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="collect-one")
|
raw["destination"] = str(root / "archive")
|
spec = TaskSpec.from_mapping(raw)
|
workflow = HiborWorkflow(spec)
|
terminal = spec.output_path / workflow._run_id() / "report_collection_terminal.json"
|
terminal.parent.mkdir(parents=True)
|
terminal.write_bytes(b"CONFLICT")
|
adb = FakeAdb()
|
result = workflow.collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True)
|
self.assertEqual(result["terminal_presence"], "I")
|
self.assertEqual(result["status"], "INTERNAL_ERROR")
|
self.assertFalse(adb.triggered)
|
|
def test_expired_work_budget_emits_terminal_before_adb(self):
|
class Clock:
|
value = 0.0
|
def __call__(self): return self.value
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
spec = TaskSpec.from_mapping(task_spec(root, mode="collect-one"))
|
clock = Clock()
|
budget = Budget(600, 100, clock=clock)
|
workflow = HiborWorkflow(spec, budget=budget)
|
clock.value = 0.51
|
adb = FakeAdb()
|
result = workflow.collect(adb=adb, ui=FakeUi(adb), execute=True, synthetic=True)
|
self.assertEqual((result["status"], result["stop_code"]),
|
("TIME_BUDGET_STOP", "DEADLINE_EXPIRED"))
|
self.assertFalse(adb.triggered)
|
|
def test_expired_batch_close_returns_in_memory_t00b_without_raise(self):
|
class Clock:
|
value = 0.0
|
def __call__(self): return self.value
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
spec = TaskSpec.from_mapping(task_spec(root, mode="collect-one"))
|
clock = Clock()
|
budget = Budget(600, 100, clock=clock)
|
workflow = HiborWorkflow(spec, budget=budget)
|
clock.value = 0.601
|
adb = FakeAdb()
|
result = workflow.collect(adb=adb, ui=FakeUi(adb),
|
execute=True, synthetic=True)
|
self.assertEqual((result["status"], result["capability_status"],
|
result["exit_code"], result["stop_code"]),
|
("TIME_BUDGET_STOP", "TIME_BUDGET_STOP", 10,
|
"DEADLINE_EXPIRED"))
|
self.assertEqual((result["terminal_presence"],
|
result["terminal_persist_attempted"]), ("N", False))
|
self.assertFalse(adb.triggered)
|
|
def test_recorded_cache_pure_reuse_does_not_touch_quota_or_ui(self):
|
class CacheAdb(FakeAdb):
|
def set_timeout_provider(self, provider): self.provider = provider
|
def list_cache(self): return (self.remote,)
|
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
raw = task_spec(root, mode="resume-postprocess")
|
raw["destination"] = str(root / "archive")
|
candidate = Candidate("CANDIDATE-001", f"{raw['query']} FAST REPORT",
|
"娴嬭瘯璇佸埜", "2026-07-29",
|
("鍒嗘瀽甯堢敳",), 1)
|
remote_path = "/sdcard/Android/data/cn.com.hibor/files/myfile/report.pdf"
|
raw["source_scope"] = {"resume_items": [{
|
"candidate_id": candidate.candidate_id,
|
"report_identity": _identity(candidate), "title": candidate.title,
|
"institution": candidate.institution, "report_date": candidate.report_date,
|
"analysts": list(candidate.analysts), "page_count": candidate.page_count,
|
"remote_path": remote_path, "remote_bytes": len(PDF),
|
"remote_sha256": hashlib.sha256(PDF).hexdigest(),
|
}]}
|
spec = TaskSpec.from_mapping(raw)
|
adb = CacheAdb()
|
result = HiborWorkflow(spec).collect(adb=adb, ui=object(), execute=True,
|
synthetic=True)
|
self.assertEqual((result["status"], result["triggered"], result["succeeded"]),
|
("SUCCESS", 0, 1))
|
self.assertTrue(result["items"][0]["reused_without_new_trigger"])
|
self.assertFalse(spec.quota_path.exists())
|
|
def test_mixed_batch_keeps_reuse_then_quota_stops_new_item_without_tap(self):
|
class NewCandidateUi(FakeUi):
|
def __init__(self, adb):
|
super().__init__(adb)
|
self.candidate = Candidate("CANDIDATE-NEW", f"{self.query if hasattr(self, 'query') else ''} NEW", None, None,
|
score=120, bounds=(1, 1, 2, 2))
|
def scan(self, **kwargs):
|
title = f"{kwargs['query']} NEW REPORT"
|
self.candidate = Candidate("CANDIDATE-NEW", title, None, None,
|
score=120, bounds=(1, 1, 2, 2))
|
return ScanOutcome("ENOUGH_CONFIRMED", (self.candidate,), 3, 1, 0)
|
def open_detail(self, candidate):
|
return Candidate(candidate.candidate_id, candidate.title, "NEW-INSTITUTION",
|
"2026-07-29", ("NEW-ANALYST",), 1,
|
candidate.score, candidate.bounds)
|
|
with tempfile.TemporaryDirectory() as temp:
|
root = Path(temp)
|
first_raw = task_spec(root, mode="collect-one")
|
first_raw["destination"] = str(root / "archive")
|
first_spec = TaskSpec.from_mapping(first_raw)
|
first_adb = FakeAdb()
|
HiborWorkflow(first_spec).collect(adb=first_adb, ui=FakeUi(first_adb),
|
execute=True, synthetic=True)
|
quota_date = QuotaLedger(first_spec.quota_path).snapshot().quota_date
|
ledger = QuotaLedger(first_spec.quota_path, quota_date=quota_date)
|
observation = AppQuotaObservation.build(
|
device_serial="emulator-5554", quota_date=quota_date,
|
captured_at_utc="2026-07-29T03:00:00Z", visible_remaining=0,
|
ui_snapshot_fingerprint="b" * 64,
|
)
|
ledger.observe_app_remaining(task_id="OBS", requester_role="R", handoff_id="H",
|
observation=observation)
|
batch_raw = task_spec(root, mode="batch", quantity=2)
|
batch_raw["task_id"] = "TASK-MIXED"
|
batch_raw["handoff_id"] = "HANDOFF-MIXED"
|
batch_raw["destination"] = str(root / "archive")
|
batch_spec = TaskSpec.from_mapping(batch_raw)
|
adb = FakeAdb()
|
result = HiborWorkflow(batch_spec).collect(adb=adb, ui=NewCandidateUi(adb),
|
execute=True, synthetic=True)
|
self.assertEqual((result["succeeded"], result["triggered"]), (1, 0))
|
self.assertIn(result["status"], {"PARTIAL_QUOTA_STOP", "PARTIAL_SUCCESS"})
|
self.assertFalse(adb.triggered)
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
