import importlib.util import os import subprocess import sys import tempfile import unittest from pathlib import Path SCRIPT = Path(__file__).parents[2] / 'agents/impl/skills/gitea-issue-rework/scripts/rework_comment.py' class ReworkReadbackTransportTests(unittest.TestCase): def module(self): path = SCRIPT.with_name('rework_readback_cli.py') self.assertTrue(path.is_file(), 'canonical live readback bridge is missing') spec = importlib.util.spec_from_file_location('transport_under_test', path) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def test_role_token_requires_private_profile_file(self): module = self.module() with tempfile.TemporaryDirectory() as directory: home = Path(directory) with self.assertRaises(ValueError): module.load_role_token(home) source = home / '.env' source.write_text('GITEA_TOKEN=' + 's' * 48 + '\n') source.chmod(0o644) with self.assertRaises(ValueError): module.load_role_token(home) source.chmod(0o600) self.assertEqual(module.load_role_token(home), 's' * 48) source.unlink() external = home / 'external' external.write_text('GITEA_TOKEN=' + 's' * 48) external.chmod(0o600) source.symlink_to(external) with self.assertRaises(ValueError): module.load_role_token(home) def test_http_transport_get_only_redirects_and_secret_echo_are_rejected(self): module = self.module() self.assertTrue(callable(getattr(module, 'RestTransport', None)), 'read-only HTTP transport missing') import http.server import threading import urllib.request requests = [] secret = 'z' * 48 class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, format, *args): pass def do_GET(self): requests.append((self.path, self.headers.get('Authorization') == 'token ' + secret)) if self.path == '/api/v1/version': self.send_response(302) self.send_header('Location', 'https://outside.invalid/user') self.end_headers() return body = (b'{"credential":"' + secret.encode() + b'"}' if self.path.endswith('/secret') else b'{"id":42,"login":"gitea-impl-agent"}') self.send_response(200) self.send_header('Content-Type', 'application/json') if self.path.endswith('/duplicate'): self.send_header('X-Total-Count', '1') self.send_header('X-Total-Count', '2') self.end_headers() self.wfile.write(body) class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() inner = urllib.request.build_opener(NoRedirect()) class Proxy: def open(self, request, timeout): self_request = urllib.request.Request( 'http://127.0.0.1:' + str(server.server_port) + request.selector, headers=dict(request.header_items()), method=request.get_method()) response = inner.open(self_request, timeout=timeout) original = request.full_url class Response: status = response.status headers = response.headers def geturl(self): return original def read1(self, n): return response.read1(n) def __enter__(self): return self def __exit__(self, *args): response.close() return Response() try: client = module.RestTransport(secret, opener=Proxy()) data, _, status = client('/user') self.assertEqual(status, 200) self.assertEqual(data['id'], 42) self.assertEqual(client('/version')[2], 302) with self.assertRaises(ValueError) as raised: client('/repos/acme/widget/secret') self.assertNotIn(secret, str(raised.exception)) with self.assertRaises(ValueError): client('/repos/acme/widget/duplicate') self.assertTrue(all(authenticated for _, authenticated in requests)) self.assertEqual(len(requests), 4) finally: server.shutdown() server.server_close() thread.join(timeout=3) def test_output_admission_preserves_receipt_on_failure_and_rejects_links(self): module = self.module() self.assertTrue(callable(getattr(module, 'OutputFile', None)), 'pinned output writer missing') with tempfile.TemporaryDirectory() as directory: receipt = Path(directory) / 'receipt.json' receipt.write_text('original') receipt.chmod(0o600) with module.OutputFile(receipt): pass self.assertEqual(receipt.read_text(), 'original') with module.OutputFile(receipt) as output: output.write({'status': 'verified'}) self.assertIn('verified', receipt.read_text()) self.assertEqual(receipt.stat().st_mode & 0o777, 0o600) foreign = Path(directory) / 'foreign' foreign.write_text('preserve') receipt.unlink() receipt.symlink_to(foreign) with self.assertRaises(ValueError): module.OutputFile(receipt) self.assertEqual(foreign.read_text(), 'preserve') def test_http_operation_budget_stops_before_next_get(self): module = self.module() from unittest.mock import patch clock = [0] calls = [] class Opener: def open(self, request, timeout): calls.append(request.full_url) class Response: status = 200 headers = {'Content-Type': 'application/json'} emitted = False def geturl(self): return request.full_url def read1(self, n): if self.emitted: return b'' self.emitted = True return b'{"id":42}' def __enter__(self): return self def __exit__(self, *args): pass return Response() with patch.object(module.time, 'monotonic', side_effect=lambda: clock[0]): client = module.RestTransport('b' * 48, opener=Opener()) self.assertEqual(client('/user')[2], 200) clock[0] = 121 with self.assertRaises(ValueError): client('/version') self.assertEqual(len(calls), 1) def test_live_bridge_rejects_unbound_provider_before_transport(self): module = self.module() self.assertTrue(callable(getattr(module, 'run_live', None)), 'live verifier bridge missing') import hashlib from types import SimpleNamespace from unittest.mock import patch with tempfile.TemporaryDirectory() as directory: root = Path(directory) expected = root / 'expected.json' expected.write_text('{}') contract = root / 'provider.json' contract.write_text('{"schema":"test-contract"}') trusted = hashlib.sha256(contract.read_bytes()).hexdigest() self.assertEqual(module.read_bound_json(contract, trusted)['schema'], 'test-contract') receipt = root / 'receipt.json' receipt.write_text('preserve') receipt.chmod(0o600) args = SimpleNamespace(cmd='snapshot', expected=str(expected), provider_contract=str(contract), provider_contract_sha256='0' * 64, actor_id=42, output=str(receipt)) with patch.object(module, 'RestTransport') as transport: with patch.object(module, 'read_bound_json', wraps=module.read_bound_json) as reader: with self.assertRaises(ValueError): module.run_live(args, lambda *a, **k: None) self.assertGreaterEqual(reader.call_count, 2) transport.assert_not_called() self.assertEqual(receipt.read_text(), 'preserve') def test_existing_command_exposes_readonly_snapshot(self): result = subprocess.run([sys.executable, '-B', str(SCRIPT), 'snapshot', '--help'], capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn('--provider-contract', result.stdout) self.assertIn('--actor-id', result.stdout) def test_existing_command_exposes_live_verification(self): result = subprocess.run([sys.executable, '-B', str(SCRIPT), 'verify-live', '--help'], capture_output=True, text=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn('--baseline-sha256', result.stdout)