genichin/gitea-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(harness): reorganize regression suites and preserve contract coverage

Browse Source
This commit is contained in:
gitea-spec-agent
2026-10-16 07:24:49 +00:00
parent b6f87db623
commit 33709ab873
108 changed files with 5869 additions and 5328 deletions
Download Patch File Download Diff File Expand all files Collapse all files
harness/tests/common/evidence/__init__.py
+1
View File
@@ -0,0 +1 @@
"""evidence regression tests."""
harness/tests/common/evidence/test_pytest_evidence_recipe.py
+86
View File
@@ -0,0 +1,86 @@
"""Conditional pytest recipe: documentation contract and real synthetic execution.
Run with an independently admitted pytest interpreter for execution coverage.
No dependency installation occurs in this test or the production workflow.
"""
from harness.tests._support.paths import ROOT as SOURCE_ROOT
import importlib.util
import json
import os
from pathlib import Path
from harness.tests._support.contracts import read_contract
import subprocess
import sys
import tempfile
import unittest
ROOT = SOURCE_ROOT
SKILL = ROOT / 'harness/agents/impl/skills/gitea-issue-rework/SKILL.md'
HEADING = '### Conditional pytest evidence recipe'
class PytestEvidenceRecipeTests(unittest.TestCase):
def recipe(self):
text = read_contract(SKILL)
self.assertIn(HEADING, text, 'conditional pytest recipe is missing')
section = text.split(HEADING, 1)[1].split('\n### ', 1)[0]
return section, json.loads(section.split('```json\n', 1)[1].split('\n```', 1)[0])
def test_documentation_contract_limits_recipe_to_existing_pytest_projects(self):
section, argv = self.recipe()
for rule in ['PYTEST_ADDOPTS', 'plugin', 'evidence_gap', 'timeout', 'output_limit']:
self.assertIn(rule, section)
self.assertIsInstance(argv, list)
self.assertEqual(argv[:4], ['{python}', '-B', '-m', 'pytest'])
@unittest.skipUnless(importlib.util.find_spec('pytest'), 'admitted pytest interpreter required')
def test_real_recipe_preserves_pass_skip_warning_summary_under_quiet_project_config(self):
_, argv = self.recipe()
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / 'pytest.ini').write_text('[pytest]\naddopts = -qq\n')
(root / 'test_sample.py').write_text(
'import pytest, warnings\n'
'def test_pass():\n warnings.warn("synthetic warning", UserWarning)\n'
'@pytest.mark.skip(reason="synthetic skip")\ndef test_skip():\n pass\n')
control = self.invoke([sys.executable, '-B', '-m', 'pytest', 'test_sample.py'], root)
self.assertEqual(control['status'], 'completed')
self.assertEqual(control['exit_code'], 0)
self.assertNotIn('1 passed', control['output'])
actual = self.invoke(self.expand(argv), root)
self.assertEqual(actual['status'], 'completed')
self.assertEqual(actual['exit_code'], 0)
self.assertRegex(actual['output'], r'1 passed, 1 skipped, 1 warning')
self.assertNotIn('\x1b[', actual['output'])
@unittest.skipUnless(importlib.util.find_spec('pytest'), 'admitted pytest interpreter required')
def test_real_recipe_preserves_failure_summary_and_exit_code(self):
_, argv = self.recipe()
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / 'pytest.ini').write_text('[pytest]\naddopts = -qq\n')
(root / 'test_sample.py').write_text('def test_fail():\n assert False, "synthetic failure"\n')
actual = self.invoke(self.expand(argv), root)
self.assertEqual(actual['status'], 'completed')
self.assertEqual(actual['exit_code'], 1)
self.assertIn('1 failed', actual['output'])
def expand(self, argv):
return [part.replace('{python}', sys.executable).replace('{tests}', 'test_sample.py') for part in argv]
def invoke(self, argv, root):
import importlib.util
path = ROOT / 'harness/agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py'
spec = importlib.util.spec_from_file_location('pytest_recipe_evidence', path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
env = {key: os.environ[key] for key in ['PATH', 'HOME', 'TMPDIR', 'LANG'] if key in os.environ}
# Isolate synthetic tests from unrelated globally installed pytest plugins.
env['PYTEST_DISABLE_PLUGIN_AUTOLOAD'] = '1'
result = module.run_bounded(argv, root, env, 65536, 30)
return {'status': result['status'], 'exit_code': result['exit_code'],
'output': result['output'].decode('utf-8', 'replace')}
if __name__ == '__main__':
unittest.main()
harness/tests/common/evidence/test_workflow_guard.py
+420
View File
@@ -0,0 +1,420 @@
import copy
import hashlib
import json
import os
import subprocess
import tempfile
import unittest
from pathlib import Path
from harness.lib import workflow_guard as guard
def inventory_digest(ids, statuses):
value = {'ids': sorted(ids), 'statuses': {key: statuses[key] for key in sorted(ids)}}
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
def make_test_inventory(ids=('case_a',), current=None):
current = list(ids if current is None else current)
ids = list(ids)
old_status = {key: {'skip': False, 'expected_failure': False} for key in ids}
new_status = {key: {'skip': False, 'expected_failure': False} for key in current}
def collection(names, statuses):
return {'source_revision': 'a' * 40, 'command_id': 'collect-tests',
'argv': ['python3', '-B', '-m', 'unittest', 'discover'],
'inventory_sha256': inventory_digest(names, statuses),
'evidence': 'fixture-collection-record'}
return {'baseline_ids': ids, 'current_ids': current,
'baseline_status': old_status, 'current_status': new_status,
'baseline_collection': collection(ids, old_status),
'current_collection': collection(current, new_status),
'authorized_changes': [], 'changed_tests': [], 'findings': []}
def contract_inventory():
return {'baseline': [{'semantic_key': 'ready-fields', 'baseline_revision': 'a' * 40,
'path': 'src/broker.py', 'meaning': 'Public ready fields are handle/socket.'}],
'current': [{'semantic_key': 'ready-fields', 'meaning': 'Public ready fields are handle/socket.',
'observation': 'observed', 'evidence': 'fixture-response-check'}],
'mappings': [{'semantic_key': 'ready-fields', 'current_keys': ['ready-fields'],
'status': 'preserved', 'authorization': '', 'verification': 'fixture-static-contract'}],
'stages': ['runtime', 'ready', 'admission'],
'precedence': [['runtime', 'ready'], ['ready', 'admission']]}
def finding_record():
def execution(code, kind):
return {'test_id': 'case_a', 'source_revision': 'a' * 40,
'scope_sha256': 'b' * 64, 'command_id': 'named-regression',
'argv': ['python3', '-B', '-m', 'unittest', 'case_a'],
'exit_code': code, 'expected_exit_code': code,
'failure_kind': kind, 'evidence': 'fixture-execution-record'}
return {'id': 'finding-a', 'test_id': 'case_a', 'classification': 'assertion',
'red': execution(1, 'assertion'), 'green': execution(0, 'pass')}
class WorkflowGuardTests(unittest.TestCase):
def repository(self, directory):
root = Path(directory) / 'repo'
root.mkdir(mode=0o700)
subprocess.run(['git', 'init', '-q', str(root)], check=True)
subprocess.run(['git', '-C', str(root), '-c', 'user.name=Fixture',
'-c', 'user.email=fixture@example.invalid', 'commit',
'--allow-empty', '-qm', 'baseline'], check=True)
(root / 'src').mkdir(mode=0o700)
(root / 'src/a.py').write_text('one')
return root
def assert_block(self, function, value):
try:
result = function(value)
except guard.GuardError:
return
self.assertEqual(result['status'], 'block')
def test_snapshot_compare_and_change_detection(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
out = Path(td) / 'snapshot.json'
guard.snapshot(root, ['src/a.py'], out)
self.assertEqual(guard.compare_source(root, out, ['src/a.py'])['status'], 'pass')
(root / 'src/a.py').write_text('two')
self.assertEqual(guard.compare_source(root, out, ['src/a.py'])['status'], 'block')
def test_rejects_traversal_and_malformed_baseline(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
with self.assertRaises(guard.GuardError):
guard.snapshot(root, ['../outside'], Path(td) / 'x.json')
bad = Path(td) / 'bad.json'
bad.write_text('{}')
with self.assertRaises(guard.GuardError):
guard.compare_source(root, bad, ['src/a.py'])
def test_test_inventory_requires_identity_and_review_evidence(self):
self.assert_block(guard.check_tests, {'baseline_ids': ['a'], 'current_ids': ['b']})
def test_contract_inventory_requires_full_coverage(self):
self.assert_block(guard.check_contracts, {'baseline': [{'semantic_key': 'x'}], 'current': [], 'mappings': []})
def test_snapshot_rejects_literal_root_alias_and_symlink_root(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
alias = Path(td) / 'alias'
alias.symlink_to(root, target_is_directory=True)
for value in (str(root) + '/./', alias):
with self.subTest(root=str(value)), self.assertRaises(guard.GuardError):
guard.snapshot(value, ['src/a.py'], Path(td) / 'snapshot.json')
def test_snapshot_rejects_symlink_output_parent(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
actual = Path(td) / 'private'
actual.mkdir(mode=0o700)
alias = Path(td) / 'alias'
alias.symlink_to(actual, target_is_directory=True)
with self.assertRaises(guard.GuardError):
guard.snapshot(root, ['src/a.py'], alias / 'snapshot.json')
self.assertFalse((actual / 'snapshot.json').exists())
def test_snapshot_exact_mode_survives_restrictive_umask(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
out = Path(td) / 'snapshot.json'
before = os.umask(0o777)
try:
guard.snapshot(root, ['src/a.py'], out)
finally:
os.umask(before)
self.assertEqual(out.stat().st_mode & 0o777, 0o600)
def test_snapshot_rejects_boolean_schema_baseline(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
out = Path(td) / 'snapshot.json'
guard.snapshot(root, ['src/a.py'], out)
value = json.loads(out.read_text())
value['schema'] = True
out.write_text(json.dumps(value))
with self.assertRaises(guard.GuardError):
guard.compare_source(root, out, ['src/a.py'])
def test_snapshot_rejects_repository_subdirectory_root(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
with self.assertRaises(guard.GuardError):
guard.snapshot(root / 'src', ['a.py'], Path(td) / 'snapshot.json')
def test_snapshot_detects_same_bytes_inode_replacement(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
out = Path(td) / 'snapshot.json'
guard.snapshot(root, ['src/a.py'], out)
replacement = root / 'src/new.py'
replacement.write_text('one')
replacement.replace(root / 'src/a.py')
self.assertEqual(guard.compare_source(root, out, ['src/a.py'])['status'], 'block')
def test_new_regression_tests_do_not_require_deletion_authorization(self):
self.assertEqual(guard.check_tests(make_test_inventory(current=['case_a', 'case_b']))['status'], 'pass')
def test_collected_inventory_requires_provenance(self):
value = make_test_inventory()
del value['baseline_collection']
self.assert_block(guard.check_tests, value)
def test_collected_inventory_digest_must_match_ids_and_statuses(self):
value = make_test_inventory()
value['baseline_collection']['inventory_sha256'] = '0' * 64
self.assert_block(guard.check_tests, value)
def test_new_skip_requires_explicit_approval(self):
value = make_test_inventory()
value['current_status']['case_a']['skip'] = True
value['current_collection']['inventory_sha256'] = inventory_digest(value['current_ids'], value['current_status'])
self.assert_block(guard.check_tests, value)
def test_new_expected_failure_requires_explicit_approval(self):
value = make_test_inventory()
value['current_status']['case_a']['expected_failure'] = True
value['current_collection']['inventory_sha256'] = inventory_digest(value['current_ids'], value['current_status'])
self.assert_block(guard.check_tests, value)
def test_rename_requires_real_replacement_and_semantic_evidence(self):
value = make_test_inventory(current=['case_b'])
value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'rename', 'replacement_ids': ['ghost'],
'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
self.assert_block(guard.check_tests, value)
def test_approved_rename_preserves_mapped_test(self):
value = make_test_inventory(current=['case_b'])
value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'rename', 'replacement_ids': ['case_b'],
'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
self.assertEqual(guard.check_tests(value)['status'], 'pass')
def test_modified_test_requires_semantic_review(self):
value = make_test_inventory()
value['changed_tests'] = [{'test_id': 'case_a', 'baseline_sha256': 'a' * 64,
'current_sha256': 'b' * 64, 'review_evidence': ''}]
self.assert_block(guard.check_tests, value)
def test_finding_truthy_strings_are_not_red_green_records(self):
value = make_test_inventory()
value['findings'] = [{'id': 'finding-a', 'test_id': 'case_a', 'classification': 'assertion',
'red': 'claimed-red', 'green': 'claimed-green'}]
self.assert_block(guard.check_tests, value)
def test_red_success_exit_cannot_demonstrate_failure(self):
value = make_test_inventory()
finding = finding_record()
finding['red']['exit_code'] = finding['red']['expected_exit_code'] = 0
value['findings'] = [finding]
self.assert_block(guard.check_tests, value)
def test_green_failure_exit_cannot_demonstrate_resolution(self):
value = make_test_inventory()
finding = finding_record()
finding['green']['exit_code'] = finding['green']['expected_exit_code'] = 1
value['findings'] = [finding]
self.assert_block(guard.check_tests, value)
def test_environment_failure_is_not_valid_red(self):
value = make_test_inventory()
finding = finding_record()
finding['red']['failure_kind'] = 'environment'
value['findings'] = [finding]
self.assert_block(guard.check_tests, value)
def test_finding_must_name_present_regression_test(self):
value = make_test_inventory()
finding = finding_record()
finding['test_id'] = 'ghost'
value['findings'] = [finding]
self.assert_block(guard.check_tests, value)
def test_named_red_green_records_with_provenance_pass(self):
value = make_test_inventory()
value['findings'] = [finding_record()]
self.assertEqual(guard.check_tests(value)['status'], 'pass')
def test_missing_and_ambiguous_contracts_cannot_be_approved_away(self):
for status in ('missing', 'ambiguous'):
with self.subTest(status=status):
value = contract_inventory()
value['mappings'][0].update(status=status, authorization='fixture-approval')
self.assert_block(guard.check_contracts, value)
def test_preserved_contract_requires_existing_current_mapping(self):
value = contract_inventory()
value['current'] = []
self.assert_block(guard.check_contracts, value)
def test_unknown_observation_is_not_actual_public_response_evidence(self):
value = contract_inventory()
value['current'][0].update(observation='unknown', evidence='')
self.assert_block(guard.check_contracts, value)
def test_preserved_contract_requires_verification(self):
value = contract_inventory()
value['mappings'][0]['verification'] = ''
self.assert_block(guard.check_contracts, value)
def test_precedence_cycle_blocks_startup_order(self):
value = contract_inventory()
value['precedence'].append(['admission', 'runtime'])
self.assert_block(guard.check_contracts, value)
def test_complete_observed_contract_and_acyclic_order_pass(self):
self.assertEqual(guard.check_contracts(contract_inventory())['status'], 'pass')
def test_duplicate_current_contract_is_not_coverage(self):
value = contract_inventory()
value['current'].append(copy.deepcopy(value['current'][0]))
self.assert_block(guard.check_contracts, value)
def test_invalid_stage_types_fail_closed_without_unhandled_typeerror(self):
value = contract_inventory()
value['stages'] = [['unhashable']]
self.assert_block(guard.check_contracts, value)
def guard_cli(self, *args):
command = ['python3', '-B', str(Path(guard.__file__).resolve()), *map(str, args)]
try:
return subprocess.run(command, capture_output=True, text=True, timeout=3)
except subprocess.TimeoutExpired:
self.fail('guard did not fail closed within 3 seconds')
def test_snapshot_fifo_protected_file_fails_fast(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
os.mkfifo(root / 'src/pipe')
paths = Path(td) / 'paths.json'
paths.write_text('["src/pipe"]')
result = self.guard_cli('snapshot', '--repo-root', root, '--paths-json', paths,
'--output', Path(td) / 'snapshot.json')
self.assertEqual(result.returncode, 2)
self.assertEqual(json.loads(result.stdout)['status'], 'block')
def test_compare_source_fifo_protected_file_fails_fast(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
out = Path(td) / 'snapshot.json'
guard.snapshot(root, ['src/a.py'], out)
(root / 'src/a.py').unlink()
os.mkfifo(root / 'src/a.py')
paths = Path(td) / 'paths.json'
paths.write_text('["src/a.py"]')
result = self.guard_cli('compare-source', '--repo-root', root, '--paths-json', paths,
'--before', out)
self.assertEqual(result.returncode, 1) # Valid baseline; observed source-type drift.
self.assertEqual(json.loads(result.stdout)['status'], 'block')
def test_snapshot_rejects_root_ending_dotdot(self):
with tempfile.TemporaryDirectory() as td:
root = self.repository(td)
with self.assertRaises(guard.GuardError):
guard.snapshot(str(root / 'src') + '/..', ['src/a.py'], Path(td) / 'snapshot.json')
def test_snapshot_rejects_ancestor_symlink_root(self):
with tempfile.TemporaryDirectory() as td:
self.repository(td)
alias = Path(td) / 'alias'
alias.symlink_to(td, target_is_directory=True)
with self.assertRaises(guard.GuardError):
guard.snapshot(alias / 'repo', ['src/a.py'], Path(td) / 'snapshot.json')
def test_json_fifo_input_fails_fast(self):
with tempfile.TemporaryDirectory() as td:
pipe = Path(td) / 'pipe.json'
os.mkfifo(pipe)
result = self.guard_cli('check-tests', '--input', pipe)
self.assertEqual(result.returncode, 2)
self.assertEqual(json.loads(result.stdout)['status'], 'block')
def test_deeply_nested_json_blocks_without_traceback(self):
with tempfile.TemporaryDirectory() as td:
bad = Path(td) / 'bad.json'
bad.write_text('[' * 2000 + ']' * 2000)
result = self.guard_cli('check-tests', '--input', bad)
self.assertEqual(result.returncode, 2)
self.assertNotIn('Traceback', result.stderr)
self.assertEqual(json.loads(result.stdout)['status'], 'block')
def test_rename_empty_replacement_cannot_stand_in_for_deletion(self):
value = make_test_inventory(current=['case_b'])
value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'rename', 'replacement_ids': [],
'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
self.assert_block(guard.check_tests, value)
def test_unclassified_empty_replacement_cannot_stand_in_for_deletion(self):
value = make_test_inventory(current=['case_b'])
value['authorized_changes'] = [{'old_id': 'case_a', 'replacement_ids': [],
'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
self.assert_block(guard.check_tests, value)
def test_explicit_approved_deletion_is_allowed(self):
value = make_test_inventory(ids=['case_a', 'case_b'], current=['case_b'])
value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'delete', 'replacement_ids': [],
'reason': 'approved obsolete case deletion', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
self.assertEqual(guard.check_tests(value)['status'], 'pass')
def test_status_approval_cannot_authorize_removal(self):
value = make_test_inventory(current=['case_b'])
value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'status', 'replacement_ids': ['case_b'],
'reason': 'approved status update', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
self.assert_block(guard.check_tests, value)
def test_explicit_status_approval_maps_same_current_id(self):
value = make_test_inventory()
value['current_status']['case_a']['skip'] = True
value['current_collection']['inventory_sha256'] = inventory_digest(value['current_ids'], value['current_status'])
value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'status', 'replacement_ids': ['case_a'],
'reason': 'approved skip', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
self.assertEqual(guard.check_tests(value)['status'], 'pass')
def test_unhashable_authorization_id_fails_closed_api(self):
value = make_test_inventory()
value['authorized_changes'] = [{'old_id': ['case_a'], 'replacement_ids': [],
'reason': 'fixture', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}]
self.assert_block(guard.check_tests, value)
def test_approved_contract_substitution_can_map_new_current_key(self):
value = contract_inventory()
value['current'][0]['semantic_key'] = 'published-ready-fields'
value['mappings'][0].update(current_keys=['published-ready-fields'],
status='intentionally_changed', authorization='fixture-approved-substitution')
self.assertEqual(guard.check_contracts(value)['status'], 'pass')
def test_approved_contract_merge_retains_each_old_semantic_mapping(self):
value = contract_inventory()
value['baseline'].append({'semantic_key': 'state-fields', 'baseline_revision': 'a' * 40,
'path': 'src/broker.py', 'meaning': 'Public state is held.'})
value['current'][0].update(semantic_key='combined-fields', meaning='Public fields are handle/socket and state held.')
value['mappings'] = [{'semantic_key': key, 'current_keys': ['combined-fields'],
'status': 'intentionally_changed', 'authorization': 'fixture-approved-merge',
'verification': 'fixture-checked-each-old-meaning'} for key in ('ready-fields', 'state-fields')]
self.assertEqual(guard.check_contracts(value)['status'], 'pass')
def test_schema_docstring_lists_nested_fields_and_exit_rules(self):
for field in ('baseline_status', 'current_status', 'skip', 'expected_failure',
'authorized_changes', 'old_id', 'kind', 'replacement_ids', 'changed_tests',
'baseline_sha256', 'current_sha256', 'review_evidence', 'findings', 'scope_sha256',
'expected_exit_code', 'failure_kind', 'baseline_revision', 'semantic_key',
'current_keys', 'observation', 'authorization', 'verification', 'precedence',
"sort_keys=True", "separators=(',', ':')", 'RED', 'GREEN', 'delete', 'rename', 'status'):
with self.subTest(field=field):
self.assertIn(field, guard.__doc__ or '')
def test_declared_evidence_shape_does_not_authenticate_records(self):
# Deliberately synthetic fixture claims: the main agent must check actual evidence separately.
self.assertEqual(guard.check_tests(make_test_inventory())['status'], 'pass')
def test_unknown_inventory_fields_are_rejected(self):
value = make_test_inventory()
value['pretend_verified'] = True
self.assert_block(guard.check_tests, value)
if __name__ == '__main__':
unittest.main()