feat: support batch repetition and selective case import

tests/unit/admin_console/test_automation.py

@@ -185,6 +185,107 @@ def test_preferences_and_required_device(client):
     assert client.post("/api/automation/batches", json=payload(device_serial="")).status_code == 422
 
 
+def test_batch_repetitions_are_ordered_independent_and_idempotent(client, isolated):
+    request = payload(repeat_count=3, request_id="repeat-test")
+    created = client.post("/api/automation/batches", json=request)
+    assert created.status_code == 200
+    batch = created.json()
+    assert (batch["case_count"], batch["repeat_count"], batch["total_executions"]) == (2, 3, 6)
+    jobs = isolated.jobs(batch["id"])
+    assert len({j["session_id"] for j in jobs}) == 6
+    assert [j["position"] for j in jobs] == list(range(1, 7))
+    assert [(j["iteration"], j["case_position"]) for j in jobs] == [
+        (1, 1),
+        (1, 2),
+        (2, 1),
+        (2, 2),
+        (3, 1),
+        (3, 2),
+    ]
+    assert [j["case"]["goal"] for j in jobs] == ["观察设置", "观察首页"] * 3
+    assert all(j["status"] == "pending" for j in jobs)
+    assert client.post("/api/automation/batches", json=request).json()["id"] == batch["id"]
+    assert len(isolated.jobs(batch["id"])) == 6
+    assert (
+        client.post("/api/automation/batches", json={**request, "repeat_count": 2}).status_code
+        == 409
+    )
+    # Each selected failure is copied once, including failures from different rounds.
+    for job in (jobs[0], jobs[4]):
+        isolated.save_job({**job, "status": "failed"})
+    retry = service.batch_action(batch["id"], "retry_failed")
+    assert retry["repeat_count"] == 1
+    assert len(isolated.jobs(retry["id"])) == 2
+    report = reports.session_report(jobs[4]["session_id"])
+    assert "第 3 / 3 轮" in report["name"]
+    archive = reports.export_batch(batch["id"])
+    with zipfile.ZipFile(archive) as z:
+        assert len([n for n in z.namelist() if n.endswith(".xlsx")]) == 7
+        assert len(set(z.namelist())) == len(z.namelist())
+
+
+@pytest.mark.parametrize("count", [0, -1, 1.5, True, "3", None])
+def test_batch_rejects_invalid_repetitions(client, isolated, count):
+    assert (
+        client.post("/api/automation/batches", json=payload(repeat_count=count)).status_code == 422
+    )
+    assert isolated.records("batch") == []
+
+
+def test_batch_limit_is_checked_before_creating_records(client, isolated, monkeypatch):
+    monkeypatch.setattr(service.settings, "ARTEMIS_BATCH_MAX_EXECUTIONS", 5)
+    assert client.get("/api/automation/preferences").json()["batch_max_executions"] == 5
+    response = client.post("/api/automation/batches", json=payload(repeat_count=3))
+    assert response.status_code == 409
+    assert "5" in response.json()["detail"]
+    assert not isolated.records("batch")
+    assert not isolated.jobs()
+
+
+@pytest.mark.asyncio
+async def test_loop_dispatch_continues_after_failure_and_resume_without_repeating(
+    isolated, monkeypatch
+):
+    from apps.admin_console import server
+    from artemis.runtime.device_pool import DeviceStatus
+
+    batch = service.create_batch(payload(repeat_count=2))
+    service.batch_action(batch["id"], "resume")
+    monkeypatch.setattr(service, "state", type("State", (), {"is_paused": False})())
+    enqueue = AsyncMock(return_value={"tasks": [{"device_serial": "test-device"}]})
+    monkeypatch.setattr(server.task_queue_service, "enqueue_tasks", enqueue)
+    monkeypatch.setattr(
+        "artemis.runtime.device_pool.try_list_devices_async",
+        AsyncMock(return_value=[DeviceStatus(serial="test-device", state="device")]),
+    )
+    for index in range(4):
+        await service.tick_batches()
+        await service.tick_batches()  # Polling must not enqueue the same execution twice.
+        assert enqueue.await_count == index + 1
+        job = isolated.jobs(batch["id"])[index]
+        assert enqueue.call_args.kwargs["session_id"] == job["session_id"]
+        assert f"第 {job['iteration']} / 2 轮" in enqueue.call_args.kwargs["display_name"]
+        isolated.save_job({**job, "status": "failed" if index == 0 else "completed"})
+        if index == 1:
+            service.batch_action(batch["id"], "pause")
+            await service.tick_batches()
+            assert enqueue.await_count == 2
+            service.batch_action(batch["id"], "resume")
+    await service.tick_batches()
+    assert isolated.get("batch", batch["id"])["status"] == "completed"
+    assert enqueue.await_count == 4
+
+
+@pytest.mark.asyncio
+async def test_repeat_fail_fast_skips_future_rounds(isolated):
+    batch = service.create_batch(payload(repeat_count=3, stop_on_failure=True))
+    first = isolated.jobs(batch["id"])[0]
+    isolated.save_job({**first, "status": "failed"})
+    service.batch_action(batch["id"], "resume")
+    await service.tick_batches()
+    assert [j["status"] for j in isolated.jobs(batch["id"])] == ["failed"] + ["skipped"] * 5
+
+
 def test_batch_delete_removes_tasks_and_files_and_prevents_request_replay(client, isolated):
     created = client.post(
         "/api/automation/batches", json=payload(request_id="delete-example")