"""Synthetic local Git pre-fetch gate checks; no provider/network operations.""" import json import os import subprocess import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[3] SCRIPT = ROOT / 'harness/lib/repository_prefetch.py' class RepositoryPrefetchTests(unittest.TestCase): def git(self, *args): result = subprocess.run(['/usr/bin/git', '-c', 'core.hooksPath=/dev/null', *map(str, args)], env=self.env, capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) return result.stdout def setUp(self): self.temp = tempfile.TemporaryDirectory(dir=os.environ.get('TMPDIR', '/opt/data/cache/scratch')) self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.env = {'PATH': '/usr/bin:/bin', 'HOME': str(self.root), 'GIT_CONFIG_NOSYSTEM': '1', 'GIT_CONFIG_GLOBAL': '/dev/null', 'GIT_TERMINAL_PROMPT': '0'} self.seed = self.root / 'seed' self.bare = self.root / 'mirror.git' self.git('init', '-q', self.seed) self.git('-C', self.seed, 'config', 'user.name', 'Synthetic') self.git('-C', self.seed, 'config', 'user.email', 'fixture@example.invalid') (self.seed / 'a').write_text('synthetic') self.git('-C', self.seed, 'add', 'a') self.git('-C', self.seed, 'commit', '-qm', 'synthetic') self.git('clone', '--mirror', self.seed, self.bare) def gate(self): self.assertTrue(SCRIPT.is_file(), 'canonical read-only prefetch helper is missing') result = subprocess.run(['python3', '-B', str(SCRIPT), '--git-dir', str(self.bare)], env=self.env, capture_output=True, text=True, timeout=15) return result.returncode, json.loads(result.stdout) def test_bare_only_passes_real_git_read_only(self): before = self.git('--git-dir', self.bare, 'show-ref') code, value = self.gate() self.assertEqual(code, 0) self.assertEqual(value, {'status': 'pass', 'attached_worktrees': [], 'reason': 'no_attached_worktrees'}) self.assertEqual(self.git('--git-dir', self.bare, 'show-ref'), before) def test_attached_blocks_exact_branch_and_path_without_mutation(self): checkout = self.root / 'attached space\npath' self.git('--git-dir', self.bare, 'worktree', 'add', '-b', 'issue-fixture', checkout, 'HEAD') self.git('--git-dir', self.bare, 'worktree', 'lock', checkout) state = checkout.parent / 'state.json' state.write_text('sentinel') before = self.fingerprint(self.bare, checkout, state) code, value = self.gate() self.assertEqual(code, 1) self.assertEqual(value, {'status': 'block', 'attached_worktrees': [ {'branch': 'refs/heads/issue-fixture', 'path': str(checkout)}], 'reason': 'attached_worktree'}) self.assertEqual(self.fingerprint(self.bare, checkout, state), before) def test_detached_locked_worktree_passes_without_mutation(self): checkout = self.root / 'detached' self.git('--git-dir', self.bare, 'worktree', 'add', '--detach', '--lock', checkout, 'HEAD') before = self.fingerprint(self.bare, checkout) code, value = self.gate() self.assertEqual(code, 0) self.assertEqual(value['attached_worktrees'], []) self.assertEqual(self.fingerprint(self.bare, checkout), before) def test_mixed_worktrees_report_all_attached_without_branch_name_policy(self): detached = self.root / 'detached' self.git('--git-dir', self.bare, 'worktree', 'add', '--detach', detached, 'HEAD') expected = [] for number, branch in enumerate(['feature/normal', 'issue-fixture']): checkout = self.root / ('attached-' + str(number)) self.git('--git-dir', self.bare, 'worktree', 'add', '-b', branch, checkout, 'HEAD') expected.append({'branch': 'refs/heads/' + branch, 'path': str(checkout)}) code, value = self.gate() self.assertEqual(code, 1) self.assertEqual(sorted(value['attached_worktrees'], key=lambda x: x['path']), expected) def test_unavailable_nonbare_and_symlink_input_fail_closed(self): self.assertEqual(self.gate()[0], 0) original = self.bare link = self.root / 'alias' link.symlink_to(original, target_is_directory=True) for target in [self.root / 'absent', self.seed / '.git', link]: with self.subTest(target=str(target)): self.bare = target code, value = self.gate() self.assertEqual(code, 2) self.assertEqual(value['status'], 'block') self.bare = original # Raw Git stderr must not become diagnostic evidence. (original / 'config').write_text('secret-like fixture invalid config [') result = subprocess.run(['python3', '-B', str(SCRIPT), '--git-dir', str(original)], env=self.env, capture_output=True, text=True, timeout=15) self.assertEqual(result.returncode, 2) self.assertNotIn('secret-like', result.stdout + result.stderr) def test_malformed_nul_listing_is_rejected_not_treated_as_empty(self): from harness.lib import repository_prefetch as module raw = self.git('--git-dir', self.bare, 'worktree', 'list', '--porcelain', '-z').encode() self.assertEqual(module.parse_worktrees(raw)[0]['kind'], 'bare') for bad in [b'', raw[:-1], raw + raw, raw.replace(b'bare', b'unknown'), b'worktree /x\0HEAD ' + b'a' * 40 + b'\0detached\0\0', raw + b'worktree /x\0HEAD ' + b'a' * 40 + b'\0branch refs/heads/../bad\0\0']: with self.subTest(raw=bad): with self.assertRaises(module.PrefetchError): module.parse_worktrees(bad) def test_reader_failure_and_malformed_output_fail_closed_after_valid_control(self): from harness.lib import repository_prefetch as module from unittest.mock import patch self.assertEqual(module.inspect_repository(str(self.bare))['status'], 'pass') bare = subprocess.CompletedProcess([], 0, b'true\n') bad = subprocess.CompletedProcess([], 0, b'invalid secret-like fixture') for responses in ([bare, bad], [bare, None]): with self.subTest(responses=str(responses)): with patch.object(module.binding, '_git', side_effect=responses): with self.assertRaises(module.PrefetchError) as error: module.inspect_repository(str(self.bare)) self.assertNotIn('secret-like', str(error.exception)) with patch.object(module.binding, '_git', side_effect=module.binding.BindingError('secret-like fixture')): with self.assertRaisesRegex(module.PrefetchError, '^git_read_unavailable$'): module.inspect_repository(str(self.bare)) def test_documentation_contract_requires_gate_before_fetch(self): text = (ROOT / 'harness/operations/repository-cache.md').read_text() section = text.split('## 기존 cache 갱신', 1)[1].split('\n## ', 1)[0] self.assertIn('repository_prefetch.py', section) self.assertLess(section.index('repository_prefetch.py'), section.index('prune/fetch')) for rule in ['--git-dir', 'repository lock', '자동 repair', 'exit 0', 'exit 1', 'exit 2']: self.assertIn(rule, section) def fingerprint(self, *roots): import hashlib result = {} for root in roots: for path in ([root] if root.is_file() else root.rglob('*')): if path.is_file(): st = path.stat() result[str(path)] = (hashlib.sha256(path.read_bytes()).hexdigest(), st.st_mode, st.st_mtime_ns, st.st_ino) return result if __name__ == '__main__': unittest.main()