Files
rea/experiments/analysis-efficiency/test_command_observation_delivery.py
T
gensokyo-dev 17f581ff75 perf(reverse): add retained GDB and objdump summaries (#5614)
Add opt-in retained GDB and objdump summaries using the shared analysis-view layer. Preserve default complete Evidence, ordered records, provenance and completion facts.

Validate real Bash disassembly, whole-client context costs, CLI publication and cleanup; retain the original measured cohorts and explicit coverage limitations.

Closes #5584.
2026-10-21 01:20:52 +08:00

82 lines
3.5 KiB
Python

"""A truncated summary must never qualify as a completed investigation."""
from copy import deepcopy
import unittest
from command_observation_delivery import verify_command_delivery
class CommandDelivery(unittest.TestCase):
def test_truncation_is_accepted_only_for_complete_default_baseline(self):
truncated = {'_invalid_json_payload': '{truncated'}
self.assertFalse(verify_command_delivery(truncated, {}, 'complete', 0))
for mode, budget in [
('summary', 0), ('summary', 30000), ('complete', 30000),
]:
with self.subTest(mode=mode, budget=budget):
with self.assertRaises(AssertionError):
verify_command_delivery(truncated, {}, mode, budget)
def test_valid_complete_answer_can_complete_under_either_budget(self):
parent = {'evidence_id': 'parent'}
for budget in [0, 30000]:
self.assertTrue(
verify_command_delivery(parent, parent, 'complete', budget)
)
with self.assertRaises(AssertionError):
verify_command_delivery({'error': 'failed command'}, parent, 'complete', 0)
def test_summary_must_preserve_every_record_and_accurate_stream_counts(self):
report = {
'command': 'disassemble /r entry', 'mi': '7^done',
'records': ['~"🧪\\n"', '=async', 'unknown', '7^done'],
'console': '🧪', 'target': '', 'log': '',
'completion_status': 'complete', 'output_truncated': False,
}
parent = {
'evidence_id': 'parent',
'parameters': {'command': report['command']},
'provider': {'id': 'gdb'}, 'analysis_profile': None,
'normalized_result': report,
}
facts = {
key: value for key, value in report.items()
if key not in ('console', 'target', 'log')
}
selected = {
'evidence_id': 'derived', 'raw_result': None, 'subject': None,
'normalized_result': {
'parent_evidence_id': 'parent', 'artifact': None,
'source_analysis_profile_digest': None,
'source_parameters': parent['parameters'],
'source_provider': parent['provider'],
'summary': {
**facts,
'available_counts': {
'records': 4, 'console_utf16_units': 2,
'target_utf16_units': 0, 'log_utf16_units': 0,
},
},
},
}
self.assertTrue(verify_command_delivery(selected, parent, 'summary', 0))
for damage in ['records', 'counts', 'parameters', 'profile']:
bad = deepcopy(selected)
result = bad['normalized_result']
if damage == 'records':
result['summary']['records'].pop(1)
elif damage == 'counts':
result['summary']['available_counts']['console_utf16_units'] = 1
elif damage == 'profile':
result['source_analysis_profile_digest'] = 'wrong'
else:
result['source_parameters'] = {}
with self.subTest(damage=damage):
with self.assertRaises(AssertionError):
verify_command_delivery(bad, parent, 'summary', 0)
parent['analysis_profile'] = {'digest': 'recorded-profile'}
selected['normalized_result']['source_analysis_profile_digest'] = 'recorded-profile'
self.assertTrue(verify_command_delivery(selected, parent, 'summary', 0))
if __name__ == '__main__':
unittest.main()