206 lines
15 KiB
Python
206 lines
15 KiB
Python
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()
|