|
|
@@ -0,0 +1,221 @@
|
|
|
+"""Exercise production-line attribution against isolated real Git histories."""
|
|
|
+
|
|
|
+import base64
|
|
|
+import importlib.util
|
|
|
+import json
|
|
|
+import os
|
|
|
+import shutil
|
|
|
+import sys
|
|
|
+from pathlib import Path
|
|
|
+import subprocess
|
|
|
+import tempfile
|
|
|
+import unittest
|
|
|
+
|
|
|
+spec = importlib.util.spec_from_file_location('blame_production', Path(__file__).with_name('blame-production.py'))
|
|
|
+blame = importlib.util.module_from_spec(spec)
|
|
|
+spec.loader.exec_module(blame)
|
|
|
+
|
|
|
+
|
|
|
+class ProductionBlameTest(unittest.TestCase):
|
|
|
+ def setUp(self):
|
|
|
+ self.directory = tempfile.TemporaryDirectory(prefix='approval-blame-')
|
|
|
+ self.addCleanup(self.directory.cleanup)
|
|
|
+ self.root = Path(self.directory.name)
|
|
|
+ self.git('init', '-q')
|
|
|
+ self.git('config', 'user.name', 'Approval Fixture')
|
|
|
+ self.git('config', 'user.email', 'fixture@example.invalid')
|
|
|
+ self.git('config', 'commit.gpgsign', 'false')
|
|
|
+ self.git('config', 'core.hooksPath', str(self.root / 'no-hooks'))
|
|
|
+
|
|
|
+ def git(self, *args):
|
|
|
+ return subprocess.run(['git', '-C', str(self.root), *args], check=True,
|
|
|
+ capture_output=True, text=True).stdout.strip()
|
|
|
+
|
|
|
+ def write(self, path, source):
|
|
|
+ file = self.root / path
|
|
|
+ file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ file.write_text(source, encoding='utf-8')
|
|
|
+
|
|
|
+ def commit(self):
|
|
|
+ self.git('add', '.')
|
|
|
+ self.git('commit', '-qm', 'fixture')
|
|
|
+ return self.git('rev-parse', 'HEAD')
|
|
|
+
|
|
|
+ def test_code_tokens_keep_strings_and_mixed_lines(self):
|
|
|
+ cases = [
|
|
|
+ ('main.ts', '// comment\n/* block\n comment */\nconst url = "https://example.test" // tail\n\n', {4}),
|
|
|
+ ('main.html', '<!-- comment -->\n<div>content</div>\n', {2}),
|
|
|
+ ('main.tsx', 'const view = <div>{/* comment */}hello</div>\n', {1}),
|
|
|
+ ('main.css', '/* block\n comment */\n.x { color: red; /* mixed */ }\n', {3}),
|
|
|
+ ('main.py', '"""module docs\nmore docs"""\n# comment\nx = "# code"\n', {4}),
|
|
|
+ ('main.c', '#include <stdio.h>\n#define VALUE 1\n', {1, 2}),
|
|
|
+ ('main.c', '/* comment */\nint main() { return 0; } // tail\n', {2}),
|
|
|
+ ('main.ts', 'const text = `first\n// string content\nlast`\n', {1, 2, 3}),
|
|
|
+ ]
|
|
|
+ for path, source, expected in cases:
|
|
|
+ with self.subTest(path=path, source=source):
|
|
|
+ self.assertEqual(blame.code_lines(path, source), expected)
|
|
|
+
|
|
|
+ def test_excludes_nonproduction_paths_and_generated_headers(self):
|
|
|
+ for path in ('docs/a.ts', 'vendor/src/a.ts', 'packages/a/tests/a.ts',
|
|
|
+ 'packages/a/src/a.spec.ts', 'packages/support/a/src/a.ts',
|
|
|
+ 'packages/a/src/generated/a.ts', 'packages/a/src/a.d.ts',
|
|
|
+ 'packages/a/src/a.md', 'scripts/a.ts',
|
|
|
+ 'packages/core/tools/src/testing.ts',
|
|
|
+ 'packages/session/session-persistence-jsonl/src/testing/generation.ts'):
|
|
|
+ self.assertFalse(blame.production_path(path), path)
|
|
|
+ self.assertTrue(blame.production_path('packages/a/src/a.ts'))
|
|
|
+ for path in ('python/sdk-runtime/runtime-bootstrap.mjs', 'apps/desktop/renderer/startup.js',
|
|
|
+ 'apps/desktop/renderer/startup.html', 'packages/experimental/code-runtime-python/py/protocol.py',
|
|
|
+ 'packages/experimental/webworker-packer/bin.js'):
|
|
|
+ self.assertTrue(blame.production_path(path), path)
|
|
|
+ self.assertEqual(blame.code_lines('a.ts', '// Generated by schema\nconst a = 1\n'), set())
|
|
|
+
|
|
|
+ def test_attributes_only_changed_old_code_at_merge_base(self):
|
|
|
+ path = 'packages/a/src/a.ts'
|
|
|
+ self.write(path, '// first\nconst a = 1\nconst b = 2\n')
|
|
|
+ first = self.commit()
|
|
|
+ self.write(path, '// first\nconst a = 1\nconst b = 3\n')
|
|
|
+ base = self.commit()
|
|
|
+ self.write(path, '// different\nconst a = 4\nconst b = 5\nconst added = 6\n')
|
|
|
+ self.write('packages/a/src/new.ts', 'const entirelyNew = 1\n')
|
|
|
+ self.write('packages/a/tests/test.ts', 'const test = 1\n')
|
|
|
+ head = self.commit()
|
|
|
+ # Base advancement must not replace the common ancestor used for attribution.
|
|
|
+ self.git('checkout', '--detach', base)
|
|
|
+ self.write('packages/a/src/unrelated.ts', 'const unrelated = 1\n')
|
|
|
+ advanced_base = self.commit()
|
|
|
+ result = blame.measure(str(self.root), advanced_base, head)
|
|
|
+ self.assertEqual(result['mergeBase'], base)
|
|
|
+ self.assertEqual(result['totalLines'], 2)
|
|
|
+ self.assertEqual(result['commitLines'], {first: 1, base: 1})
|
|
|
+
|
|
|
+ def test_renames_deletions_and_unusual_paths(self):
|
|
|
+ path = 'packages/a/src/quoted " name.ts'
|
|
|
+ self.write(path, '// header\nconst a = 1\nconst b = 2\nconst c = 3\n')
|
|
|
+ self.write('packages/a/src/deleted.ts', 'const gone = 1\n// comment\n')
|
|
|
+ base = self.commit()
|
|
|
+ renamed = 'packages/a/src/renamed.ts'
|
|
|
+ self.git('mv', path, renamed)
|
|
|
+ self.write(renamed, '// header\nconst a = 1\nconst b = 2\nconst c = 4\n')
|
|
|
+ self.git('rm', 'packages/a/src/deleted.ts')
|
|
|
+ result = blame.measure(str(self.root), base, self.commit())
|
|
|
+ self.assertEqual(result['totalLines'], 2)
|
|
|
+ self.assertEqual(result['commitLines'], {base: 2})
|
|
|
+
|
|
|
+ def test_pure_additions_and_comment_only_changes_have_zero_denominator(self):
|
|
|
+ self.write('packages/a/src/a.ts', '// comment\nconst a = 1\n')
|
|
|
+ base = self.commit()
|
|
|
+ self.write('packages/a/src/a.ts', '// revised\nconst a = 1\nconst newLine = 2\n')
|
|
|
+ result = blame.measure(str(self.root), base, self.commit())
|
|
|
+ self.assertEqual(result['totalLines'], 0)
|
|
|
+ self.assertEqual(result['commitLines'], {})
|
|
|
+
|
|
|
+ def test_binary_marked_replacement_does_not_hide_old_production_lines(self):
|
|
|
+ path = 'packages/a/src/a.ts'
|
|
|
+ self.write(path, 'const old = 1\n')
|
|
|
+ base = self.commit()
|
|
|
+ self.write(path, '\0binary replacement\n')
|
|
|
+ self.assertEqual(blame.measure(str(self.root), base, self.commit())['totalLines'], 1)
|
|
|
+
|
|
|
+ def test_pure_rename_has_zero_denominator(self):
|
|
|
+ self.write('packages/a/src/a.ts', 'const a = 1\n')
|
|
|
+ base = self.commit()
|
|
|
+ self.git('mv', 'packages/a/src/a.ts', 'packages/a/src/b.ts')
|
|
|
+ self.assertEqual(blame.measure(str(self.root), base, self.commit())['totalLines'], 0)
|
|
|
+
|
|
|
+ def test_publisher_fetches_history_without_checking_out_pr_code(self):
|
|
|
+ self.write('packages/a/src/a.ts', 'const old = 1\n')
|
|
|
+ self.write('packages/a/src/base.ts', 'const base = 1\n')
|
|
|
+ base = self.commit()
|
|
|
+ self.write('packages/a/src/a.ts', 'throw new Error("PR code must not run")\n')
|
|
|
+ branch = self.commit()
|
|
|
+ self.git('checkout', '--detach', base)
|
|
|
+ self.write('packages/a/src/base.ts', 'const base = 2\n')
|
|
|
+ advanced = self.commit()
|
|
|
+ self.git('checkout', '--detach', branch)
|
|
|
+ self.git('merge', '--no-edit', advanced)
|
|
|
+ head = self.git('rev-parse', 'HEAD')
|
|
|
+ remote = self.root / 'remote' / 'owner' / 'repo.git'
|
|
|
+ remote.parent.mkdir(parents=True)
|
|
|
+ self.git('clone', '--bare', str(self.root), str(remote))
|
|
|
+ subprocess.run(['git', '-C', str(remote), 'update-ref', 'refs/heads/trusted', advanced], check=True)
|
|
|
+ subprocess.run(['git', '-C', str(remote), 'symbolic-ref', 'HEAD', 'refs/heads/trusted'], check=True)
|
|
|
+ subprocess.run(['git', '-C', str(remote), 'update-ref', 'refs/pull/42/head', advanced], check=True)
|
|
|
+ checkout = self.root / 'checkout'
|
|
|
+ self.git('clone', '--depth=1', remote.as_uri(), str(checkout))
|
|
|
+ wrappers = self.root / 'wrappers'
|
|
|
+ wrappers.mkdir()
|
|
|
+ trace = self.root / 'git-environment.json'
|
|
|
+ git_wrapper = wrappers / 'git'
|
|
|
+ git_wrapper.write_text(f"#!{sys.executable}\nimport json, os, sys\n"
|
|
|
+ f"if 'fetch' in sys.argv: open({str(trace)!r}, 'w').write(json.dumps({{key: value for key, value in os.environ.items() if key.startswith('GIT_CONFIG_')}}))\n"
|
|
|
+ f"os.execv({shutil.which('git')!r}, ['git', *sys.argv[1:]])\n")
|
|
|
+ git_wrapper.chmod(0o755)
|
|
|
+ module = Path(__file__).with_name('blame-ownership.mjs').resolve().as_uri()
|
|
|
+ program = f"""
|
|
|
+ import {{ productionOwnership }} from {json.dumps(module)};
|
|
|
+ const result = await productionOwnership({{repository:'owner/repo', number:42, headSha:{json.dumps(head)}}},
|
|
|
+ async path => path === '/graphql'
|
|
|
+ ? {{data:{{repository:{{c0:{{author:{{user:{{login:'writer'}}}}}}}}}}}}
|
|
|
+ : {{base:{{sha:{json.dumps(base)},ref:'trusted'}},head:{{sha:{json.dumps(head)}}}}});
|
|
|
+ console.log(JSON.stringify(result));
|
|
|
+ """
|
|
|
+ result = subprocess.run(['node', '--input-type=module', '-e', program], cwd=checkout,
|
|
|
+ check=True, capture_output=True, text=True,
|
|
|
+ env={**os.environ, 'GITHUB_TOKEN': 'fixture-token',
|
|
|
+ 'GITHUB_SERVER_URL': (self.root / 'remote').as_uri(),
|
|
|
+ 'PATH': str(wrappers) + os.pathsep + str(Path(sys.executable).parent) + os.pathsep + os.environ['PATH']})
|
|
|
+ self.assertEqual(json.loads(result.stdout), {'totalLines': 1, 'reviewerLines': {'writer': 1}})
|
|
|
+ self.assertEqual(subprocess.check_output(['git', '-C', str(checkout), 'rev-parse', 'HEAD'], text=True).strip(), advanced)
|
|
|
+ fetch_environment = json.loads(trace.read_text())
|
|
|
+ self.assertEqual(fetch_environment['GIT_CONFIG_COUNT'], '1')
|
|
|
+ self.assertEqual(fetch_environment['GIT_CONFIG_KEY_0'],
|
|
|
+ f"http.{(self.root / 'remote').as_uri()}/.extraheader")
|
|
|
+ self.assertEqual(fetch_environment['GIT_CONFIG_VALUE_0'],
|
|
|
+ 'AUTHORIZATION: basic ' + base64.b64encode(b'x-access-token:fixture-token').decode())
|
|
|
+ config = (checkout / '.git' / 'config').read_text()
|
|
|
+ self.assertNotIn('fixture-token', config)
|
|
|
+ self.assertNotIn('AUTHORIZATION', config)
|
|
|
+
|
|
|
+ def test_merge_forward_excludes_base_only_edits(self):
|
|
|
+ path = 'packages/a/src/a.ts'
|
|
|
+ self.write(path, 'const a = 1\n')
|
|
|
+ self.write('packages/a/src/base.ts', 'const base = 1\n')
|
|
|
+ fork = self.commit()
|
|
|
+ self.write(path, 'const a = 2\n')
|
|
|
+ branch = self.commit()
|
|
|
+ self.git('checkout', '--detach', fork)
|
|
|
+ self.write('packages/a/src/base.ts', 'const base = 2\n')
|
|
|
+ advanced = self.commit()
|
|
|
+ self.git('checkout', '--detach', branch)
|
|
|
+ self.git('merge', '--no-edit', advanced)
|
|
|
+ result = blame.measure(str(self.root), advanced, self.git('rev-parse', 'HEAD'))
|
|
|
+ self.assertEqual(result['mergeBase'], advanced)
|
|
|
+ self.assertEqual(result['commitLines'], {fork: 1})
|
|
|
+
|
|
|
+ def test_non_utf8_blobs_and_commit_metadata_keep_old_lines(self):
|
|
|
+ path = 'packages/a/src/a.ts'
|
|
|
+ self.write(path, 'const a = "old"\n')
|
|
|
+ (self.root / path).write_bytes(b'const a = "\xff"\n')
|
|
|
+ self.git('add', '.')
|
|
|
+ subprocess.run(['git', '-C', str(self.root), '-c', 'i18n.commitEncoding=ISO-8859-1',
|
|
|
+ 'commit', '-q', '-F', '-'], input=b'metadata \xff', check=True, capture_output=True)
|
|
|
+ base = self.git('rev-parse', 'HEAD')
|
|
|
+ (self.root / path).write_bytes(b'\0new \xfe\n')
|
|
|
+ self.assertEqual(blame.measure(str(self.root), base, self.commit())['commitLines'], {base: 1})
|
|
|
+
|
|
|
+ def test_generated_words_in_code_do_not_exclude_handwritten_files(self):
|
|
|
+ self.assertEqual(blame.code_lines('a.ts', 'const text = "generated by an agent"\n'), {1})
|
|
|
+ self.assertEqual(blame.code_lines('a.ts', 'const a = 1\n// do not edit user data\n'), {1})
|
|
|
+
|
|
|
+ def test_rejects_shallow_history(self):
|
|
|
+ self.write('packages/a/src/a.ts', 'const a = 1\n')
|
|
|
+ base = self.commit()
|
|
|
+ (self.root / '.git' / 'shallow').write_text(base + '\n')
|
|
|
+ with self.assertRaisesRegex(ValueError, 'complete history'):
|
|
|
+ blame.measure(str(self.root), base, base)
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ unittest.main()
|
Turtle