from harness.tests._support.paths import HARNESS as SOURCE_HARNESS 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 = SOURCE_HARNESS 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()), [])