chore(project-info): archive Bilibili dynamic ce9be69f16d9

MB-X Bilibili Pipeline

6 days ago 5c75bc6b3fe4eda9b1210574230d22249ecf949f

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