gabriel / muse public
test_hub_body_file_assignee.py python
841 lines 33.4 KB
Raw
sha256:ff478cfdcdd4b7fd6de89cb68896601a981f945634463275ec333bd20ca36402 Merge branch 'dev' into main Human 21 days ago
1 """8-tier tests for hub.py ergonomics: --body-file and --assignee.
2
3 Covers:
4 Tier 1 — Shape / Schema
5 Tier 2 — Round-Trip / Integration
6 Tier 3 — Edge Cases
7 Tier 4 — Stress
8 Tier 5 — Data Integrity
9 Tier 6 — Performance
10 Tier 7 — Security (extra vigilant — all inputs are user-supplied)
11 Tier 8 — Docstrings / API Contract
12 """
13
14 from __future__ import annotations
15 from collections.abc import Mapping
16
17 import io
18 import json
19 import pathlib
20 import textwrap
21 import time
22 import unittest.mock
23
24 import pytest
25 from tests.cli_test_helper import CliRunner
26
27 from muse._version import __version__
28 from muse.core.paths import commits_dir, heads_dir, muse_dir, objects_dir, snapshots_dir
29 from muse.cli.commands.hub.connection import (
30 _MAX_HANDLE_LEN,
31 _HANDLE_RE,
32 _resolve_body,
33 _validate_assignee,
34 )
35 from muse.cli.config import set_hub_url
36 from muse.core.types import MsgpackDict
37 from muse.core.identity import IdentityEntry
38
39 type _HubBody = Mapping[str, str | bool | list[str] | None] | None
40 type _HubResponse = MsgpackDict
41 from muse.core.errors import ExitCode
42 from muse.core.identity import IdentityEntry, save_identity
43
44 cli = None
45 runner = CliRunner()
46
47
48 # ---------------------------------------------------------------------------
49 # Shared fixtures
50 # ---------------------------------------------------------------------------
51
52
53 @pytest.fixture()
54 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
55 """Minimal .muse/ repo with identity wired up."""
56 heads_dir(tmp_path).mkdir(parents=True, exist_ok=True)
57 objects_dir(tmp_path).mkdir(parents=True, exist_ok=True)
58 commits_dir(tmp_path).mkdir(parents=True, exist_ok=True)
59 snapshots_dir(tmp_path).mkdir(parents=True, exist_ok=True)
60 (muse_dir(tmp_path) / "repo.json").write_text(
61 json.dumps({
62 "repo_id": "test-repo",
63 "schema_version": __version__,
64 "domain": "midi",
65 })
66 )
67 (muse_dir(tmp_path) / "HEAD").write_text("ref: refs/heads/main\n")
68 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
69 monkeypatch.chdir(tmp_path)
70 return tmp_path
71
72
73 def _setup_auth(repo: pathlib.Path, hub_url: str = "https://localhost:1337/gabriel/muse") -> None:
74 set_hub_url(hub_url, repo)
75 identity = IdentityEntry(
76 type="human",
77 handle="gabriel",
78 key_path=str(repo / "fake_home" / ".muse" / "keys" / "key.pem"),
79 algorithm="ed25519",
80 fingerprint="deadbeef",
81 )
82 save_identity(hub_url, identity)
83
84
85 def _hub_patches(calls: list[tuple], response: _HubResponse | None = None) -> unittest.mock._patch:
86 """Context manager that mocks hub network helpers and records calls."""
87 mock_resp = response or {"issueId": "abc-123", "number": 1, "title": "t"}
88
89 def _fake_api(hub_url: str, identity: IdentityEntry, method: str, path: str, **kw: str) -> _HubResponse:
90 calls.append((method, path, kw))
91 return mock_resp
92
93 return unittest.mock.patch.multiple(
94 "muse.cli.commands.hub",
95 _hub_api=unittest.mock.MagicMock(side_effect=_fake_api),
96 _get_hub_and_identity=unittest.mock.MagicMock(
97 return_value=("https://localhost:1337/gabriel/muse", unittest.mock.MagicMock())
98 ),
99 _resolve_repo_id=unittest.mock.MagicMock(return_value="test-repo-id"),
100 )
101
102
103 # ===========================================================================
104 # Tier 1 — Shape / Schema
105 # ===========================================================================
106
107
108 class TestShape:
109 """Parser flags exist; helpers are importable with correct signatures."""
110
111 def test_resolve_body_importable(self) -> None:
112 """_resolve_body must be importable from hub."""
113 from muse.cli.commands.hub import _resolve_body
114 assert callable(_resolve_body)
115
116 def test_validate_assignee_importable(self) -> None:
117 """_validate_assignee must be importable from hub."""
118 from muse.cli.commands.hub import _validate_assignee
119 assert callable(_validate_assignee)
120
121 def test_handle_re_is_compiled_pattern(self) -> None:
122 """_HANDLE_RE must be a compiled regex."""
123 import re
124 assert isinstance(_HANDLE_RE, re.Pattern)
125
126 def test_max_handle_len_positive_int(self) -> None:
127 """_MAX_HANDLE_LEN must be a positive integer."""
128 assert isinstance(_MAX_HANDLE_LEN, int)
129 assert _MAX_HANDLE_LEN > 0
130
131 def test_assignee_flag_present_on_issue_create(self, repo: pathlib.Path) -> None:
132 """--assignee must appear in the issue create help text."""
133 _setup_auth(repo)
134 result = runner.invoke(cli, ["hub", "issue", "create", "--help"])
135 assert "--assignee" in result.output, (
136 f"--assignee flag missing from 'hub issue create --help': {result.output}"
137 )
138
139 def test_body_file_flag_present_on_issue_create(self, repo: pathlib.Path) -> None:
140 """--body-file must appear in the issue create help text."""
141 _setup_auth(repo)
142 result = runner.invoke(cli, ["hub", "issue", "create", "--help"])
143 assert "--body-file" in result.output
144
145 def test_body_file_flag_present_on_issue_update(self, repo: pathlib.Path) -> None:
146 """--body-file must appear in the issue update help text."""
147 _setup_auth(repo)
148 result = runner.invoke(cli, ["hub", "issue", "update", "--help"])
149 assert "--body-file" in result.output
150
151 def test_body_file_flag_present_on_issue_comment(self, repo: pathlib.Path) -> None:
152 """--body-file must appear in the issue comment help text."""
153 _setup_auth(repo)
154 result = runner.invoke(cli, ["hub", "issue", "comment", "--help"])
155 assert "--body-file" in result.output
156
157 def test_handle_re_matches_valid_handles(self) -> None:
158 """_HANDLE_RE must match well-formed handles."""
159 valid = [
160 "gabriel",
161 "aaronrene",
162 "mix-engine-7",
163 "studio_9",
164 "A",
165 "a1",
166 "z" * _MAX_HANDLE_LEN,
167 ]
168 for h in valid:
169 if len(h) <= _MAX_HANDLE_LEN:
170 assert _HANDLE_RE.match(h), f"_HANDLE_RE should match {h!r}"
171
172 def test_handle_re_rejects_leading_hyphen(self) -> None:
173 assert not _HANDLE_RE.match("-gabriel")
174
175 def test_handle_re_rejects_spaces(self) -> None:
176 assert not _HANDLE_RE.match("gab riel")
177
178 def test_handle_re_rejects_at_symbol(self) -> None:
179 assert not _HANDLE_RE.match("@gabriel")
180
181 def test_handle_re_rejects_slash(self) -> None:
182 assert not _HANDLE_RE.match("gabriel/muse")
183
184
185 # ===========================================================================
186 # Tier 2 — Round-Trip / Integration
187 # ===========================================================================
188
189
190 class TestRoundTrip:
191 """CLI invocations produce correct sequences of API calls."""
192
193 def test_issue_create_with_assignee_calls_create_then_assign(
194 self, repo: pathlib.Path
195 ) -> None:
196 """create + --assignee must POST to /issues then POST to /assign."""
197 _setup_auth(repo)
198 calls: list[tuple] = []
199 with _hub_patches(calls):
200 result = runner.invoke(
201 cli,
202 ["hub", "issue", "create", "--title", "Test", "--assignee", "aaronrene"],
203 )
204 assert result.exit_code == 0, result.output
205 methods_and_paths = [(m, p) for m, p, _ in calls]
206 assert any("/issues" in p and m == "POST" for m, p in methods_and_paths), (
207 f"Expected POST /issues; got {methods_and_paths}"
208 )
209 assert any("/assign" in p for _, p in methods_and_paths), (
210 f"Expected /assign call; got {methods_and_paths}"
211 )
212
213 def test_issue_create_without_assignee_does_not_call_assign(
214 self, repo: pathlib.Path
215 ) -> None:
216 """create without --assignee must NOT dispatch an /assign call."""
217 _setup_auth(repo)
218 calls: list[tuple] = []
219 with _hub_patches(calls):
220 result = runner.invoke(
221 cli, ["hub", "issue", "create", "--title", "No assignee"]
222 )
223 assert result.exit_code == 0, result.output
224 paths = [p for _, p, _ in calls]
225 assert not any("/assign" in p for p in paths), (
226 f"/assign was called even though no --assignee was given: {paths}"
227 )
228
229 def test_issue_create_with_body_file_sends_correct_body(
230 self, tmp_path: pathlib.Path, repo: pathlib.Path
231 ) -> None:
232 """Body read from --body-file must appear verbatim in the API payload."""
233 _setup_auth(repo)
234 body_text = "# My Issue\n\nHello `world`\n"
235 body_file = tmp_path / "body.md"
236 body_file.write_text(body_text, encoding="utf-8")
237
238 payloads: list[dict] = []
239
240 def _capturing_api(hub_url: str, identity: IdentityEntry, method: str, path: str, *, body: _HubBody = None, **kw: str) -> _HubResponse:
241 if body:
242 payloads.append(body)
243 return {"issueId": "x", "number": 1, "title": "t"}
244
245 with unittest.mock.patch.multiple(
246 "muse.cli.commands.hub",
247 _hub_api=unittest.mock.MagicMock(side_effect=_capturing_api),
248 _get_hub_and_identity=unittest.mock.MagicMock(
249 return_value=("https://localhost:1337/gabriel/muse", unittest.mock.MagicMock())
250 ),
251 _resolve_repo_id=unittest.mock.MagicMock(return_value="test-repo-id"),
252 ):
253 result = runner.invoke(
254 cli,
255 ["hub", "issue", "create", "--title", "t", "--body-file", str(body_file)],
256 )
257 assert result.exit_code == 0, result.output
258 assert payloads, "No API payload captured"
259 assert payloads[0]["body"] == body_text
260
261 def test_issue_update_with_body_file_sends_correct_body(
262 self, tmp_path: pathlib.Path, repo: pathlib.Path
263 ) -> None:
264 """issue update --body-file must PATCH with the file contents."""
265 _setup_auth(repo)
266 body_text = "Updated body `with backticks`"
267 body_file = tmp_path / "update.md"
268 body_file.write_text(body_text, encoding="utf-8")
269
270 payloads: list[dict] = []
271
272 def _capturing_api(hub_url: str, identity: IdentityEntry, method: str, path: str, *, body: _HubBody = None, **kw: str) -> _HubResponse:
273 if body:
274 payloads.append(body)
275 return {"number": 1, "title": "x", "state": "open"}
276
277 with unittest.mock.patch.multiple(
278 "muse.cli.commands.hub",
279 _hub_api=unittest.mock.MagicMock(side_effect=_capturing_api),
280 _get_hub_and_identity=unittest.mock.MagicMock(
281 return_value=("https://localhost:1337/gabriel/muse", unittest.mock.MagicMock())
282 ),
283 _resolve_repo_id=unittest.mock.MagicMock(return_value="test-repo-id"),
284 ):
285 result = runner.invoke(
286 cli,
287 ["hub", "issue", "update", "1", "--body-file", str(body_file)],
288 )
289 assert result.exit_code == 0, result.output
290 assert payloads and payloads[0]["body"] == body_text
291
292 def test_issue_assign_validates_handle_before_api_call(
293 self, repo: pathlib.Path
294 ) -> None:
295 """issue assign with a bad handle must fail before any network I/O."""
296 _setup_auth(repo)
297 calls: list[tuple] = []
298 with _hub_patches(calls):
299 result = runner.invoke(
300 cli,
301 ["hub", "issue", "assign", "1", "--assignee", "bad handle!"],
302 )
303 assert result.exit_code != 0
304 assert not calls, "API was called even though the handle was invalid"
305
306
307 # ===========================================================================
308 # Tier 3 — Edge Cases
309 # ===========================================================================
310
311
312 class TestEdgeCases:
313 """Boundary conditions and unusual-but-valid inputs."""
314
315 def test_body_file_wins_over_body_when_both_given(
316 self, tmp_path: pathlib.Path, repo: pathlib.Path
317 ) -> None:
318 """When both --body and --body-file are given, --body-file wins."""
319 _setup_auth(repo)
320 file_text = "from file"
321 body_file = tmp_path / "b.txt"
322 body_file.write_text(file_text, encoding="utf-8")
323
324 payloads: list[dict] = []
325
326 def _cap(hub_url: str, identity: IdentityEntry, method: str, path: str, *, body: _HubBody = None, **kw: str) -> _HubResponse:
327 if body:
328 payloads.append(body)
329 return {"issueId": "x", "number": 1, "title": "t"}
330
331 with unittest.mock.patch.multiple(
332 "muse.cli.commands.hub",
333 _hub_api=unittest.mock.MagicMock(side_effect=_cap),
334 _get_hub_and_identity=unittest.mock.MagicMock(
335 return_value=("https://localhost:1337/gabriel/muse", unittest.mock.MagicMock())
336 ),
337 _resolve_repo_id=unittest.mock.MagicMock(return_value="test-repo-id"),
338 ):
339 result = runner.invoke(
340 cli,
341 ["hub", "issue", "create", "--title", "t",
342 "--body", "from inline",
343 "--body-file", str(body_file)],
344 )
345 assert result.exit_code == 0, result.output
346 assert payloads[0]["body"] == file_text, (
347 f"Expected file content, got {payloads[0]['body']!r}"
348 )
349
350 def test_body_file_missing_exits_with_user_error(
351 self, tmp_path: pathlib.Path, repo: pathlib.Path
352 ) -> None:
353 """--body-file pointing at a non-existent file must exit USER_ERROR."""
354 _setup_auth(repo)
355 result = runner.invoke(
356 cli,
357 ["hub", "issue", "create", "--title", "t",
358 "--body-file", str(tmp_path / "does_not_exist.md")],
359 )
360 assert result.exit_code == ExitCode.USER_ERROR
361
362 def test_validate_assignee_empty_allow_empty_true_ok(self) -> None:
363 """An empty handle is valid when allow_empty=True (unassign path)."""
364 _validate_assignee("", allow_empty=True) # must not raise
365
366 def test_validate_assignee_empty_allow_empty_false_raises(self) -> None:
367 """An empty handle is invalid when allow_empty=False (create path)."""
368 with pytest.raises(SystemExit) as exc_info:
369 _validate_assignee("", allow_empty=False)
370 assert exc_info.value.code == ExitCode.USER_ERROR
371
372 def test_validate_assignee_single_char_valid(self) -> None:
373 """A single alphanumeric char is a valid handle."""
374 _validate_assignee("a")
375
376 def test_issue_create_assignee_shown_in_success_output(
377 self, repo: pathlib.Path
378 ) -> None:
379 """After creation with --assignee, the assignee name should appear in output."""
380 _setup_auth(repo)
381 calls: list[tuple] = []
382 with _hub_patches(calls):
383 result = runner.invoke(
384 cli,
385 ["hub", "issue", "create", "--title", "t", "--assignee", "aaronrene"],
386 )
387 assert result.exit_code == 0, result.output
388 assert "aaronrene" in result.stderr
389
390 def test_resolve_body_neither_body_nor_body_file_returns_empty(self) -> None:
391 """When args has neither body nor body_file, _resolve_body returns ''."""
392 import argparse
393 args = argparse.Namespace(body=None, body_file=None)
394 assert _resolve_body(args) == ""
395
396 def test_resolve_body_body_only_returns_body(self) -> None:
397 import argparse
398 args = argparse.Namespace(body="hello", body_file=None)
399 assert _resolve_body(args) == "hello"
400
401 def test_resolve_body_body_file_none_returns_body_string(self) -> None:
402 import argparse
403 args = argparse.Namespace(body="hello", body_file=None)
404 assert _resolve_body(args) == "hello"
405
406 def test_resolve_body_body_file_path_returns_file_contents(
407 self, tmp_path: pathlib.Path
408 ) -> None:
409 import argparse
410 f = tmp_path / "b.txt"
411 f.write_text("file body", encoding="utf-8")
412 args = argparse.Namespace(body="", body_file=str(f))
413 assert _resolve_body(args) == "file body"
414
415 def test_resolve_body_stdin_sentinel(self, monkeypatch: pytest.MonkeyPatch) -> None:
416 """Passing '-' for body_file must read from stdin."""
417 import argparse
418 import io
419 monkeypatch.setattr("sys.stdin", io.StringIO("stdin body"))
420 args = argparse.Namespace(body="", body_file="-")
421 assert _resolve_body(args) == "stdin body"
422
423
424 # ===========================================================================
425 # Tier 4 — Stress
426 # ===========================================================================
427
428
429 class TestStress:
430 """Max-length inputs and bulk operations work without error."""
431
432 def test_max_length_valid_handle_accepted(self) -> None:
433 """A handle exactly _MAX_HANDLE_LEN chars long must be accepted."""
434 handle = "a" * _MAX_HANDLE_LEN
435 _validate_assignee(handle) # must not raise
436
437 def test_handle_one_over_max_rejected(self) -> None:
438 """A handle one char over _MAX_HANDLE_LEN must be rejected."""
439 handle = "a" * (_MAX_HANDLE_LEN + 1)
440 with pytest.raises(SystemExit) as exc_info:
441 _validate_assignee(handle)
442 assert exc_info.value.code == ExitCode.USER_ERROR
443
444 def test_large_body_file_read_correctly(self, tmp_path: pathlib.Path) -> None:
445 """A 200KB body file must be read in full without truncation."""
446 import argparse
447 large_body = "# heading\n\n" + ("line of content\n" * 12_000)
448 f = tmp_path / "large.md"
449 f.write_text(large_body, encoding="utf-8")
450 args = argparse.Namespace(body="", body_file=str(f))
451 result = _resolve_body(args)
452 assert result == large_body
453
454 def test_validate_assignee_called_1000_times_fast(self) -> None:
455 """_validate_assignee on a valid handle 1000× must complete quickly."""
456 start = time.monotonic()
457 for _ in range(1000):
458 _validate_assignee("gabriel")
459 elapsed = time.monotonic() - start
460 assert elapsed < 0.5, f"1000 validations took {elapsed:.3f}s — too slow"
461
462 def test_issue_create_with_many_labels_and_assignee(
463 self, repo: pathlib.Path
464 ) -> None:
465 """Multiple labels combined with --assignee must succeed."""
466 _setup_auth(repo)
467 calls: list[tuple] = []
468 with _hub_patches(calls):
469 result = runner.invoke(
470 cli,
471 [
472 "hub", "issue", "create",
473 "--title", "Big issue",
474 "--label", "bug",
475 "--label", "enhancement",
476 "--label", "phase-1",
477 "--assignee", "aaronrene",
478 ],
479 )
480 assert result.exit_code == 0, result.output
481 assert any("/assign" in p for _, p, _ in calls)
482
483
484 # ===========================================================================
485 # Tier 5 — Data Integrity
486 # ===========================================================================
487
488
489 class TestDataIntegrity:
490 """Payloads sent to the API match exactly what the user supplied."""
491
492 def test_assignee_payload_matches_flag_value(self, repo: pathlib.Path) -> None:
493 """The handle POSTed to /assign must be exactly what --assignee received."""
494 _setup_auth(repo)
495 captured_assign_body: list[dict] = []
496
497 def _cap(hub_url: str, identity: IdentityEntry, method: str, path: str, *, body: _HubBody = None, **kw: str) -> _HubResponse:
498 if "/assign" in path and body:
499 captured_assign_body.append(body)
500 return {"issueId": "x", "number": 1, "title": "t"}
501
502 with unittest.mock.patch.multiple(
503 "muse.cli.commands.hub",
504 _hub_api=unittest.mock.MagicMock(side_effect=_cap),
505 _get_hub_and_identity=unittest.mock.MagicMock(
506 return_value=("https://localhost:1337/gabriel/muse", unittest.mock.MagicMock())
507 ),
508 _resolve_repo_id=unittest.mock.MagicMock(return_value="test-repo-id"),
509 ):
510 runner.invoke(
511 cli,
512 ["hub", "issue", "create", "--title", "t", "--assignee", "aaronrene"],
513 )
514 assert captured_assign_body, "No assign payload captured"
515 assert captured_assign_body[0]["assignee"] == "aaronrene"
516
517 def test_body_file_bytes_match_payload_exactly(
518 self, tmp_path: pathlib.Path, repo: pathlib.Path
519 ) -> None:
520 """Non-ASCII in body file (UTF-8 encoded) must round-trip unchanged."""
521 _setup_auth(repo)
522 body_text = "Header\n\n```python\nprint('hello')\n```\n\n— em-dash ✓"
523 body_file = tmp_path / "body.md"
524 body_file.write_text(body_text, encoding="utf-8")
525
526 payloads: list[dict] = []
527
528 def _cap(hub_url: str, identity: IdentityEntry, method: str, path: str, *, body: _HubBody = None, **kw: str) -> _HubResponse:
529 if body:
530 payloads.append(body)
531 return {"issueId": "x", "number": 1, "title": "t"}
532
533 with unittest.mock.patch.multiple(
534 "muse.cli.commands.hub",
535 _hub_api=unittest.mock.MagicMock(side_effect=_cap),
536 _get_hub_and_identity=unittest.mock.MagicMock(
537 return_value=("https://localhost:1337/gabriel/muse", unittest.mock.MagicMock())
538 ),
539 _resolve_repo_id=unittest.mock.MagicMock(return_value="test-repo-id"),
540 ):
541 runner.invoke(
542 cli,
543 ["hub", "issue", "create", "--title", "t",
544 "--body-file", str(body_file)],
545 )
546 assert payloads and payloads[0]["body"] == body_text
547
548 def test_assign_called_with_correct_issue_number(
549 self, repo: pathlib.Path
550 ) -> None:
551 """The /assign path must contain the issue number returned by the create endpoint."""
552 _setup_auth(repo)
553 assign_paths: list[str] = []
554
555 def _cap(hub_url: str, identity: IdentityEntry, method: str, path: str, *, body: _HubBody = None, **kw: str) -> _HubResponse:
556 if "/assign" in path:
557 assign_paths.append(path)
558 return {"issueId": "x", "number": 42, "title": "t"}
559
560 with unittest.mock.patch.multiple(
561 "muse.cli.commands.hub",
562 _hub_api=unittest.mock.MagicMock(side_effect=_cap),
563 _get_hub_and_identity=unittest.mock.MagicMock(
564 return_value=("https://localhost:1337/gabriel/muse", unittest.mock.MagicMock())
565 ),
566 _resolve_repo_id=unittest.mock.MagicMock(return_value="test-repo-id"),
567 ):
568 runner.invoke(
569 cli,
570 ["hub", "issue", "create", "--title", "t", "--assignee", "gabriel"],
571 )
572 assert assign_paths, "No /assign call found"
573 assert "42" in assign_paths[0], (
574 f"/assign path {assign_paths[0]!r} should contain issue number 42"
575 )
576
577
578 # ===========================================================================
579 # Tier 6 — Performance
580 # ===========================================================================
581
582
583 class TestPerformance:
584 """_resolve_body and _validate_assignee run in sub-millisecond time."""
585
586 def test_resolve_body_from_file_single_read(
587 self, tmp_path: pathlib.Path
588 ) -> None:
589 """_resolve_body must read the file exactly once (no double reads)."""
590 import argparse
591 f = tmp_path / "b.txt"
592 f.write_text("content", encoding="utf-8")
593
594 read_count = 0
595 real_open = open
596
597 def _counting_open(path: pathlib.Path | str, *a: str, **kw: str) -> "io.IOBase":
598 nonlocal read_count
599 if str(path) == str(f):
600 read_count += 1
601 return real_open(path, *a, **kw)
602
603 with unittest.mock.patch("builtins.open", side_effect=_counting_open):
604 args = argparse.Namespace(body="", body_file=str(f))
605 _resolve_body(args)
606
607 assert read_count == 1, f"File was read {read_count} times; expected exactly 1"
608
609 def test_validate_assignee_valid_under_1ms(self) -> None:
610 """Single _validate_assignee call on a valid handle must finish under 1 ms."""
611 start = time.monotonic()
612 _validate_assignee("gabriel")
613 elapsed = time.monotonic() - start
614 assert elapsed < 0.001, f"took {elapsed*1000:.2f} ms"
615
616 def test_validate_assignee_invalid_under_1ms(self) -> None:
617 """Rejection path must also complete under 1 ms."""
618 start = time.monotonic()
619 with pytest.raises(SystemExit):
620 _validate_assignee("bad handle!")
621 elapsed = time.monotonic() - start
622 assert elapsed < 0.001, f"took {elapsed*1000:.2f} ms"
623
624
625 # ===========================================================================
626 # Tier 7 — Security
627 # ===========================================================================
628
629
630 class TestSecurity:
631 """All user-supplied handle input is rigorously rejected for malformed values.
632
633 Threat model (see _validate_assignee docstring):
634 * Terminal injection via ANSI/control sequences
635 * Null-byte truncation attacks
636 * Newline injection — makes error messages look like success
637 * Unicode confusable characters impersonating ASCII handles
638 * Oversized input for DoS
639 * Shell metacharacters (defense-in-depth; args go into JSON anyway)
640 """
641
642 @pytest.mark.parametrize("bad_handle", [
643 "gab\x00riel", # null byte
644 "gab\nriel", # newline
645 "gab\rriel", # carriage return
646 "gab\triel", # tab
647 "\x01gabriel", # SOH control char
648 "gabriel\x1b[31mred", # ANSI escape — terminal injection
649 "gabriel\x7f", # DEL
650 "\x0cgabriel", # form feed
651 "gabriel\x0b", # vertical tab
652 "gabriel\x08extra", # backspace — visual overwrite attack
653 ])
654 def test_control_characters_rejected(self, bad_handle: str) -> None:
655 """Any handle with a control character must be rejected."""
656 with pytest.raises(SystemExit) as exc_info:
657 _validate_assignee(bad_handle)
658 assert exc_info.value.code == ExitCode.USER_ERROR, (
659 f"Expected USER_ERROR for handle {bad_handle!r}"
660 )
661
662 @pytest.mark.parametrize("bad_handle", [
663 "gаbriel", # Cyrillic 'а' (U+0430) — looks like 'a'
664 "aaronrenё", # Cyrillic 'ё' (U+0451)
665 "gábríel", # Latin accented chars
666 "gabriel™", # trademark symbol
667 "gabriel→muse", # arrow
668 "𝕘𝕒𝕓𝕣𝕚𝕖𝕝", # mathematical double-struck
669 ])
670 def test_non_ascii_unicode_rejected(self, bad_handle: str) -> None:
671 """Non-ASCII Unicode handles must be rejected (confusable / RTL risk)."""
672 with pytest.raises(SystemExit) as exc_info:
673 _validate_assignee(bad_handle)
674 assert exc_info.value.code == ExitCode.USER_ERROR, (
675 f"Expected USER_ERROR for handle {bad_handle!r}"
676 )
677
678 @pytest.mark.parametrize("bad_handle", [
679 "gabriel muse", # space
680 "gabriel@muse", # at sign
681 "gabriel/muse", # slash — path traversal appearance
682 "gabriel.muse", # dot
683 "gabriel!", # bang
684 "gabriel;rm -rf /", # shell injection attempt
685 "gabriel$(whoami)", # command substitution
686 "gabriel`id`", # backtick injection
687 "gabriel|cat /etc/passwd", # pipe
688 "gabriel&&echo pwned", # logical AND
689 "gabriel>>/etc/crontab", # redirection
690 ])
691 def test_shell_metacharacters_rejected(self, bad_handle: str) -> None:
692 """Shell metacharacters and non-alphanumeric symbols must be rejected."""
693 with pytest.raises(SystemExit) as exc_info:
694 _validate_assignee(bad_handle)
695 assert exc_info.value.code == ExitCode.USER_ERROR
696
697 def test_extremely_long_handle_rejected(self) -> None:
698 """A handle of 1000 chars must be rejected — DoS guard."""
699 with pytest.raises(SystemExit) as exc_info:
700 _validate_assignee("a" * 1000)
701 assert exc_info.value.code == ExitCode.USER_ERROR
702
703 def test_empty_handle_not_allowed_on_create(
704 self, repo: pathlib.Path
705 ) -> None:
706 """issue create --assignee '' must fail before any API call."""
707 _setup_auth(repo)
708 calls: list[tuple] = []
709 with _hub_patches(calls):
710 result = runner.invoke(
711 cli,
712 ["hub", "issue", "create", "--title", "t", "--assignee", ""],
713 )
714 assert result.exit_code != 0
715 assert not any("/issues" in p for _, p, _ in calls), (
716 "API was called even though the assignee was empty"
717 )
718
719 def test_invalid_handle_rejects_before_network_io(
720 self, repo: pathlib.Path
721 ) -> None:
722 """An invalid handle must fail before ANY network call is made."""
723 _setup_auth(repo)
724 calls: list[tuple] = []
725 with _hub_patches(calls):
726 result = runner.invoke(
727 cli,
728 ["hub", "issue", "create", "--title", "t",
729 "--assignee", "bad handle!"],
730 )
731 assert result.exit_code != 0
732 assert not calls, f"Network was called despite invalid handle: {calls}"
733
734 def test_body_file_path_not_leaked_on_error(
735 self, tmp_path: pathlib.Path, repo: pathlib.Path
736 ) -> None:
737 """Error message for missing --body-file must contain the path for
738 actionability, but must not expose sensitive filesystem layout beyond
739 what was explicitly given."""
740 _setup_auth(repo)
741 # Use a path that doesn't exist
742 missing = str(tmp_path / "secret_dir" / "body.md")
743 result = runner.invoke(
744 cli,
745 ["hub", "issue", "create", "--title", "t", "--body-file", missing],
746 )
747 assert result.exit_code == ExitCode.USER_ERROR
748 # The path given by the user may appear in the error (for debugging),
749 # but no other paths from the filesystem should be exposed.
750 assert "secret_dir" in result.stderr or "body.md" in result.stderr, (
751 "Error message should mention the problematic path for debuggability"
752 )
753
754 def test_issue_assign_invalid_handle_rejected(
755 self, repo: pathlib.Path
756 ) -> None:
757 """issue assign with control-char handle must fail before any API call."""
758 _setup_auth(repo)
759 calls: list[tuple] = []
760 with _hub_patches(calls):
761 result = runner.invoke(
762 cli,
763 ["hub", "issue", "assign", "1", "--assignee", "bad\x00handle"],
764 )
765 assert result.exit_code != 0
766 assert not calls
767
768 def test_issue_update_invalid_assign_rejected_before_network(
769 self, repo: pathlib.Path
770 ) -> None:
771 """issue update --assign with bad handle must fail before API call."""
772 _setup_auth(repo)
773 calls: list[tuple] = []
774 with _hub_patches(calls):
775 result = runner.invoke(
776 cli,
777 ["hub", "issue", "update", "1", "--assign", "bad handle!"],
778 )
779 assert result.exit_code != 0
780 assign_calls = [p for _, p, _ in calls if "/assign" in p]
781 assert not assign_calls
782
783
784 # ===========================================================================
785 # Tier 8 — Docstrings / API Contract
786 # ===========================================================================
787
788
789 class TestDocstrings:
790 """Key symbols have complete, accurate docstrings."""
791
792 def test_resolve_body_has_docstring(self) -> None:
793 """_resolve_body must have a non-empty docstring."""
794 assert _resolve_body.__doc__, "_resolve_body has no docstring"
795
796 def test_resolve_body_docstring_mentions_body_file(self) -> None:
797 assert "body_file" in _resolve_body.__doc__ or "body-file" in _resolve_body.__doc__
798
799 def test_resolve_body_docstring_mentions_stdin(self) -> None:
800 assert "-" in _resolve_body.__doc__, (
801 "Docstring should mention '-' as the stdin sentinel"
802 )
803
804 def test_resolve_body_docstring_has_args_section(self) -> None:
805 assert "Args:" in _resolve_body.__doc__
806
807 def test_resolve_body_docstring_has_returns_section(self) -> None:
808 assert "Returns:" in _resolve_body.__doc__
809
810 def test_resolve_body_docstring_has_raises_section(self) -> None:
811 assert "Raises:" in _resolve_body.__doc__
812
813 def test_validate_assignee_has_docstring(self) -> None:
814 assert _validate_assignee.__doc__, "_validate_assignee has no docstring"
815
816 def test_validate_assignee_docstring_mentions_allow_empty(self) -> None:
817 assert "allow_empty" in _validate_assignee.__doc__
818
819 def test_validate_assignee_docstring_has_threat_model(self) -> None:
820 doc = _validate_assignee.__doc__
821 assert any(
822 kw in doc
823 for kw in ("terminal", "injection", "null", "control", "Threat")
824 ), "Docstring should describe the threat model"
825
826 def test_validate_assignee_docstring_has_raises_section(self) -> None:
827 assert "Raises:" in _validate_assignee.__doc__
828
829 def test_validate_assignee_docstring_mentions_user_error(self) -> None:
830 assert "USER_ERROR" in _validate_assignee.__doc__
831
832 def test_handle_re_constant_is_named_consistently(self) -> None:
833 """_HANDLE_RE must follow the module constant naming convention."""
834 from muse.cli.commands import hub
835 assert hasattr(hub, "_HANDLE_RE"), "_HANDLE_RE not found in hub module"
836
837 def test_max_handle_len_constant_documented_in_code(self) -> None:
838 """_MAX_HANDLE_LEN must exist as a module-level constant."""
839 from muse.cli.commands import hub
840 assert hasattr(hub, "_MAX_HANDLE_LEN")
841 assert isinstance(hub._MAX_HANDLE_LEN, int)
File History 1 commit
sha256:ff478cfdcdd4b7fd6de89cb68896601a981f945634463275ec333bd20ca36402 Merge branch 'dev' into main Human 21 days ago