384 lines
25 KiB
Python
384 lines
25 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 TestNativePublication(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_payload_is_exact_bound_native_create(self):
|
|
m = self.module()
|
|
p = plan_fixture()
|
|
self.assertEqual(m.build_payload(p), {'commit_id': p['target']['head'],
|
|
'body': p['body'], 'event': 'APPROVED', 'comments': []})
|
|
bad = copy.deepcopy(p); bad['target']['pull'] = True
|
|
with self.assertRaises(m.ValidationError):
|
|
m.build_payload(bad)
|
|
bad = copy.deepcopy(p); bad['body'] = bad['body'].replace('a' * 40, 'd' * 40)
|
|
with self.assertRaises(m.ValidationError):
|
|
m.build_payload(bad)
|
|
def test_verify_exact_target_and_bound_provider_changes(self):
|
|
m = self.module()
|
|
self.assertTrue(callable(getattr(m, 'verify_publication', None)), 'readback verifier is missing')
|
|
p, before, after, exact = publication_fixture()
|
|
previous = after['reviews'][0]
|
|
previous.update(official=False, dismissed=True, updated_at=exact['submitted_at'])
|
|
after['commits'][0]['files'].reverse()
|
|
result = m.verify_publication(p, before, after, exact)
|
|
self.assertEqual(result['review_id'], 10)
|
|
self.assertEqual(result['projected_review_ids'], [8])
|
|
self.assertTrue(result['commit_file_order_normalized'])
|
|
for mutation in ['head', 'body', 'unrelated_review', 'commit_file']:
|
|
bad = copy.deepcopy(after)
|
|
if mutation == 'head': bad['pr']['head']['sha'] = 'e' * 40
|
|
elif mutation == 'body': bad['reviews'][-1]['body'] += ' altered'
|
|
elif mutation == 'unrelated_review': bad['reviews'][1]['body'] += ' altered'
|
|
else: bad['commits'][0]['files'][0]['status'] = 'removed'
|
|
with self.subTest(mutation=mutation), self.assertRaises(m.ValidationError):
|
|
m.verify_publication(p, before, bad, exact)
|
|
def test_publish_once_and_reconcile_durable_attempt_without_reposting(self):
|
|
m = self.module()
|
|
self.assertTrue(callable(getattr(m, 'publish', None)), 'shared publisher is missing')
|
|
p, before, after, exact = publication_fixture()
|
|
api = FakeAPI(p, before, after, exact); lease = FakeLease()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
result = m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'})
|
|
self.assertEqual(result['mode'], 'created')
|
|
self.assertEqual(api.posts, 1)
|
|
self.assertEqual(m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'})['mode'], 'no_op')
|
|
self.assertEqual(api.posts, 1)
|
|
self.assertNotIn('body', json.loads((self.canonical(m, p) / 'intent.json').read_text()))
|
|
for applies in [True, False]:
|
|
m = self.module(); p, before, after, exact = publication_fixture()
|
|
api = FakeAPI(p, before, after, exact); api.timeout = True; api.apply = applies
|
|
with self.subTest(applies=applies), tempfile.TemporaryDirectory() as td:
|
|
if applies:
|
|
self.assertEqual(m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'})['mode'], 'reconciled')
|
|
self.assertEqual(m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'})['mode'], 'no_op')
|
|
else:
|
|
for _ in range(2):
|
|
with self.assertRaises(m.UncertainPublication) as error:
|
|
m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'})
|
|
self.assertNotIn('secret transport detail', str(error.exception))
|
|
self.assertEqual(api.posts, 1)
|
|
def test_empty_review_comments_no_header_contract_is_explicit(self):
|
|
text = (ROOT / 'harness/agents/review/skills/gitea-code-review/references/native-review.md').read_text()
|
|
self.assertIn('### Empty review-comments readback without a count header', text)
|
|
section = text.split('### Empty review-comments readback without a count header', 1)[1].split('\n### ', 1)[0]
|
|
for rule in ['X-Total-Count', 'comments_count=0', 'JSON `[]`', 'declared_total=None', 'rel=next']:
|
|
self.assertIn(rule, section)
|
|
def test_cli_preflight_and_saved_wire_verification_are_read_only(self):
|
|
m = self.module(); p, before, after, exact = publication_fixture()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root = Path(td)
|
|
for name, value in [('plan', p), ('before', before), ('after', after), ('exact', exact)]:
|
|
(root / (name + '.json')).write_text(json.dumps(value, ensure_ascii=False))
|
|
base = [sys.executable, str(SCRIPT)]
|
|
env = {'PATH': os.environ['PATH'], 'PYTHONDONTWRITEBYTECODE': '1'}
|
|
pre = subprocess.run(base + ['preflight', '--plan', str(root / 'plan.json'),
|
|
'--baseline', str(root / 'before.json')], capture_output=True, text=True, env=env)
|
|
self.assertEqual(pre.returncode, 0, pre.stderr)
|
|
self.assertIn('"ready": true', pre.stdout, 'read-only preflight CLI is missing')
|
|
verified = subprocess.run(base + ['verify', '--plan', str(root / 'plan.json'),
|
|
'--before', str(root / 'before.json'), '--after', str(root / 'after.json'),
|
|
'--exact', str(root / 'exact.json')], capture_output=True, text=True, env=env)
|
|
self.assertEqual(verified.returncode, 0, verified.stderr)
|
|
self.assertTrue(json.loads(verified.stdout)['read_only'])
|
|
self.assertEqual(json.loads(verified.stdout)['review_id'], 10)
|
|
unsupported = subprocess.run(base + ['publish'], capture_output=True, text=True, env=env)
|
|
self.assertNotEqual(unsupported.returncode, 0)
|
|
def test_spec_approval_may_reference_unchanged_surrounding_test_paths(self):
|
|
m = self.module(); p, before, after, exact = publication_fixture()
|
|
p['spec_paths'] = ['src/a.py', 'tests/existing.py']
|
|
spec_result = support.parse_review(before['reviews'][1]['body'], {'src/a.py'})
|
|
spec_result['criteria'][0]['test_paths'] = ['tests/existing.py']
|
|
before['reviews'][1]['body'] = support.render_review(spec_result, set(p['spec_paths']), declaration={'mode': 'none'})
|
|
try:
|
|
m.validate_snapshot(p, before)
|
|
except m.ValidationError:
|
|
self.fail('spec approval must use its declared scope, not only changed files')
|
|
def test_malformed_historical_actor_and_missing_exact_review_flags_fail_closed(self):
|
|
m = self.module(); p, before, after, exact = publication_fixture()
|
|
bad = copy.deepcopy(before); bad['reviews'][0]['user'] = {'id': True, 'login': 'x'}
|
|
with self.assertRaises(m.ValidationError):
|
|
m.validate_snapshot(p, bad)
|
|
p, before, after, exact = publication_fixture()
|
|
del exact['official']; del after['reviews'][-1]['official']
|
|
with self.assertRaises(m.ValidationError):
|
|
m.verify_publication(p, before, after, exact)
|
|
def test_invalid_plan_bindings_are_not_normalized(self):
|
|
m = self.module()
|
|
mutations = [lambda p: p['target'].__setitem__('issue', True),
|
|
lambda p: p['target'].__setitem__('head', 'A' * 40),
|
|
lambda p: p['target'].__setitem__('owner', 'acme/other'),
|
|
lambda p: p.__setitem__('verdict', 'blocked'),
|
|
lambda p: p.__setitem__('body', p['body'] + '- Verdict: `approve`\n'),
|
|
lambda p: p['provider_contract'].__setitem__('evidence', []),
|
|
lambda p: p.__setitem__('spec_paths', ['../secret']),
|
|
lambda p: p.__setitem__('spec_paths', ['src/a.py', 'src/a.py']),
|
|
lambda p: p['reviewer'].__setitem__('id', p['spec_reviewer']['id'])]
|
|
for i, mutate in enumerate(mutations):
|
|
with self.subTest(case=i):
|
|
p = plan_fixture(); mutate(p)
|
|
with self.assertRaises(m.ValidationError): m.build_payload(p)
|
|
def test_wrong_role_and_expired_lease_prevent_any_post(self):
|
|
m = self.module()
|
|
for kind in ['wrong_role', 'expired']:
|
|
with self.subTest(kind=kind), tempfile.TemporaryDirectory() as td:
|
|
p, before, after, exact = publication_fixture(); api = FakeAPI(p, before, after, exact); lease = FakeLease()
|
|
if kind == 'expired': lease.valid = False
|
|
else: api.identity = lambda: p['spec_reviewer']
|
|
with self.assertRaises(m.ValidationError): m.publish(p, before, api, lease, self.canonical(m, p), declaration={'mode': 'none'})
|
|
self.assertEqual(api.posts, 0)
|
|
self.assertFalse((self.canonical(m, p) / 'intent.json').exists())
|
|
def test_fresh_head_contract_and_source_linkage_drift_prevent_post(self):
|
|
m = self.module()
|
|
changes = [lambda s: s['pr']['head'].__setitem__('sha', 'b' * 40),
|
|
lambda s: s['issue'].__setitem__('body', 'changed contract'),
|
|
lambda s: s['pr'].__setitem__('body', 'Closes #4'),
|
|
lambda s: s['pr']['base'].__setitem__('ref', 'other'),
|
|
lambda s: s['pr'].__setitem__('merged', True)]
|
|
for i, change in enumerate(changes):
|
|
with self.subTest(case=i), tempfile.TemporaryDirectory() as td:
|
|
p, before, after, exact = publication_fixture(); api = FakeAPI(p, copy.deepcopy(before), after, exact)
|
|
change(api.before)
|
|
with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'})
|
|
self.assertEqual(api.posts, 0)
|
|
def test_pagination_total_duplicate_and_truncation_are_hard_errors(self):
|
|
m = self.module()
|
|
changes = [lambda s: s['pagination']['files'][0].__setitem__('declared_total', 2),
|
|
lambda s: s['pagination']['reviews'][0].__setitem__('link', '<x>; rel="next"'),
|
|
lambda s: s['pagination']['commits'][0].__setitem__('count', True),
|
|
lambda s: s['files'].append(copy.deepcopy(s['files'][0])),
|
|
lambda s: s['review_comments'].pop('8')]
|
|
for i, change in enumerate(changes):
|
|
with self.subTest(case=i):
|
|
p, before, after, exact = publication_fixture(); change(before)
|
|
with self.assertRaises(m.ValidationError): m.validate_snapshot(p, before)
|
|
def test_spec_approval_body_head_role_and_state_are_bound(self):
|
|
m = self.module()
|
|
changes = [lambda s: s['reviews'][1].__setitem__('state', 'REQUEST_CHANGES'),
|
|
lambda s: s['reviews'][1].__setitem__('commit_id', 'b' * 40),
|
|
lambda s: s['reviews'][1].__setitem__('dismissed', True),
|
|
lambda s: s['reviews'][1].__setitem__('body', s['reviews'][1]['body'].replace('충족', '임의 변경', 1)),
|
|
lambda s: s['reviews'][1]['user'].__setitem__('id', 23)]
|
|
for i, change in enumerate(changes):
|
|
with self.subTest(case=i):
|
|
p, before, after, exact = publication_fixture(); change(before)
|
|
with self.assertRaises(m.ValidationError): m.validate_snapshot(p, before)
|
|
def test_normal_supersession_is_allowed_without_projection_opt_in(self):
|
|
for already_dismissed in (False, True):
|
|
with self.subTest(already_dismissed=already_dismissed):
|
|
m = self.module()
|
|
p, before, after, exact = publication_fixture()
|
|
p['provider_contract'].update(supersede_same_author=False,
|
|
retouch_dismissed=False,
|
|
unordered_commit_files=False, evidence=[])
|
|
if already_dismissed:
|
|
before['reviews'][0].update(official=False, dismissed=True)
|
|
after['reviews'][0].update(official=False, dismissed=True,
|
|
updated_at=exact['submitted_at'])
|
|
saved_plan = copy.deepcopy(p)
|
|
saved_before = copy.deepcopy(before)
|
|
try:
|
|
result = m.verify_publication(p, before, after, exact)
|
|
except m.ValidationError as exc:
|
|
self.fail(f'normal same-author replacement was rejected: {exc}')
|
|
self.assertTrue(result['verified'])
|
|
self.assertEqual(result['projected_review_ids'], [8])
|
|
self.assertEqual(p, saved_plan)
|
|
self.assertEqual(before, saved_before)
|
|
def test_legacy_replacement_flags_do_not_require_external_evidence(self):
|
|
m = self.module()
|
|
p, before, after, exact = publication_fixture()
|
|
p['provider_contract'].update(unordered_commit_files=False, evidence=[])
|
|
after['reviews'][0].update(official=False, dismissed=True,
|
|
updated_at=exact['submitted_at'])
|
|
try:
|
|
result = m.verify_publication(p, before, after, exact)
|
|
except m.ValidationError as exc:
|
|
self.fail(f'compatibility-only replacement flags required an opt-in: {exc}')
|
|
self.assertTrue(result['verified'])
|
|
self.assertEqual(m.build_payload(p)['body'], p['body'])
|
|
# Unordered file comparisons remain a separate, evidence-backed exception.
|
|
p['provider_contract']['unordered_commit_files'] = True
|
|
with self.assertRaisesRegex(m.ValidationError, 'provider projection requires evidence'):
|
|
m.validate_plan(p)
|
|
def test_normal_replacement_does_not_allow_historical_content_mutation(self):
|
|
m = self.module()
|
|
for field, value in [('body', 'altered'), ('commit_id', 'e' * 40),
|
|
('state', 'APPROVED'), ('user', {'id': 72, 'login': 'other'}),
|
|
('submitted_at', '2026-10-04T00:03:00Z'),
|
|
('official', True), ('dismissed', False),
|
|
('updated_at', '2026-10-04T00:03:00Z')]:
|
|
with self.subTest(field=field):
|
|
p, before, after, exact = publication_fixture()
|
|
p['provider_contract'].update(supersede_same_author=False,
|
|
retouch_dismissed=False,
|
|
unordered_commit_files=False, evidence=[])
|
|
after['reviews'][0].update(official=False, dismissed=True,
|
|
updated_at=exact['submitted_at'])
|
|
after['reviews'][0][field] = value
|
|
with self.assertRaises(m.ValidationError):
|
|
m.verify_publication(p, before, after, exact)
|
|
p, before, after, exact = publication_fixture()
|
|
p['provider_contract'].update(supersede_same_author=False, retouch_dismissed=False,
|
|
unordered_commit_files=False, evidence=[])
|
|
after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at'])
|
|
for snap in (before, after):
|
|
snap['review_comments']['8'] = [{'id': 71, 'body': 'original'}]
|
|
snap['reviews'][0]['comments_count'] = 1
|
|
snap['pagination']['review_comments_8'][0].update(count=1, declared_total=1)
|
|
after['review_comments']['8'][0]['body'] = 'altered'
|
|
with self.assertRaisesRegex(m.ValidationError, 'historical review comments changed'):
|
|
m.verify_publication(p, before, after, exact)
|
|
def test_normal_replacement_reconciles_original_intent_without_second_post(self):
|
|
m = self.module()
|
|
p, before, after, exact = publication_fixture()
|
|
p['provider_contract'].update(supersede_same_author=False, retouch_dismissed=False,
|
|
unordered_commit_files=False, evidence=[])
|
|
after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at'])
|
|
api = FakeAPI(p, before, after, exact)
|
|
original_read = api.read_review
|
|
api.read_review = lambda target, rid: (_ for _ in ()).throw(TimeoutError('unavailable'))
|
|
with self.assertRaises(m.ValidationError):
|
|
m.publish(p, before, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'})
|
|
self.assertEqual(api.posts, 1)
|
|
journal = self.canonical(m, p)
|
|
intent = (journal / 'intent.json').read_bytes()
|
|
self.assertFalse((journal / 'verified.json').exists())
|
|
api.read_review = original_read
|
|
result = m.publish(p, before, api, FakeLease(), journal, declaration={'mode': 'none'})
|
|
self.assertTrue(result['verified'])
|
|
self.assertFalse(result['wrote'])
|
|
self.assertEqual(result['projected_review_ids'], [8])
|
|
self.assertEqual(api.posts, 1)
|
|
self.assertEqual((journal / 'intent.json').read_bytes(), intent)
|
|
self.assertEqual(json.loads(intent)['fingerprint'],
|
|
__import__('hashlib').sha256(m.canonical(p)).hexdigest())
|
|
self.assertTrue(json.loads((journal / 'verified.json').read_text())['verified'])
|
|
def test_provider_projection_requires_explicit_bound_evidence(self):
|
|
m = self.module()
|
|
for kind in ['wrong_initial_flags', 'wrong_timestamp', 'wrong_author']:
|
|
with self.subTest(kind=kind):
|
|
p, before, after, exact = publication_fixture()
|
|
after['reviews'][0].update(official=False, dismissed=True, updated_at=exact['submitted_at'])
|
|
if kind == 'wrong_initial_flags': before['reviews'][0].update(official=True, dismissed=True)
|
|
elif kind == 'wrong_timestamp': after['reviews'][0]['updated_at'] = '2026-10-04T00:02:00Z'
|
|
else:
|
|
before['reviews'][0]['user'] = {'id': 72, 'login': 'other'}
|
|
after['reviews'][0]['user'] = {'id': 72, 'login': 'other'}
|
|
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
|
def test_order_normalization_does_not_mask_file_contents_or_parent_order(self):
|
|
m = self.module()
|
|
p, before, after, exact = publication_fixture()
|
|
after['commits'][0]['files'].reverse()
|
|
self.assertTrue(m.verify_publication(p, before, after, exact)['commit_file_order_normalized'])
|
|
p['provider_contract']['unordered_commit_files'] = False
|
|
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
|
p, before, after, exact = publication_fixture()
|
|
after['commits'][0]['files'][0]['status'] = 'removed'
|
|
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
|
p, before, after, exact = publication_fixture()
|
|
before['commits'][0]['parents'] = [{'sha': 'e' * 40}, {'sha': 'f' * 40}]
|
|
after['commits'][0]['parents'] = list(reversed(before['commits'][0]['parents']))
|
|
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
|
def test_historical_head_state_body_and_comments_remain_immutable(self):
|
|
m = self.module()
|
|
changes = [lambda s: s['reviews'][0].__setitem__('body', 'altered'),
|
|
lambda s: s['reviews'][0].__setitem__('commit_id', 'e' * 40),
|
|
lambda s: s['reviews'][0].__setitem__('state', 'APPROVED')]
|
|
for i, change in enumerate(changes):
|
|
with self.subTest(case=i):
|
|
p, before, after, exact = publication_fixture(); change(after)
|
|
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
|
p, before, after, exact = publication_fixture()
|
|
for snap in [before, after]:
|
|
snap['review_comments']['8'] = [{'id': 71, 'body': 'original'}]
|
|
snap['reviews'][0]['comments_count'] = 1
|
|
snap['pagination']['review_comments_8'][0].update(count=1, declared_total=1)
|
|
after['review_comments']['8'][0]['body'] = 'altered'
|
|
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
|
def test_existing_canonical_review_is_write_free_no_op(self):
|
|
m = self.module(); p, before, after, exact = publication_fixture()
|
|
api = FakeAPI(p, before, after, exact); api.created = True
|
|
with tempfile.TemporaryDirectory() as td:
|
|
result = m.publish(p, after, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'})
|
|
self.assertEqual(result['mode'], 'no_op'); self.assertFalse(result['wrote'])
|
|
self.assertEqual(api.posts, 0)
|
|
self.assertFalse((self.canonical(m, p) / 'intent.json').exists())
|
|
def test_duplicate_or_conflicting_canonical_review_is_not_reposted(self):
|
|
m = self.module()
|
|
for kind in ['body', 'duplicate']:
|
|
with self.subTest(kind=kind), tempfile.TemporaryDirectory() as td:
|
|
p, before, after, exact = publication_fixture()
|
|
if kind == 'body': after['reviews'][-1]['body'] += 'other'
|
|
else:
|
|
extra = copy.deepcopy(exact); extra['id'] = 11; after['reviews'].append(extra)
|
|
after['review_comments']['11'] = []
|
|
after['pagination']['reviews'][0].update(count=4, declared_total=4)
|
|
after['pagination']['review_comments_11'] = [{'page': 1, 'count': 0, 'declared_total': 0, 'link': None}]
|
|
api = FakeAPI(p, before, after, exact); api.created = True
|
|
with self.assertRaises(m.ValidationError): m.publish(p, before, api, FakeLease(), self.canonical(m, p), declaration={'mode': 'none'})
|
|
self.assertEqual(api.posts, 0)
|
|
def test_issue_repository_mismatch_is_not_accepted(self):
|
|
m = self.module(); p, before, after, exact = publication_fixture()
|
|
before['issue']['repository']['full_name'] = 'other/repo'
|
|
with self.assertRaises(m.ValidationError): m.validate_snapshot(p, before)
|
|
def test_fork_head_and_nonofficial_native_states_preserve_accepted_contract(self):
|
|
m = self.module(); p, before, after, exact = publication_fixture()
|
|
for snapshot in [before, after]:
|
|
snapshot['pr']['head']['repo']['full_name'] = 'contributor/repo'
|
|
snapshot['reviews'][1]['official'] = False
|
|
after['reviews'][-1]['official'] = exact['official'] = False
|
|
self.assertTrue(m.verify_publication(p, before, after, exact)['verified'])
|
|
after['pr']['head']['repo']['full_name'] = 'different/repo'
|
|
with self.assertRaises(m.ValidationError): m.verify_publication(p, before, after, exact)
|
|
|
|
def test_alternate_path_is_rejected_before_publication(self):
|
|
m = load(SCRIPT, 'native_review_alternate_test')
|
|
fixtures = load(ROOT / 'harness/tests/_support/review.py', 'native_review_alternate_fixture_test')
|
|
plan, before, after, exact = fixtures.publication_fixture()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
with self.assertRaises(m.ValidationError):
|
|
m.publish(plan, before, fixtures.FakeAPI(plan, before, after, exact),
|
|
fixtures.FakeLease(), td, declaration={'mode': 'none'})
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|