190 lines
11 KiB
Python
190 lines
11 KiB
Python
import json
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from harness.lib.managed_lock import ManagedLockSet
|
|
from harness.lib import runtime_root as runtime
|
|
|
|
|
|
class RuntimeRootTests(unittest.TestCase):
|
|
def cache(self, root):
|
|
repo = root / 'repositories' / 'owner' / 'repo'
|
|
for p in [root / 'repositories', root / 'repositories' / 'owner', repo, repo / 'locks']:
|
|
p.mkdir(mode=0o700)
|
|
return repo
|
|
|
|
def prepare(self, root, run='workflow-1'):
|
|
with ManagedLockSet.for_issue(root, 'owner', 'repo', 17) as locks:
|
|
with locks.repository_scope():
|
|
return runtime.prepare_issue_runtime(root, 'owner', 'repo', 17, run, lock_set=locks)
|
|
|
|
def test_new_runtime_is_private_and_preserves_existing_cache(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); repo = self.cache(root)
|
|
os.chmod(root / 'repositories', 0o755)
|
|
before = repo.stat()
|
|
result = self.prepare(root)
|
|
try:
|
|
self.assertEqual(result.path, runtime.issue_runtime_path(root, 'owner', 'repo', 17, 'workflow-1'))
|
|
self.assertEqual(stat.S_IMODE(os.fstat(result.fd).st_mode), 0o700)
|
|
self.assertEqual(os.fstat(result.fd).st_ino, result.path.stat().st_ino)
|
|
finally:
|
|
os.close(result.fd)
|
|
self.assertEqual(repo.stat().st_ino, before.st_ino)
|
|
self.assertEqual(stat.S_IMODE((root / 'repositories').stat().st_mode), 0o755)
|
|
self.assertFalse((result.path.parents[1] / 'state.json').exists())
|
|
|
|
def test_missing_or_wrong_lock_set_cannot_create_directories(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); repo = self.cache(root)
|
|
locks = ManagedLockSet.for_issue(root, 'owner', 'repo', 17)
|
|
with self.assertRaises(ValueError):
|
|
runtime.prepare_issue_runtime(root, 'owner', 'repo', 17, 'run', lock_set=locks)
|
|
self.assertFalse((repo / 'worktrees').exists())
|
|
with locks:
|
|
with self.assertRaises(ValueError):
|
|
runtime.prepare_issue_runtime(root, 'owner', 'repo', 17, 'run', lock_set=locks)
|
|
with locks.repository_scope():
|
|
with self.assertRaises(ValueError):
|
|
runtime.prepare_issue_runtime(root, 'owner', 'repo', 18, 'run', lock_set=locks)
|
|
self.assertFalse((repo / 'worktrees').exists())
|
|
|
|
def test_kernel_unlocked_wrappers_cannot_authorize_runtime_creation(self):
|
|
import fcntl
|
|
for resource in ('issue', 'repository'):
|
|
with self.subTest(resource=resource), tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); repo = self.cache(root)
|
|
with ManagedLockSet.for_issue(root, 'owner', 'repo', 17) as locks:
|
|
with locks.repository_scope():
|
|
fcntl.flock(getattr(locks, resource).fd, fcntl.LOCK_UN)
|
|
self.assertTrue(getattr(locks, resource).held)
|
|
with self.assertRaises(ValueError):
|
|
runtime.prepare_issue_runtime(root, 'owner', 'repo', 17, 'run', lock_set=locks)
|
|
self.assertFalse((repo / 'worktrees').exists())
|
|
|
|
def test_run_collision_preserves_competing_run_and_parent_chain(self):
|
|
from unittest.mock import patch
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); self.cache(root)
|
|
path = runtime.issue_runtime_path(root, 'owner', 'repo', 17, 'collision')
|
|
mkdir = os.mkdir
|
|
def competing_run(name, mode=0o777, *, dir_fd=None):
|
|
mkdir(name, mode, dir_fd=dir_fd)
|
|
if name == 'collision':
|
|
raise FileExistsError('competing allocator created run')
|
|
with patch.object(runtime.os, 'mkdir', side_effect=competing_run):
|
|
with self.assertRaises(ValueError): self.prepare(root, 'collision')
|
|
self.assertTrue(path.is_dir())
|
|
self.assertTrue(path.parent.is_dir())
|
|
self.assertTrue(path.parent.parent.is_dir())
|
|
self.assertEqual(list(path.iterdir()), [])
|
|
|
|
def test_stale_private_run_is_rejected_without_reuse(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); self.cache(root)
|
|
result = self.prepare(root); os.close(result.fd)
|
|
marker = result.path / 'marker'; marker.write_text('preserve')
|
|
before = result.path.stat()
|
|
with self.assertRaises(ValueError): self.prepare(root)
|
|
self.assertEqual(result.path.stat().st_ino, before.st_ino)
|
|
self.assertEqual(marker.read_text(), 'preserve')
|
|
|
|
def test_unsafe_parent_is_not_repaired(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); repo = self.cache(root)
|
|
p = repo / 'worktrees'; p.mkdir(mode=0o755)
|
|
with self.assertRaises(ValueError): self.prepare(root)
|
|
self.assertEqual(stat.S_IMODE(p.stat().st_mode), 0o755)
|
|
self.assertEqual(list(p.iterdir()), [])
|
|
|
|
def test_symlink_parent_cannot_redirect_creation(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); repo = self.cache(root)
|
|
other = root / 'elsewhere'; other.mkdir(mode=0o700)
|
|
(repo / 'worktrees').symlink_to(other)
|
|
with self.assertRaises((ValueError, OSError)): self.prepare(root)
|
|
self.assertEqual(list(other.iterdir()), [])
|
|
|
|
def test_invalid_identifiers_and_literal_traversal_are_rejected(self):
|
|
for root, owner, repository, issue, run in [
|
|
('/tmp/../root', 'owner', 'repo', 17, 'run'),
|
|
('/tmp/root', '../owner', 'repo', 17, 'run'),
|
|
('/tmp/root', 'owner', 'repo', True, 'run'),
|
|
('/tmp/root', 'owner', 'repo', 17, '..'),
|
|
('/tmp/root', 'owner', 'repo', 17, 'x' * 129),
|
|
]:
|
|
with self.subTest(root=root, owner=owner, issue=issue, run=run), self.assertRaises(ValueError):
|
|
runtime.issue_runtime_path(root, owner, repository, issue, run)
|
|
self.assertIn('o' * 255, str(runtime.issue_runtime_path('/tmp/root', 'o' * 255, 'repo', 17, 'run')))
|
|
|
|
def git(self, *args):
|
|
result = subprocess.run(['git', *map(str, args)], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
return result.stdout
|
|
|
|
def test_runtime_only_container_supports_real_new_git_checkout_and_removal(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); repo = self.cache(root)
|
|
seed = root / 'seed'
|
|
self.git('init', seed)
|
|
(seed / 'file').write_text('fixture')
|
|
self.git('-C', seed, 'add', 'file')
|
|
self.git('-C', seed, '-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid', 'commit', '-m', 'fixture')
|
|
self.git('clone', '--bare', seed, repo / 'repository.git')
|
|
source = self.git('-C', seed, 'rev-parse', 'HEAD').strip()
|
|
run = self.prepare(root); os.close(run.fd)
|
|
issue = repo / 'worktrees' / 'issue-17'
|
|
self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'runtime-only')
|
|
self.git('--git-dir', repo / 'repository.git', 'worktree', 'add', '--detach', issue / 'checkout', source)
|
|
self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'invalid')
|
|
state = issue / 'state.json'; state.write_text(json.dumps({'schema': 'gitea-agent.worktree/v1'})); state.chmod(0o600)
|
|
self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'checkout-candidate')
|
|
self.git('--git-dir', repo / 'repository.git', 'worktree', 'remove', issue / 'checkout')
|
|
state.unlink()
|
|
self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'runtime-only')
|
|
self.assertTrue(run.path.exists())
|
|
(issue / 'unknown').write_text('preserve')
|
|
self.assertEqual(runtime.inspect_issue_worktree_admission(root, 'owner', 'repo', 17), 'invalid')
|
|
|
|
def test_inspection_cli_requires_live_repository_lease_and_uses_real_git(self):
|
|
import sys
|
|
from harness.lib import managed_lock_broker as broker
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp); repo = self.cache(root)
|
|
self.git('init', '--bare', repo / 'repository.git')
|
|
base = Path(__file__).resolve().parents[2]
|
|
cmd = [sys.executable, '-B', str(base / 'lib' / 'managed_lock_broker.py'), 'serve',
|
|
'--root', str(root), '--owner', 'owner', '--repository', 'repo', '--issue', '17',
|
|
'--role', 'gitea-impl-agent', '--run-id', 'cli-workflow', '--ttl-seconds', '30']
|
|
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
try:
|
|
line = process.stdout.readline()
|
|
self.assertTrue(line, 'broker failed before ready')
|
|
ready = json.loads(line)
|
|
inspect = [sys.executable, '-B', str(base / 'lib' / 'runtime_root.py'), 'inspect-worktree',
|
|
'--root', str(root), '--owner', 'owner', '--repository', 'repo', '--issue', '17',
|
|
'--handle', ready['handle']]
|
|
result = subprocess.run(inspect, capture_output=True, text=True, timeout=10)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
broker.request_handle(ready['handle'], 'acquire_repository')
|
|
result = subprocess.run(inspect, capture_output=True, text=True, timeout=10)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(json.loads(result.stdout)['container'], 'runtime-only')
|
|
broker.request_handle(ready['handle'], 'release_repository')
|
|
broker.request_handle(ready['handle'], 'release')
|
|
self.assertEqual(process.wait(timeout=10), 0)
|
|
finally:
|
|
if process.poll() is None:
|
|
process.terminate(); process.wait(timeout=10)
|
|
process.stdout.close(); process.stderr.close()
|
|
|
|
def test_inspection_rejects_missing_cache_instead_of_creating_it(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
with self.assertRaises((ValueError, OSError)):
|
|
runtime.inspect_issue_worktree_admission(Path(tmp), 'owner', 'repo', 17)
|
|
self.assertEqual(list(Path(tmp).iterdir()), [])
|