244 lines
13 KiB
Python
244 lines
13 KiB
Python
"""Actual DOM observer tests for explicit reply targets, never nickname guesses."""
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from backend.browser import tools
|
|
|
|
|
|
FIXTURE = Path(__file__).with_name('reply_dom_fixture_d41.cjs')
|
|
PROFILE = lambda letter: 'https://www.xiaohongshu.com/user/profile/' + letter * 24
|
|
|
|
|
|
def node(identity='', classes='', text='', *children, tag='div', **attrs):
|
|
return {'tag': tag, 'attrs': {'id': identity, 'class': classes, **attrs}, 'text': text, 'children': list(children)}
|
|
|
|
|
|
def speaker(identity, person, content, *, kind='reply', extra=None, **attrs):
|
|
author = node(identity + '-author', 'name', '同名发言人', tag='a', href=PROFILE(person))
|
|
return node(identity, 'comment-item' if kind == 'comment' else 'comment-item-sub', '',
|
|
node('', 'author-wrapper', '', author), content, *(extra or []), **{'data-comment-id': identity + '-key', **attrs})
|
|
|
|
|
|
def content(identity, text='独立原话', *children):
|
|
return node(identity + '-content', 'content', '', node(identity + '-text', 'note-text', text, *children))
|
|
|
|
|
|
def targeted(identity='child', person='c', *, href=True, target_attrs=None):
|
|
target = node('target', 'user-name', '同名发言人', tag='a' if href else 'span',
|
|
**({'href': PROFILE(person)} if href else {}), **(target_attrs or {}))
|
|
return content(identity, '回复 ', target, node('', '', ': 新的独立需求'))
|
|
|
|
|
|
def thread(*replies, identity='thread', root_person='b'):
|
|
return node(identity, 'parent-comment', '',
|
|
speaker(identity + '-root', root_person, content(identity + '-root'), kind='comment'),
|
|
node('', 'reply-container', '', *replies))
|
|
|
|
|
|
def observe(*groups, selector='#child-content', verified_target_identity=None):
|
|
tree = node('noteContainer', '', '', node('detail-desc', '', '正文'), *groups)
|
|
arg = {'selector': selector, 'verified_target_identity': verified_target_identity} if verified_target_identity else selector
|
|
result = subprocess.run(['node', str(FIXTURE)], input=json.dumps(
|
|
{'code': tools._OBSERVE_SOURCE, 'arg': arg, 'tree': tree}, ensure_ascii=False),
|
|
text=True, capture_output=True, check=True, timeout=5)
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def parent_reply(identity='parent', person='c'):
|
|
return speaker(identity, person, content(identity))
|
|
|
|
|
|
def test_leading_target_profile_uniquely_resolves_a_flat_parent_reply_with_same_display_names():
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', targeted())))
|
|
assert result['parent_source_selector'] == '#parent-content'
|
|
assert result['root_source_selector'] == '#thread-root-content'
|
|
assert result['parent_resolution'] == 'unique_target_profile'
|
|
assert result['parent_source_key'] == 'parent-key'
|
|
assert result['reply_to'] == '同名发言人' and result['reply_to_selector'] == '#target'
|
|
assert result['relation_diagnostics']['target_profile_links'] == 1
|
|
|
|
|
|
def test_two_comments_by_the_target_person_are_ambiguous_even_if_the_profile_is_known():
|
|
result = observe(thread(parent_reply(), parent_reply('another'), speaker('child', 'd', targeted())))
|
|
assert result['parent_source_selector'] == ''
|
|
assert result['parent_resolution'] == 'ambiguous_or_missing_target_comment'
|
|
assert result['relation_diagnostics']['matching_parent_comments'] == 2
|
|
|
|
|
|
def test_a_same_named_root_cannot_replace_a_target_profile_absent_from_the_thread():
|
|
result = observe(thread(parent_reply(person='e'), speaker('child', 'd', targeted())))
|
|
assert result['root_source_selector'] == '#thread-root-content'
|
|
assert result['parent_source_selector'] == ''
|
|
|
|
|
|
def test_explicit_user_identity_on_a_leading_target_span_is_observed_identity():
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', targeted(href=False, target_attrs={'data-user-id': 'c' * 24}))))
|
|
assert result['parent_source_selector'] == '#parent-content'
|
|
assert result['parent_resolution'] == 'unique_target_profile'
|
|
|
|
|
|
def test_plain_target_span_has_a_real_selector_and_display_value_but_no_guessed_identity():
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', targeted(href=False))))
|
|
assert result['parent_source_selector'] == ''
|
|
assert result['parent_resolution'] == 'target_identity_missing'
|
|
assert result['reply_to'] == '同名发言人' and result['reply_to_selector'] == '#target'
|
|
assert result['relation_diagnostics']['target_elements'] == 1
|
|
assert result['relation_diagnostics']['target_identities'] == 0
|
|
|
|
|
|
def test_ordinary_profile_mention_after_the_message_is_not_a_reply_target():
|
|
quote = content('child', '我提到一个人 ', node('mention', '', '同名发言人', tag='a', href=PROFILE('c')))
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', quote)))
|
|
assert result['parent_source_selector'] == '#thread-root-content'
|
|
assert result['parent_resolution'] == 'thread_root'
|
|
assert result['reply_to'] == '' and result['reply_to_selector'] == ''
|
|
|
|
|
|
def test_external_profile_lookalike_does_not_establish_target_identity():
|
|
quote = targeted()
|
|
quote['children'][0]['children'][0]['attrs']['href'] = 'https://evil.test/user/profile/' + 'c' * 24
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', quote)))
|
|
assert result['parent_source_selector'] == ''
|
|
assert result['relation_diagnostics']['target_profile_links'] == 0
|
|
|
|
|
|
def test_explicit_comment_reference_conflicting_with_the_observed_target_profile_is_rejected():
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', targeted(person='e'), **{'data-parent-comment-id': 'parent-key'})))
|
|
assert result['parent_source_selector'] == ''
|
|
assert result['parent_resolution'] == 'conflicting_or_missing_comment_reference'
|
|
|
|
|
|
def test_cross_thread_target_person_does_not_establish_parent_in_the_current_thread():
|
|
result = observe(thread(parent_reply(), identity='other'), thread(speaker('child', 'd', targeted()), identity='current'))
|
|
assert result['parent_source_selector'] == ''
|
|
assert result['root_source_selector'] == '#current-root-content'
|
|
|
|
|
|
def test_target_profile_and_explicit_comment_id_can_agree():
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', targeted(), **{'data-parent-comment-id': 'parent-key'})))
|
|
assert result['parent_source_selector'] == '#parent-content'
|
|
assert result['parent_resolution'] == 'explicit_comment_id'
|
|
|
|
|
|
def test_nested_root_wrapper_does_not_override_an_explicit_reply_to_another_child():
|
|
first = parent_reply()
|
|
second = speaker('child', 'd', targeted())
|
|
root = speaker('root', 'b', content('root'), kind='comment', extra=[first, second])
|
|
result = observe(root)
|
|
assert result['parent_source_selector'] == '#parent-content'
|
|
assert result['root_source_selector'] == '#root-content'
|
|
|
|
|
|
def test_nested_wrapper_with_only_a_target_name_does_not_guess_parent():
|
|
marker = node('target', 'reply-to', '同名发言人')
|
|
child = speaker('child', 'd', content('child'), extra=[marker])
|
|
root = speaker('root', 'b', content('root'), kind='comment', extra=[child])
|
|
result = observe(root)
|
|
assert result['parent_source_selector'] == ''
|
|
assert result['root_source_selector'] == '#root-content'
|
|
assert result['parent_resolution'] == 'target_identity_missing'
|
|
|
|
|
|
def test_explicit_target_marker_outside_content_can_supply_a_unique_profile():
|
|
marker = node('target', 'reply-to-name', '同名发言人', tag='a', href=PROFILE('c'))
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', content('child'), extra=[marker])))
|
|
assert result['parent_source_selector'] == '#parent-content'
|
|
assert result['reply_to'] == '同名发言人'
|
|
|
|
|
|
def test_conflicting_target_profile_and_user_attribute_remain_unknown():
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', targeted(target_attrs={'data-user-id': 'e' * 24}))))
|
|
assert result['parent_source_selector'] == ''
|
|
assert result['parent_resolution'] == 'conflicting_target_profiles'
|
|
|
|
|
|
def test_program_native_profile_identity_can_resolve_a_real_leading_span_without_href():
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', targeted(href=False))), verified_target_identity='c' * 24)
|
|
assert result['parent_resolution'] == 'native_target_profile'
|
|
assert result['parent_source_selector'] == '#parent-content'
|
|
assert result['parent_source_key'] == 'parent-key'
|
|
|
|
|
|
def test_native_profile_identity_does_not_override_a_conflicting_real_target_link():
|
|
result = observe(thread(parent_reply(), speaker('child', 'd', targeted())), verified_target_identity='e' * 24)
|
|
assert result['parent_source_selector'] == ''
|
|
assert result['parent_resolution'] == 'conflicting_target_profiles'
|
|
|
|
|
|
def test_native_identity_still_rejects_multiple_comments_by_the_same_target_person():
|
|
result = observe(thread(parent_reply(), parent_reply('another'), speaker('child', 'd', targeted(href=False))),
|
|
verified_target_identity='c' * 24)
|
|
assert result['parent_source_selector'] == ''
|
|
assert result['parent_resolution'] == 'ambiguous_or_missing_target_comment'
|
|
|
|
|
|
class Page:
|
|
def __init__(self, source_type, text, **metadata):
|
|
self.source_type, self.text, self.metadata = source_type, text, metadata
|
|
|
|
async def evaluate(self, _code, _selector):
|
|
return json.dumps({'url': 'https://www.xiaohongshu.com/explore/' + 'a' * 24,
|
|
'text': self.text, 'source_type': self.source_type, **self.metadata})
|
|
|
|
|
|
class Session:
|
|
def __init__(self, page): self.page = page
|
|
async def get_current_page(self): return self.page
|
|
|
|
|
|
@pytest.fixture
|
|
def context(tmp_path, monkeypatch):
|
|
task_id = 'reply_attribution_d42'
|
|
monkeypatch.setattr(tools, 'task_doc_dir', lambda _: tmp_path)
|
|
tools.set_prospecting_context(task_id)
|
|
yield task_id
|
|
tools.clear_task_context(task_id)
|
|
|
|
|
|
async def save(task, kind, selector, quote, **metadata):
|
|
return await tools._save_source(task, Session(Page(kind, quote, **metadata)),
|
|
source_type=kind, source_selector=selector, content=quote)
|
|
|
|
|
|
async def test_saving_a_later_parent_does_not_backfill_old_reply_without_reobservation(context):
|
|
await save(context, 'note', '#note', '正文')
|
|
reply, _ = await save(context, 'reply', '#reply', '回复原话', source_key='reply-key',
|
|
parent_source_selector='#parent', root_source_selector='#parent', parent_resolution='unique_target_profile')
|
|
old_fact = tools.get_task_artifacts(context)['operation_evidence'][-1]
|
|
root, _ = await save(context, 'comment', '#parent', '父评论')
|
|
assert reply['parent_source_id'] is None and reply['root_source_id'] is None
|
|
again, created = await save(context, 'reply', '#reply', '回复原话', source_key='reply-key',
|
|
parent_source_selector='#parent', root_source_selector='#parent', parent_resolution='unique_target_profile')
|
|
assert not created and again['source_id'] == reply['source_id']
|
|
assert again['parent_source_id'] == again['root_source_id'] == root['source_id']
|
|
assert old_fact['outcome'] == 'success' and old_fact['details']['post_snapshot']['parent_source_id'] is None
|
|
assert old_fact['details']['post_snapshot']['relations_checked'] is False
|
|
assert tools.get_task_artifacts(context)['operation_evidence'][-1]['outcome'] == 'success'
|
|
|
|
|
|
async def test_two_saved_parent_excerpts_at_the_same_selector_are_not_arbitrarily_linked(context):
|
|
await save(context, 'note', '#note', '正文')
|
|
await save(context, 'comment', '#parent', '原话一', source_key='key-one')
|
|
await save(context, 'comment', '#parent', '原话二', source_key='key-two')
|
|
reply, _ = await save(context, 'reply', '#reply', '回复', parent_source_selector='#parent', root_source_selector='#parent')
|
|
assert reply['parent_source_id'] is None and reply['root_source_id'] is None
|
|
|
|
|
|
async def test_dom_reordered_selector_does_not_reuse_a_different_saved_parent_key(context):
|
|
await save(context, 'note', '#note', '正文')
|
|
root, _ = await save(context, 'comment', '#parent', '原根评论', source_key='root-key')
|
|
reply, _ = await save(context, 'reply', '#reply', '新的回复', parent_source_selector='#parent',
|
|
root_source_selector='#parent', parent_source_key='different-parent-key', root_source_key='root-key')
|
|
assert reply['parent_source_id'] is None
|
|
assert reply['root_source_id'] == root['source_id']
|
|
|
|
|
|
async def test_same_quote_and_selector_with_changed_comment_key_is_a_different_observation(context):
|
|
await save(context, 'note', '#note', '正文')
|
|
first, _ = await save(context, 'comment', '#parent', '相同原话', source_key='first-key')
|
|
second, created = await save(context, 'comment', '#parent', '相同原话', source_key='second-key')
|
|
assert created and second['source_id'] != first['source_id']
|