mirror of
https://github.com/storytold/craft-common-actions.git
synced 2026-11-15 15:02:24 +00:00
FredAppsandClaude Opus 5.5
e09a018c65
Anyone can commit under an agent email, so a free-form Git name such as "Private Person <noreply@anthropic.com>" became a published model title, which the apps would ship in their About windows. Each provider now lists the words a model title may use. A name with any other word credits the agent's default model and repeats none of it; versions are bounded too. Across the co-author trailers in the 13 apps' history, 18,099 keep their model row and 7 unusual names now fall back to the default. Also pin that the PR lookup reads the merge commit rather than the PR head. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
592 lines
35 KiB
Python
592 lines
35 KiB
Python
"""Exercise contributor attribution, name consent and atomic release snapshots."""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "actions/refresh-contributors/refresh_contributors.py"
|
|
SPEC = importlib.util.spec_from_file_location("refresh_contributors", SCRIPT)
|
|
credits = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(credits)
|
|
|
|
|
|
class ContributorRefreshTests(unittest.TestCase):
|
|
"""Use real temporary Git history and fixture GitHub responses."""
|
|
|
|
def setUp(self):
|
|
"""Create a repository whose Git author name must never become a public name."""
|
|
self.directory = tempfile.TemporaryDirectory(prefix="caller checkout ")
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.root = Path(self.directory.name)
|
|
self.git("init", "--initial-branch=main")
|
|
self.git("config", "user.name", "Private Git Author")
|
|
self.git("config", "user.email", "private@example.test")
|
|
(self.root / "contributors").mkdir()
|
|
self.people = self.root / "contributors/people.toml"
|
|
self.people.write_text('[people.alice]\nsync_github_name = true\n\n'
|
|
'[people.bob]\ndisplay_name = "Unconsented name"\nsync_github_name = true\n')
|
|
self.output = self.root / "contributors/contributors.json"
|
|
self.output.write_text('{"old": true}\n')
|
|
(self.root / "text.txt").write_text("first\n")
|
|
(self.root / "old.bin").write_bytes(b"\0old")
|
|
self.identities = {}
|
|
self.coauthors = {}
|
|
self.pulls = []
|
|
self.associated = {}
|
|
self.associated_lookups = []
|
|
self.linked = {}
|
|
self.permissions = {}
|
|
self.api_calls = []
|
|
self.commit("Initial fixture", "alice")
|
|
|
|
def git(self, *args):
|
|
"""Run Git in the fixture, failing the test on a setup error."""
|
|
return subprocess.check_output(["git", *args], cwd=self.root, stderr=subprocess.PIPE).decode().strip()
|
|
|
|
def commit(self, message, login, author=None):
|
|
"""Record the GitHub identity separately from untrusted Git author text."""
|
|
self.git("add", ".")
|
|
self.git(*(["-c", f"user.name={author[0]}", "-c", f"user.email={author[1]}"] if author else []),
|
|
"commit", "-m", message)
|
|
sha = self.git("rev-parse", "HEAD")
|
|
self.identities[sha] = login
|
|
return sha
|
|
|
|
def api(self, root, endpoint, query, paginate=False):
|
|
"""Supply projected API records without a network connection."""
|
|
self.api_calls.append((endpoint, query, paginate))
|
|
if "/pulls?" in endpoint:
|
|
self.assertTrue(paginate)
|
|
return self.pulls
|
|
if "/collaborators/" in endpoint:
|
|
login = endpoint.split("/")[-2]
|
|
return [dict(value=self.permissions.get(login, "read"))]
|
|
if endpoint.casefold() == "users/alice":
|
|
return [dict(value="Alice Public")]
|
|
self.fail(f"Unexpected API request: {endpoint}")
|
|
|
|
def pull_openers(self, root, repo, shas, reachable):
|
|
"""Answer commit-to-PR lookups from the fixture's merged-PR openers."""
|
|
self.associated_lookups.append(set(shas))
|
|
return {sha: login for sha, login in self.associated.items() if sha in shas}
|
|
|
|
def collect(self):
|
|
"""Collect through the real Git parser and fixture API."""
|
|
authors = {sha: self.coauthors.get(sha, {login.casefold(): login} if login else {})
|
|
for sha, login in self.identities.items()}
|
|
linked = {sha: self.linked.get(sha, login) for sha, login in self.identities.items()}
|
|
with patch.object(credits, "commit_authors", return_value=(self.identities, authors, linked)), \
|
|
patch.object(credits, "pull_openers", side_effect=self.pull_openers), \
|
|
patch.object(credits, "api", side_effect=self.api):
|
|
return credits.collect(self.root, "owner/app")
|
|
|
|
def history_page(self, *nodes):
|
|
"""One GraphQL history page holding the given commit nodes."""
|
|
return json.dumps(dict(data=dict(repository=dict(object=dict(history=dict(
|
|
nodes=list(nodes), pageInfo=dict(hasNextPage=False, endCursor=None)))))))
|
|
|
|
def test_agent_and_bot_accounts_are_never_people(self):
|
|
"""GitHub links agent trailers to accounts such as claude; only people get rows."""
|
|
node = dict(oid="a" * 40,
|
|
author=dict(email="198982749+Copilot@users.noreply.github.com", user=dict(login="Copilot")),
|
|
signature=dict(state="VALID", signer=dict(login="Copilot")),
|
|
authors=dict(nodes=[dict(email="noreply@anthropic.com", user=dict(login="claude")),
|
|
dict(email="codex@openai.com", user=dict(login="codex")),
|
|
dict(email="1+dependabot[bot]@users.noreply.github.com",
|
|
user=dict(login="dependabot[bot]")),
|
|
dict(email="alice@anthropic.com", user=dict(login="alice"))],
|
|
pageInfo=dict(hasNextPage=False)))
|
|
with patch.object(credits, "run", return_value=self.history_page(node)):
|
|
identities, authors, linked = credits.commit_authors(self.root, "owner/app", "a" * 40)
|
|
self.assertEqual(authors["a" * 40], {"alice": "alice"})
|
|
self.assertIsNone(identities["a" * 40])
|
|
self.assertIsNone(linked["a" * 40])
|
|
|
|
def test_agent_authored_commit_credits_its_model_and_the_pr_opener(self):
|
|
"""An agent's own commit counts for its model and for the person whose PR merged it."""
|
|
(self.root / "text.txt").write_text("agent change\n")
|
|
sha = self.commit("Agent change", None, author=("Claude", "noreply@anthropic.com"))
|
|
(self.root / "text.txt").write_text("cursor change\n")
|
|
cursor = self.commit("Cursor change\n\nCo-authored-by: Claude Opus 5.5 <noreply@anthropic.com>", None,
|
|
author=("Cursor Agent", "cursoragent@cursor.com"))
|
|
self.associated = {sha: "bob", cursor: "Copilot"}
|
|
result = self.collect()
|
|
rows = {row["login"]: row for row in result["contributors"]}
|
|
self.assertEqual(set(rows), {"alice", "bob"})
|
|
self.assertEqual(rows["bob"]["commits"], 1)
|
|
self.assertEqual(result["unattributed_commits"], 1)
|
|
self.assertEqual({row["id"]: row["commits"] for row in result["models"]},
|
|
{"anthropic/claude": 1, "anysphere/cursor": 1, "anthropic/claude-opus/5.5": 1})
|
|
self.assertNotIn("Claude <", json.dumps(result))
|
|
|
|
def test_merge_commit_pr_opener_verifies_their_own_names(self):
|
|
"""An unsigned self-added entry merged with a merge commit is verified through its PR."""
|
|
self.people.write_text('[people.alice]\nsync_github_name = true\n\n'
|
|
'[people.bob]\ndisplay_name = "Bob chose this"\n')
|
|
sha = self.commit("Add my name", "bob")
|
|
self.identities[sha] = None
|
|
self.coauthors[sha] = {"bob": "bob"}
|
|
for opener, expected in ((None, False), ("mallory", False), ("bob", True)):
|
|
with self.subTest(opener=opener):
|
|
self.associated = {sha: opener} if opener else {}
|
|
rows = {row["login"]: row for row in self.collect()["contributors"]}
|
|
self.assertEqual(rows["bob"]["people_entry"], expected)
|
|
self.assertIn(sha, self.associated_lookups[-1])
|
|
|
|
def test_writer_carrying_the_persons_own_commit_verifies_their_names(self):
|
|
"""An integration PR keeps the contributor's authorship; its writer-opener vouches for it."""
|
|
self.people.write_text('[people.alice]\nsync_github_name = true\n\n'
|
|
'[people.bob]\ndisplay_name = "Bob chose this"\n')
|
|
sha = self.commit("Add my name", "bob")
|
|
self.identities[sha] = None
|
|
self.coauthors[sha] = {"bob": "bob"}
|
|
self.associated = {sha: "maintainer"}
|
|
for author, permission, expected in (("bob", "write", True), ("bob", "read", False),
|
|
("mallory", "write", False)):
|
|
with self.subTest(author=author, permission=permission):
|
|
self.linked = {sha: author}
|
|
self.permissions["maintainer"] = permission
|
|
rows = {row["login"]: row for row in self.collect()["contributors"]}
|
|
self.assertEqual(rows["bob"]["people_entry"], expected)
|
|
|
|
def test_pull_openers_trust_only_one_merged_reachable_opener(self):
|
|
"""Ignore unmerged, unreachable, bot-opened and truncated PR lists; disagreeing openers stay unknown."""
|
|
shas = [c * 40 for c in "abcdef"] + ["f" * 39 + "e"]
|
|
merged = lambda login, oid="1" * 40, kind="User": dict(
|
|
merged=True, author=dict(__typename=kind, login=login), mergeCommit=dict(oid=oid))
|
|
repository = dict(
|
|
c0=dict(associatedPullRequests=dict(totalCount=2, nodes=[merged("alice"), merged("alice")])),
|
|
c1=dict(associatedPullRequests=dict(totalCount=2, nodes=[merged("alice"), merged("mallory")])),
|
|
c2=dict(associatedPullRequests=dict(totalCount=1, nodes=[dict(merged=False, author=dict(login="bob"),
|
|
mergeCommit=None)])),
|
|
c3=dict(associatedPullRequests=dict(totalCount=1, nodes=[merged("bob", "9" * 40)])),
|
|
c4=dict(associatedPullRequests=dict(totalCount=11, nodes=[merged("bob")])),
|
|
c5=dict(associatedPullRequests=dict(totalCount=1, nodes=[merged("dependabot", kind="Bot")])),
|
|
c6=None)
|
|
with patch.object(credits, "run", return_value=json.dumps(dict(data=dict(repository=repository)))) as mocked:
|
|
openers = credits.pull_openers(self.root, "owner/app", shas, {"1" * 40})
|
|
self.assertEqual(openers, {shas[0]: "alice"})
|
|
body = json.loads(mocked.call_args.kwargs["input_text"])
|
|
self.assertIn(f'object(oid: "{shas[0]}")', body["query"])
|
|
self.assertEqual(body["variables"], dict(owner="owner", name="app"))
|
|
with self.assertRaisesRegex(ValueError, "Invalid commit identity"):
|
|
credits.pull_openers(self.root, "owner/app", ['") { evil }'], set())
|
|
|
|
def test_pulls_opened_by_agents_or_bots_add_no_people(self):
|
|
"""A merged PR opened by an agent or bot account never creates a contributor row."""
|
|
sha = self.git("rev-parse", "HEAD")
|
|
self.pulls = [dict(number=1, sha=sha, login="Copilot"), dict(number=2, sha=sha, login="dependabot[bot]")]
|
|
self.assertEqual([row["login"] for row in self.collect()["contributors"]], ["alice"])
|
|
|
|
def test_verified_human_coauthors_are_credited_once(self):
|
|
"""Credit API-linked co-authors without interpreting human names or email text."""
|
|
sha = self.git("rev-parse", "HEAD")
|
|
self.coauthors[sha] = {"alice": "alice", "bob": "bob"}
|
|
rows = {row["login"]: row for row in self.collect()["contributors"]}
|
|
self.assertEqual(rows["alice"]["commits"], 1)
|
|
self.assertEqual(rows["bob"]["commits"], 1)
|
|
self.assertFalse(rows["bob"]["people_entry"])
|
|
|
|
def test_name_preferences_stay_bound_to_the_captured_commit(self):
|
|
"""Concurrent working-tree edits cannot change the release's consent preferences."""
|
|
original_api = self.api
|
|
original_head = self.git("rev-parse", "HEAD")
|
|
|
|
def edit_during_collection(root, endpoint, query, paginate=False):
|
|
if "/pulls?" in endpoint:
|
|
self.people.write_text('[people.alice]\nsync_github_name = false\n')
|
|
return original_api(root, endpoint, query, paginate)
|
|
|
|
with patch.object(self, "api", side_effect=edit_during_collection):
|
|
result = self.collect()
|
|
row = result["contributors"][0]
|
|
self.assertEqual(result["head"], original_head)
|
|
self.assertTrue(row["people_entry"])
|
|
self.assertEqual(row["github_name"], "Alice Public")
|
|
self.assertIn("false", self.people.read_text())
|
|
|
|
def test_author_history_pagination_preserves_primary_and_coauthor_identities(self):
|
|
"""Follow every cursor and retain GitHub-linked co-authors separately from editors."""
|
|
pages = []
|
|
for sha, login, following in (("a" * 40, "alice", "next"), ("b" * 40, "bob", None)):
|
|
node = dict(oid=sha, author=dict(user=dict(login=login)),
|
|
signature=dict(state="VALID", signer=dict(login=login)),
|
|
authors=dict(nodes=[dict(user=dict(login="alice")), dict(user=dict(login="bob"))],
|
|
pageInfo=dict(hasNextPage=False)))
|
|
pages.append(json.dumps(dict(data=dict(repository=dict(object=dict(history=dict(
|
|
nodes=[node], pageInfo=dict(hasNextPage=bool(following), endCursor=following))))))))
|
|
with patch.object(credits, "run", side_effect=pages) as mocked:
|
|
primary, authors, linked = credits.commit_authors(self.root, "owner/app", "a" * 40)
|
|
self.assertEqual(linked, {"a" * 40: "alice", "b" * 40: "bob"})
|
|
self.assertEqual(primary, {"a" * 40: "alice", "b" * 40: "bob"})
|
|
self.assertEqual(authors["b" * 40], {"alice": "alice", "bob": "bob"})
|
|
bodies = [json.loads(call.kwargs["input_text"]) for call in mocked.call_args_list]
|
|
self.assertIn("signature { state signer { login } }", bodies[0]["query"])
|
|
self.assertIsNone(bodies[0]["variables"]["cursor"])
|
|
self.assertEqual(bodies[1]["variables"]["cursor"], "next")
|
|
|
|
def test_consent_identity_requires_a_valid_matching_signer(self):
|
|
"""Spoofable email attribution cannot authorize names; signed authors or PR openers can."""
|
|
sha = self.git("rev-parse", "HEAD")
|
|
signatures = [
|
|
(None, False),
|
|
(dict(state="INVALID", signer=dict(login="alice")), False),
|
|
(dict(state="VALID", signer=None), False),
|
|
(dict(state="VALID", signer=dict(login="mallory")), False),
|
|
(dict(state="VALID", signer=dict(login="ALICE")), True),
|
|
]
|
|
real_run = credits.run
|
|
for signature, expected in signatures:
|
|
with self.subTest(signature=signature):
|
|
node = dict(oid=sha, author=dict(user=dict(login="alice")), signature=signature,
|
|
authors=dict(nodes=[dict(user=dict(login="alice"))],
|
|
pageInfo=dict(hasNextPage=False)))
|
|
page = json.dumps(dict(data=dict(repository=dict(object=dict(history=dict(
|
|
nodes=[node], pageInfo=dict(hasNextPage=False, endCursor=None)))))))
|
|
|
|
def response(args, root, input_text=None):
|
|
if args[:3] == ["gh", "api", "graphql"]:
|
|
return page
|
|
return real_run(args, root, input_text)
|
|
|
|
for opener in (None, "alice", "mallory"):
|
|
via_pull = opener == "alice"
|
|
self.pulls = [dict(number=1, sha=sha, login=opener)] if opener else []
|
|
self.api_calls.clear()
|
|
with patch.object(credits, "run", side_effect=response), \
|
|
patch.object(credits, "api", side_effect=self.api):
|
|
result = credits.collect(self.root, "owner/app")
|
|
row = result["contributors"][0]
|
|
self.assertEqual(row["login"], "alice")
|
|
self.assertEqual(row["commits"], 1)
|
|
self.assertEqual(row["people_entry"], expected or via_pull)
|
|
self.assertEqual(row["github_name"], "Alice Public" if expected or via_pull else None)
|
|
self.assertEqual(any(endpoint == "users/alice" for endpoint, _, _ in self.api_calls),
|
|
expected or via_pull)
|
|
|
|
def test_invalid_people_field_is_an_actionable_error(self):
|
|
"""Reject a malformed opt-in instead of guessing its intended value."""
|
|
self.people.write_text('[people.alice]\nsync_github_name = "false"\n')
|
|
self.commit("Malformed preference", "alice")
|
|
with self.assertRaisesRegex(ValueError, "must be a boolean"):
|
|
credits.load_people(self.root)
|
|
|
|
def test_stalled_history_pagination_is_rejected(self):
|
|
"""Do not loop forever or publish an incomplete history if a cursor repeats."""
|
|
page = json.dumps(dict(data=dict(repository=dict(object=dict(history=dict(
|
|
nodes=[], pageInfo=dict(hasNextPage=True, endCursor="same")))))))
|
|
with patch.object(credits, "run", return_value=page) as mocked:
|
|
with self.assertRaisesRegex(ValueError, "pagination did not advance"):
|
|
credits.commit_authors(self.root, "owner/app", "a" * 40)
|
|
self.assertEqual(mocked.call_count, 2)
|
|
|
|
def test_truncated_coauthor_list_is_rejected(self):
|
|
"""Fail clearly rather than silently omitting authors beyond the page limit."""
|
|
node = dict(oid="a" * 40, author=dict(user=dict(login="alice")),
|
|
authors=dict(nodes=[], pageInfo=dict(hasNextPage=True)))
|
|
page = json.dumps(dict(data=dict(repository=dict(object=dict(history=dict(
|
|
nodes=[node], pageInfo=dict(hasNextPage=False, endCursor=None)))))))
|
|
with patch.object(credits, "run", return_value=page):
|
|
with self.assertRaisesRegex(ValueError, "refusing incomplete credits"):
|
|
credits.commit_authors(self.root, "owner/app", "a" * 40)
|
|
|
|
def test_empty_scalar_api_response_is_an_actionable_error(self):
|
|
"""Do not index an absent profile or repository-permission response."""
|
|
with patch.object(credits, "api", return_value=[]):
|
|
with self.assertRaisesRegex(ValueError, "Expected one API value"):
|
|
credits.single_value(self.root, "users/alice", ".name")
|
|
|
|
def test_username_casing_matches_the_ui_registry_lookup(self):
|
|
"""Preserve the table key so the UI can find the verified entry by row login."""
|
|
self.people.write_text('[people.AlIcE]\nsync_github_name = true\n')
|
|
self.commit("Keep my chosen table casing", "alice")
|
|
row = self.collect()["contributors"][0]
|
|
self.assertEqual(row["login"], "AlIcE")
|
|
self.assertTrue(row["people_entry"])
|
|
self.assertEqual(row["github_name"], "Alice Public")
|
|
|
|
def test_duplicate_account_casing_is_rejected(self):
|
|
"""Do not silently choose between conflicting preferences for one account."""
|
|
self.people.write_text('[people.alice]\ndisplay_name = "One"\n'
|
|
'[people.ALICE]\ndisplay_name = "Two"\n')
|
|
self.commit("Conflicting preferences", "alice")
|
|
with self.assertRaisesRegex(ValueError, "duplicate account casing"):
|
|
credits.load_people(self.root)
|
|
|
|
def test_head_change_during_collection_aborts(self):
|
|
"""Reject a moved checkout rather than describing it with stale commit metadata."""
|
|
original_api = self.api
|
|
original_snapshot = self.output.read_bytes()
|
|
|
|
def commit_during_collection(root, endpoint, query, paginate=False):
|
|
if "/pulls?" in endpoint:
|
|
(self.root / "text.txt").write_text("concurrent commit\n")
|
|
self.commit("Concurrent change", "alice")
|
|
return original_api(root, endpoint, query, paginate)
|
|
|
|
with patch.object(self, "api", side_effect=commit_during_collection):
|
|
with self.assertRaisesRegex(ValueError, "HEAD changed"):
|
|
self.collect()
|
|
self.assertEqual(self.output.read_bytes(), original_snapshot)
|
|
|
|
def test_statistics_models_and_no_inferred_human_names(self):
|
|
"""Attribute text/binary changes and deduplicate explicit model co-authors."""
|
|
(self.root / "old.bin").unlink()
|
|
(self.root / "new.bin").write_bytes(b"\0new")
|
|
(self.root / "text.txt").write_text("first\nsecond\n")
|
|
(self.root / "tab\tname.txt").write_text("quoted path\n")
|
|
self.commit("Change fixture\n\nCo-authored-by: Claude Opus 5.5 <noreply@anthropic.com>\n"
|
|
"Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>\n"
|
|
"Co-authored-by: OpenAI Codex <noreply@openai.com>", "bob")
|
|
result = self.collect()
|
|
rows = {row["login"]: row for row in result["contributors"]}
|
|
self.assertEqual(result["head"], self.git("rev-parse", "HEAD"))
|
|
self.assertEqual(result["commits"], 2)
|
|
self.assertEqual(result["unattributed_commits"], 0)
|
|
self.assertEqual(rows["bob"]["lines_added"], 2)
|
|
self.assertEqual(rows["bob"]["lines_deleted"], 0)
|
|
self.assertEqual(rows["bob"]["binary_added"], 1)
|
|
self.assertEqual(rows["bob"]["binary_deleted"], 1)
|
|
self.assertEqual({row["id"]: row["commits"] for row in result["models"]},
|
|
{"anthropic/claude-opus/5.5": 1, "openai/codex": 1})
|
|
self.assertNotIn("Private Git Author", json.dumps(result))
|
|
self.assertNotIn("private@example.test", json.dumps(result))
|
|
|
|
def test_supported_model_formats_and_unknown_providers(self):
|
|
"""Keep the documented model families aligned with accepted explicit trailers."""
|
|
accepted = {
|
|
"Claude Sonnet 5.5 <noreply@anthropic.com>": ("anthropic/claude-sonnet/5.5", "Claude Sonnet", "5.5"),
|
|
"Claude Mythos 5.1 <noreply@anthropic.com>": ("anthropic/claude-mythos/5.1", "Claude Mythos", "5.1"),
|
|
"Claude <noreply@anthropic.com>": ("anthropic/claude", "Claude", ""),
|
|
"Codex <codex@openai.com>": ("openai/codex", "Codex", ""),
|
|
"OpenAI Codex <noreply@openai.com>": ("openai/codex", "Codex", ""),
|
|
"ChatGPT <noreply@openai.com>": ("openai/chatgpt", "ChatGPT", ""),
|
|
"GPT-5.1 <noreply@openai.com>": ("openai/gpt/5.1", "GPT", "5.1"),
|
|
"GPT-6.1 Sol <noreply@openai.com>": ("openai/gpt-sol/6.1", "GPT Sol", "6.1"),
|
|
"DeepSeek V4.1 Flash <noreply@deepseek.com>": ("deepseek/deepseek-flash/4.1", "DeepSeek Flash", "4.1"),
|
|
"GLM-5.3-Flash <noreply@z.ai>": ("zhipu/glm-flash/5.3", "GLM Flash", "5.3"),
|
|
"Qwen 3.8 Flash <noreply@qwen.ai>": ("alibaba/qwen-flash/3.8", "Qwen Flash", "3.8"),
|
|
"Codex (GPT-6) <noreply@openai.com>": ("openai/gpt/6", "GPT", "6"),
|
|
"Claude Opus 5.5 (1M context) <noreply@anthropic.com>": ("anthropic/claude-opus/5.5", "Claude Opus", "5.5"),
|
|
"Cursor Agent <cursoragent@cursor.com>": ("anysphere/cursor", "Cursor", ""),
|
|
"Copilot App <223556219+Copilot@users.noreply.github.com>": ("github/copilot", "Copilot", ""),
|
|
"Devin <1+devin-ai-integration[bot]@users.noreply.github.com>": ("cognition/devin", "Devin", ""),
|
|
}
|
|
for trailer, (expected, title, version) in accepted.items():
|
|
with self.subTest(trailer=trailer):
|
|
identity = credits.model(trailer)
|
|
self.assertEqual((identity["id"], identity["model"], identity["version"]), (expected, title, version))
|
|
for trailer in ("Another Model <noreply@other.example>", "Claude Opus 5.5 <unrelated@example.test>",
|
|
"A Person <someone@anthropic.com>", "Bob <1+bob@users.noreply.github.com>", "not a trailer"):
|
|
with self.subTest(trailer=trailer):
|
|
self.assertIsNone(credits.model(trailer))
|
|
|
|
def test_free_text_git_names_never_become_model_titles(self):
|
|
"""Anyone can commit under an agent email; only known words reach the apps."""
|
|
claude, codex = ("anthropic/claude", "Claude", ""), ("openai/codex", "Codex", "")
|
|
for trailer, expected in {
|
|
"Private Person <noreply@anthropic.com>": claude,
|
|
"Claude Buy This 5.5 <noreply@anthropic.com>": claude,
|
|
"Claude 123456789 <noreply@anthropic.com>": claude,
|
|
"Mallory <codex@openai.com>": codex,
|
|
"GPT Mallory 6 <noreply@openai.com>": codex,
|
|
"Mallory (nothing here) <noreply@openai.com>": codex,
|
|
"Mallory (GPT-6) <noreply@openai.com>": ("openai/gpt/6", "GPT", "6"),
|
|
"Mallory <noreply@deepseek.com>": ("deepseek/deepseek", "DeepSeek", ""),
|
|
}.items():
|
|
with self.subTest(trailer=trailer):
|
|
identity = credits.model(trailer)
|
|
self.assertEqual((identity["id"], identity["model"], identity["version"]), expected)
|
|
|
|
def test_a_person_committing_under_an_agent_email_is_not_published(self):
|
|
"""The commit credits the agent's default model and repeats nothing of the Git name."""
|
|
(self.root / "text.txt").write_text("spoofed\n")
|
|
self.commit("Spoofed author", None, author=("Private Person", "noreply@anthropic.com"))
|
|
(self.root / "text.txt").write_text("spoofed trailer\n")
|
|
self.commit("Spoofed trailer\n\nCo-authored-by: Other Person <noreply@openai.com>", None)
|
|
result = self.collect()
|
|
self.assertEqual({row["id"]: row["commits"] for row in result["models"]}, {"anthropic/claude": 1, "openai/codex": 1})
|
|
self.assertNotIn("Private Person", json.dumps(result))
|
|
self.assertNotIn("Other Person", json.dumps(result))
|
|
|
|
def test_profiles_are_requested_only_for_verified_opt_in_entries(self):
|
|
"""A copied people entry never enables name synchronization."""
|
|
result = self.collect()
|
|
rows = {row["login"]: row for row in result["contributors"]}
|
|
self.assertTrue(rows["alice"]["people_entry"])
|
|
self.assertEqual(rows["alice"]["github_name"], "Alice Public")
|
|
self.assertFalse(credits.verified_people(
|
|
self.root, "owner/app", {"bob": {"sync_github_name": True}}, self.identities, {})["bob"])
|
|
self.assertEqual([call[0] for call in self.api_calls if call[0].startswith("users/")],
|
|
["users/alice"])
|
|
|
|
def test_self_authored_name_replaces_an_unverified_entry(self):
|
|
"""Verify actual name lines after the contributor changes an existing table."""
|
|
self.people.write_text('[people.alice]\nsync_github_name = true\n\n'
|
|
'[people.bob]\ndisplay_name = "Bob chose this"\n')
|
|
self.commit("Choose my display name", "bob")
|
|
rows = {row["login"]: row for row in self.collect()["contributors"]}
|
|
self.assertTrue(rows["bob"]["people_entry"])
|
|
|
|
def test_recorded_consent_requires_repository_write_permission(self):
|
|
"""A consent_source alone cannot authorize an unrelated contributor's names."""
|
|
self.people.write_text('[people.alice]\nsync_github_name = true\n\n'
|
|
'[people.bob]\ndisplay_name = "Requested name"\n'
|
|
'consent_source = "https://github.com/owner/app/issues/1#issuecomment-2"\n')
|
|
self.commit("Record requested names", "alice")
|
|
for permission, expected in (("read", False), ("write", True), ("admin", True)):
|
|
with self.subTest(permission=permission):
|
|
self.permissions["alice"] = permission
|
|
with patch.object(credits, "api", side_effect=self.api):
|
|
verified = credits.verified_people(
|
|
self.root, "owner/app", {"bob": {"consent_source": "recorded request"}},
|
|
self.identities, {})
|
|
self.assertEqual(verified["bob"], expected)
|
|
|
|
def test_only_merged_pulls_reachable_from_head_count(self):
|
|
"""Exclude future/other-branch merges and deduplicate PR records."""
|
|
sha = self.git("rev-parse", "HEAD")
|
|
self.pulls = [dict(number=1, sha=sha, login="alice"),
|
|
dict(number=1, sha=sha, login="alice"),
|
|
dict(number=2, sha="f" * 40, login="outsider")]
|
|
result = self.collect()
|
|
pull_queries = [query for endpoint, query, _ in self.api_calls if "/pulls?" in endpoint]
|
|
self.assertEqual(len(pull_queries), 1)
|
|
self.assertIn("select(.merged_at != null)", pull_queries[0])
|
|
self.assertIn("sha: .merge_commit_sha", pull_queries[0])
|
|
self.assertEqual(len(result["contributors"]), 1)
|
|
self.assertEqual(result["contributors"][0]["prs"], 1)
|
|
|
|
@unittest.skipUnless(shutil.which("jq"), "needs jq, the language of gh's --jq")
|
|
def test_pull_projection_selects_the_merge_commit_not_the_head(self):
|
|
"""A squash merge's head commit is not on the base branch; only the merge commit is."""
|
|
self.collect()
|
|
query = next(query for endpoint, query, _ in self.api_calls if "/pulls?" in endpoint)
|
|
pulls = [dict(number=7, merged_at="2026-11-01T00:00:00Z", merge_commit_sha="a" * 40,
|
|
head=dict(sha="b" * 40), user=dict(login="alice")),
|
|
dict(number=8, merged_at=None, merge_commit_sha="c" * 40,
|
|
head=dict(sha="d" * 40), user=dict(login="bob"))]
|
|
output = subprocess.check_output(["jq", "-c", query], input=json.dumps(pulls), text=True)
|
|
self.assertEqual([json.loads(line) for line in output.splitlines()],
|
|
[dict(number=7, sha="a" * 40, login="alice")])
|
|
|
|
def test_unknown_identity_is_not_guessed_from_git_author(self):
|
|
"""Keep unmapped commits visible as unattributed statistics."""
|
|
(self.root / "text.txt").write_text("unmapped\n")
|
|
self.commit("No GitHub identity", None)
|
|
result = self.collect()
|
|
self.assertEqual(result["unattributed_commits"], 1)
|
|
self.assertEqual(result["contributors"][0]["commits"], 1)
|
|
|
|
def test_missing_identity_records_fail_instead_of_publishing_partial_data(self):
|
|
"""Require the complete API history, including explicitly unmapped identities."""
|
|
self.identities.clear()
|
|
with self.assertRaisesRegex(ValueError, "missing 1 reachable commits"):
|
|
self.collect()
|
|
|
|
def test_uncommitted_preferences_fail_clearly_before_api_reads(self):
|
|
"""Explain how to resolve dirty preferences without logging their private values."""
|
|
original = self.output.read_bytes()
|
|
self.people.write_text('[people.alice]\ndisplay_name = "Private local preference"\n')
|
|
with self.assertRaisesRegex(ValueError, "commit or restore") as error:
|
|
self.collect()
|
|
self.assertNotIn("Private local preference", str(error.exception))
|
|
self.assertEqual(self.api_calls, [])
|
|
self.assertEqual(self.output.read_bytes(), original)
|
|
|
|
def test_missing_or_untracked_preferences_fail_before_api_access(self):
|
|
"""An absent committed preference file must not trigger collection or leak local names."""
|
|
self.git("rm", "contributors/people.toml")
|
|
self.git("commit", "-m", "Remove preference file")
|
|
original = self.output.read_bytes()
|
|
for untracked in (False, True):
|
|
with self.subTest(untracked=untracked):
|
|
if untracked:
|
|
self.people.write_text('[people.alice]\ndisplay_name = "Private untracked name"\n')
|
|
with patch.object(credits, "commit_authors") as mocked_history, \
|
|
patch.object(credits, "api") as mocked_api:
|
|
with self.assertRaisesRegex(ValueError, "commit or restore") as error:
|
|
credits.collect(self.root, "owner/app")
|
|
self.assertNotIn("Private untracked name", str(error.exception))
|
|
mocked_history.assert_not_called()
|
|
mocked_api.assert_not_called()
|
|
self.assertEqual(self.output.read_bytes(), original)
|
|
|
|
def test_atomic_write_retains_previous_snapshot_on_replace_failure(self):
|
|
"""Preserve the last good file when disk replacement fails."""
|
|
original = self.output.read_bytes()
|
|
with patch.object(credits.os, "replace", side_effect=OSError("disk full")):
|
|
with self.assertRaisesRegex(OSError, "disk full"):
|
|
credits.write_snapshot(self.output, {"new": True})
|
|
self.assertEqual(self.output.read_bytes(), original)
|
|
self.assertEqual(list(self.root.glob("contributors/.contributors-*")), [])
|
|
|
|
def test_cli_writes_deterministic_snapshot_and_preserves_it_on_api_failure(self):
|
|
"""Run the real CLI twice, then fail gh before a replacement can be written."""
|
|
tools = self.root / "tools"
|
|
tools.mkdir()
|
|
records = self.root / "api.json"
|
|
records.write_text(json.dumps(dict(
|
|
commits=[dict(sha=sha, login=login) for sha, login in self.identities.items()])))
|
|
gh = tools / "gh"
|
|
gh.write_text(f"#!{sys.executable}\n" + '''import json, os, sys
|
|
from pathlib import Path
|
|
if os.environ.get("FAIL_GH") == "yes":
|
|
print("fixture API unavailable", file=sys.stderr)
|
|
raise SystemExit(23)
|
|
endpoint = sys.argv[2]
|
|
if endpoint == "graphql":
|
|
commits = json.loads(Path(os.environ["API_FIXTURE"]).read_text())["commits"]
|
|
nodes = [dict(oid=row["sha"], author=dict(user=dict(login=row["login"])),
|
|
signature=dict(state="VALID", signer=dict(login=row["login"])),
|
|
authors=dict(nodes=[dict(user=dict(login=row["login"]))],
|
|
pageInfo=dict(hasNextPage=False))) for row in commits]
|
|
print(json.dumps(dict(data=dict(repository=dict(object=dict(history=dict(
|
|
nodes=nodes, pageInfo=dict(hasNextPage=False, endCursor=None))))))))
|
|
raise SystemExit(0)
|
|
elif "/pulls?" in endpoint:
|
|
rows = []
|
|
elif endpoint == "users/alice":
|
|
rows = [dict(value="Alice Public")]
|
|
else:
|
|
raise SystemExit("Unexpected endpoint: " + endpoint)
|
|
for row in rows:
|
|
print(json.dumps(row))
|
|
''')
|
|
gh.chmod(0o755)
|
|
env = dict(os.environ, PATH=str(tools) + os.pathsep + os.environ["PATH"],
|
|
API_FIXTURE=str(records))
|
|
self.assertFalse((self.root / "scripts/refresh-contributors.py").exists())
|
|
command = [sys.executable, str(SCRIPT), "--repo", "owner/app", "--directory", str(self.root)]
|
|
first = subprocess.run(command, env=env, capture_output=True, text=True, timeout=30)
|
|
self.assertEqual(first.returncode, 0, first.stderr)
|
|
snapshot = self.output.read_bytes()
|
|
self.assertEqual(json.loads(snapshot)["head"], self.git("rev-parse", "HEAD"))
|
|
again = subprocess.run(command, env=env, capture_output=True, text=True, timeout=30)
|
|
self.assertEqual(again.returncode, 0, again.stderr)
|
|
self.assertEqual(self.output.read_bytes(), snapshot)
|
|
failed = subprocess.run(command, env={**env, "FAIL_GH": "yes"},
|
|
capture_output=True, text=True, timeout=30)
|
|
self.assertEqual(failed.returncode, 1)
|
|
self.assertIn("fixture API unavailable", failed.stderr)
|
|
self.assertEqual(self.output.read_bytes(), snapshot)
|
|
|
|
def test_shallow_checkout_is_rejected_before_api_access(self):
|
|
"""Require full history even when a truncated checkout can resolve HEAD."""
|
|
clone = self.root / "shallow"
|
|
self.git("clone", "--depth=1", "--no-local", self.root.as_uri(), str(clone))
|
|
with patch.object(credits, "api") as mocked_api:
|
|
with self.assertRaisesRegex(ValueError, "full Git history"):
|
|
credits.collect(clone, "owner/app")
|
|
mocked_api.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|