183 lines
10 KiB
Python
183 lines
10 KiB
Python
"""Preflight completion is based on current program receipts, not a model."""
|
||
from copy import deepcopy
|
||
|
||
import pytest
|
||
from backend.config import settings
|
||
from backend.prospecting.preflight import delivery_projection, summarize_preflight
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def enable_reply_relation_preflight(monkeypatch):
|
||
monkeypatch.setattr(settings, 'reply_relation_preflight_enabled', True)
|
||
|
||
RUN = 'a'*32
|
||
N1, N2, N3, EXCLUDED = '1'*24, '2'*24, '3'*24, '6adaf8a6000000001901d400'
|
||
GOAL = f'本轮只做二级回复关系的只读预检，不保存正文或任何来源，最多2篇新笔记，排除ID为{EXCLUDED}的旧笔记。'
|
||
|
||
|
||
def observation(note=N1, *, checked=0, eligible=0, status=None, discussion='visible_comments', folded=True, index=1, **updates):
|
||
row = {'id': 'preflight_'+f'{index:032x}', 'version': 1, 'operation': 'inspect_reply_relations',
|
||
'run_id': RUN, 'goal_version': 2, 'note_id': note, 'observed_at': '2026-10-29T20:00:00+08:00',
|
||
'status': status or ('checked' if checked else 'no_visible_reply'), 'checked_count': checked,
|
||
'eligible_count': eligible, 'classification_counts': {'unique_dom_deep_reply': eligible,
|
||
'needs_native_identity': 0, 'ambiguous': checked-eligible, 'unknown': 0},
|
||
'discussion_state': discussion, 'visible_comment_count': checked,
|
||
'has_unexpanded_replies': folded, 'history_truncated': False}
|
||
row.update(updates)
|
||
return row
|
||
|
||
|
||
def navigation(note=N1, *, status='entered_note', index=1, **updates):
|
||
row = {'id': 'preflight_nav_'+f'{index:032x}', 'version': 1, 'run_id': RUN, 'goal_version': 2,
|
||
'status': status, 'action': 'observe', 'note_id': note, 'observed_at': '2026-10-29T20:00:00+08:00'}
|
||
row.update(updates)
|
||
return row
|
||
|
||
|
||
def task(*, observations=None, navigation_rows=None):
|
||
return {'status': 'completed', 'current_step': 'preflight_results',
|
||
'inputs': {'goal': GOAL, '_goal_version': 2, 'intent': 'inspect', 'max_notes': 2},
|
||
'artifacts': {'browser_verdict': 'success', 'browser_history': {'is_successful': True, 'final_result': 'model says zero eligible'},
|
||
'preflight_run': {'run_id': RUN, 'goal_version': 2},
|
||
'preflight_observations': observations if observations is not None else [observation()],
|
||
'preflight_navigation': navigation_rows if navigation_rows is not None else [navigation()]}}
|
||
|
||
|
||
def test_zero_eligible_is_a_valid_preflight_result_and_maximum_is_not_minimum():
|
||
row = task()
|
||
before = deepcopy(row)
|
||
summary = summarize_preflight(row)
|
||
assert summary['completed'] is True and summary['scope_passed'] is True
|
||
assert summary['inspected_notes'] == summary['visited_notes'] == 1
|
||
assert summary['max_notes'] == 2 and summary['eligible_count'] == summary['checked_count'] == 0
|
||
assert summary['folded_notes'] == 1 and '可见范围未发现合格样本' in summary['headline']
|
||
assert '未展开' in summary['detail'] and '未知' in summary['detail']
|
||
delivery = delivery_projection(row)
|
||
assert delivery['status'] == 'fulfilled' and delivery['fulfillment']['complete'] is True
|
||
assert delivery['fulfillment']['target'] is None and delivery['fulfillment']['current'] == 0
|
||
assert delivery['artifacts']['delivered'] is True and delivery['artifacts']['list_available'] is False
|
||
assert row == before
|
||
|
||
|
||
def test_counts_use_latest_current_receipt_per_note_and_never_count_old_runs_or_goals():
|
||
row = task(observations=[observation(checked=5, eligible=3, index=1),
|
||
observation(checked=2, eligible=1, index=2), observation(N2, checked=1, eligible=0, index=3),
|
||
observation(N3, checked=5, eligible=5, index=4, run_id='b'*32),
|
||
observation(EXCLUDED, checked=5, eligible=5, index=5, goal_version=1)],
|
||
navigation_rows=[navigation(), navigation(N1, status='revisited_note', index=2), navigation(N2, index=3),
|
||
navigation(N3, index=4, run_id='b'*32)])
|
||
summary = summarize_preflight(row)
|
||
assert summary['completed'] is True and summary['scope_passed'] is True
|
||
assert summary['visited_notes'] == summary['inspected_notes'] == 2
|
||
assert summary['checked_count'] == 3 and summary['eligible_count'] == 1
|
||
assert summary['classification_counts']['unique_dom_deep_reply'] == 1
|
||
assert summary['classification_counts']['ambiguous'] == 2
|
||
|
||
|
||
@pytest.mark.parametrize('state,status', [('comments_not_in_view', 'no_visible_reply'), ('unknown', 'unknown'),
|
||
('no_visible_discussion', 'no_visible_reply')])
|
||
def test_no_viewport_or_unknown_observation_is_not_inspected_even_if_model_claims_success(state, status):
|
||
summary = summarize_preflight(task(observations=[observation(discussion=state, status=status, folded=None)]))
|
||
assert summary['completed'] is False and summary['inspected_notes'] == 0
|
||
assert summary['checked_count'] == summary['eligible_count'] == 0
|
||
assert summary['not_in_view_notes'] + summary['unknown_notes'] == 1
|
||
assert summary['folded_unknown_notes'] == 1
|
||
|
||
|
||
def test_later_unknown_supersedes_an_earlier_successful_observation():
|
||
summary = summarize_preflight(task(observations=[observation(checked=2, eligible=1),
|
||
observation(status='unknown', discussion='unknown', folded=None, index=2)]))
|
||
assert summary['completed'] is False and summary['inspected_notes'] == 0
|
||
assert summary['eligible_count'] == 0 and summary['unknown_notes'] == 1
|
||
|
||
|
||
@pytest.mark.parametrize('change', ['missing_marker', 'old_marker', 'new_empty_run', 'no_receipts', 'no_navigation'])
|
||
def test_missing_current_program_proof_cannot_reuse_historical_model_summary(change):
|
||
row = task()
|
||
if change == 'missing_marker':
|
||
row['artifacts'].pop('preflight_run')
|
||
elif change == 'old_marker':
|
||
row['artifacts']['preflight_run']['goal_version'] = 1
|
||
elif change == 'new_empty_run':
|
||
row['artifacts']['preflight_run']['run_id'] = 'b'*32
|
||
elif change == 'no_receipts':
|
||
row['artifacts']['preflight_observations'] = []
|
||
else:
|
||
row['artifacts']['preflight_navigation'] = []
|
||
summary = summarize_preflight(row)
|
||
assert summary['completed'] is False
|
||
if change != 'no_receipts':
|
||
assert summary['visited_known'] is False and '实际访问数量未记录' in summary['detail']
|
||
assert '实际访问0篇' not in summary['detail']
|
||
if change != 'no_receipts':
|
||
assert summary['scope_passed'] is None
|
||
if change != 'no_navigation':
|
||
assert summary['inspected_notes'] == 0
|
||
|
||
|
||
def test_scope_violation_is_actual_navigation_not_a_preemptively_blocked_entry():
|
||
blocked = task(navigation_rows=[navigation(), navigation(EXCLUDED, status='excluded_note_blocked', index=2),
|
||
navigation(N2, status='note_limit_blocked', index=3), navigation(None, status='unknown_click_blocked', index=4)])
|
||
assert summarize_preflight(blocked)['completed'] is True
|
||
assert summarize_preflight(blocked)['visited_notes'] == 1
|
||
assert '目标无法事前核对的入口' in summarize_preflight(blocked)['detail']
|
||
actual = task(navigation_rows=[navigation(), navigation(EXCLUDED, status='unexpected_note_rejected', index=2)])
|
||
summary = summarize_preflight(actual)
|
||
assert summary['scope_passed'] is False and summary['completed'] is False
|
||
|
||
|
||
def test_actual_excluded_note_and_over_limit_observations_cannot_become_completed():
|
||
excluded = task(observations=[observation(EXCLUDED)], navigation_rows=[navigation(EXCLUDED)])
|
||
assert summarize_preflight(excluded)['scope_passed'] is False
|
||
over = task(observations=[observation(N1), observation(N2, index=2), observation(N3, index=3)],
|
||
navigation_rows=[navigation(N1), navigation(N2, index=2), navigation(N3, index=3)])
|
||
assert summarize_preflight(over)['scope_passed'] is False
|
||
|
||
|
||
@pytest.mark.parametrize('change', ['untracked', 'truncated', 'extra_field', 'bad_count', 'bad_navigation'])
|
||
def test_untracked_truncated_and_malformed_current_metadata_preserve_unknown_scope(change):
|
||
row = task()
|
||
if change == 'untracked':
|
||
row['artifacts']['preflight_navigation'].append(navigation(None, status='scope_history_untracked', index=2))
|
||
elif change == 'truncated':
|
||
row['artifacts']['preflight_observations'][0]['history_truncated'] = True
|
||
elif change == 'extra_field':
|
||
row['artifacts']['preflight_observations'][0]['raw_text'] = 'must not pass through'
|
||
elif change == 'bad_count':
|
||
row['artifacts']['preflight_observations'][0]['eligible_count'] = 1
|
||
else:
|
||
row['artifacts']['preflight_navigation'][0]['action'] = {'raw': 'not a declared action'}
|
||
summary = summarize_preflight(row)
|
||
assert summary['completed'] is False and summary['scope_passed'] is None
|
||
assert 'raw_text' not in repr(summary) and 'must not pass through' not in repr(summary)
|
||
|
||
|
||
def test_one_valid_note_does_not_hide_another_visited_note_without_a_real_preflight():
|
||
summary = summarize_preflight(task(navigation_rows=[navigation(), navigation(N2, index=2)]))
|
||
assert summary['completed'] is False and summary['inspected_notes'] == 1
|
||
assert summary['unknown_notes'] == 1 and summary['visited_notes'] == 2
|
||
|
||
|
||
def test_folded_state_unknown_is_reported_without_inventing_a_total_reply_count():
|
||
summary = summarize_preflight(task(observations=[observation(folded=None)]))
|
||
assert summary['completed'] is True and summary['folded_unknown_notes'] == 1
|
||
assert summary['folded_notes'] == 0 and '折叠状态尚未确认' in summary['detail']
|
||
|
||
|
||
def test_failed_browser_run_is_not_upgraded_by_valid_receipts():
|
||
row = task()
|
||
row['artifacts']['browser_verdict'] = 'failed'
|
||
assert summarize_preflight(row)['completed'] is False
|
||
assert delivery_projection(row)['status'] == 'partial'
|
||
|
||
|
||
def test_projection_does_not_apply_to_regular_collection_or_use_sources_as_preflight_proof():
|
||
row = task()
|
||
row['inputs']['goal'] = '请阅读当前笔记并保存来源。'
|
||
assert summarize_preflight(row) is None and delivery_projection(row) is None
|
||
row = task(observations=[], navigation_rows=[])
|
||
row['artifacts']['sources'] = [{'source_id': 'old', 'content': 'old evidence'}]
|
||
row['artifacts']['leads'] = [{'source_id': 'old', 'quote': 'old evidence'}]
|
||
assert delivery_projection(row)['status'] == 'no_evidence'
|
||
assert delivery_projection(row)['counts']['eligible'] == 0
|