from __future__ import annotations import csv import json from pathlib import Path import sys import tempfile import unittest PROJECT_DEV = Path(__file__).resolve().parents[1] if str(PROJECT_DEV) not in sys.path: sys.path.insert(0, str(PROJECT_DEV)) import hibor_batch_executor as batch_executor STUB_CLI = r''' from __future__ import annotations import argparse import json from pathlib import Path import sys parser = argparse.ArgumentParser() parser.add_argument("--task", type=Path, required=True) parser.add_argument("--execute", action="store_true") args = parser.parse_args() task = json.loads(args.task.read_text(encoding="utf-8")) index = int(task["task_id"].rsplit("-", 1)[1]) fail_index = int(task["source_scope"].get("stub_fail_index", 0)) failed = index == fail_index run_root = Path(task["output_root"]) / f"RUN-STUB-{index:02d}" run_root.mkdir(parents=True) manifest = run_root / "manifest.csv" delivery = run_root / "delivery.md" timing = run_root / "timing.json" terminal_path = run_root / "report_collection_terminal.json" manifest.write_text("item_id,status\nITEM-001," + ("FAILED" if failed else "SUCCESS") + "\n", encoding="utf-8") delivery.write_text("stub\n", encoding="utf-8") timing.write_text(json.dumps({"rows": [ {"stage": "preflight", "elapsed_ms": 10}, {"stage": "ui_scan", "elapsed_ms": 20}, {"stage": "item_1_detail", "elapsed_ms": 30}, {"stage": "item_1_cache", "elapsed_ms": 40}, {"stage": "item_1_copy", "elapsed_ms": 5}, {"stage": "item_1_validation_publish", "elapsed_ms": 6}, {"stage": "manifest", "elapsed_ms": 1}, {"stage": "delivery", "elapsed_ms": 1} ]}, ensure_ascii=False), encoding="utf-8") terminal = { "status": "VALIDATION_FAILED" if failed else "SUCCESS", "stop_code": "PDF_MAGIC_INVALID" if failed else None, "blocker": None, "manifest_path": str(manifest), "delivery_path": str(delivery), "timing_path": str(timing), "terminal_path": str(terminal_path), "terminal_present": True, } terminal_path.write_text(json.dumps(terminal, ensure_ascii=False), encoding="utf-8") sys.stdout.write(json.dumps(terminal, ensure_ascii=False, separators=(",", ":"))) raise SystemExit(16 if failed else 0) ''' def source_scope(fail_index: int = 0) -> dict: return { "ui": { "search_entry_texts": ["研究报告"], "search_input_resource_id": "cn.com.hibor:id/editText1", "search_submit_texts": ["搜索"], "result_resource_id": "cn.com.hibor:id/research_tv_title", "open_texts": ["在线浏览报告原文"], "institution_resource_id": "", "date_resource_id": "cn.com.hibor:id/detail_tv_time", "analysts_resource_id": "cn.com.hibor:id/tv_author", "page_count_resource_id": "", "swipe": [360, 1080, 360, 420, 350], "quota_remaining_resource_id": "", "quota_remaining_pattern": "", }, "reuse_manifest_paths": [], "resume_items": [], "stub_fail_index": fail_index, } def batch_value(batch_id: str, count: int, *, fail_index: int = 0) -> dict: items = [] for index in range(1, count + 1): item = { "item_id": f"report-{index:02d}", "query": f"精确报告{index}", "title": f"精确报告{index}", "publisher": "测试证券", "publication_date": "2026-07-31", "source_id": f"SRC-HIBOR-TEST-{index:02d}", "raw_destination": f"ana-data/cases/test/raw/SRC-HIBOR-TEST-{index:02d}", } if index != 1: item["pages"] = index + 5 items.append(item) return { "schema_version": batch_executor.BATCH_SCHEMA, "batch_id": batch_id, "source_thread_id": "thread-source", "reply_thread_id": "thread-reply", "runtime": { "quota_ledger": "ana-data/tmp/report-collection-control/daily_quota_test.csv", "adb_executable": "adb", "pdfinfo_executable": "pdfinfo", "device_serial": "127.0.0.1:21503", "source_scope": source_scope(fail_index), }, "items": items, } class HiborBatchExecutorTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) (self.root / "ana-data" / "cases").mkdir(parents=True) (self.root / "ana-data" / "tmp").mkdir(parents=True) self.stub = self.root / "stub_cli.py" self.stub.write_text(STUB_CLI, encoding="utf-8") def tearDown(self) -> None: self.temp.cleanup() def run_batch(self, value: dict) -> dict: path = self.root / f"{value['batch_id']}.json" path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8") return batch_executor.execute_batch( path, project_root=self.root, kernel_root=self.root, python_executable=sys.executable, kernel_entry=self.stub, ) def test_all_success_creates_exact_tasks_and_batch_outputs(self): terminal = self.run_batch(batch_value("ok-batch", 3)) batch_root = self.root / "ana-data" / "tmp" / "hibor-runs" / "ok-batch" self.assertEqual((terminal["status"], terminal["succeeded"], terminal["failed"]), ("SUCCESS", 3, 0)) self.assertTrue((batch_root / "batch_terminal.json").is_file()) self.assertTrue((batch_root / "batch_manifest.csv").is_file()) self.assertTrue((batch_root / "batch_delivery.md").is_file()) self.assertTrue((batch_root / "batch_timing.json").is_file()) self.assertEqual(terminal["max_generated_path_chars"], max( len(str(path.resolve())) for path in batch_root.rglob("*") if path.is_file() )) self.assertLessEqual(terminal["max_generated_path_chars"], 220) task = json.loads((batch_root / "t" / "01.json").read_text(encoding="utf-8")) self.assertEqual(task["mode"], "collect-one") self.assertEqual((task["min_screens"], task["normal_max_screens"]), (1, 1)) self.assertEqual(task["expected_reports"][0]["title"], "精确报告1") self.assertEqual(task["expected_reports"][0]["page_count"], 1) with (batch_root / "batch_manifest.csv").open(encoding="utf-8", newline="") as handle: rows = list(csv.DictReader(handle)) self.assertEqual([row["status"] for row in rows], ["SUCCESS"] * 3) self.assertEqual(rows[0]["pages_assumed"], "true") def test_one_failure_continues_and_returns_partial_success(self): terminal = self.run_batch(batch_value("partial", 3, fail_index=2)) batch_root = self.root / "ana-data" / "tmp" / "hibor-runs" / "partial" self.assertEqual((terminal["status"], terminal["exit_code"]), ("PARTIAL_SUCCESS", 2)) self.assertEqual((terminal["succeeded"], terminal["failed"]), (2, 1)) self.assertEqual([item["status"] for item in terminal["items"]], ["SUCCESS", "FAILED", "SUCCESS"]) self.assertTrue((batch_root / "r" / "03").is_dir()) self.assertTrue((batch_root / "batch_terminal.json").is_file()) def test_existing_batch_output_is_never_overwritten(self): value = batch_value("no-overwrite", 1) first = self.run_batch(value) terminal_path = Path(first["terminal_path"]) before = terminal_path.read_bytes() input_path = self.root / "no-overwrite.json" with self.assertRaises(batch_executor.BatchInputError): batch_executor.execute_batch( input_path, project_root=self.root, kernel_root=self.root, python_executable=sys.executable, kernel_entry=self.stub, ) self.assertEqual(terminal_path.read_bytes(), before) def test_invalid_count_duplicate_and_destination_escape_fail_fast(self): value = batch_value("invalid", 1) value["items"] = [] with self.assertRaises(batch_executor.BatchInputError): batch_executor.validate_batch(value, self.root) value = batch_value("duplicate", 2) value["items"][1]["item_id"] = value["items"][0]["item_id"] with self.assertRaises(batch_executor.BatchInputError): batch_executor.validate_batch(value, self.root) value = batch_value("escape", 1) value["items"][0]["raw_destination"] = "../outside" with self.assertRaises(batch_executor.BatchInputError): batch_executor.validate_batch(value, self.root) def test_timing_summary_groups_item_stages(self): timing_one = self.root / "timing-one.json" timing_two = self.root / "timing-two.json" timing_one.write_text(json.dumps({"rows": [ {"stage": "ui_scan", "elapsed_ms": 7}, {"stage": "item_1_detail", "elapsed_ms": 11}, ]}), encoding="utf-8") timing_two.write_text(json.dumps({"rows": [ {"stage": "ui_scan", "elapsed_ms": 13}, {"stage": "item_1_detail", "elapsed_ms": 17}, ]}), encoding="utf-8") summary = batch_executor.summarize_timing_files([timing_one, timing_two]) self.assertEqual(summary["files_read"], 2) self.assertEqual(summary["stage_totals_ms"], {"ui_scan": 20, "detail": 28}) if __name__ == "__main__": unittest.main()