surshar / claude-code

Claude Code is an agentic coding tool that lives in your terminal, understands your codebase, and helps you code faster by executing routine tasks, explaining complex code, and handling git workflows - all through natural language commands.

Built by Anthropic · github.com/anthropics

claude-code / plugins / security-guidance / tests / test_repo_resolution.py

Claudesecurity-guidance: sync with claude-plugins-official (2.0.0 → 2.0.12) 968f8b3 · 15 days ago History

621 lines · 29.1 KBRaw

1	import json
2	import os
3	import subprocess
4	import threading
5	import time
6
7	import pytest
8
9	from conftest import (
10	GIT_ENV, HOOKS_DIR, STUB_VULN, VULN_PY, bash_payload, commit_file, edit_payload,
11	git, make_repo, metrics_of, run_hook, stop_payload, ups_payload,
12	)
13
14	import gitutil
15	import reporesolve as rr
16	import security_reminder_hook as hook
17
18
19
20	class TestDirsFromCommand:
21	def test_cd_and_and(self, workspace):
22	ws, repo = workspace
23	assert rr.dirs_from_command("cd sub && git commit -m x", str(ws)) == [str(repo)]
24
25	def test_cd_semicolon_subshell(self, workspace):
26	ws, repo = workspace
27	assert rr.dirs_from_command("(cd sub; git commit -m x)", str(ws)) == [str(repo)]
28
29	def test_git_dash_C(self, workspace):
30	ws, repo = workspace
31	assert rr.dirs_from_command("git -C sub commit -q -m x", str(ws)) == [str(repo)]
32
33	def test_git_dash_C_absolute_and_quoted(self, tmp_path):
34	d = tmp_path / "my repo"
35	d.mkdir()
36	assert rr.dirs_from_command(f'git -C "{d}" commit -m "a b"', "/nonexistent") == [str(d)]
37
38	def test_env_prefix_and_config_opts(self, workspace):
39	ws, repo = workspace
40	cmd = "FOO=1 git -c user.name=x -C sub commit -m x"
41	assert rr.dirs_from_command(cmd, str(ws)) == [str(repo)]
42
43	def test_git_dir_work_tree(self, workspace):
44	ws, repo = workspace
45	cmd = "git --git-dir=sub/.git --work-tree=sub commit -m x"
46	assert rr.dirs_from_command(cmd, str(ws)) == [str(repo)]
47	cmd = "git --git-dir sub/.git commit -m x"
48	assert rr.dirs_from_command(cmd, str(ws)) == [str(repo)]
49
50	def test_subcommand_filter(self, workspace):
51	ws, repo = workspace
52	cmd = "git -C other status && git -C sub commit -m x && git -C third push"
53	assert rr.dirs_from_command(cmd, str(ws), rr.COMMIT_SUBCOMMANDS) == [str(repo)]
54	assert rr.dirs_from_command(cmd, str(ws), rr.PUSH_SUBCOMMANDS) == [str(ws / "third")]
55
56	def test_gt(self, workspace):
57	ws, repo = workspace
58	assert rr.dirs_from_command("cd sub && gt create -m x", str(ws), rr.COMMIT_SUBCOMMANDS) == [str(repo)]
59
60	def test_pushd_and_relative_chain(self, workspace):
61	ws, repo = workspace
62	cmd = "pushd sub && cd .. && cd ./sub && git commit -m x"
63	assert rr.dirs_from_command(cmd, str(ws)) == [str(repo)]
64
65	@pytest.mark.parametrize("cmd", [
66	'git -C "unterminated commit -m x',
67	"git -C=sub commit -m x",
68	"cd && git commit",
69	"cd - && git commit -m x",
70	"git",
71	"git -C",
72	"cd /definitely/not/here && git commit -m x",
73	"echo 'git commit' \| cat",
74	"git commit -m \"$(cat <<'EOF'\nmsg\nEOF\n)\"",
75	"",
76	None,
77	])
78	def test_odd_inputs_do_not_raise(self, workspace, cmd):
79	ws, _ = workspace
80	out = rr.dirs_from_command(cmd, str(ws))
81	assert isinstance(out, list)
82	assert rr.toplevel_from_command(cmd, str(ws)) in (None, str(ws / "sub"))
83
84	def test_toplevel_from_command_nonexistent_dir(self, workspace):
85	ws, _ = workspace
86	assert rr.toplevel_from_command("cd nope && git commit -m x", str(ws)) is None
87
88	def test_toplevel_from_subdir_of_repo(self, workspace):
89	ws, repo = workspace
90	(repo / "pkg").mkdir()
91	assert rr.toplevel_from_command("cd sub/pkg && git commit -m x", str(ws)) == str(repo)
92
93	def test_windows_backslash_paths_survive_tokenizing(self, workspace, monkeypatch):
94	ws, _ = workspace
95	monkeypatch.setattr(rr.os, "sep", "\\")
96	toks = rr._tokenize(r'git -C C:\Users\me\repo commit -m x && git -C "D:\a b\r" push')
97	assert r"C:\Users\me\repo" in toks and r"D:\a b\r" in toks
98	toks = rr._tokenize(r"git -C \\srv\share\repo commit -m x")
99	assert r"\\srv\share\repo" in toks
100
101
102	class TestShaScan:
103	def test_finds_repo_containing_commit(self, workspace):
104	ws, repo = workspace
105	sha, _ = commit_file(repo, "app.py", VULN_PY)
106	assert rr.repo_containing_commit(sha[:7], [str(ws)]) == str(repo)
107	assert rr.repo_containing_commit(sha, [str(ws)]) == str(repo)
108
109	def test_unknown_sha(self, workspace):
110	ws, _ = workspace
111	assert rr.repo_containing_commit("deadbeefdeadbeef", [str(ws)]) is None
112
113	def test_invalid_sha(self, workspace):
114	ws, _ = workspace
115	assert rr.repo_containing_commit("HEAD; rm -rf x", [str(ws)]) is None
116	assert rr.repo_containing_commit(None, [str(ws)]) is None
117
118	def test_skips_ignored_dirs_and_respects_depth(self, tmp_path):
119	ws = tmp_path / "ws"
120	deep = make_repo(ws / "a" / "b" / "c" / "d" / "repo")
121	nm = make_repo(ws / "node_modules" / "pkg")
122	found = list(rr.iter_git_repos([str(ws)], max_depth=3))
123	assert str(deep) not in found and str(nm) not in found
124	found = list(rr.iter_git_repos([str(ws)], max_depth=6))
125	assert str(deep) in found and str(nm) not in found
126
127	def test_max_repos_cap(self, tmp_path):
128	ws = tmp_path / "ws"
129	for i in range(4):
130	make_repo(ws / f"r{i}")
131	assert len(list(rr.iter_git_repos([str(ws)], max_repos=2))) == 2
132
133
134	class TestReposFromPaths:
135	def test_groups_by_toplevel_and_orders_by_count(self, tmp_path):
136	ws = tmp_path / "ws"
137	a = make_repo(ws / "a")
138	b = make_repo(ws / "b")
139	paths = [str(b / "x.py"), str(a / "one.py"), str(a / "pkg" / "two.py"),
140	str(ws / "loose.txt"), "", None, 42]
141	assert rr.repos_from_paths(paths, str(ws)) == [str(a), str(b)]
142
143	def test_relative_paths_resolve_against_cwd(self, workspace):
144	ws, repo = workspace
145	assert rr.repos_from_paths(["sub/app.py"], str(ws)) == [str(repo)]
146
147	def test_missing_parent_dirs(self, workspace):
148	ws, repo = workspace
149	assert rr.repos_from_paths([str(repo / "new" / "deeper" / "f.py")], str(ws)) == [str(repo)]
150
151
152	class TestResolveRepoRoot:
153	def test_cwd_is_repo(self, workspace):
154	ws, repo = workspace
155	assert rr.resolve_repo_root(str(repo), "cd /tmp && git commit") == (str(repo), rr.RES_CWD)
156
157	def test_cwd_is_repo_same_repo_in_command_stays_cwd(self, workspace):
158	ws, repo = workspace
159	(repo / "pkg").mkdir()
160	for cmd in ("git commit -m x", "cd pkg && git commit -m x", f"git -C {repo} commit -m x",
161	"git -C pkg commit -m x", "git -C nope commit -m x"):
162	assert rr.resolve_repo_root(str(repo), cmd, rr.COMMIT_SUBCOMMANDS) == (str(repo), rr.RES_CWD), cmd
163
164	def test_explicit_other_repo_in_command_beats_cwd(self, workspace):
165	ws, repo = workspace
166	other = make_repo(ws / "other")
167	assert rr.resolve_repo_root(str(repo), f"git -C {other} commit -m x", rr.COMMIT_SUBCOMMANDS) == (str(other), rr.RES_COMMAND)
168	assert rr.resolve_repo_root(str(repo), "git -C ../other push", rr.PUSH_SUBCOMMANDS) == (str(other), rr.RES_COMMAND)
169	assert rr.resolve_repo_root(str(repo), "cd ../other && git commit -m x", rr.COMMIT_SUBCOMMANDS) == (str(other), rr.RES_COMMAND)
170	assert rr.resolve_repo_root(str(repo), "git --git-dir=../other/.git --work-tree=../other commit -m x", rr.COMMIT_SUBCOMMANDS) == (str(other), rr.RES_COMMAND)
171	assert rr.resolve_repo_root(str(repo), f"git -C {other} status && git commit -m x", rr.COMMIT_SUBCOMMANDS) == (str(repo), rr.RES_CWD)
172
173	def test_order_command_then_paths_then_sha(self, workspace):
174	ws, repo = workspace
175	other = make_repo(ws / "other")
176	sha, _ = commit_file(repo, "app.py", VULN_PY)
177	assert rr.resolve_repo_root(str(ws), "git -C other commit", rr.COMMIT_SUBCOMMANDS,
178	sha=sha, touched_paths=[str(repo / "app.py")]) == (str(other), rr.RES_COMMAND)
179	assert rr.resolve_repo_root(str(ws), "git commit", rr.COMMIT_SUBCOMMANDS,
180	sha=sha, touched_paths=[str(repo / "app.py")]) == (str(repo), rr.RES_TOUCHED_PATHS)
181	assert rr.resolve_repo_root(str(ws), "git commit", rr.COMMIT_SUBCOMMANDS,
182	sha=sha) == (str(repo), rr.RES_SHA_SCAN)
183	assert rr.resolve_repo_root(str(ws), "git commit") == (None, rr.RES_NONE)
184
185
186	class TestRegexes:
187	@pytest.mark.parametrize("cmd", [
188	"git commit -m x",
189	"git -C sub commit -m x",
190	'git -C "a b" commit -m x',
191	"git -c core.editor=true -C sub commit --amend",
192	"git --git-dir=x/.git --work-tree=x commit -m x",
193	"git --git-dir x/.git commit -m x",
194	"cd sub && git commit -m x",
195	"gt create -m x",
196	])
197	def test_commit_re(self, cmd):
198	assert hook._GIT_COMMIT_RE.search(cmd)
199
200	@pytest.mark.parametrize("cmd", [
201	"git push", "git -C sub push origin main", 'git -C "a b" push',
202	"git --work-tree x push -u origin HEAD", "gt submit",
203	])
204	def test_push_re(self, cmd):
205	assert hook._GIT_PUSH_RE.search(cmd)
206
207	@pytest.mark.parametrize("cmd", ["git status", "git log --oneline", "echo commit"])
208	def test_commit_re_negative(self, cmd):
209	assert not hook._GIT_COMMIT_RE.search(cmd)
210
211	@pytest.mark.parametrize("unit", ['-c "a"', "-c 'a'", "--no-a=b", "--a=b=c=d", "--git-dir x"])
212	@pytest.mark.parametrize("regex,verb", [("_GIT_COMMIT_RE", "commit"), ("_GIT_PUSH_RE", "push")])
213	def test_global_option_prefix_is_linear(self, regex, verb, unit):
214	cre = getattr(hook, regex)
215	prefix = "git " + " ".join([unit] * 40)
216	t0 = time.perf_counter()
217	assert not cre.search(prefix + " " + verb + "foo")
218	assert time.perf_counter() - t0 < 0.05
219	assert cre.search(prefix + " " + verb + " -m x")
220
221
222	class TestHooksJson:
223	def test_matchers_and_events(self):
224	cfg = json.loads((HOOKS_DIR / "hooks.json").read_text())
225	# Stop already reviews what subagents changed during the turn.
226	assert "SubagentStop" not in cfg["hooks"]
227	stop = cfg["hooks"]["Stop"][0]["hooks"][0]
228	assert stop["command"].endswith('security_reminder_hook.py"')
229	assert stop["asyncRewake"] is True
230	bash = [g for g in cfg["hooks"]["PostToolUse"] if g.get("matcher") == "Bash"][0]
231	ifs = {h.get("if") for h in bash["hooks"]}
232	assert {"Bash(git commit:)", "Bash(git push:)", "Bash(git -C * commit *)",
233	"Bash(git -C * push)", "Bash(gt create:)", "Bash(gt modify:*)",
234	"Bash(gt submit:*)"} <= ifs
235
236
237
238	def _read_state(env):
239	d = env["SECURITY_WARNINGS_STATE_DIR"]
240	files = [e.path for e in os.scandir(d) if e.name.endswith(".json")]
241	assert files, os.listdir(d)
242	with open(files[0]) as f:
243	return json.load(f)
244
245
246	def _read_state_or_empty(env):
247	d = env["SECURITY_WARNINGS_STATE_DIR"]
248	if not any(e.name.endswith(".json") for e in os.scandir(d)):
249	return {}
250	return _read_state(env)
251
252
253	class TestCommitReview:
254	def test_cwd_is_repo_unchanged(self, workspace, hook_env, stub_api):
255	ws, repo = workspace
256	sha, out = commit_file(repo, "app.py", VULN_PY)
257	rc, so, se = run_hook(bash_payload(repo, "git commit -m change", out), hook_env)
258	m = metrics_of(so)
259	assert m["commit_review"] is True
260	assert "cwd_is_repo" not in m and "repo_resolution" not in m
261	assert m.get("files_reviewed") == 1 and m.get("skip_reason") is None
262	assert stub_api.calls
263
264	def test_cwd_is_repo_cd_subdir_unchanged(self, workspace, hook_env, stub_api):
265	ws, repo = workspace
266	(repo / "pkg").mkdir()
267	sha, out = commit_file(repo, "pkg/app.py", VULN_PY)
268	rc, so, se = run_hook(bash_payload(repo, "cd pkg && git commit -m change", out), hook_env)
269	m = metrics_of(so)
270	assert "cwd_is_repo" not in m and "repo_resolution" not in m
271	assert m.get("files_reviewed") == 1 and m.get("skip_reason") is None
272
273	def test_cwd_is_repo_dash_C_other_repo_reviews_other_repo(self, workspace, hook_env, stub_api):
274	ws, repo = workspace
275	other = make_repo(ws / "other")
276	sha, out = commit_file(other, "srv.py", VULN_PY)
277	rc, so, se = run_hook(bash_payload(repo, f"git -C {other} commit -m change", out), hook_env)
278	m = metrics_of(so)
279	assert m.get("skip_reason") is None, m
280	assert "cwd_is_repo" not in m and m["repo_resolution"] == rr.RES_COMMAND
281	assert m["files_reviewed"] == 1
282	assert stub_api.calls
283	assert (other / ".git" / "sg-reviewed-shas").exists()
284	assert not (repo / ".git" / "sg-reviewed-shas").exists()
285
286	def test_workspace_cwd_cd_sub(self, workspace, hook_env, stub_api):
287	ws, repo = workspace
288	sha, out = commit_file(repo, "app.py", VULN_PY)
289	rc, so, se = run_hook(bash_payload(ws, "cd sub && git commit -m change", out), hook_env)
290	m = metrics_of(so)
291	assert m.get("skip_reason") is None, m
292	assert m["cwd_is_repo"] is False and m["repo_resolution"] == rr.RES_COMMAND
293	assert m["files_reviewed"] == 1
294	assert stub_api.calls
295
296	def test_workspace_cwd_git_dash_C_quiet_uses_reflog(self, workspace, hook_env, stub_api):
297	ws, repo = workspace
298	sha, _ = commit_file(repo, "app.py", VULN_PY)
299	rc, so, se = run_hook(bash_payload(ws, "git -C sub commit -q -m change", ""), hook_env)
300	m = metrics_of(so)
301	assert m.get("skip_reason") is None, m
302	assert m["repo_resolution"] == rr.RES_COMMAND and m["sha_via_reflog"] is True
303	assert m["files_reviewed"] == 1
304
305	def test_workspace_cwd_sha_scan(self, workspace, hook_env, stub_api):
306	ws, repo = workspace
307	sha, out = commit_file(repo, "app.py", VULN_PY)
308	rc, so, se = run_hook(bash_payload(ws, "git commit -m change", out), hook_env)
309	m = metrics_of(so)
310	assert m.get("skip_reason") is None, m
311	assert m["repo_resolution"] == rr.RES_SHA_SCAN and m["files_reviewed"] == 1
312	assert "repo_root_hint" not in _read_state_or_empty(hook_env)
313
314	def test_workspace_cwd_sha_scan_via_project_dir(self, workspace, hook_env, tmp_path):
315	ws, repo = workspace
316	sha, out = commit_file(repo, "app.py", VULN_PY)
317	elsewhere = tmp_path / "elsewhere"
318	elsewhere.mkdir()
319	env = {**hook_env, "CLAUDE_PROJECT_DIR": str(ws)}
320	rc, so, se = run_hook(bash_payload(elsewhere, "git commit -m change", out), env)
321	m = metrics_of(so)
322	assert m.get("skip_reason") is None, m
323	assert m["repo_resolution"] == rr.RES_SHA_SCAN
324
325	def test_hint_saved_and_used(self, workspace, hook_env):
326	ws, repo = workspace
327	sha, out = commit_file(repo, "app.py", VULN_PY)
328	run_hook(bash_payload(ws, "cd sub && git commit -m change", out), hook_env)
329	state = _read_state(hook_env)
330	assert state.get("repo_root_hint") == str(repo)
331	assert rr.resolve_repo_root(str(ws), "git commit") == (None, rr.RES_NONE)
332	os.environ["SECURITY_WARNINGS_STATE_DIR"] = hook_env["SECURITY_WARNINGS_STATE_DIR"]
333	try:
334	assert rr.load_repo_hint("s1") == str(repo)
335	finally:
336	os.environ.pop("SECURITY_WARNINGS_STATE_DIR", None)
337
338	def test_hint_used_for_quiet_commit_without_dir(self, workspace, hook_env):
339	ws, repo = workspace
340	sha, out = commit_file(repo, "app.py", VULN_PY)
341	run_hook(bash_payload(ws, "cd sub && git commit -m change", out), hook_env)
342	commit_file(repo, "app.py", VULN_PY + "# 2\n")
343	rc, so, se = run_hook(bash_payload(ws, "git commit -q -m change", ""), hook_env)
344	m = metrics_of(so)
345	assert m.get("skip_reason") is None, m
346	assert m["repo_resolution"] == rr.RES_HINT and m["sha_via_reflog"] is True
347
348	def test_unresolvable_still_skips_26(self, workspace, hook_env, tmp_path):
349	ws, repo = workspace
350	sha, out = commit_file(repo, "app.py", VULN_PY)
351	empty = tmp_path / "empty"
352	empty.mkdir()
353	rc, so, se = run_hook(bash_payload(empty, "git commit -m change", out), hook_env)
354	m = metrics_of(so)
355	assert m["skip_reason"] == 26 and m["cwd_is_repo"] is False and m["repo_resolution"] == rr.RES_NONE
356
357	def test_no_credentials_gate_precedes_repo_resolution(self, workspace, hook_env):
358	ws, repo = workspace
359	sha, out = commit_file(repo, "app.py", VULN_PY)
360	env = {k: v for k, v in hook_env.items() if k != "ANTHROPIC_API_KEY"}
361	rc, so, se = run_hook(bash_payload(ws, "cd sub && git commit -m change", out), env)
362	m = metrics_of(so)
363	assert m["skip_reason"] == 22 and m["repo_resolution"] == rr.RES_COMMAND
364
365	def test_dedup_sentinel_uses_resolved_repo(self, workspace, hook_env):
366	ws, repo = workspace
367	sha, out = commit_file(repo, "app.py", VULN_PY)
368	p = bash_payload(ws, "git -C sub commit -m change && git -C sub push", out, tool_use_id="toolu_1")
369	rc1, so1, _ = run_hook(p, hook_env)
370	rc2, so2, _ = run_hook(p, hook_env)
371	assert metrics_of(so1).get("commit_review") is True
372	assert metrics_of(so2) == {"bash_hook_dedup": True}
373
374
375	class TestPushSweep:
376	def test_workspace_cwd_git_dash_C_push(self, workspace, hook_env, tmp_path):
377	ws, repo = workspace
378	remote = tmp_path / "remote.git"
379	git(ws, "init", "-q", "--bare", str(remote))
380	git(repo, "remote", "add", "origin", str(remote))
381	git(repo, "push", "-q", "-u", "origin", "main")
382	base = git(repo, "rev-parse", "HEAD").strip()
383	sha, _ = commit_file(repo, "app.py", VULN_PY)
384	out = git(repo, "push", "--porcelain", "origin", "main")
385	push_stdout = f"To {remote}\n {base[:7]}..{sha[:7]} main -> main\n"
386	rc, so, se = run_hook(bash_payload(ws, "git -C sub push origin main", push_stdout), hook_env)
387	m = metrics_of(so)
388	assert m["push_sweep"] is True
389	assert m["cwd_is_repo"] is False and m["repo_resolution"] == rr.RES_COMMAND
390	assert m.get("skip_reason") != 26
391	assert m.get("pushed") == 1
392
393	def test_workspace_cwd_sha_scan_push_leaves_no_hint(self, workspace, hook_env, tmp_path):
394	ws, repo = workspace
395	remote = tmp_path / "remote.git"
396	git(ws, "init", "-q", "--bare", str(remote))
397	git(repo, "remote", "add", "origin", str(remote))
398	git(repo, "push", "-q", "-u", "origin", "main")
399	base = git(repo, "rev-parse", "HEAD").strip()
400	sha, _ = commit_file(repo, "app.py", VULN_PY)
401	git(repo, "push", "-q", "origin", "main")
402	push_stdout = f"To {remote}\n {base[:7]}..{sha[:7]} main -> main\n"
403	rc, so, se = run_hook(bash_payload(ws, "git push origin main", push_stdout), hook_env)
404	m = metrics_of(so)
405	assert m["push_sweep"] is True and m["repo_resolution"] == rr.RES_SHA_SCAN, m
406	assert m.get("skip_reason") != 26
407	assert "repo_root_hint" not in _read_state_or_empty(hook_env)
408
409	def test_cwd_is_repo_dash_C_other_repo(self, workspace, hook_env, tmp_path):
410	ws, repo = workspace
411	other = make_repo(ws / "other")
412	remote = tmp_path / "remote.git"
413	git(ws, "init", "-q", "--bare", str(remote))
414	git(other, "remote", "add", "origin", str(remote))
415	git(other, "push", "-q", "-u", "origin", "main")
416	base = git(other, "rev-parse", "HEAD").strip()
417	sha, _ = commit_file(other, "srv.py", VULN_PY)
418	git(other, "push", "-q", "origin", "main")
419	push_stdout = f"To {remote}\n {base[:7]}..{sha[:7]} main -> main\n"
420	rc, so, se = run_hook(bash_payload(repo, f"git -C {other} push origin main", push_stdout), hook_env)
421	m = metrics_of(so)
422	assert m["push_sweep"] is True and "cwd_is_repo" not in m
423	assert m["repo_resolution"] == rr.RES_COMMAND and m.get("pushed") == 1
424
425
426	class TestStop:
427	def _touch(self, ws, repo, hook_env, session_id="s1"):
428	(repo / "app.py").write_text(VULN_PY)
429	run_hook(edit_payload(ws, repo / "app.py", VULN_PY, session_id=session_id), hook_env)
430
431	def test_cwd_is_repo_unchanged(self, workspace, hook_env, stub_api):
432	ws, repo = workspace
433	run_hook(ups_payload(repo), hook_env)
434	self._touch(repo, repo, hook_env)
435	rc, so, se = run_hook(stop_payload(repo), hook_env)
436	m = metrics_of(so)
437	assert m.get("skip_reason") is None, m
438	assert "repo_resolution" not in m and m["files_reviewed"] == 1
439	assert stub_api.calls
440
441	def test_workspace_cwd_resolves_from_touched_paths(self, workspace, hook_env, stub_api):
442	ws, repo = workspace
443	run_hook(ups_payload(ws), hook_env)
444	self._touch(ws, repo, hook_env)
445	rc, so, se = run_hook(stop_payload(ws), hook_env)
446	m = metrics_of(so)
447	assert m.get("skip_reason") is None, m
448	assert m["cwd_is_repo"] is False and m["repo_resolution"] == rr.RES_TOUCHED_PATHS
449	assert m["files_reviewed"] == 1 and m["review_set_count"] == 1
450	assert stub_api.calls
451
452	def test_subagent_stop_reviews_without_consuming_session_state(self, workspace, hook_env, stub_api):
453	ws, repo = workspace
454	run_hook(ups_payload(ws), hook_env)
455	self._touch(ws, repo, hook_env)
456	before = _read_state(hook_env)
457	rc, so, se = run_hook(stop_payload(ws, event="SubagentStop"), hook_env)
458	m = metrics_of(so)
459	assert m.get("skip_reason") is None, m
460	assert m["repo_resolution"] == rr.RES_TOUCHED_PATHS and m["files_reviewed"] == 1
461	after = _read_state(hook_env)
462	assert after["touched_paths"] == before["touched_paths"] != []
463	assert after.get("baseline_sha") == before.get("baseline_sha")
464	assert after.get("reviewed_diff_hash")
465	assert "repo_root_hint" not in after
466	n_calls = len(stub_api.calls)
467	rc, so, se = run_hook(stop_payload(ws), hook_env)
468	m2 = metrics_of(so)
469	assert m2["skip_reason"] == 12, m2
470	assert m2["repo_resolution"] == rr.RES_TOUCHED_PATHS
471	assert len(stub_api.calls) == n_calls
472	final = _read_state(hook_env)
473	assert final["touched_paths"] == [] and "reviewed_diff_hash" not in final
474
475	def test_main_edits_during_subagent_review_are_reviewed_at_stop(self, workspace, hook_env, stub_api):
476	ws, repo = workspace
477	run_hook(ups_payload(repo), hook_env)
478	self._touch(repo, repo, hook_env)
479	stub_api.delay = 3
480	res = {}
481	t = threading.Thread(target=lambda: res.update(
482	sub=run_hook(stop_payload(repo, event="SubagentStop"), hook_env)))
483	t.start()
484	time.sleep(1.5)
485	(repo / "b.py").write_text("import os\nos.system(input())\n")
486	run_hook(edit_payload(repo, repo / "b.py", "x"), hook_env)
487	t.join()
488	stub_api.delay = 0
489	m_sub = metrics_of(res["sub"][1])
490	assert m_sub.get("skip_reason") is None and m_sub["files_reviewed"] == 1, m_sub
491	n_calls = len(stub_api.calls)
492	rc, so, se = run_hook(stop_payload(repo), hook_env)
493	m = metrics_of(so)
494	assert m.get("skip_reason") is None, m
495	assert m["files_reviewed"] == 2 and m["touched_paths_count"] == 2
496	assert len(stub_api.calls) > n_calls
497
498	def test_subagent_stop_findings_do_not_advance_baseline_or_fire_count(self, workspace, hook_env, stub_api):
499	ws, repo = workspace
500	run_hook(ups_payload(repo), hook_env)
501	self._touch(repo, repo, hook_env)
502	before = _read_state(hook_env)
503	stub_api.vulns = [STUB_VULN]
504	rc, so, se = run_hook(stop_payload(repo, event="SubagentStop"), hook_env)
505	m = metrics_of(so)
506	assert rc == 2 and m["vulns_found"] == 1, (rc, m)
507	assert "repo_resolution" not in m and "cwd_is_repo" not in m
508	after = _read_state(hook_env)
509	assert after.get("baseline_sha") == before.get("baseline_sha")
510	assert not after.get("stop_hook_fire_count")
511	assert after["touched_paths"] == before["touched_paths"]
512	assert len(after.get("previous_findings", [])) == 1 and after.get("reviewed_diff_hash")
513	rc, so, se = run_hook(stop_payload(repo), hook_env)
514	assert metrics_of(so)["skip_reason"] == 12
515	(repo / "app.py").write_text(VULN_PY + "x = 1\n")
516	run_hook(edit_payload(repo, repo / "app.py", "x"), hook_env)
517	rc, so, se = run_hook(stop_payload(repo), hook_env)
518	m3 = metrics_of(so)
519	assert rc == 2 and m3.get("skip_reason") is None and m3["files_reviewed"] == 1, m3
520	assert _read_state(hook_env)["stop_hook_fire_count"] == 1
521
522	def test_subagent_stop_in_other_worktree_is_skipped(self, workspace, hook_env, stub_api, tmp_path):
523	ws, repo = workspace
524	wt = tmp_path / "wt"
525	git(repo, "worktree", "add", "-q", "-b", "agent", str(wt))
526	run_hook(ups_payload(repo), hook_env)
527	self._touch(repo, repo, hook_env)
528	before = _read_state(hook_env)
529	(wt / "new.py").write_text("import pickle\npickle.loads(b)\n")
530	run_hook(edit_payload(wt, wt / "new.py", "x"), hook_env)
531	env = {**hook_env, "CLAUDE_PROJECT_DIR": str(repo)}
532	rc, so, se = run_hook(stop_payload(wt, event="SubagentStop"), env)
533	m = metrics_of(so)
534	assert m["skip_reason"] == 11, m
535	assert not stub_api.calls
536	after = _read_state(hook_env)
537	assert after.get("baseline_sha") == before.get("baseline_sha")
538	assert after.get("head_at_capture") == before.get("head_at_capture")
539	rc, so, se = run_hook(stop_payload(repo), env)
540	m2 = metrics_of(so)
541	assert m2.get("skip_reason") is None and m2["files_reviewed"] == 1, m2
542
543	def test_subagent_stop_in_other_worktree_from_workspace_cwd(self, workspace, hook_env, stub_api, tmp_path):
544	ws, repo = workspace
545	wt = tmp_path / "wt"
546	git(repo, "worktree", "add", "-q", "-b", "agent", str(wt))
547	run_hook(ups_payload(ws), hook_env)
548	(wt / "new.py").write_text("import pickle\npickle.loads(b)\n")
549	run_hook(edit_payload(ws, wt / "new.py", "x"), hook_env)
550	env = {**hook_env, "CLAUDE_PROJECT_DIR": str(repo)}
551	rc, so, se = run_hook(stop_payload(ws, event="SubagentStop"), env)
552	m = metrics_of(so)
553	assert m["skip_reason"] == 11 and m["repo_resolution"] == rr.RES_TOUCHED_PATHS, m
554	assert not stub_api.calls
555	assert "repo_root_hint" not in _read_state(hook_env)
556
557	def test_repository_config_cannot_run_programs(self, workspace, hook_env, stub_api, tmp_path):
558	ws, repo = workspace
559	marker = tmp_path / "marker"
560	mon = tmp_path / "mon.sh"
561	mon.write_text(f"#!/bin/sh\necho ran >> '{marker}'\n")
562	mon.chmod(0o755)
563	git(repo, "config", "core.fsmonitor", str(mon))
564	clean = {k: v for k, v in os.environ.items() if not k.startswith("GIT_CONFIG_")}
565	subprocess.run(["git", "status"], cwd=repo, env={clean, GIT_ENV}, capture_output=True)
566	assert marker.exists()
567	marker.unlink()
568	run_hook(ups_payload(ws), hook_env)
569	self._touch(ws, repo, hook_env)
570	rc, so, se = run_hook(stop_payload(ws), hook_env)
571	m = metrics_of(so)
572	assert m.get("skip_reason") is None and m["files_reviewed"] == 1, m
573	assert not marker.exists()
574	sha, out = commit_file(repo, "app.py", VULN_PY + "z = 3\n")
575	marker.unlink(missing_ok=True)
576	rc, so, se = run_hook(bash_payload(ws, "cd sub && git commit -m change", out), hook_env)
577	assert metrics_of(so).get("files_reviewed") == 1
578	assert not marker.exists()
579
580	def test_safe_git_env_extends_existing_config_count(self):
581	base = {"GIT_CONFIG_COUNT": "2", "GIT_CONFIG_KEY_0": "a.b", "GIT_CONFIG_VALUE_0": "1",
582	"GIT_CONFIG_KEY_1": "c.d", "GIT_CONFIG_VALUE_1": "2"}
583	env = gitutil.git_config_env(gitutil.SAFE_GIT_CONFIG, base=base)
584	assert env["GIT_CONFIG_COUNT"] == str(2 + len(gitutil.SAFE_GIT_CONFIG))
585	assert "GIT_CONFIG_KEY_0" not in env and "GIT_CONFIG_KEY_1" not in env
586	got = {env[f"GIT_CONFIG_KEY_{i}"]: env[f"GIT_CONFIG_VALUE_{i}"]
587	for i in range(2, int(env["GIT_CONFIG_COUNT"]))}
588	assert got == dict(gitutil.SAFE_GIT_CONFIG)
589	assert gitutil.git_config_env((("x.y", "z"),), base={"GIT_CONFIG_COUNT": "junk"})["GIT_CONFIG_COUNT"] == "1"
590
591	def test_subagent_stop_same_repo_with_project_dir_reviews(self, workspace, hook_env, stub_api):
592	ws, repo = workspace
593	(repo / "pkg").mkdir()
594	run_hook(ups_payload(repo), hook_env)
595	self._touch(repo, repo, hook_env)
596	env = {**hook_env, "CLAUDE_PROJECT_DIR": str(repo)}
597	rc, so, se = run_hook(stop_payload(repo / "pkg", event="SubagentStop"), env)
598	m = metrics_of(so)
599	assert m.get("skip_reason") is None and m["files_reviewed"] == 1, m
600	env_ws = {**hook_env, "CLAUDE_PROJECT_DIR": str(ws)}
601	(repo / "app.py").write_text(VULN_PY + "y = 2\n")
602	rc, so, se = run_hook(stop_payload(repo, event="SubagentStop"), env_ws)
603	assert metrics_of(so).get("skip_reason") is None
604
605	def test_ups_uses_hint_for_baseline(self, workspace, hook_env):
606	ws, repo = workspace
607	run_hook(ups_payload(ws), hook_env)
608	self._touch(ws, repo, hook_env)
609	run_hook(stop_payload(ws), hook_env)
610	(repo / "app.py").write_text(VULN_PY + "\n# more\n")
611	run_hook(ups_payload(ws, session_id="s1"), hook_env)
612	state = _read_state(hook_env)
613	assert state.get("baseline_sha") and state.get("repo_root_hint") == str(repo)
614
615	def test_workspace_cwd_nothing_touched_skips(self, workspace, hook_env, stub_api):
616	ws, repo = workspace
617	run_hook(ups_payload(ws), hook_env)
618	rc, so, se = run_hook(stop_payload(ws), hook_env)
619	m = metrics_of(so)
620	assert m["skip_reason"] == 9 and m["repo_resolution"] == rr.RES_NONE
621	assert not stub_api.calls