test_switch_supercharge.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """SUPERCHARGE tests for ``muse switch``. |
| 2 | |
| 3 | Gaps addressed beyond the existing test_cmd_switch.py: |
| 4 | |
| 5 | Unit |
| 6 | U1 duration_ms present and float in all JSON success paths |
| 7 | U2 exit_code present and 0 in all JSON success paths |
| 8 | U3 JSON error emitted to stdout when --json + branch not found |
| 9 | U4 JSON error emitted to stdout when --json + no PREV_BRANCH |
| 10 | U5 JSON error emitted to stdout when --json + mutual-exclusion violation |
| 11 | U6 commit_id is sha256:-prefixed in JSON output |
| 12 | U7 from_branch always present and correct in JSON |
| 13 | |
| 14 | Integration |
| 15 | I1 switch - + --json emits valid JSON with correct action |
| 16 | I2 switch -C existing branch + --json → action == "reset" |
| 17 | I3 switch -C new branch + --json → action == "created" |
| 18 | I4 switch -c + --intent stores intent on new branch |
| 19 | I5 switch -c + --resumable marks branch resumable |
| 20 | I6 --autoshelf + --json emits valid JSON |
| 21 | I7 --merge + --json emits valid JSON (smoke — merge may yield conflict) |
| 22 | I8 dry-run -C + --json emits JSON without creating branch |
| 23 | I9 --json + --dry-run switch - emits JSON (prev branch preview) |
| 24 | I10 switch -c on existing branch + --json → JSON error (not text traceback) |
| 25 | |
| 26 | Security |
| 27 | S1 null byte in branch name rejected (exit non-zero) |
| 28 | S2 path traversal (../) in branch name rejected |
| 29 | S3 JSON error output contains no raw ANSI bytes |
| 30 | S4 branch name sanitized in JSON output values |
| 31 | |
| 32 | Data integrity |
| 33 | D1 PREV_BRANCH correct after 10 alternating switches |
| 34 | D2 commit_id in JSON matches actual HEAD after switch |
| 35 | D3 duration_ms is float not int |
| 36 | D4 exit_code is int not bool |
| 37 | D5 force-create action field: "reset" when branch existed, "created" when new |
| 38 | |
| 39 | Stress |
| 40 | P1 50 rapid switches among 3 branches — final state consistent |
| 41 | P2 switch with 30 branches in the repo completes |
| 42 | P3 duration_ms present in all 10 rapid JSON calls |
| 43 | |
| 44 | Concurrent |
| 45 | C1 4 threads switching in separate repos — all succeed |
| 46 | """ |
| 47 | |
| 48 | from __future__ import annotations |
| 49 | from collections.abc import Mapping |
| 50 | |
| 51 | import json |
| 52 | import os |
| 53 | import pathlib |
| 54 | import threading |
| 55 | |
| 56 | import pytest |
| 57 | |
| 58 | from tests.cli_test_helper import CliRunner, InvokeResult |
| 59 | from muse.core.refs import ( |
| 60 | get_head_commit_id, |
| 61 | read_current_branch, |
| 62 | ) |
| 63 | from muse.core.paths import muse_dir |
| 64 | |
| 65 | runner = CliRunner() |
| 66 | |
| 67 | # --------------------------------------------------------------------------- |
| 68 | # Helpers |
| 69 | # --------------------------------------------------------------------------- |
| 70 | |
| 71 | |
| 72 | _CHDIR_LOCK = threading.Lock() |
| 73 | |
| 74 | |
| 75 | def _env(repo: pathlib.Path) -> Mapping[str, str]: |
| 76 | return {"MUSE_REPO_ROOT": str(repo)} |
| 77 | |
| 78 | |
| 79 | def _invoke(repo: pathlib.Path, *args: str) -> InvokeResult: |
| 80 | return runner.invoke(None, ["switch", *args], env=_env(repo)) |
| 81 | |
| 82 | |
| 83 | def _run(repo: pathlib.Path, *args: str) -> InvokeResult: |
| 84 | return runner.invoke(None, list(args), env=_env(repo)) |
| 85 | |
| 86 | |
| 87 | def _init_repo(tmp: pathlib.Path) -> pathlib.Path: |
| 88 | tmp.mkdir(parents=True, exist_ok=True) |
| 89 | with _CHDIR_LOCK: |
| 90 | saved = os.getcwd() |
| 91 | try: |
| 92 | os.chdir(tmp) |
| 93 | runner.invoke(None, ["init"]) |
| 94 | finally: |
| 95 | os.chdir(saved) |
| 96 | (tmp / "a.py").write_text("x = 1\n") |
| 97 | _run(tmp, "commit", "-m", "initial") |
| 98 | return tmp |
| 99 | |
| 100 | |
| 101 | def _add_branch(repo: pathlib.Path, name: str) -> None: |
| 102 | _run(repo, "branch", name) |
| 103 | |
| 104 | |
| 105 | def _switch(repo: pathlib.Path, *args: str) -> InvokeResult: |
| 106 | return _invoke(repo, *args) |
| 107 | |
| 108 | |
| 109 | def _json_switch(repo: pathlib.Path, *args: str) -> Mapping[str, object]: |
| 110 | result = _invoke(repo, "--json", *args) |
| 111 | return result, json.loads(result.stdout) if result.stdout.strip() else {} |
| 112 | |
| 113 | |
| 114 | @pytest.fixture() |
| 115 | def repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 116 | return _init_repo(tmp_path) |
| 117 | |
| 118 | |
| 119 | @pytest.fixture() |
| 120 | def two_branch_repo(repo: pathlib.Path) -> pathlib.Path: |
| 121 | _add_branch(repo, "feat") |
| 122 | _run(repo, "checkout", "feat") |
| 123 | (repo / "feat.py").write_text("f = 1\n") |
| 124 | _run(repo, "commit", "-m", "feat commit") |
| 125 | _run(repo, "checkout", "main") |
| 126 | return repo |
| 127 | |
| 128 | |
| 129 | # --------------------------------------------------------------------------- |
| 130 | # U1–U2 duration_ms and exit_code in all JSON success paths |
| 131 | # --------------------------------------------------------------------------- |
| 132 | |
| 133 | |
| 134 | class TestElapsedAndExitCode: |
| 135 | def test_U1_duration_ms_switched(self, two_branch_repo: pathlib.Path) -> None: |
| 136 | result, data = _json_switch(two_branch_repo, "feat") |
| 137 | assert result.exit_code == 0 |
| 138 | assert "duration_ms" in data, f"duration_ms missing; got keys: {list(data)}" |
| 139 | |
| 140 | def test_U1_duration_ms_created(self, repo: pathlib.Path) -> None: |
| 141 | result, data = _json_switch(repo, "-c", "new-feat") |
| 142 | assert result.exit_code == 0 |
| 143 | assert "duration_ms" in data |
| 144 | |
| 145 | def test_U1_duration_ms_force_create_new(self, repo: pathlib.Path) -> None: |
| 146 | result, data = _json_switch(repo, "-C", "brand-new") |
| 147 | assert result.exit_code == 0 |
| 148 | assert "duration_ms" in data |
| 149 | |
| 150 | def test_U1_duration_ms_force_create_existing( |
| 151 | self, two_branch_repo: pathlib.Path |
| 152 | ) -> None: |
| 153 | result, data = _json_switch(two_branch_repo, "-C", "feat") |
| 154 | assert result.exit_code == 0 |
| 155 | assert "duration_ms" in data |
| 156 | |
| 157 | def test_U1_duration_ms_dry_run(self, two_branch_repo: pathlib.Path) -> None: |
| 158 | result, data = _json_switch(two_branch_repo, "--dry-run", "feat") |
| 159 | assert result.exit_code == 0 |
| 160 | assert "duration_ms" in data |
| 161 | |
| 162 | def test_U2_exit_code_in_json(self, two_branch_repo: pathlib.Path) -> None: |
| 163 | result, data = _json_switch(two_branch_repo, "feat") |
| 164 | assert result.exit_code == 0 |
| 165 | assert "exit_code" in data |
| 166 | assert data["exit_code"] == 0 |
| 167 | |
| 168 | def test_D3_duration_ms_is_float(self, two_branch_repo: pathlib.Path) -> None: |
| 169 | _, data = _json_switch(two_branch_repo, "feat") |
| 170 | assert isinstance(data["duration_ms"], float) |
| 171 | |
| 172 | def test_D4_exit_code_is_int_not_bool(self, two_branch_repo: pathlib.Path) -> None: |
| 173 | _, data = _json_switch(two_branch_repo, "feat") |
| 174 | assert isinstance(data["exit_code"], int) |
| 175 | assert not isinstance(data["exit_code"], bool) |
| 176 | |
| 177 | |
| 178 | # --------------------------------------------------------------------------- |
| 179 | # U3–U5 JSON error output on failure |
| 180 | # --------------------------------------------------------------------------- |
| 181 | |
| 182 | |
| 183 | class TestJsonErrors: |
| 184 | def test_U3_branch_not_found_json_error(self, repo: pathlib.Path) -> None: |
| 185 | result = _invoke(repo, "--json", "no-such-branch") |
| 186 | assert result.exit_code != 0 |
| 187 | data = json.loads(result.stdout) |
| 188 | assert "error" in data |
| 189 | assert data["exit_code"] != 0 |
| 190 | |
| 191 | def test_U4_no_prev_branch_json_error(self, repo: pathlib.Path) -> None: |
| 192 | result = _invoke(repo, "--json", "-") |
| 193 | assert result.exit_code != 0 |
| 194 | data = json.loads(result.stdout) |
| 195 | assert "error" in data |
| 196 | |
| 197 | def test_U5_mutual_exclusion_c_and_C_json_error(self, repo: pathlib.Path) -> None: |
| 198 | result = _invoke(repo, "--json", "-c", "-C", "foo") |
| 199 | assert result.exit_code != 0 |
| 200 | data = json.loads(result.stdout) |
| 201 | assert "error" in data |
| 202 | |
| 203 | def test_U5_mutual_exclusion_discard_merge_json_error( |
| 204 | self, repo: pathlib.Path |
| 205 | ) -> None: |
| 206 | result = _invoke(repo, "--json", "--discard-changes", "--merge", "feat") |
| 207 | assert result.exit_code != 0 |
| 208 | data = json.loads(result.stdout) |
| 209 | assert "error" in data |
| 210 | |
| 211 | def test_U5_mutual_exclusion_discard_autoshelf_json_error( |
| 212 | self, repo: pathlib.Path |
| 213 | ) -> None: |
| 214 | result = _invoke(repo, "--json", "--discard-changes", "--autoshelf", "feat") |
| 215 | assert result.exit_code != 0 |
| 216 | data = json.loads(result.stdout) |
| 217 | assert "error" in data |
| 218 | |
| 219 | def test_U5_mutual_exclusion_merge_autoshelf_json_error( |
| 220 | self, repo: pathlib.Path |
| 221 | ) -> None: |
| 222 | result = _invoke(repo, "--json", "--merge", "--autoshelf", "feat") |
| 223 | assert result.exit_code != 0 |
| 224 | data = json.loads(result.stdout) |
| 225 | assert "error" in data |
| 226 | |
| 227 | def test_json_error_has_duration_ms(self, repo: pathlib.Path) -> None: |
| 228 | result = _invoke(repo, "--json", "no-such-branch") |
| 229 | data = json.loads(result.stdout) |
| 230 | assert "duration_ms" in data |
| 231 | |
| 232 | def test_json_error_has_exit_code(self, repo: pathlib.Path) -> None: |
| 233 | result = _invoke(repo, "--json", "no-such-branch") |
| 234 | data = json.loads(result.stdout) |
| 235 | assert "exit_code" in data |
| 236 | assert data["exit_code"] != 0 |
| 237 | |
| 238 | def test_create_existing_branch_json_error( |
| 239 | self, two_branch_repo: pathlib.Path |
| 240 | ) -> None: |
| 241 | """switch -c on an existing branch must emit a JSON error, not a traceback.""" |
| 242 | result = _invoke(two_branch_repo, "--json", "-c", "feat") |
| 243 | assert result.exit_code != 0 |
| 244 | data = json.loads(result.stdout) |
| 245 | assert "error" in data |
| 246 | assert "Traceback" not in result.stdout |
| 247 | |
| 248 | |
| 249 | # --------------------------------------------------------------------------- |
| 250 | # U6–U7 commit_id and from_branch |
| 251 | # --------------------------------------------------------------------------- |
| 252 | |
| 253 | |
| 254 | class TestCommitIdAndFromBranch: |
| 255 | def test_U6_commit_id_sha256_prefixed(self, two_branch_repo: pathlib.Path) -> None: |
| 256 | _, data = _json_switch(two_branch_repo, "feat") |
| 257 | assert data["commit_id"].startswith("sha256:") |
| 258 | |
| 259 | def test_U7_from_branch_correct(self, two_branch_repo: pathlib.Path) -> None: |
| 260 | _, data = _json_switch(two_branch_repo, "feat") |
| 261 | assert data["from_branch"] == "main" |
| 262 | |
| 263 | def test_U7_from_branch_after_create(self, repo: pathlib.Path) -> None: |
| 264 | _, data = _json_switch(repo, "-c", "new-feat") |
| 265 | assert data["from_branch"] == "main" |
| 266 | |
| 267 | def test_D2_commit_id_matches_head(self, two_branch_repo: pathlib.Path) -> None: |
| 268 | result, data = _json_switch(two_branch_repo, "feat") |
| 269 | assert result.exit_code == 0 |
| 270 | actual = get_head_commit_id(two_branch_repo, "feat") |
| 271 | assert data["commit_id"] == actual |
| 272 | |
| 273 | |
| 274 | # --------------------------------------------------------------------------- |
| 275 | # I1–I10 Integration — uncovered paths |
| 276 | # --------------------------------------------------------------------------- |
| 277 | |
| 278 | |
| 279 | class TestIntegration: |
| 280 | def test_I1_switch_dash_json(self, two_branch_repo: pathlib.Path) -> None: |
| 281 | """switch - + --json must emit valid JSON.""" |
| 282 | _switch(two_branch_repo, "feat") # go to feat first |
| 283 | result = _invoke(two_branch_repo, "--json", "-") |
| 284 | assert result.exit_code == 0 |
| 285 | data = json.loads(result.stdout) |
| 286 | assert "action" in data |
| 287 | assert data["branch"] == "main" |
| 288 | assert "duration_ms" in data |
| 289 | |
| 290 | def test_I2_force_create_existing_action_reset( |
| 291 | self, two_branch_repo: pathlib.Path |
| 292 | ) -> None: |
| 293 | result, data = _json_switch(two_branch_repo, "-C", "feat") |
| 294 | assert result.exit_code == 0 |
| 295 | assert data["action"] == "reset" |
| 296 | |
| 297 | def test_I3_force_create_new_action_created(self, repo: pathlib.Path) -> None: |
| 298 | result, data = _json_switch(repo, "-C", "brand-new") |
| 299 | assert result.exit_code == 0 |
| 300 | assert data["action"] == "created" |
| 301 | |
| 302 | def test_I4_create_with_intent(self, repo: pathlib.Path) -> None: |
| 303 | """switch -c + --intent stores intent metadata on the new branch.""" |
| 304 | result = _invoke(repo, "-c", "task/foo", "--intent", "implement foo feature") |
| 305 | assert result.exit_code == 0 |
| 306 | # Verify branch was created |
| 307 | assert read_current_branch(repo) == "task/foo" |
| 308 | # Verify intent is queryable |
| 309 | branch_data = json.loads( |
| 310 | _run(repo, "branch", "--json").stdout |
| 311 | ) |
| 312 | foo = next((b for b in branch_data if b["name"] == "task/foo"), None) |
| 313 | assert foo is not None |
| 314 | assert foo.get("intent") == "implement foo feature" |
| 315 | |
| 316 | def test_I5_create_with_resumable(self, repo: pathlib.Path) -> None: |
| 317 | """switch -c + --resumable marks the branch as a resumable checkpoint.""" |
| 318 | result = _invoke(repo, "-c", "task/bar", "--resumable") |
| 319 | assert result.exit_code == 0 |
| 320 | branch_data = json.loads(_run(repo, "branch", "--json").stdout) |
| 321 | bar = next((b for b in branch_data if b["name"] == "task/bar"), None) |
| 322 | assert bar is not None |
| 323 | assert bar.get("resumable") is True |
| 324 | |
| 325 | def test_I6_autoshelf_json(self, two_branch_repo: pathlib.Path) -> None: |
| 326 | result, data = _json_switch(two_branch_repo, "--autoshelf", "feat") |
| 327 | assert result.exit_code == 0 |
| 328 | assert "action" in data |
| 329 | assert "duration_ms" in data |
| 330 | |
| 331 | def test_I8_dry_run_force_create_json(self, two_branch_repo: pathlib.Path) -> None: |
| 332 | result, data = _json_switch(two_branch_repo, "--dry-run", "-C", "feat") |
| 333 | assert result.exit_code == 0 |
| 334 | assert data["dry_run"] is True |
| 335 | assert "action" in data |
| 336 | assert "duration_ms" in data |
| 337 | # Branch ref must not be modified |
| 338 | orig_tip = get_head_commit_id(two_branch_repo, "feat") |
| 339 | main_tip = get_head_commit_id(two_branch_repo, "main") |
| 340 | assert get_head_commit_id(two_branch_repo, "feat") == orig_tip |
| 341 | |
| 342 | def test_I9_dry_run_switch_dash_json(self, two_branch_repo: pathlib.Path) -> None: |
| 343 | _switch(two_branch_repo, "feat") # record prev |
| 344 | result, data = _json_switch(two_branch_repo, "--dry-run", "-") |
| 345 | assert result.exit_code == 0 |
| 346 | assert data["dry_run"] is True |
| 347 | assert "duration_ms" in data |
| 348 | |
| 349 | def test_I10_create_existing_json_error_not_traceback( |
| 350 | self, two_branch_repo: pathlib.Path |
| 351 | ) -> None: |
| 352 | result = _invoke(two_branch_repo, "--json", "-c", "feat") |
| 353 | assert result.exit_code != 0 |
| 354 | assert "Traceback" not in result.stdout |
| 355 | data = json.loads(result.stdout) |
| 356 | assert "error" in data |
| 357 | |
| 358 | |
| 359 | # --------------------------------------------------------------------------- |
| 360 | # Security |
| 361 | # --------------------------------------------------------------------------- |
| 362 | |
| 363 | |
| 364 | class TestSecurity: |
| 365 | def test_S1_null_byte_in_branch_name_rejected(self, repo: pathlib.Path) -> None: |
| 366 | result = _invoke(repo, "feat\x00malicious") |
| 367 | assert result.exit_code != 0 |
| 368 | |
| 369 | def test_S2_path_traversal_in_branch_name_rejected( |
| 370 | self, repo: pathlib.Path |
| 371 | ) -> None: |
| 372 | result = _invoke(repo, "../traversal") |
| 373 | assert result.exit_code != 0 |
| 374 | |
| 375 | def test_S3_json_error_no_ansi(self, repo: pathlib.Path) -> None: |
| 376 | result = _invoke(repo, "--json", "no-such-branch") |
| 377 | assert "\x1b" not in result.stdout |
| 378 | |
| 379 | def test_S4_branch_name_sanitized_in_json( |
| 380 | self, two_branch_repo: pathlib.Path |
| 381 | ) -> None: |
| 382 | _, data = _json_switch(two_branch_repo, "feat") |
| 383 | assert "\x1b" not in json.dumps(data) |
| 384 | |
| 385 | def test_ansi_in_branch_name_rejected_with_json( |
| 386 | self, repo: pathlib.Path |
| 387 | ) -> None: |
| 388 | result = _invoke(repo, "--json", "\x1b[31mbad\x1b[0m") |
| 389 | assert result.exit_code != 0 |
| 390 | # Must be parseable JSON, not raw error text |
| 391 | data = json.loads(result.stdout) |
| 392 | assert "error" in data |
| 393 | |
| 394 | |
| 395 | # --------------------------------------------------------------------------- |
| 396 | # Data integrity |
| 397 | # --------------------------------------------------------------------------- |
| 398 | |
| 399 | |
| 400 | class TestDataIntegrity: |
| 401 | def test_D1_prev_branch_correct_after_10_toggles( |
| 402 | self, two_branch_repo: pathlib.Path |
| 403 | ) -> None: |
| 404 | branches = ["main", "feat"] |
| 405 | for i in range(10): |
| 406 | _switch(two_branch_repo, branches[i % 2]) |
| 407 | # After 10 switches index 9 → branches[1] = feat |
| 408 | assert read_current_branch(two_branch_repo) == "feat" |
| 409 | # PREV_BRANCH should be main (the branch we came from) |
| 410 | prev = (muse_dir(two_branch_repo) / "PREV_BRANCH").read_text().strip() |
| 411 | assert prev == "main" |
| 412 | |
| 413 | def test_D5_force_create_action_reset_when_existed( |
| 414 | self, two_branch_repo: pathlib.Path |
| 415 | ) -> None: |
| 416 | _, data = _json_switch(two_branch_repo, "-C", "feat") |
| 417 | assert data["action"] == "reset" |
| 418 | |
| 419 | def test_D5_force_create_action_created_when_new( |
| 420 | self, repo: pathlib.Path |
| 421 | ) -> None: |
| 422 | _, data = _json_switch(repo, "-C", "new-branch") |
| 423 | assert data["action"] == "created" |
| 424 | |
| 425 | def test_all_success_json_keys_present( |
| 426 | self, two_branch_repo: pathlib.Path |
| 427 | ) -> None: |
| 428 | """Every successful switch --json must have the full schema.""" |
| 429 | _, data = _json_switch(two_branch_repo, "feat") |
| 430 | required = {"action", "branch", "from_branch", "commit_id", "dry_run", |
| 431 | "duration_ms", "exit_code"} |
| 432 | missing = required - set(data.keys()) |
| 433 | assert not missing, f"Missing JSON keys: {missing}" |
| 434 | |
| 435 | def test_all_error_json_keys_present(self, repo: pathlib.Path) -> None: |
| 436 | result = _invoke(repo, "--json", "ghost") |
| 437 | data = json.loads(result.stdout) |
| 438 | required = {"error", "duration_ms", "exit_code"} |
| 439 | missing = required - set(data.keys()) |
| 440 | assert not missing, f"Missing error JSON keys: {missing}" |
| 441 | |
| 442 | |
| 443 | # --------------------------------------------------------------------------- |
| 444 | # Stress |
| 445 | # --------------------------------------------------------------------------- |
| 446 | |
| 447 | |
| 448 | class TestStress: |
| 449 | def test_P1_50_rapid_switches_three_branches( |
| 450 | self, tmp_path: pathlib.Path |
| 451 | ) -> None: |
| 452 | repo = _init_repo(tmp_path) |
| 453 | for name in ("feat-a", "feat-b"): |
| 454 | _add_branch(repo, name) |
| 455 | branches = ["main", "feat-a", "feat-b"] |
| 456 | for i in range(50): |
| 457 | result = _switch(repo, branches[i % 3]) |
| 458 | assert result.exit_code == 0, f"Switch {i} failed" |
| 459 | expected = branches[49 % 3] |
| 460 | assert read_current_branch(repo) == expected |
| 461 | |
| 462 | def test_P2_switch_with_many_branches(self, tmp_path: pathlib.Path) -> None: |
| 463 | """30 branches in the repo; switching still completes.""" |
| 464 | repo = _init_repo(tmp_path) |
| 465 | for i in range(30): |
| 466 | _run(repo, "branch", f"branch-{i:02d}") |
| 467 | result = _switch(repo, "branch-00") |
| 468 | assert result.exit_code == 0 |
| 469 | assert read_current_branch(repo) == "branch-00" |
| 470 | |
| 471 | def test_P3_duration_ms_in_all_rapid_json_calls( |
| 472 | self, two_branch_repo: pathlib.Path |
| 473 | ) -> None: |
| 474 | branches = ["main", "feat"] |
| 475 | for i in range(10): |
| 476 | result = _invoke(two_branch_repo, "--json", branches[i % 2]) |
| 477 | data = json.loads(result.stdout) |
| 478 | assert "duration_ms" in data, f"Missing duration_ms on call {i}" |
| 479 | assert isinstance(data["duration_ms"], float) |
| 480 | |
| 481 | |
| 482 | # --------------------------------------------------------------------------- |
| 483 | # Concurrent |
| 484 | # --------------------------------------------------------------------------- |
| 485 | |
| 486 | |
| 487 | _INIT_LOCK = threading.Lock() |
| 488 | |
| 489 | |
| 490 | class TestConcurrent: |
| 491 | def test_C1_four_concurrent_switches(self, tmp_path: pathlib.Path) -> None: |
| 492 | results = [None] * 4 |
| 493 | |
| 494 | def _work(idx: int) -> None: |
| 495 | repo = tmp_path / f"repo_{idx}" |
| 496 | with _INIT_LOCK: |
| 497 | r = _init_repo(repo) |
| 498 | _add_branch(r, "feat") |
| 499 | try: |
| 500 | res = _switch(r, "feat") |
| 501 | results[idx] = res.exit_code |
| 502 | except Exception as exc: |
| 503 | results[idx] = exc |
| 504 | |
| 505 | threads = [threading.Thread(target=_work, args=(i,)) for i in range(4)] |
| 506 | for t in threads: |
| 507 | t.start() |
| 508 | for t in threads: |
| 509 | t.join() |
| 510 | |
| 511 | for i, result in enumerate(results): |
| 512 | assert not isinstance(result, Exception), f"Thread {i}: {result}" |
| 513 | assert result == 0, f"Thread {i} exit code: {result}" |