File size: 4,699 Bytes
4c0ecfa
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
from __future__ import annotations

import json
import subprocess
import tempfile
import unittest
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "run-task-batches.ps1"
DEFAULT_CONFIG = ROOT / "config" / "task-batch-runner.json"


def run_script(config: dict) -> subprocess.CompletedProcess[str]:
    with tempfile.TemporaryDirectory() as tmp:
        config_path = Path(tmp) / "task-batch-runner.json"
        config = {"loadLocalConfig": False, **config}
        config_path.write_text(json.dumps(config), encoding="utf-8")
        return subprocess.run(
            [
                "powershell",
                "-NoProfile",
                "-ExecutionPolicy",
                "Bypass",
                "-File",
                str(SCRIPT),
                "-ConfigPath",
                str(config_path),
                "-DryRun",
                "-PlanJson",
            ],
            cwd=ROOT,
            text=True,
            encoding="utf-8",
            errors="replace",
            capture_output=True,
            check=False,
        )


class TaskBatchRunnerTests(unittest.TestCase):
    def test_dry_run_builds_pipeline_command_with_max_concurrency(self) -> None:
        result = run_script(
            {
                "tasks": ["a", "b", "c", "d", "e", "f", "g"],
                "batchSize": 3,
                "runsDir": "output/custom-runs",
                "maxRounds": 4,
                "timeoutSeconds": 99,
                "timestampPrefix": "unit",
            }
        )

        self.assertEqual(result.returncode, 0, result.stderr)
        plan = json.loads(result.stdout)

        self.assertEqual(plan["tasks"], ["a", "b", "c", "d", "e", "f", "g"])
        self.assertEqual(plan["maxConcurrentTasks"], 3)
        self.assertIs(plan["continueOnFailure"], True)
        command = plan["command"]
        for task in ["a", "b", "c", "d", "e", "f", "g"]:
            self.assertIn(f"'--task' '{task}'", command)
        self.assertIn("'--concurrency' '3'", command)
        self.assertIn("'--runs-dir' 'output/custom-runs'", command)
        self.assertIn("'--max-rounds' '4'", command)
        self.assertIn("'--timeout-seconds' '99'", command)

    def test_rejects_empty_task_list(self) -> None:
        result = run_script({"tasks": [], "batchSize": 3})

        self.assertNotEqual(result.returncode, 0)
        self.assertIn("tasks", result.stderr.lower())

    def test_mri_example_config_runs_requested_tasks_in_one_batch(self) -> None:
        result = run_script(
            {
                "tasks": ["mri_sense", "mri_tv"],
                "batchSize": 3,
                "timestampPrefix": "mri_sense_tv_rerun",
            }
        )

        self.assertEqual(result.returncode, 0, result.stderr)
        plan = json.loads(result.stdout)
        self.assertEqual(plan["tasks"], ["mri_sense", "mri_tv"])
        self.assertEqual(plan["maxConcurrentTasks"], 3)
        self.assertIn("'--task' 'mri_sense'", plan["command"])
        self.assertIn("'--task' 'mri_tv'", plan["command"])

    def test_default_config_is_valid_task_set(self) -> None:
        config = json.loads(DEFAULT_CONFIG.read_text(encoding="utf-8"))

        self.assertIsInstance(config.get("tasks"), list)
        self.assertGreater(len(config["tasks"]), 0)
        self.assertTrue(all(isinstance(task, str) and task.strip() for task in config["tasks"]))
        self.assertEqual(config.get("batchSize"), 3)
        self.assertIs(config.get("continueOnFailure", True), True)

    def test_default_config_dry_run_matches_configured_pipeline(self) -> None:
        config = json.loads(DEFAULT_CONFIG.read_text(encoding="utf-8"))
        tasks = config["tasks"]
        batch_size = config.get("batchSize", 3)

        result = subprocess.run(
            [
                "powershell",
                "-NoProfile",
                "-ExecutionPolicy",
                "Bypass",
                "-File",
                str(SCRIPT),
                "-DryRun",
                "-PlanJson",
            ],
            cwd=ROOT,
            text=True,
            encoding="utf-8",
            errors="replace",
            capture_output=True,
            check=False,
        )

        self.assertEqual(result.returncode, 0, result.stderr)
        plan = json.loads(result.stdout)
        self.assertEqual(plan["tasks"], tasks)
        self.assertEqual(plan["maxConcurrentTasks"], batch_size)
        self.assertIs(plan["continueOnFailure"], True)
        self.assertIn(f"'--task' '{tasks[0]}'", plan["command"])
        self.assertIn(f"'--task' '{tasks[-1]}'", plan["command"])


if __name__ == "__main__":
    unittest.main()