import copy import importlib.util import json import os import subprocess import sys import tempfile import time import unittest from pathlib import Path from harness.tests._support.paths import ROOT SCRIPT = ROOT / 'harness/agents/review/skills/gitea-code-review/scripts/native_review.py' def load(path, name): spec = importlib.util.spec_from_file_location(name, path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module from harness.tests._support.review import ( FakeAPI, FakeLease, plan_fixture, publication_fixture, support, ) class TestNativeLease(unittest.TestCase): def module(self): self.assertTrue(SCRIPT.is_file(), 'shared native review adapter is missing') m = load(SCRIPT, 'native_review_test_' + str(id(self))) # Every test gets a private registry root; production has no test-only # root override, so isolate through the journal module seam. private_root = tempfile.TemporaryDirectory() self.addCleanup(private_root.cleanup) root = Path(private_root.name) / 'native-review-journals' m._journal.root_for = lambda anchor, root=root: root return m @staticmethod def canonical(m, plan): return m.canonical_journal_path(plan) def test_held_lease_checks_real_registered_worktree_and_expiry(self): import fcntl m = self.module() self.assertTrue(callable(getattr(m, 'HeldIssueLease', None)), 'real Issue lease guard is missing') with tempfile.TemporaryDirectory() as td: root = Path(td); policy = root / 'harness/operations/repository-cache.md' policy.parent.mkdir(parents=True); policy.write_text('canonical cache fixture') rs = root / 'repositories'; owner = rs / 'acme'; cache = owner / 'repo'; locks = cache / 'locks' for directory in [rs, owner, cache, locks]: directory.mkdir(mode=0o700) bare = cache / 'repository.git'; seed = root / 'seed' def git(*args): out = subprocess.run(['git', '-c', 'core.hooksPath=/dev/null', *map(str, args)], capture_output=True, text=True, env={'PATH': os.environ['PATH'], 'GIT_CONFIG_GLOBAL': '/dev/null', 'GIT_CONFIG_NOSYSTEM': '1'}) self.assertEqual(out.returncode, 0, out.stderr) return out.stdout.strip() git('init', '--bare', bare); git('init', seed) git('-C', seed, 'config', 'user.name', 'Synthetic'); git('-C', seed, 'config', 'user.email', 'fixture@example.test') (seed / 'a.txt').write_text('synthetic') git('-C', seed, 'add', 'a.txt'); git('-C', seed, 'commit', '-m', 'fixture') git('--git-dir=' + str(bare), 'fetch', seed, 'HEAD:refs/heads/main') wt = cache / 'worktrees/issue-3/checkout'; wt.parent.mkdir(parents=True, mode=0o700) git('--git-dir=' + str(bare), 'worktree', 'add', '--detach', '--lock', wt, 'main') head = git('-C', wt, 'rev-parse', 'HEAD') p = plan_fixture(); p['target']['head'] = head; p['body'] = p['body'].replace('a' * 40, head) state = wt.parent / 'state.json' state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': 'gitea-impl-agent', 'mode': 'detached'})) state.chmod(0o600) path = locks / 'issue-3.lock'; fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600) try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) lease = m.HeldIssueLease(fd, policy, time.monotonic() + 30) canonical_bytes = state.read_bytes(); canonical_stat = state.stat() lease.check(p) self.assertEqual(state.read_bytes(), canonical_bytes) self.assertEqual((state.stat().st_mode, state.stat().st_mtime_ns, state.stat().st_ino), (canonical_stat.st_mode, canonical_stat.st_mtime_ns, canonical_stat.st_ino)) # The bounded compatibility admission is read-only: mutate only # this scratch fixture from canonical producer state to legacy impl. state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})) state.chmod(0o600) legacy_bytes = state.read_bytes(); legacy_stat = state.stat() try: lease.check(p) except m.ValidationError as exc: self.fail(f'valid legacy impl registration must be admitted: {exc}') after_stat = state.stat() self.assertEqual(state.read_bytes(), legacy_bytes) self.assertEqual((after_stat.st_mode, after_stat.st_mtime_ns, after_stat.st_ino), (legacy_stat.st_mode, legacy_stat.st_mtime_ns, legacy_stat.st_ino)) for bad_role in [None, True, False, 1, [], {}, '', ' impl', 'impl ', 'Impl', 'IMPL', 'gitea-impl-agent ', 'review', 'gitea-review-agent', 'gitea-spec-agent', 'gitea-release-agent', 'other']: state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': bad_role, 'mode': 'detached'})) state.chmod(0o600) with self.subTest(bad_role=repr(bad_role)), self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) missing_role = json.loads(legacy_bytes); missing_role.pop('created_by_role') state.write_text(json.dumps(missing_role)); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) bool_issue = json.loads(legacy_bytes); bool_issue['issue'] = True state.write_text(json.dumps(bool_issue)); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'other', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'other', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 4, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'wrong/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'attached'})); state.chmod(0o600) with self.assertRaisesRegex(m.ValidationError, 'worktree registration identity mismatch'): lease.check(p) state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'created_by_role': 'impl', 'mode': 'detached'})); state.chmod(0o600) wrong_head = copy.deepcopy(p); wrong_head['target']['head'] = 'b' * 40 wrong_head['body'] = wrong_head['body'].replace(head, 'b' * 40) with self.assertRaisesRegex(m.ValidationError, 'worktree HEAD mismatch'): lease.check(wrong_head) fcntl.flock(fd, fcntl.LOCK_UN) with self.assertRaises(m.ValidationError): lease.check(p) self.assertNotIn('lock:', Path(f'/proc/self/fdinfo/{fd}').read_text()) fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) from unittest.mock import patch fakebin = root / 'fakebin'; fakebin.mkdir() marker = root / 'path-git-executed' fakegit = fakebin / 'git' fakegit.write_text(f'#!/bin/sh\n/usr/bin/touch "{marker}"\nexit 1\n'); fakegit.chmod(0o700) with patch.dict(os.environ, {'PATH': str(fakebin)}): lease.check(p) self.assertFalse(marker.exists()) lease.deadline_monotonic = time.monotonic() - 1 with self.assertRaises(m.ValidationError): lease.check(p) lease.deadline_monotonic = time.monotonic() + 30 (wt / 'a.txt').write_text('dirty') with self.assertRaises(m.ValidationError): lease.check(p) # Every failed source validation must close all transient FDs. before_fds = len(list((Path('/proc/self/fd')).iterdir())) for _ in range(20): with self.assertRaises(m.ValidationError): lease.check(p) after_fds = len(list((Path('/proc/self/fd')).iterdir())) self.assertLessEqual(after_fds, before_fds + 1) (wt / 'a.txt').unlink() for directory in [wt.parent.parent, wt.parent, wt]: directory.chmod(0o770) with self.subTest(directory=directory), self.assertRaises(m.ValidationError): lease.check(p) directory.chmod(0o700) finally: os.close(fd) def test_false_lease_result_does_not_authorize_publication(self): m = self.module(); p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact) lease = FakeLease(); lease.check = lambda plan: False with tempfile.TemporaryDirectory() as td: with self.assertRaises(m.ValidationError): m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'}) self.assertEqual(api.posts, 0) def test_expired_lease_does_not_create_new_journal_directory(self): m = self.module(); p, before, after, exact = publication_fixture() lease = FakeLease(); lease.valid = False with tempfile.TemporaryDirectory() as td: directory = Path(td) / 'not-created' with self.assertRaises(m.ValidationError): m.publish(p, before, FakeAPI(p, before, after, exact), lease, directory) self.assertFalse(directory.exists()) def test_broker_issue_lease_accepts_authenticated_matching_status(self): m = self.module(); p, before, after, exact = publication_fixture() now = time.monotonic() lease = m.BrokerIssueLease('/trusted/handle.json', '/trusted/policy.md', now + 30) handle = {'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': now + 20, 'token': 'not-returned', 'socket': '/trusted/broker.sock'} status = {'status': 'held', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': handle['deadline']} from unittest.mock import patch with patch.object(m._broker, 'read_handle_metadata', return_value={k: v for k, v in handle.items() if k != 'token'}), \ patch.object(m._broker, 'request_handle', return_value=status), \ patch.object(m, '_validate_registered_source') as source: self.assertIsNone(lease.check(p)) source.assert_called_once_with(p, '/trusted/policy.md', lease.deadline_monotonic) def test_broker_issue_lease_rejects_wrong_identity_status_or_deadline(self): m = self.module(); p, before, after, exact = publication_fixture(); now = time.monotonic() lease = m.BrokerIssueLease('/trusted/handle.json', '/trusted/policy.md', now + 30) handle = {'pid': 1234, 'deadline': now + 20, 'token': 'hidden', 'socket': '/trusted/broker.sock'} cases = [ {'status': 'held', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-spec-agent', 'pid': 1234, 'deadline': handle['deadline']}, {'status': 'held', 'owner': 'other', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': handle['deadline']}, {'status': 'released', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': handle['deadline']}, {'status': 'held', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 9999, 'deadline': handle['deadline']}, {'status': 'held', 'owner': 'acme', 'repository': 'repo', 'issue': 3, 'role': 'gitea-review-agent', 'pid': 1234, 'deadline': now + 40}, ] from unittest.mock import patch for status in cases: with self.subTest(status=status): with patch.object(m._broker, 'read_handle_metadata', return_value={k: v for k, v in handle.items() if k != 'token'}), \ patch.object(m._broker, 'request_handle', return_value=status), \ patch.object(m, '_validate_registered_source') as source: with self.assertRaises(m.ValidationError): lease.check(p) source.assert_not_called() if __name__ == '__main__': unittest.main()