|
|
@@ -0,0 +1,161 @@
|
|
|
+"""Exercise production-line attribution against isolated real Git histories."""
|
|
|
+
|
|
|
+import importlib.util
|
|
|
+import json
|
|
|
+import os
|
|
|
+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.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'):
|
|
|
+ self.assertFalse(blame.production_path(path), path)
|
|
|
+ self.assertTrue(blame.production_path('packages/a/src/a.ts'))
|
|
|
+ self.assertTrue(blame.production_path('python/sdk-runtime/runtime-bootstrap.mjs'))
|
|
|
+ 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')
|
|
|
+ base = self.commit()
|
|
|
+ self.write('packages/a/src/a.ts', 'throw new Error("PR code must not run")\n')
|
|
|
+ head = self.commit()
|
|
|
+ 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', base], 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', head], check=True)
|
|
|
+ checkout = self.root / 'checkout'
|
|
|
+ self.git('clone', '--depth=1', remote.as_uri(), str(checkout))
|
|
|
+ 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)}}},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(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(), base)
|
|
|
+ config = (checkout / '.git' / 'config').read_text()
|
|
|
+ self.assertNotIn('fixture-token', config)
|
|
|
+ self.assertNotIn('AUTHORIZATION', config)
|
|
|
+
|
|
|
+ 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()
|
turtle1999