210 lines
9.4 KiB
Python
210 lines
9.4 KiB
Python
"""Reveal a proved expand entry with bounded native input before its baseline."""
|
|
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')
|
|
URL = 'https://www.xiaohongshu.com/explore/' + 'a' * 24
|
|
|
|
|
|
def rect(top, bottom, left=0, right=400):
|
|
return dict(left=left, right=right, top=top, bottom=bottom,
|
|
width=right-left, height=bottom-top)
|
|
|
|
|
|
def discussion(*, target_top=500, document_scroll=False, occluded=False):
|
|
control = {'attrs': {'id': 'expand'}, 'text': '展开1条回复',
|
|
'rect': rect(target_top, target_top+30, 20, 200)}
|
|
root = {'attrs': {'id': 'root', 'class': 'comment-item'}, 'children': [
|
|
{'attrs': {'id': 'root-text', 'class': 'content'}, 'text': '根评论'}]}
|
|
thread = {'attrs': {'id': 'thread', 'class': 'parent-comment'}, 'children': [
|
|
root, {'attrs': {'class': 'reply-container'}, 'children': [control]}]}
|
|
inner = {'attrs': {'id': 'inner-scroll'}, 'style': {'overflowY': 'auto'},
|
|
'rect': rect(100, 300), 'clientHeight': 200, 'scrollHeight': 1200,
|
|
'scrollTop': 100, 'children': [thread]}
|
|
outer = {'attrs': {'id': 'outer-scroll'}, 'style': {'overflowY': 'auto'},
|
|
'rect': rect(0, 500), 'clientHeight': 500, 'scrollHeight': 2000,
|
|
'scrollTop': 100, 'children': [inner]}
|
|
result = {'attrs': {'id': 'noteContainer'}, 'rect': rect(0, 1200),
|
|
'children': [thread if document_scroll else outer],
|
|
'hit_target_id': 'noteContainer' if document_scroll else 'inner-scroll',
|
|
'occluded': occluded}
|
|
if document_scroll:
|
|
result['document_scroll'] = {'scrollTop': 100, 'clientHeight': 600,
|
|
'scrollHeight': 2000, 'rect': rect(0, 0)}
|
|
return result
|
|
|
|
|
|
def observe(value):
|
|
result = subprocess.run(['node', str(FIXTURE)], input=json.dumps({
|
|
'code': tools._OBSERVE_REPLY_CONTROL, 'arg': {'selector': '#expand'}, 'tree': value}),
|
|
capture_output=True, text=True, check=True, timeout=5)
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def test_hidden_control_uses_deepest_actual_scroll_range_with_owned_anchor():
|
|
result = observe(discussion())
|
|
assert result['owner_identity'] == 'root'
|
|
assert result['owner_selector'] == '#thread'
|
|
assert result['error'] == 'reply_control_occluded'
|
|
reveal = result['reveal_scroll']
|
|
assert reveal['selector'] == '#inner-scroll'
|
|
assert 100 <= reveal['anchor_y'] < 300
|
|
assert 0 < reveal['delta_y'] <= 550
|
|
|
|
|
|
def test_above_clipped_control_scrolls_up_from_actual_inner_range():
|
|
reveal = observe(discussion(target_top=50))['reveal_scroll']
|
|
assert reveal['selector'] == '#inner-scroll'
|
|
assert -550 <= reveal['delta_y'] < 0
|
|
|
|
|
|
def test_zero_height_document_boxes_use_viewport_owned_anchor():
|
|
result = observe(discussion(target_top=900, document_scroll=True))
|
|
assert result['error'] == 'reply_control_not_visible'
|
|
reveal = result['reveal_scroll']
|
|
assert reveal['selector'] == 'html'
|
|
assert 0 <= reveal['anchor_y'] < 600
|
|
assert 0 < reveal['delta_y'] <= 550
|
|
|
|
|
|
def test_occluding_overlay_cannot_supply_a_scroll_anchor():
|
|
result = observe(discussion(occluded=True))
|
|
assert result['error'] == 'reply_control_occluded'
|
|
assert result['reveal_error'] == 'reply_control_reveal_unavailable'
|
|
assert 'reveal_scroll' not in result
|
|
|
|
|
|
class RevealPage:
|
|
def __init__(self, mode='one'):
|
|
self.mode = mode
|
|
self.wheels = []
|
|
self.clicks = []
|
|
self.control_reads = 0
|
|
self.baselines = 0
|
|
|
|
async def get_url(self):
|
|
return URL.replace('a'*24, 'b'*24) if self.mode == 'stale_note' and self.wheels else URL
|
|
|
|
def owner(self):
|
|
return {'url': URL, 'owner_selector': '#thread',
|
|
'owner_identity': 'other-root' if self.mode == 'stale_owner' and self.wheels else 'root'}
|
|
|
|
def reply(self, selector):
|
|
return {'source_type': 'reply', 'source_selector': selector, 'content': selector,
|
|
'author': '作者', 'url': URL}
|
|
|
|
async def evaluate(self, code, arg=None):
|
|
if code == tools._OBSERVE_REPLY_CONTROL:
|
|
self.control_reads += 1
|
|
if not self.wheels or self.mode in {'many_scrolls', 'no_move'}:
|
|
step = len(self.wheels) if self.mode == 'many_scrolls' else 0
|
|
scroll = {'selector': '#scroll', 'anchor_x': 150, 'anchor_y': 200,
|
|
'delta_y': -200 if self.mode == 'up' else 200,
|
|
'viewport_width': 400, 'viewport_height': 600,
|
|
'scroll_top': 100+step*100, 'target_top': 800-step*20,
|
|
'target_bottom': 830-step*20}
|
|
if self.mode == 'invalid_point':
|
|
scroll['anchor_x'] = 400
|
|
return {**self.owner(), 'error': 'reply_control_not_visible', 'reveal_scroll': scroll}
|
|
fingerprint = 'changed' if self.mode == 'preclick_changed' and self.control_reads >= 3 else 'ready'
|
|
return {**self.owner(), 'fingerprint': fingerprint, 'text': '展开1条回复',
|
|
'reply_count': 1+bool(self.clicks), 'reply_snapshot_complete': True,
|
|
'visible_reply_selectors': ['#old-after-reveal']+(['#new'] if self.clicks else []),
|
|
'click_point': {'x': 150, 'y': 80, 'viewport_width': 400, 'viewport_height': 600}}
|
|
if code == tools._OBSERVE_REPLY_OWNER:
|
|
return {**self.owner(), 'reply_count': 2, 'reply_snapshot_complete': True,
|
|
'visible_reply_selectors': ['#old-after-reveal', '#new']}
|
|
if code == tools._INSPECT_SOURCE_CANDIDATES:
|
|
assert self.wheels, 'Source baseline must be taken after native reveal'
|
|
self.baselines += 1
|
|
return {'url': URL, 'candidates': [self.reply('#old-after-reveal')]+
|
|
([self.reply('#new')] if self.clicks else [])}
|
|
if code == tools._OBSERVE_SOURCE:
|
|
return {'url': URL, 'source_type': 'reply', 'text': '#new'}
|
|
raise AssertionError('Unexpected observer')
|
|
|
|
async def get_elements_by_css_selector(self, selector):
|
|
assert selector in {'#scroll', '#expand'}
|
|
async def actor_click(*_args, **_kwargs):
|
|
raise AssertionError('Implicit actor scroll/click is forbidden')
|
|
return [SimpleNamespace(click=actor_click)]
|
|
|
|
@property
|
|
async def mouse(self):
|
|
async def scroll(**gesture):
|
|
self.wheels.append(gesture)
|
|
async def click(**point):
|
|
assert self.wheels
|
|
assert point == {'x': 150, 'y': 80}
|
|
self.clicks.append(point)
|
|
return SimpleNamespace(scroll=scroll, click=click)
|
|
|
|
|
|
@pytest.fixture
|
|
def reveal_context(tmp_path, monkeypatch):
|
|
task_id = 'reply-reveal-d42'
|
|
tools.clear_task_context(task_id)
|
|
tools.set_prospecting_context(task_id)
|
|
monkeypatch.setattr(tools, 'task_doc_dir', lambda _: tmp_path)
|
|
yield task_id
|
|
tools.clear_task_context(task_id)
|
|
|
|
|
|
async def expand(task_id, page):
|
|
async def current():
|
|
return page
|
|
return await tools._expand_replies(task_id, SimpleNamespace(get_current_page=current), '#expand')
|
|
|
|
|
|
@pytest.mark.parametrize('mode', ['one', 'up'])
|
|
async def test_native_reveal_then_post_reveal_baseline_then_current_point_click(reveal_context, mode):
|
|
page = RevealPage(mode)
|
|
result = await expand(reveal_context, page)
|
|
assert len(page.wheels) == len(page.clicks) == 1
|
|
assert result['reveal_scroll_count'] == 1
|
|
assert result['new_visible_replies'] == 1
|
|
facts = tools.get_task_artifacts(reveal_context)['operation_evidence']
|
|
assert facts[-1]['outcome'] == 'success'
|
|
before = facts[-1]['details']['pre_snapshot']
|
|
assert before['visible_reply_selectors'] == ['#old-after-reveal']
|
|
assert before['reveal_scroll_count'] == 1
|
|
|
|
|
|
async def test_current_expansion_can_prove_revealing_an_already_saved_reply_without_duplicating_it(reveal_context):
|
|
tools._task_artifacts.setdefault(reveal_context, {})['sources'] = [{
|
|
'source_id': 'existing-child', 'note_id': 'a'*24, 'source_type': 'reply',
|
|
'source_selector': '#new', 'content': '#new'}]
|
|
page = RevealPage()
|
|
result = await expand(reveal_context, page)
|
|
assert result['new_visible_replies'] == 1
|
|
assert tools.get_task_artifacts(reveal_context)['operation_evidence'][-1]['outcome'] == 'success'
|
|
assert len(tools.get_task_artifacts(reveal_context)['sources']) == 1
|
|
async def current():
|
|
return page
|
|
public = await tools._inspect_source_candidates(SimpleNamespace(get_current_page=current), task_id=reveal_context)
|
|
assert [row['source_selector'] for row in public['candidates']] == ['#old-after-reveal']
|
|
|
|
|
|
@pytest.mark.parametrize('mode,reason,wheel_count', [
|
|
('no_move', 'reply_control_reveal_not_moved', 1),
|
|
('stale_owner', 'reply_control_reveal_changed', 1),
|
|
('many_scrolls', 'reply_control_reveal_limit', 3),
|
|
('invalid_point', 'reply_control_reveal_unavailable', 0),
|
|
('stale_note', '展开前笔记已变化', 1),
|
|
('preclick_changed', '点击前回复入口或讨论已变化', 1),
|
|
])
|
|
async def test_unsafe_reveal_is_bounded_and_never_clicked(reveal_context, mode, reason, wheel_count):
|
|
page = RevealPage(mode)
|
|
with pytest.raises(ValueError, match=reason):
|
|
await expand(reveal_context, page)
|
|
assert len(page.wheels) == wheel_count
|
|
assert not page.clicks
|
|
assert not tools.get_task_artifacts(reveal_context).get('operation_evidence')
|