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()
|
