From 2042980bbf75b0eb72b725048536a878ca028ab2 Mon Sep 17 00:00:00 2001
From: Cai <cai@nbcai.cc>
Date: Sat, 22 Aug 2026 10:03:18 +0800
Subject: [PATCH] chore: 记录2026-08-21估值周度增量复评
---
dev/project-dev/test/test_hibor_batch_executor.py | 119 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 117 insertions(+), 2 deletions(-)
diff --git a/dev/project-dev/test/test_hibor_batch_executor.py b/dev/project-dev/test/test_hibor_batch_executor.py
index 4b1b791..bfb8ec5 100644
--- a/dev/project-dev/test/test_hibor_batch_executor.py
+++ b/dev/project-dev/test/test_hibor_batch_executor.py
@@ -3,6 +3,7 @@
import csv
import json
from pathlib import Path
+import subprocess
import sys
import tempfile
import unittest
@@ -110,7 +111,7 @@
"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",
+ "device_serial": "emulator-5554",
"source_scope": source_scope(fail_index),
},
"items": items,
@@ -125,11 +126,32 @@
(self.root / "ana-data" / "tmp").mkdir(parents=True)
self.stub = self.root / "stub_cli.py"
self.stub.write_text(STUB_CLI, encoding="utf-8")
+ self.adb_calls: list[tuple[list[str], Path, int]] = []
def tearDown(self) -> None:
self.temp.cleanup()
- def run_batch(self, value: dict) -> dict:
+ def make_device_runner(
+ self,
+ stdout: bytes = b"List of devices attached\nemulator-5554\tdevice product:sdk\n",
+ *,
+ returncode: int = 0,
+ stderr: bytes = b"",
+ ):
+ def device_runner(command, cwd, environment, timeout_seconds):
+ self.adb_calls.append((list(command), cwd, timeout_seconds))
+ return subprocess.CompletedProcess(
+ list(command), returncode, stdout=stdout, stderr=stderr
+ )
+
+ return device_runner
+
+ def run_batch(
+ self,
+ value: dict,
+ *,
+ adb_stdout: bytes = b"List of devices attached\nemulator-5554\tdevice product:sdk\n",
+ ) -> 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(
@@ -138,6 +160,7 @@
kernel_root=self.root,
python_executable=sys.executable,
kernel_entry=self.stub,
+ device_runner=self.make_device_runner(adb_stdout),
)
def test_all_success_creates_exact_tasks_and_batch_outputs(self):
@@ -159,6 +182,10 @@
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)
+ self.assertEqual(task["device_serial"], "emulator-5554")
+ self.assertEqual(self.adb_calls, [
+ (["adb", "devices", "-l"], self.root, batch_executor.ADB_PREFLIGHT_TIMEOUT_SECONDS)
+ ])
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)
@@ -221,6 +248,94 @@
self.assertEqual(summary["files_read"], 2)
self.assertEqual(summary["stage_totals_ms"], {"ui_scan": 20, "detail": 28})
+ def test_explicit_and_auto_device_resolution(self):
+ multiple = (
+ b"List of devices attached\n"
+ b"emulator-5554\tdevice product:sdk\n"
+ b"emulator-5556\tdevice product:sdk\n"
+ )
+ explicit = batch_executor.resolve_device_serial(
+ "adb",
+ "emulator-5554",
+ cwd=self.root,
+ environment={},
+ runner=self.make_device_runner(multiple),
+ )
+ self.assertEqual(explicit, "emulator-5554")
+
+ value = batch_value("auto-single", 1)
+ value["runtime"]["device_serial"] = "auto"
+ terminal = self.run_batch(value)
+ self.assertEqual(terminal["status"], "SUCCESS")
+ task = json.loads(
+ (self.root / "ana-data" / "tmp" / "hibor-runs" / "auto-single" / "t" / "01.json")
+ .read_text(encoding="utf-8")
+ )
+ self.assertEqual(task["device_serial"], "emulator-5554")
+
+ def test_auto_zero_and_multiple_devices_fail_closed_before_output(self):
+ cases = {
+ "zero": b"List of devices attached\n\n",
+ "multiple": (
+ b"List of devices attached\n"
+ b"emulator-5554\tdevice\n"
+ b"emulator-5556\tdevice\n"
+ ),
+ }
+ for batch_id, stdout in cases.items():
+ with self.subTest(batch_id=batch_id):
+ value = batch_value(batch_id, 1)
+ value["runtime"]["device_serial"] = "auto"
+ path = self.root / f"{batch_id}.json"
+ path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
+ with self.assertRaises(batch_executor.BatchInputError):
+ batch_executor.execute_batch(
+ path,
+ project_root=self.root,
+ kernel_root=self.root,
+ python_executable=sys.executable,
+ kernel_entry=self.stub,
+ device_runner=self.make_device_runner(stdout),
+ )
+ self.assertFalse(
+ (self.root / "ana-data" / "tmp" / "hibor-runs" / batch_id).exists()
+ )
+
+ def test_unready_missing_and_failed_adb_preflight_fail_closed(self):
+ cases = (
+ (
+ "emulator-5554",
+ b"List of devices attached\nemulator-5554\toffline\n",
+ 0,
+ b"",
+ ),
+ (
+ "emulator-5554",
+ b"List of devices attached\nemulator-5556\tdevice\n",
+ 0,
+ b"",
+ ),
+ (
+ "auto",
+ b"List of devices attached\nemulator-5554\tunauthorized\n",
+ 0,
+ b"",
+ ),
+ ("auto", b"", 1, b"adb unavailable"),
+ )
+ for requested, stdout, returncode, stderr in cases:
+ with self.subTest(requested=requested, stdout=stdout, returncode=returncode):
+ with self.assertRaises(batch_executor.BatchInputError):
+ batch_executor.resolve_device_serial(
+ "adb",
+ requested,
+ cwd=self.root,
+ environment={},
+ runner=self.make_device_runner(
+ stdout, returncode=returncode, stderr=stderr
+ ),
+ )
+
if __name__ == "__main__":
unittest.main()
--
Gitblit v1.9.3