214 lines
13 KiB
Python
214 lines
13 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 TestNativeTransport(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_api_adapter_empty_comments_without_header_verifies_publication(self):
|
|
m = self.module()
|
|
p, before, after, exact = publication_fixture()
|
|
|
|
def collect(source, comments=None, headers=None):
|
|
calls = []
|
|
def request(path, payload=None):
|
|
self.assertIsNone(payload, 'readback regression must be GET-only')
|
|
calls.append(path)
|
|
direct = {'/user': p['reviewer'], '/users/gitea-spec-agent': p['spec_reviewer'],
|
|
'/repos/acme/repo': source['repository'],
|
|
'/repos/acme/repo/issues/3': source['issue'],
|
|
'/repos/acme/repo/pulls/7': source['pr'],
|
|
'/repos/acme/repo/pulls/7/reviews/10': exact}
|
|
if path in direct:
|
|
return direct[path], {}, 200
|
|
from urllib.parse import urlsplit
|
|
endpoint = urlsplit(path).path
|
|
for name, suffix in [('files', '/pulls/7/files'), ('commits', '/pulls/7/commits'),
|
|
('reviews', '/pulls/7/reviews'), ('issue_comments', '/issues/3/comments'),
|
|
('pr_comments', '/issues/7/comments'),
|
|
('statuses', '/statuses/' + p['target']['head'])]:
|
|
if endpoint == '/repos/acme/repo' + suffix:
|
|
return copy.deepcopy(source[name]), {'X-Total-Count': str(len(source[name]))}, 200
|
|
if '/reviews/' in endpoint and endpoint.endswith('/comments'):
|
|
return ([] if comments is None else comments), ({} if headers is None else headers), 200
|
|
raise AssertionError('unexpected synthetic readback endpoint')
|
|
api = m.ApiAdapter(request, origin=m.ORIGIN + '/api/v1',
|
|
credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
|
|
snapshot = api.snapshot(p['target'])
|
|
self.assertEqual(calls[0], '/user')
|
|
self.assertTrue(any('/reviews/' in path and '/comments?' in path for path in calls))
|
|
return api, snapshot
|
|
|
|
_, initial = collect(before)
|
|
api, final = collect(after)
|
|
m.validate_snapshot(p, initial)
|
|
m.validate_snapshot(p, final)
|
|
authoritative = api.read_review(p['target'], exact['id'])
|
|
m.verify_publication(p, initial, final, authoritative)
|
|
for snapshot in [initial, final]:
|
|
for rid in snapshot['review_comments']:
|
|
self.assertEqual(snapshot['review_comments'][rid], [])
|
|
self.assertEqual(snapshot['pagination']['review_comments_' + rid],
|
|
[{'page': 1, 'count': 0, 'declared_total': None, 'link': None}])
|
|
|
|
# The admitted control above distinguishes intentional rejection from
|
|
# a broken fixture. Missing headers never waive the review count or
|
|
# the complete-collection and newly-created no-inline-comment checks.
|
|
for count in [1, None, True]:
|
|
with self.subTest(count=count):
|
|
drift = copy.deepcopy(final)
|
|
drift['reviews'][-1]['comments_count'] = count
|
|
with self.assertRaisesRegex(m.ValidationError, 'review comment cardinality mismatch'):
|
|
m.validate_snapshot(p, drift)
|
|
_, unexpected = collect(after, comments=[{'id': 999}])
|
|
for review in unexpected['reviews']:
|
|
review['comments_count'] = 1
|
|
m.validate_snapshot(p, unexpected)
|
|
exact_with_inline = copy.deepcopy(authoritative)
|
|
exact_with_inline['comments_count'] = 1
|
|
with self.assertRaisesRegex(m.ValidationError, 'exact review binding mismatch'):
|
|
m.verify_publication(p, initial, unexpected, exact_with_inline)
|
|
with self.assertRaisesRegex(m.ValidationError, 'truncated pages'):
|
|
collect(after, headers={'X-Total-Count': '1'})
|
|
truncated = copy.deepcopy(final)
|
|
truncated['pagination']['review_comments_' + str(exact['id'])][0]['link'] = '<next>; rel=next'
|
|
with self.assertRaisesRegex(m.ValidationError, 'truncated pagination'):
|
|
m.validate_snapshot(p, truncated)
|
|
with self.assertRaisesRegex(m.ValidationError, 'truncated pagination'):
|
|
m.check_pages([{'id': i + 1} for i in range(50)],
|
|
[{'page': 1, 'count': 50, 'declared_total': None, 'link': None}])
|
|
def test_api_adapter_checks_identity_first_and_collects_all_pages(self):
|
|
m = self.module()
|
|
self.assertTrue(callable(getattr(m, 'ApiAdapter', None)), 'REST bridge is missing')
|
|
p, before, after, exact = publication_fixture()
|
|
calls = []
|
|
def request(path, payload=None):
|
|
calls.append((path, payload))
|
|
if path == '/user': return p['reviewer'], {}, 200
|
|
if path == '/users/gitea-spec-agent': return p['spec_reviewer'], {}, 200
|
|
if path == '/repos/acme/repo': return before['repository'], {}, 200
|
|
if path == '/repos/acme/repo/issues/3': return before['issue'], {}, 200
|
|
if path == '/repos/acme/repo/pulls/7': return before['pr'], {}, 200
|
|
from urllib.parse import parse_qs, urlsplit
|
|
parsed = urlsplit(path); page = int(parse_qs(parsed.query).get('page', ['1'])[0])
|
|
for name, suffix in [('files', '/pulls/7/files'), ('commits', '/pulls/7/commits'),
|
|
('reviews', '/pulls/7/reviews'), ('issue_comments', '/issues/3/comments'),
|
|
('pr_comments', '/issues/7/comments'), ('statuses', '/statuses/' + p['target']['head'])]:
|
|
if parsed.path == '/repos/acme/repo' + suffix:
|
|
if name == 'reviews' and payload is not None: return exact, {}, 201
|
|
return before[name], {'x-total-count': str(len(before[name]))}, 200
|
|
if parsed.path.endswith('/comments'): return [], {'X-Total-Count': '0'}, 200
|
|
raise AssertionError('unexpected request')
|
|
adapter = m.ApiAdapter(request, origin=m.ORIGIN + '/api/v1',
|
|
credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
|
|
snapshot = adapter.snapshot(p['target'])
|
|
m.validate_snapshot(p, snapshot)
|
|
self.assertEqual(calls[0][0], '/user')
|
|
self.assertEqual(adapter.create_review(p['target'], m.build_payload(p))['id'], 10)
|
|
self.assertEqual(calls[-1][1]['comments'], [])
|
|
with self.assertRaises(m.ValidationError):
|
|
m.ApiAdapter(request, origin='https://other.test/api/v1', credential_source='unknown')
|
|
def test_transport_origin_credential_source_and_exception_text_are_guarded(self):
|
|
m = self.module()
|
|
with self.assertRaises(m.ValidationError): m.ApiAdapter(lambda *args: None, origin='https://other/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
|
|
with self.assertRaises(m.ValidationError): m.ApiAdapter(lambda *args: None, origin=m.ORIGIN + '/api/v1', credential_source='other-profile')
|
|
def bad(*args): raise TimeoutError('DO_NOT_ECHO_TRANSPORT_SECRET')
|
|
api = m.ApiAdapter(bad, origin=m.ORIGIN + '/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
|
|
with self.assertRaises(m.ValidationError) as raised: api.identity()
|
|
self.assertNotIn('DO_NOT_ECHO_TRANSPORT_SECRET', str(raised.exception))
|
|
self.assertTrue(raised.exception.__suppress_context__)
|
|
def test_live_rest_bridge_over_loopback_publishes_once_then_no_ops(self):
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from urllib.request import Request, urlopen
|
|
m = self.module(); p, before, after, exact = publication_fixture()
|
|
after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at'])
|
|
after['commits'][0]['files'].reverse()
|
|
state = {'posted': False, 'posts': 0}
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args): pass
|
|
def respond(self, value, status=200, total=None):
|
|
body = json.dumps(value, ensure_ascii=False).encode()
|
|
self.send_response(status); self.send_header('Content-Type', 'application/json')
|
|
if total is not None: self.send_header('X-Total-Count', str(total))
|
|
self.send_header('Content-Length', str(len(body))); self.end_headers(); self.wfile.write(body)
|
|
def do_GET(self):
|
|
path = self.path.split('?')[0]; s = after if state['posted'] else before
|
|
if path == '/user': return self.respond(p['reviewer'])
|
|
if path == '/users/gitea-spec-agent': return self.respond(p['spec_reviewer'])
|
|
prefix = '/repos/acme/repo'; pull = prefix + '/pulls/7'
|
|
singles = {prefix: s['repository'], prefix + '/issues/3': s['issue'], pull: s['pr']}
|
|
if path in singles: return self.respond(singles[path])
|
|
if path == pull + '/reviews/10': return self.respond(exact)
|
|
arrays = {pull + '/files': s['files'], pull + '/commits': s['commits'], pull + '/reviews': s['reviews'],
|
|
prefix + '/issues/3/comments': s['issue_comments'], prefix + '/issues/7/comments': s['pr_comments'],
|
|
prefix + '/statuses/' + p['target']['head']: s['statuses']}
|
|
for rid, comments in s['review_comments'].items(): arrays[pull + '/reviews/' + rid + '/comments'] = comments
|
|
if path in arrays: return self.respond(arrays[path], total=len(arrays[path]))
|
|
self.respond({}, 404)
|
|
def do_POST(self):
|
|
self.server.received = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
|
|
state['posts'] += 1; state['posted'] = True
|
|
self.respond(exact, 201)
|
|
server = HTTPServer(('127.0.0.1', 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True); thread.start()
|
|
def request(path, payload=None):
|
|
wire = None if payload is None else json.dumps(payload, ensure_ascii=False).encode()
|
|
req = Request('http://127.0.0.1:' + str(server.server_port) + path, data=wire,
|
|
headers={'Content-Type': 'application/json'})
|
|
with urlopen(req, timeout=5) as response:
|
|
return json.loads(response.read()), dict(response.headers), response.status
|
|
try:
|
|
# Controlled local fixture transport, not a real role credential/origin assertion.
|
|
api = m.ApiAdapter(request, origin=m.ORIGIN + '/api/v1', credential_source='$HERMES_HOME/.env:GITEA_TOKEN')
|
|
baseline = api.snapshot(p['target'])
|
|
with tempfile.TemporaryDirectory() as td:
|
|
result = m.publish(p, baseline, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'})
|
|
self.assertEqual(result['mode'], 'created'); self.assertEqual(result['projected_review_ids'], [8])
|
|
self.assertTrue(result['commit_file_order_normalized'])
|
|
second = m.publish(p, baseline, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'})
|
|
self.assertEqual(second['mode'], 'no_op'); self.assertFalse(second['wrote'])
|
|
self.assertEqual(state['posts'], 1); self.assertEqual(server.received, m.build_payload(p))
|
|
finally:
|
|
server.shutdown(); server.server_close(); thread.join(timeout=5)
|
|
self.assertFalse(thread.is_alive())
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|