import copy import json import subprocess import sys import tempfile from pathlib import Path import unittest from harness.tests._support import spec as fixture class RequestedReviewerProjectionTests(unittest.TestCase): def snapshots(self): before, after, review, result, paths = fixture.publication_fixture() rich_user = { 'id': 42, 'login': 'spec-agent', 'full_name': 'Spec Agent', 'metadata': { 'active': True, 'roles': ['reviewer', {'scope': 'implementation-spec'}], }, } review['user'] = copy.deepcopy(rich_user) after['reviews'][0]['user'] = copy.deepcopy(rich_user) after['timeline'][0]['user'] = copy.deepcopy(rich_user) people = [copy.deepcopy(rich_user), {'id': 7, 'login': 'other', 'metadata': {'active': False, 'roles': ['observer']}}] before['pr']['requested_reviewers'] = copy.deepcopy(people) after['pr']['requested_reviewers'] = copy.deepcopy(people) return before, after, review, result, paths def contract(self): return {'version': {'version': '1.27.3'}, 'requested_reviewers_presentation_order': True} def test_existing_reviewer_permutation_requires_explicit_versioned_contract(self): before, after, review, result, paths = self.snapshots() fixture.verify_fixture(before, after, review, result, paths, self.contract()) after['pr']['requested_reviewers'].reverse() try: actual = fixture.verify_fixture(before, after, review, result, paths, self.contract()) except fixture.mod.ValidationError as exc: self.fail('versioned presentation-only permutation was rejected: ' + str(exc)) self.assertEqual(actual['review_id'], 11) def test_missing_false_or_unrelated_projection_flags_remain_order_strict(self): before, after, review, result, paths = self.snapshots() after['pr']['requested_reviewers'].reverse() for contract in (None, {'version': {'version': '1.27.3'}}, {'version': {'version': '1.27.3'}, 'requested_reviewers_presentation_order': False}, {'version': {'version': '1.27.3'}, 'supersession_same_author_only': True}): with self.subTest(contract=contract), self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(before, after, review, result, paths, contract) def test_new_reviewer_addition_can_reorder_preserved_members(self): before, after, review, result, paths = self.snapshots() old = [{'id': 5, 'login': 'five'}, {'id': 7, 'login': 'seven'}] before['pr']['requested_reviewers'] = copy.deepcopy(old) after['pr']['requested_reviewers'] = [old[1], copy.deepcopy(review['user']), old[0]] originals = copy.deepcopy((before, after)) self.assertEqual(fixture.verify_fixture(before, after, review, result, paths, self.contract())['review_id'], 11) self.assertEqual((before, after), originals) # Without opt-in only an addition preserving previous order is allowed. with self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(before, after, review, result, paths) after['pr']['requested_reviewers'] = old + [copy.deepcopy(review['user'])] fixture.verify_fixture(before, after, review, result, paths) def test_opt_in_new_reviewer_must_match_exact_review_user_record(self): before, after, review, result, paths = self.snapshots() old = [{'id': 5, 'login': 'five'}, {'id': 7, 'login': 'seven'}] before['pr']['requested_reviewers'] = copy.deepcopy(old) expected = copy.deepcopy(review['user']) for forged in ( {'id': 42}, {'id': 42, 'login': 'forged', 'metadata': expected['metadata']}, {**expected, 'extra': 'unexpected'}, {**expected, 'metadata': {'active': True, 'roles': ['reviewer', {'scope': 'other'}]}}, {'id': 42, 'login': expected['login'], 'metadata': expected['metadata'], 'name': None}, ): bad = copy.deepcopy(after) bad['pr']['requested_reviewers'] = [old[1], forged, old[0]] with self.subTest(forged=forged), self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(before, bad, review, result, paths, self.contract()) def test_removals_unknown_additions_and_all_existing_record_drift_fail(self): before, after, review, result, paths = self.snapshots() changes = [] for people in ([{'id': 42}], after['pr']['requested_reviewers'] + [{'id': 43}], after['pr']['requested_reviewers'] + [copy.deepcopy(after['pr']['requested_reviewers'][0])]): bad = copy.deepcopy(after); bad['pr']['requested_reviewers'] = people; changes.append(bad) for field in ('login', 'metadata', 'extra'): bad = copy.deepcopy(after); bad['pr']['requested_reviewers'][0][field] = 'changed'; changes.append(bad) bad = copy.deepcopy(after); del bad['pr']['requested_reviewers'][0]['metadata']; changes.append(bad) for bad in changes: with self.subTest(case=changes.index(bad)), self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(before, bad, review, result, paths, self.contract()) bad_before = copy.deepcopy(before) bad_before['pr']['requested_reviewers'].append(copy.deepcopy(bad_before['pr']['requested_reviewers'][0])) with self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(bad_before, after, review, result, paths, self.contract()) def test_contract_version_and_flag_types_fail_closed(self): before, after, review, result, paths = self.snapshots() for value in (None, 0, 1, 'true', [], {}): contract = self.contract(); contract['requested_reviewers_presentation_order'] = value with self.subTest(flag=value), self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(before, after, review, result, paths, contract) for version in ('1.27.4', '1.27.3+patched', None, True): contract = self.contract(); contract['version']['version'] = version with self.subTest(version=version), self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(before, after, review, result, paths, contract) def test_invalid_id_types_and_null_collections_fail_closed(self): before, after, review, result, paths = self.snapshots() for value in (True, 0, -1, '42', 42.0, None): bad = copy.deepcopy(after); bad['pr']['requested_reviewers'][0]['id'] = value with self.subTest(id=value), self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(before, bad, review, result, paths, self.contract()) for side in ('before', 'after'): b, a = copy.deepcopy(before), copy.deepcopy(after) (b if side == 'before' else a)['pr']['requested_reviewers'] = None with self.subTest(side=side), self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(b, a, review, result, paths, self.contract()) def test_other_meaningful_orders_remain_strict(self): before, after, review, result, paths = self.snapshots() comments = [{'id': 1, 'body': 'one'}, {'id': 2, 'body': 'two'}] for field in ('issue_comments', 'pr_comments'): b, a = copy.deepcopy(before), copy.deepcopy(after) b[field] = copy.deepcopy(comments); a[field] = list(reversed(comments)) with self.subTest(field=field), self.assertRaises(fixture.mod.ValidationError): fixture.verify_fixture(b, a, review, result, paths, self.contract()) # Historical review order must stay intact, even with reviewer opt-in. for ident in (8, 9): old = copy.deepcopy(review); old['id'] = ident before['reviews'].append(old); after['reviews'].insert(0, copy.deepcopy(old)) with self.assertRaisesRegex(fixture.mod.ValidationError, 'previous review order'): fixture.verify_fixture(before, after, review, result, paths, self.contract()) def test_cli_uses_opt_in_without_mutating_input_files(self): before, after, review, result, paths = self.snapshots() after['pr']['requested_reviewers'].reverse() with tempfile.TemporaryDirectory(dir='/opt/data/cache/scratch') as td: root = Path(td) values = {'before': before, 'after': after, 'exact-review': review, 'result': result, 'allowed-paths': list(paths), 'inline-comments': [], 'provider-contract': self.contract()} args = [sys.executable, '-B', str(fixture.MODPATH), 'verify-publication'] for name, value in values.items(): path = root / (name + '.json'); path.write_text(json.dumps(value)) args += ['--' + name, str(path)] args += ['--reviewer-id', '42', '--expected-pr-url', 'https://example.test/acme/repo/pulls/7'] original = {p.name: p.read_bytes() for p in root.iterdir()} checked = subprocess.run(args, capture_output=True, text=True, timeout=15) self.assertEqual(checked.returncode, 0, checked.stderr) self.assertEqual(json.loads(checked.stdout)['review_id'], 11) self.assertEqual({p.name: p.read_bytes() for p in root.iterdir()}, original) path = root / 'provider-contract.json' contract = self.contract(); contract['requested_reviewers_presentation_order'] = False path.write_text(json.dumps(contract)) denied = subprocess.run(args, capture_output=True, text=True, timeout=15) self.assertNotEqual(denied.returncode, 0) def test_document_declares_version_default_and_narrow_collection_scope(self): skill = fixture.MODPATH.parent.parent / 'SKILL.md' text = skill.read_text() for token in ('requested_reviewers_presentation_order', '1.27.3'): with self.subTest(token=token): self.assertIn(token, text) if __name__ == '__main__': unittest.main()