test_core_ci.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
| 1 | """Tests for muse.core.ci — CI gate runner and .muse/ci.toml loading. |
| 2 | |
| 3 | Coverage: |
| 4 | - load_ci_config returns DEFAULT_CONFIG when file missing. |
| 5 | - load_ci_config parses a valid .muse/ci.toml correctly. |
| 6 | - load_ci_config raises ValueError on invalid TOML. |
| 7 | - load_ci_config raises ValueError when gate command is not a list of strings. |
| 8 | - _parse_gate validates required 'command' field. |
| 9 | - _parse_settings applies defaults for missing fields. |
| 10 | - run_ci executes all gates and returns CiRunResult. |
| 11 | - run_ci marks overall as passed when all required gates pass. |
| 12 | - run_ci marks overall as failed when any required gate fails. |
| 13 | - run_ci skips non-required gate failure in overall result. |
| 14 | - GateResult has all required fields. |
| 15 | - timeout enforcement marks gate as timed_out. |
| 16 | """ |
| 17 | |
| 18 | from __future__ import annotations |
| 19 | |
| 20 | import pathlib |
| 21 | import sys |
| 22 | |
| 23 | import pytest |
| 24 | |
| 25 | from muse.core.ci import ( |
| 26 | CiConfig, |
| 27 | CiGate, |
| 28 | CiRunResult, |
| 29 | CiSettings, |
| 30 | GateResult, |
| 31 | _DEFAULT_CONFIG, |
| 32 | _parse_gate, |
| 33 | _parse_settings, |
| 34 | load_ci_config, |
| 35 | run_ci, |
| 36 | ) |
| 37 | from muse.core.store import MsgpackValue |
| 38 | |
| 39 | |
| 40 | # --------------------------------------------------------------------------- |
| 41 | # Unit tests — _parse_gate |
| 42 | # --------------------------------------------------------------------------- |
| 43 | |
| 44 | |
| 45 | class TestParseGate: |
| 46 | def test_minimal_gate(self) -> None: |
| 47 | raw: MsgpackValue = {"command": ["echo", "hi"], "name": "echo"} |
| 48 | gate = _parse_gate(raw, 0) |
| 49 | assert gate["name"] == "echo" |
| 50 | assert gate["command"] == ["echo", "hi"] |
| 51 | assert gate["required"] is True |
| 52 | assert gate["timeout_s"] == 0.0 |
| 53 | |
| 54 | def test_full_gate(self) -> None: |
| 55 | raw: MsgpackValue = { |
| 56 | "name": "tests", |
| 57 | "command": [sys.executable, "-m", "pytest", "tests/"], |
| 58 | "timeout_s": 300.0, |
| 59 | "required": False, |
| 60 | } |
| 61 | gate = _parse_gate(raw, 0) |
| 62 | assert gate["name"] == "tests" |
| 63 | assert gate["timeout_s"] == 300.0 |
| 64 | assert gate["required"] is False |
| 65 | |
| 66 | def test_missing_command_raises(self) -> None: |
| 67 | raw: MsgpackValue = {"name": "bad"} |
| 68 | with pytest.raises(ValueError, match="'command'"): |
| 69 | _parse_gate(raw, 0) |
| 70 | |
| 71 | def test_empty_command_raises(self) -> None: |
| 72 | raw: MsgpackValue = {"name": "empty", "command": []} |
| 73 | with pytest.raises(ValueError, match="must not be empty"): |
| 74 | _parse_gate(raw, 0) |
| 75 | |
| 76 | def test_non_string_command_raises(self) -> None: |
| 77 | raw: MsgpackValue = {"name": "bad", "command": [123, "pytest"]} |
| 78 | with pytest.raises(ValueError, match="list of strings"): |
| 79 | _parse_gate(raw, 0) |
| 80 | |
| 81 | def test_non_dict_raises(self) -> None: |
| 82 | with pytest.raises(ValueError, match="must be a TOML table"): |
| 83 | non_dict: MsgpackValue = "not a dict" |
| 84 | _parse_gate(non_dict, 0) |
| 85 | |
| 86 | def test_default_name_when_missing(self) -> None: |
| 87 | raw: MsgpackValue = {"command": ["echo", "hi"]} |
| 88 | gate = _parse_gate(raw, 5) |
| 89 | assert gate["name"] == "gate-5" |
| 90 | |
| 91 | |
| 92 | # --------------------------------------------------------------------------- |
| 93 | # Unit tests — _parse_settings |
| 94 | # --------------------------------------------------------------------------- |
| 95 | |
| 96 | |
| 97 | class TestParseSettings: |
| 98 | def test_defaults_on_empty(self) -> None: |
| 99 | settings = _parse_settings({}) |
| 100 | assert settings["test_budget_s"] == 300.0 |
| 101 | assert settings["workers"] == 1 |
| 102 | assert settings["env_allowlist"] == [] |
| 103 | |
| 104 | def test_parses_values(self) -> None: |
| 105 | raw: MsgpackValue = { |
| 106 | "test_budget_s": 120.0, |
| 107 | "workers": 4, |
| 108 | "env_allowlist": ["CI", "MY_VAR"], |
| 109 | } |
| 110 | settings = _parse_settings(raw) |
| 111 | assert settings["test_budget_s"] == 120.0 |
| 112 | assert settings["workers"] == 4 |
| 113 | assert settings["env_allowlist"] == ["CI", "MY_VAR"] |
| 114 | |
| 115 | def test_non_dict_returns_defaults(self) -> None: |
| 116 | settings = _parse_settings(None) |
| 117 | assert settings["workers"] == 1 |
| 118 | |
| 119 | |
| 120 | # --------------------------------------------------------------------------- |
| 121 | # Integration tests — load_ci_config |
| 122 | # --------------------------------------------------------------------------- |
| 123 | |
| 124 | |
| 125 | class TestLoadCiConfig: |
| 126 | def test_missing_file_returns_default(self, tmp_path: pathlib.Path) -> None: |
| 127 | (tmp_path / ".muse").mkdir() |
| 128 | config = load_ci_config(tmp_path) |
| 129 | assert config["version"] == _DEFAULT_CONFIG["version"] |
| 130 | assert len(config["gates"]) == len(_DEFAULT_CONFIG["gates"]) |
| 131 | |
| 132 | def test_parses_valid_toml(self, tmp_path: pathlib.Path) -> None: |
| 133 | muse_dir = tmp_path / ".muse" |
| 134 | muse_dir.mkdir() |
| 135 | toml_content = """\ |
| 136 | version = 1 |
| 137 | |
| 138 | [settings] |
| 139 | test_budget_s = 60 |
| 140 | workers = 2 |
| 141 | env_allowlist = ["CI"] |
| 142 | |
| 143 | [[gate]] |
| 144 | name = "echo-check" |
| 145 | command = ["echo", "hello"] |
| 146 | timeout_s = 5 |
| 147 | required = true |
| 148 | """ |
| 149 | (muse_dir / "ci.toml").write_text(toml_content) |
| 150 | config = load_ci_config(tmp_path) |
| 151 | assert config["version"] == 1 |
| 152 | assert config["settings"]["workers"] == 2 |
| 153 | assert config["settings"]["test_budget_s"] == 60.0 |
| 154 | assert len(config["gates"]) == 1 |
| 155 | gate = config["gates"][0] |
| 156 | assert gate["name"] == "echo-check" |
| 157 | assert gate["command"] == ["echo", "hello"] |
| 158 | assert gate["timeout_s"] == 5.0 |
| 159 | assert gate["required"] is True |
| 160 | |
| 161 | def test_invalid_toml_raises(self, tmp_path: pathlib.Path) -> None: |
| 162 | muse_dir = tmp_path / ".muse" |
| 163 | muse_dir.mkdir() |
| 164 | (muse_dir / "ci.toml").write_text("this is [ not valid toml !!!{{}}") |
| 165 | with pytest.raises(ValueError, match="Failed to parse"): |
| 166 | load_ci_config(tmp_path) |
| 167 | |
| 168 | def test_non_table_toml_raises(self, tmp_path: pathlib.Path) -> None: |
| 169 | """A TOML file that doesn't produce a dict at top level raises.""" |
| 170 | # Note: TOML always produces a dict at top level; this tests the |
| 171 | # error path for defensive code. |
| 172 | muse_dir = tmp_path / ".muse" |
| 173 | muse_dir.mkdir() |
| 174 | # Write syntactically valid TOML (always a dict, so test the guard |
| 175 | # is unreachable, but let's ensure we parse it without error). |
| 176 | (muse_dir / "ci.toml").write_text("version = 1\n") |
| 177 | config = load_ci_config(tmp_path) |
| 178 | assert config["version"] == 1 |
| 179 | |
| 180 | |
| 181 | # --------------------------------------------------------------------------- |
| 182 | # Integration tests — run_ci |
| 183 | # --------------------------------------------------------------------------- |
| 184 | |
| 185 | |
| 186 | class TestRunCi: |
| 187 | def _make_config( |
| 188 | self, gates: list[CiGate], settings: CiSettings | None = None |
| 189 | ) -> CiConfig: |
| 190 | return CiConfig( |
| 191 | version=1, |
| 192 | settings=settings or CiSettings( |
| 193 | test_budget_s=300.0, |
| 194 | workers=1, |
| 195 | env_allowlist=[], |
| 196 | ), |
| 197 | gates=gates, |
| 198 | ) |
| 199 | |
| 200 | def test_all_gates_pass(self, tmp_path: pathlib.Path) -> None: |
| 201 | config = self._make_config([ |
| 202 | CiGate( |
| 203 | name="echo1", |
| 204 | command=["echo", "pass"], |
| 205 | timeout_s=5.0, |
| 206 | required=True, |
| 207 | ), |
| 208 | ]) |
| 209 | result = run_ci(tmp_path, config) |
| 210 | assert result["passed"] is True |
| 211 | assert len(result["gates"]) == 1 |
| 212 | assert result["gates"][0]["passed"] is True |
| 213 | |
| 214 | def test_required_gate_failure_marks_overall_failed( |
| 215 | self, tmp_path: pathlib.Path |
| 216 | ) -> None: |
| 217 | config = self._make_config([ |
| 218 | CiGate( |
| 219 | name="fail-gate", |
| 220 | command=[sys.executable, "-c", "raise SystemExit(1)"], |
| 221 | timeout_s=5.0, |
| 222 | required=True, |
| 223 | ), |
| 224 | ]) |
| 225 | result = run_ci(tmp_path, config) |
| 226 | assert result["passed"] is False |
| 227 | assert result["gates"][0]["passed"] is False |
| 228 | |
| 229 | def test_non_required_gate_failure_does_not_fail_overall( |
| 230 | self, tmp_path: pathlib.Path |
| 231 | ) -> None: |
| 232 | config = self._make_config([ |
| 233 | CiGate( |
| 234 | name="pass-gate", |
| 235 | command=["echo", "ok"], |
| 236 | timeout_s=5.0, |
| 237 | required=True, |
| 238 | ), |
| 239 | CiGate( |
| 240 | name="warn-gate", |
| 241 | command=[sys.executable, "-c", "raise SystemExit(1)"], |
| 242 | timeout_s=5.0, |
| 243 | required=False, |
| 244 | ), |
| 245 | ]) |
| 246 | result = run_ci(tmp_path, config) |
| 247 | assert result["passed"] is True |
| 248 | warn = next(g for g in result["gates"] if g["name"] == "warn-gate") |
| 249 | assert warn["passed"] is False |
| 250 | assert "warning" in warn |
| 251 | |
| 252 | def test_gate_result_structure(self, tmp_path: pathlib.Path) -> None: |
| 253 | config = self._make_config([ |
| 254 | CiGate( |
| 255 | name="echo", |
| 256 | command=["echo", "hello"], |
| 257 | timeout_s=5.0, |
| 258 | required=True, |
| 259 | ), |
| 260 | ]) |
| 261 | result = run_ci(tmp_path, config) |
| 262 | g = result["gates"][0] |
| 263 | assert isinstance(g["name"], str) |
| 264 | assert isinstance(g["command"], list) |
| 265 | assert isinstance(g["exit_code"], int) |
| 266 | assert isinstance(g["duration_ms"], float) |
| 267 | assert isinstance(g["passed"], bool) |
| 268 | assert isinstance(g["timed_out"], bool) |
| 269 | assert isinstance(g["stdout"], str) |
| 270 | assert isinstance(g["stderr"], str) |
| 271 | |
| 272 | def test_ci_run_result_structure(self, tmp_path: pathlib.Path) -> None: |
| 273 | config = self._make_config([ |
| 274 | CiGate( |
| 275 | name="echo", |
| 276 | command=["echo", "hi"], |
| 277 | timeout_s=5.0, |
| 278 | required=True, |
| 279 | ), |
| 280 | ]) |
| 281 | result = run_ci(tmp_path, config) |
| 282 | assert isinstance(result["passed"], bool) |
| 283 | assert isinstance(result["gates"], list) |
| 284 | assert isinstance(result["total_duration_ms"], float) |
| 285 | assert isinstance(result["timestamp"], str) |
| 286 | assert result["total_duration_ms"] >= 0.0 |
| 287 | |
| 288 | def test_timeout_marks_timed_out(self, tmp_path: pathlib.Path) -> None: |
| 289 | """A gate that exceeds timeout_s is marked timed_out.""" |
| 290 | import time |
| 291 | config = self._make_config([ |
| 292 | CiGate( |
| 293 | name="slow", |
| 294 | command=[sys.executable, "-c", "import time; time.sleep(30)"], |
| 295 | timeout_s=0.1, |
| 296 | required=True, |
| 297 | ), |
| 298 | ]) |
| 299 | result = run_ci(tmp_path, config) |
| 300 | slow_gate = result["gates"][0] |
| 301 | assert slow_gate["timed_out"] is True |
| 302 | assert result["passed"] is False |
| 303 | |
| 304 | def test_empty_gates_passes(self, tmp_path: pathlib.Path) -> None: |
| 305 | """CI with no gates is trivially passing.""" |
| 306 | config = self._make_config([]) |
| 307 | result = run_ci(tmp_path, config) |
| 308 | assert result["passed"] is True |
| 309 | assert result["gates"] == [] |
| 310 | |
| 311 | def test_multiple_gates_all_executed(self, tmp_path: pathlib.Path) -> None: |
| 312 | """All gates run even when earlier ones fail (full picture collection).""" |
| 313 | config = self._make_config([ |
| 314 | CiGate( |
| 315 | name="gate-1", |
| 316 | command=[sys.executable, "-c", "raise SystemExit(1)"], |
| 317 | timeout_s=5.0, |
| 318 | required=True, |
| 319 | ), |
| 320 | CiGate( |
| 321 | name="gate-2", |
| 322 | command=["echo", "still runs"], |
| 323 | timeout_s=5.0, |
| 324 | required=True, |
| 325 | ), |
| 326 | ]) |
| 327 | result = run_ci(tmp_path, config) |
| 328 | assert len(result["gates"]) == 2 |
| 329 | names = {g["name"] for g in result["gates"]} |
| 330 | assert "gate-1" in names |
| 331 | assert "gate-2" in names |
File History
5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e
Merge branch 'dev' into main
Human
29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce
Merge branch 'dev' into main
Human
48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa
feat: Muse — version control for the agent era
Human
100 days ago