243 lines
12 KiB
Python
243 lines
12 KiB
Python
"""Execute real read-only expansion observers on separate discussion groups."""
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from backend.browser import tools
|
|
|
|
|
|
FIXTURE = Path(__file__).with_name('reply_dom_fixture_d41.cjs')
|
|
|
|
|
|
def el(identity, classes='', text='', *children):
|
|
return {'attrs': {'id': identity, 'class': classes}, 'text': text, 'children': list(children)}
|
|
|
|
|
|
def record(identity, reply=False):
|
|
return el(identity, 'comment-item-sub' if reply else 'comment-item', '',
|
|
el(identity + '-content', 'content', '', el(identity + '-text', 'note-text', '独立原话')))
|
|
|
|
|
|
def group(identity, replies=1):
|
|
return el(identity, 'parent-comment', '', record(identity + '-root'),
|
|
el(identity + '-list', 'reply-container', '',
|
|
*(record(identity + '-reply-' + str(i), True) for i in range(replies)),
|
|
el(identity + '-expand', 'show-more', '展开3条回复')))
|
|
|
|
|
|
def observe(code, arg, *groups):
|
|
tree = el('noteContainer', '', '', el('detail-desc', '', '正文'), *groups)
|
|
result = subprocess.run(['node', str(FIXTURE)], text=True, capture_output=True, check=True, timeout=5,
|
|
input=json.dumps({'code': code, 'arg': arg, 'tree': tree}, ensure_ascii=False))
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def test_div_control_in_sibling_reply_list_is_discovered_without_text_click_guess():
|
|
result = observe(tools._INSPECT_SOURCE_CANDIDATES, {'max_comments': 5}, group('one'))
|
|
assert result['reply_controls'] == [{'source_selector': '#one-expand', 'text': '展开3条回复', 'in_view': True}]
|
|
|
|
|
|
def test_expansion_observes_the_whole_unique_thread_not_only_its_root_record():
|
|
before = observe(tools._OBSERVE_REPLY_CONTROL, {'selector': '#one-expand'}, group('one'), group('two', 4))
|
|
after = observe(tools._OBSERVE_REPLY_OWNER, {'selector': before['owner_selector']}, group('one', 2), group('two', 7))
|
|
assert before['owner_selector'] == '#one'
|
|
assert before['owner_identity'] == after['owner_identity'] == 'one-root'
|
|
assert before['reply_count'] == 1 and after['reply_count'] == 2
|
|
|
|
|
|
def test_candidate_delta_is_scoped_to_observed_discussion_even_with_other_new_replies():
|
|
result = observe(tools._INSPECT_SOURCE_CANDIDATES,
|
|
{'max_comments': 5, 'discussion_selector': '#one'}, group('one'), group('two', 4))
|
|
assert {row['source_selector'] for row in result['candidates']} == {'#one-root-content', '#one-reply-0-content'}
|
|
assert len(result['reply_controls']) == 1
|
|
|
|
|
|
def test_ambiguous_thread_does_not_choose_a_root_by_position():
|
|
ambiguous = el('bad', 'parent-comment', '', record('first'), record('second'),
|
|
el('expand', 'show-more', '展开3条回复'))
|
|
result = observe(tools._OBSERVE_REPLY_CONTROL, {'selector': '#expand'}, ambiguous)
|
|
assert result['error'] == 'discussion_owner_unresolved'
|
|
|
|
|
|
def test_reply_scope_disappeared_is_rejected_instead_of_reading_all_discussions():
|
|
result = observe(tools._INSPECT_SOURCE_CANDIDATES,
|
|
{'max_comments': 5, 'discussion_selector': '#absent'}, group('two'))
|
|
assert result['error'] == 'discussion_scope_not_unique'
|
|
|
|
|
|
def test_collapse_label_cannot_be_used_as_a_new_expansion():
|
|
expanded = group('one')
|
|
expanded['children'][1]['children'][-1]['text'] = '收起回复'
|
|
result = observe(tools._OBSERVE_REPLY_CONTROL, {'selector': '#one-expand'}, expanded)
|
|
assert result['error'] == 'not_expand_reply_control'
|
|
|
|
|
|
def test_rendered_offscreen_control_is_only_a_navigation_target_not_a_clickable_claim():
|
|
tree = group('one')
|
|
tree['children'][1]['children'][-1]['rect'] = {'left': 0, 'right': 300, 'top': 800,
|
|
'bottom': 835, 'width': 300, 'height': 35}
|
|
result = observe(tools._INSPECT_SOURCE_CANDIDATES, {'max_comments': 5}, tree)
|
|
assert result['reply_controls'] == [{'source_selector': '#one-expand', 'text': '展开3条回复', 'in_view': False, 'position': 'below'}]
|
|
assert result['reply_control_diagnostics']['rendered_matches'] > 0
|
|
assert result['reply_control_diagnostics']['viewport_matches'] == 0
|
|
click_guard = observe(tools._OBSERVE_REPLY_CONTROL, {'selector': '#one-expand'}, tree)
|
|
assert click_guard['error'] == 'reply_control_not_visible'
|
|
|
|
|
|
def test_full_before_snapshot_contains_more_replies_than_one_five_source_batch():
|
|
result = observe(tools._OBSERVE_REPLY_CONTROL, {'selector': '#one-expand'}, group('one', 8))
|
|
assert result['reply_snapshot_complete'] is True
|
|
assert result['reply_count'] == 8
|
|
assert set(result['visible_reply_selectors']) == {'#one-reply-' + str(i) + '-content' for i in range(8)}
|
|
|
|
|
|
def test_before_snapshot_over_hundred_records_is_explicitly_incomplete():
|
|
result = observe(tools._OBSERVE_REPLY_CONTROL, {'selector': '#one-expand'}, group('one', 101))
|
|
assert result['reply_count'] == 101 and len(result['visible_reply_selectors']) == 100
|
|
assert result['reply_snapshot_complete'] is False
|
|
|
|
|
|
def test_duplicate_root_text_cannot_stand_in_for_stable_discussion_identity():
|
|
first, second = group('one'), group('two')
|
|
first['children'][0]['attrs']['id'] = ''
|
|
result = observe(tools._OBSERVE_REPLY_CONTROL, {'selector': '#one-expand'}, first, second)
|
|
assert result['error'] == 'discussion_owner_unresolved'
|
|
|
|
|
|
def test_unique_root_text_can_identify_a_thread_without_a_root_record_id():
|
|
first, second = group('one'), group('two')
|
|
first['children'][0]['attrs']['id'] = ''
|
|
second['children'][0]['children'][0]['children'][0]['text'] = '其他根评论'
|
|
result = observe(tools._OBSERVE_REPLY_CONTROL, {'selector': '#one-expand'}, first, second)
|
|
assert result['owner_identity'] == 'unique-root-text:独立原话'
|
|
assert result['owner_selector'] == '#one'
|
|
|
|
|
|
class BatchedExpansionPage:
|
|
"""A reading batch can omit old visible replies that the full snapshot saw."""
|
|
url = 'https://www.xiaohongshu.com/explore/' + 'a' * 24
|
|
|
|
def __init__(self, before_complete=True, after_complete=True):
|
|
self.clicked = False
|
|
self.before_complete, self.after_complete = before_complete, after_complete
|
|
|
|
async def get_url(self):
|
|
return self.url
|
|
|
|
async def evaluate(self, code, options):
|
|
if code in {tools._OBSERVE_REPLY_CONTROL, tools._OBSERVE_REPLY_OWNER}:
|
|
return json.dumps({'url': self.url, 'text': '展开1条回复', 'owner_selector': '#thread',
|
|
'click_point': {'x':100,'y':80,'viewport_width':400,'viewport_height':600},
|
|
'owner_identity': 'stable-root', 'fingerprint': str(self.clicked),
|
|
'reply_count': 6 + int(self.clicked),
|
|
'visible_reply_selectors': ['#old-' + str(i) for i in range(6)] + (['#new'] if self.clicked else []),
|
|
'reply_snapshot_complete': self.after_complete if self.clicked else self.before_complete})
|
|
assert code == tools._INSPECT_SOURCE_CANDIDATES
|
|
assert options['discussion_selector'] == '#thread'
|
|
candidates = [{'source_type': 'reply', 'source_selector': '#old-' + str(i), 'content': '旧回复' + str(i)} for i in range(5)]
|
|
if self.clicked:
|
|
candidates = [{'source_type': 'reply', 'source_selector': '#old-5', 'content': '旧回复5'},
|
|
{'source_type': 'reply', 'source_selector': '#new', 'content': '实际新增'}]
|
|
return json.dumps({'url': self.url, 'candidates': candidates})
|
|
|
|
async def get_elements_by_css_selector(self, selector):
|
|
assert selector == '#expand'
|
|
async def click():
|
|
raise AssertionError('Native current-point click must not implicitly scroll')
|
|
return [SimpleNamespace(click=click)]
|
|
|
|
@property
|
|
async def mouse(self):
|
|
async def click(**point):
|
|
assert point == {'x':100,'y':80}
|
|
self.clicked = True
|
|
return SimpleNamespace(click=click)
|
|
|
|
|
|
@pytest.fixture
|
|
def expansion_context(tmp_path, monkeypatch):
|
|
task_id = 'd41-expansion-snapshot-fixture'
|
|
monkeypatch.setattr(tools, 'task_doc_dir', lambda _: tmp_path)
|
|
tools.clear_task_context(task_id)
|
|
tools.set_prospecting_context(task_id)
|
|
yield task_id
|
|
tools.clear_task_context(task_id)
|
|
|
|
|
|
async def test_old_reply_outside_reading_batch_is_not_new_expansion_evidence(expansion_context):
|
|
page = BatchedExpansionPage()
|
|
async def current():
|
|
return page
|
|
result = await tools._expand_replies(expansion_context, SimpleNamespace(get_current_page=current), '#expand')
|
|
assert result['new_visible_replies'] == 1
|
|
fact = tools.get_task_artifacts(expansion_context)['operation_evidence'][-1]
|
|
assert fact['outcome'] == 'success'
|
|
assert fact['details']['pre_snapshot']['reply_snapshot_complete'] is True
|
|
import hashlib
|
|
assert fact['details']['post_snapshot']['new_reply_fingerprints'] == [hashlib.sha256('#new实际新增'.encode()).hexdigest()]
|
|
|
|
|
|
@pytest.mark.parametrize('before_complete,after_complete', [(False, True), (True, False), (False, False)])
|
|
async def test_incomplete_discussion_snapshot_never_credits_expansion_success(expansion_context, before_complete, after_complete):
|
|
page = BatchedExpansionPage(before_complete, after_complete)
|
|
async def current():
|
|
return page
|
|
await tools._expand_replies(expansion_context, SimpleNamespace(get_current_page=current), '#expand')
|
|
fact = tools.get_task_artifacts(expansion_context)['operation_evidence'][-1]
|
|
assert fact['outcome'] == 'unknown'
|
|
|
|
|
|
class ContainerObservedExpansionPage(BatchedExpansionPage):
|
|
def __init__(self, failure=None):
|
|
super().__init__()
|
|
self.failure = failure
|
|
self.container_observations = 0
|
|
|
|
async def evaluate(self, code, options):
|
|
if code != tools._OBSERVE_SOURCE:
|
|
return await super().evaluate(code, options)
|
|
assert options == '#new'
|
|
self.container_observations += 1
|
|
if self.failure == 'exception':
|
|
raise RuntimeError('isolated observer unavailable')
|
|
return json.dumps({'url': self.url if self.failure != 'note' else 'https://www.xiaohongshu.com/explore/' + 'b' * 24,
|
|
'text': '不相关原话' if self.failure == 'quote' else '实际新增，\n还有完整上下文。',
|
|
'source_type': 'comment' if self.failure == 'kind' else 'reply'})
|
|
|
|
|
|
async def test_expansion_program_records_full_normalized_container_fingerprint(expansion_context):
|
|
page = ContainerObservedExpansionPage()
|
|
async def current():
|
|
return page
|
|
await tools._expand_replies(expansion_context, SimpleNamespace(get_current_page=current), '#expand')
|
|
fact = tools.get_task_artifacts(expansion_context)['operation_evidence'][-1]
|
|
import hashlib
|
|
expected = hashlib.sha256(('a' * 24 + '\n#new\n实际新增，还有完整上下文。').encode()).hexdigest()
|
|
assert fact['details']['post_snapshot']['new_reply_container_fingerprints'] == [expected]
|
|
assert page.container_observations == 1 and page.clicked
|
|
|
|
|
|
@pytest.mark.parametrize('failure', ['exception', 'note', 'quote', 'kind'])
|
|
async def test_failed_extra_container_observation_does_not_retry_or_abort_native_expansion(expansion_context, failure):
|
|
page = ContainerObservedExpansionPage(failure)
|
|
async def current():
|
|
return page
|
|
result = await tools._expand_replies(expansion_context, SimpleNamespace(get_current_page=current), '#expand')
|
|
fact = tools.get_task_artifacts(expansion_context)['operation_evidence'][-1]
|
|
assert result['status'] == 'expanded' and fact['outcome'] == 'success'
|
|
assert fact['details']['post_snapshot']['new_reply_container_fingerprints'] == []
|
|
assert page.container_observations == 1 and page.clicked
|
|
|
|
|
|
def test_visible_control_has_priority_over_ten_earlier_offscreen_controls():
|
|
groups = [group('thread-' + str(i)) for i in range(11)]
|
|
for tree in groups[:10]:
|
|
tree['children'][1]['children'][-1]['rect'] = {'left':0,'right':300,'top':800,'bottom':835,'width':300,'height':35}
|
|
result = observe(tools._INSPECT_SOURCE_CANDIDATES, {'max_comments': 5}, *groups)
|
|
assert len(result['reply_controls']) == 10
|
|
assert result['reply_controls'][0]['source_selector'] == '#thread-10-expand'
|
|
assert result['reply_controls'][0]['in_view'] is True
|