131 lines
6.4 KiB
Python
131 lines
6.4 KiB
Python
"""An explicitly requested reply-to-reply branch needs new, linked evidence."""
|
|
from copy import deepcopy
|
|
import hashlib
|
|
|
|
import pytest
|
|
|
|
from backend.prospecting.discussion_scope import conditions, requirements
|
|
|
|
|
|
NOTE = 'a' * 24
|
|
URL = 'https://www.xiaohongshu.com/explore/' + NOTE
|
|
GOAL = '先保存目前已显示的父回复，再展开同组回复，保存一条明确回复该父回复的新增子回复。'
|
|
|
|
|
|
def task(goal=GOAL, **inputs):
|
|
return {'inputs':{'goal':goal,'intent':'inspect','subject_urls':[URL],'target_candidates':None,
|
|
'_goal_version':1,'max_comments_per_note':5,**inputs},
|
|
'artifacts':{'messages':[{'role':'user','text':goal}],'sources':[],'operation_evidence':[]}}
|
|
|
|
|
|
def source(row, sid, kind, parent=None, root=None, note_id=NOTE):
|
|
item = {'source_id':sid,'source_type':kind,'note_id':note_id,'source_url':URL,
|
|
'source_selector':'#'+sid,'content':'原话'+sid,'parent_source_id':parent,'root_source_id':root}
|
|
row['artifacts']['sources'].append(item)
|
|
if kind != 'note':
|
|
row['artifacts']['operation_evidence'].append({
|
|
'capability':'save_source','source_kind':kind,'outcome':'success',
|
|
'verifier':'source_container_quote_and_persisted','evidence_refs':{'pre':'before','post':'after'},
|
|
'details':{'trace':{'goal_version':1},'pre_snapshot':{'note_id':note_id,'container_selector':'#'+sid,'quote_matched':True},
|
|
'post_snapshot':{'source_id':sid,'source_type':kind,'persisted':True,'relations_checked':True,
|
|
'parent_source_id':parent,'root_source_id':root}}})
|
|
return item
|
|
|
|
|
|
def expansion(row, reply):
|
|
fact = {'capability':'expand_replies','source_kind':'reply','outcome':'success','verifier':'reply_delta_and_parent',
|
|
'evidence_refs':{'pre':'before-expand','post':'after-expand'},
|
|
'details':{'trace':{'goal_version':1},'pre_snapshot':{'note_id':NOTE},
|
|
'post_snapshot':{'note_id':NOTE,'parent_scope_checked':True,'status':'expanded','owner_reply_delta':1,
|
|
'new_visible_replies':1,'new_reply_fingerprints':[
|
|
hashlib.sha256((reply['source_selector']+reply['content']).encode()).hexdigest()]}}}
|
|
row['artifacts']['operation_evidence'].append(fact)
|
|
return fact
|
|
|
|
|
|
def ready(*, nested=True):
|
|
row = task()
|
|
source(row,'note','note')
|
|
source(row,'root','comment')
|
|
source(row,'parent','reply',parent='root',root='root')
|
|
child = source(row,'child','reply',parent='parent' if nested else 'root',root='root')
|
|
expansion(row,child)
|
|
return row
|
|
|
|
|
|
def depth(row):
|
|
return next(c for c in conditions(row) if c['key']=='discussion_reply_to_reply')
|
|
|
|
|
|
@pytest.mark.parametrize('goal', [GOAL,'保存一条回复另一条回复的新增子回复。','保存回复父回复的子回复。'])
|
|
def test_explicit_depth_wording_requires_this_specific_branch(goal):
|
|
row = task(goal)
|
|
assert 'reply_to_reply' in requirements(row)
|
|
assert depth(row)['required'] and not depth(row)['met']
|
|
|
|
|
|
@pytest.mark.parametrize('goal', ['核对完整回复链。','核对发言人、直接父评论或父回复、根讨论。','保存父评论和子回复。'])
|
|
def test_generic_complete_chain_or_optional_parent_kind_does_not_add_depth(goal):
|
|
assert 'reply_to_reply' not in requirements(task(goal))
|
|
|
|
|
|
def test_root_to_direct_reply_does_not_fulfill_reply_to_reply_branch():
|
|
row = ready(nested=False)
|
|
assert next(c for c in conditions(row) if c['key']=='discussion_expansion')['met']
|
|
assert not depth(row)['met']
|
|
|
|
|
|
def test_actual_new_child_and_observed_saved_parent_prove_depth_without_mutation():
|
|
row = ready()
|
|
before = deepcopy(row)
|
|
assert depth(row)['met'] and depth(row)['current']==depth(row)['target']==1
|
|
assert row==before and row['inputs']['max_comments_per_note']==5
|
|
|
|
|
|
@pytest.mark.parametrize('change', [
|
|
lambda row: row['artifacts']['operation_evidence'][-1]['details']['trace'].update(goal_version=0),
|
|
lambda row: row['artifacts']['operation_evidence'][-1]['details']['post_snapshot'].update(new_reply_fingerprints=['not-this-child']),
|
|
lambda row: row['artifacts']['operation_evidence'][2]['details']['trace'].update(goal_version=0),
|
|
lambda row: row['artifacts']['operation_evidence'][1]['details']['trace'].update(goal_version=0),
|
|
lambda row: row['artifacts']['operation_evidence'][2]['details']['post_snapshot'].update(relations_checked=False),
|
|
lambda row: row['artifacts']['operation_evidence'][1].update(outcome='unknown'),
|
|
lambda row: row['artifacts']['operation_evidence'][2]['details']['post_snapshot'].update(parent_source_id='root'),
|
|
lambda row: row['artifacts']['sources'][-2].update(parent_source_id='child'),
|
|
lambda row: row['artifacts']['sources'][-2].update(note_id='other-note'),
|
|
lambda row: row['artifacts']['operation_evidence'].pop(),
|
|
])
|
|
def test_old_unrelated_weak_or_cyclic_proof_cannot_establish_requested_depth(change):
|
|
row = ready()
|
|
change(row)
|
|
assert not depth(row)['met']
|
|
|
|
|
|
def test_expanding_a_different_direct_reply_does_not_prove_already_visible_nested_reply_is_new():
|
|
row = ready()
|
|
old = source(row,'new-direct','reply',parent='root',root='root')
|
|
row['artifacts']['operation_evidence'] = [f for f in row['artifacts']['operation_evidence'] if f['capability']!='expand_replies']
|
|
expansion(row,old)
|
|
assert not depth(row)['met']
|
|
|
|
|
|
@pytest.mark.parametrize('goal', ['不要保存回复该父回复的新增子回复。','只保存正文。比如回复另一条回复。',
|
|
'只保存正文。\n> 保存回复父回复的子回复。','只保存正文。请忽略引用“回复该父回复”。'])
|
|
def test_negated_or_quoted_depth_does_not_expand_user_scope(goal):
|
|
assert 'reply_to_reply' not in requirements(task(goal))
|
|
|
|
|
|
def test_later_cancellation_removes_only_explicit_depth():
|
|
row = ready()
|
|
row['artifacts']['messages'].append({'role':'user','text':'不用保存回复该父回复的新增子回复。'})
|
|
assert 'reply_to_reply' not in requirements(row)
|
|
assert 'expansion' in requirements(row)
|
|
|
|
|
|
def test_depth_requires_each_saved_note_in_current_requested_scope():
|
|
row = ready()
|
|
other_note='b'*24
|
|
row['inputs']['subject_urls'].append('https://www.xiaohongshu.com/explore/'+other_note)
|
|
source(row,'second-note','note',note_id=other_note)
|
|
assert not depth(row)['met']
|
|
assert (depth(row)['current'],depth(row)['target'])==(1,2)
|