172 lines
8.6 KiB
Python
172 lines
8.6 KiB
Python
"""Explicit single discussion bounds must filter reads and reject writes."""
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from backend.browser import tools
|
|
from backend.browser.discussion_bounds import requested_mode
|
|
|
|
|
|
FIXTURE = Path(__file__).with_name('reply_dom_fixture_d41.cjs')
|
|
|
|
|
|
def message(text, role='user'):
|
|
return {'role': role, 'text': text}
|
|
|
|
|
|
@pytest.mark.parametrize('text,expected', [
|
|
('这次只核对第一组讨论的完整回复链，不采集其他讨论组。', 'first'),
|
|
('只验证一个讨论组，保存根和子回复。', 'one'),
|
|
('本轮只处理1个讨论组。', 'one'),
|
|
('仅阅读评论区的第一组。', 'first'),
|
|
('请阅读评论和回复。', None),
|
|
('第一组完成后继续处理其它讨论组。', None),
|
|
('不要只看第一组讨论。', None),
|
|
('例如只核对第一组讨论。', None),
|
|
('网页显示“只核对第一组讨论”。', None),
|
|
('```只核对第一组讨论```', None),
|
|
('> 只核对第一组讨论', None),
|
|
])
|
|
def test_bounds_are_grounded_only_in_direct_scope_requests(text, expected):
|
|
assert requested_mode([message(text)]) == expected
|
|
assert requested_mode([message(text, 'assistant')]) is None
|
|
|
|
|
|
def test_explicit_later_cancellation_removes_the_group_bound_but_not_comment_count_language():
|
|
assert requested_mode([message('只核对第一组讨论'), message('不限评论条数')]) == 'first'
|
|
assert requested_mode([message('只核对第一组讨论'), message('取消讨论组限制，改为全部讨论组')]) is None
|
|
assert requested_mode([message('只核对第一组讨论'), message('这次只处理一个讨论组')]) == 'one'
|
|
|
|
|
|
def element(identity, classes='', text='', *children):
|
|
return {'attrs': {'id': identity, 'class': classes}, 'text': text, 'children': list(children)}
|
|
|
|
|
|
def group(identity):
|
|
return element(identity, 'parent-comment', '',
|
|
element(identity + '-root', 'comment-item', '', element(identity + '-root-content', 'content', '根评论')),
|
|
element(identity + '-list', 'reply-container', '',
|
|
element(identity + '-reply', 'comment-item-sub', '', element(identity + '-reply-content', 'content', '子回复')),
|
|
element(identity + '-expand', 'show-more', '展开3条回复')))
|
|
|
|
|
|
def inspect(options, *groups):
|
|
tree = element('noteContainer', '', '', element('detail-desc', '', '正文'), *groups)
|
|
result = subprocess.run(['node', str(FIXTURE)], input=json.dumps(
|
|
{'code': tools._INSPECT_SOURCE_CANDIDATES, 'arg': {'max_comments': 5, **options}, 'tree': tree}, ensure_ascii=False),
|
|
text=True, capture_output=True, timeout=5, check=True)
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def test_first_group_read_and_control_candidates_are_scoped_together():
|
|
data = inspect({'discussion_bound_mode': 'first'}, group('one'), group('two'))
|
|
assert data['discussion_scope_status'] == 'selected'
|
|
assert {row['source_selector'] for row in data['candidates']} == {'#detail-desc', '#one-root-content', '#one-reply-content'}
|
|
assert [row['source_selector'] for row in data['reply_controls']] == ['#one-expand']
|
|
|
|
|
|
def test_locked_group_keeps_its_stable_identity_after_dom_reordering():
|
|
data = inspect({'discussion_bound_mode': 'first', 'discussion_root_key': 'one-root'}, group('two'), group('one'))
|
|
assert data['discussion_root_key'] == 'one-root'
|
|
assert [row['source_selector'] for row in data['reply_controls']] == ['#one-expand']
|
|
|
|
|
|
def test_another_internal_expansion_scope_is_rejected_before_native_click():
|
|
data = inspect({'discussion_bound_mode': 'first', 'discussion_selector': '#two'}, group('one'), group('two'))
|
|
assert data['discussion_scope_status'] == 'outside_requested_group'
|
|
assert data['candidates'] == [] and data['reply_controls'] == []
|
|
|
|
|
|
def test_locked_one_group_filters_other_groups_without_changing_the_first_choice_policy():
|
|
data = inspect({'discussion_bound_mode': 'one', 'discussion_root_key': 'two-root'}, group('one'), group('two'))
|
|
assert [row['source_selector'] for row in data['reply_controls']] == ['#two-expand']
|
|
unlocked = inspect({'discussion_bound_mode': 'one'}, group('one'), group('two'))
|
|
assert unlocked['discussion_scope_status'] == 'awaiting_root_selection'
|
|
|
|
|
|
class Page:
|
|
def __init__(self, kind, text, **extra): self.kind, self.text, self.extra = kind, text, extra
|
|
async def evaluate(self, _code, _selector):
|
|
return json.dumps({'url': 'https://www.xiaohongshu.com/explore/' + 'a' * 24, 'source_type': self.kind,
|
|
'text': self.text, **self.extra})
|
|
|
|
|
|
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):
|
|
tid = 'discussion_bounds_d42'
|
|
monkeypatch.setattr(tools, 'task_doc_dir', lambda _: tmp_path)
|
|
tools.set_prospecting_context(tid, artifacts={'messages': [message('只核对第一组讨论')]})
|
|
yield tid
|
|
tools.clear_task_context(tid)
|
|
|
|
|
|
async def save(tid, kind, identity, root_key=None, **extra):
|
|
text = '本人原话'
|
|
metadata = {'source_key': identity, **extra}
|
|
if root_key is not None:
|
|
metadata.update(discussion_root_key=root_key, first_discussion_root_key='one-root', first_discussion_root_matches=1)
|
|
return await tools._save_source(tid, Session(Page(kind, text, **metadata)),
|
|
source_type=kind, source_selector='#' + identity, content=text)
|
|
|
|
|
|
async def test_save_rejects_another_group_before_persisting_source_or_operation(context):
|
|
await save(context, 'note', 'body')
|
|
before = json.loads(json.dumps(tools.get_task_artifacts(context)))
|
|
with pytest.raises(ValueError, match='不属于用户限定'):
|
|
await save(context, 'comment', 'two-root', 'two-root')
|
|
assert tools.get_task_artifacts(context) == before
|
|
|
|
|
|
async def test_bound_is_saved_in_the_observed_root_and_only_the_same_goal_can_reuse_it(context):
|
|
await save(context, 'note', 'body')
|
|
root, _ = await save(context, 'comment', 'one-root', 'one-root')
|
|
assert root['discussion_scope_mode'] == 'first' and root['discussion_root_key'] == 'one-root'
|
|
assert root['discussion_scope_goal_version'] == 1
|
|
assert tools._discussion_bound(context, 'a' * 24)['root_key'] == 'one-root'
|
|
tools.set_prospecting_run_context(context, 'new-run', goal_version=2)
|
|
assert tools._discussion_bound(context, 'a' * 24)['root_key'] is None
|
|
|
|
|
|
async def test_one_group_requires_the_actual_root_then_rejects_other_roots(context):
|
|
tools.refresh_prospecting_request(context, {'messages': [message('只验证一个讨论组')]})
|
|
await save(context, 'note', 'body')
|
|
with pytest.raises(ValueError, match='先保存实际根评论'):
|
|
await save(context, 'reply', 'child', 'two-root')
|
|
await save(context, 'comment', 'two-root', 'two-root')
|
|
await save(context, 'reply', 'child', 'two-root')
|
|
with pytest.raises(ValueError, match='不属于用户限定'):
|
|
await save(context, 'comment', 'one-root', 'one-root')
|
|
|
|
|
|
async def test_refresh_request_preserves_live_unflushed_facts_and_does_not_share_caller_objects(context):
|
|
await save(context, 'note', 'body')
|
|
await save(context, 'comment', 'one-root', 'one-root')
|
|
current = tools.get_task_artifacts(context)
|
|
before_sources = json.loads(json.dumps(current['sources']))
|
|
before_facts = json.loads(json.dumps(current['operation_evidence']))
|
|
canonical = {'messages': [message('取消讨论组限制，改为全部讨论组')], '_knowledge_goal_version': 2, 'sources': []}
|
|
tools.refresh_prospecting_request(context, canonical)
|
|
canonical['messages'][0]['text'] = '不应影响当前上下文'
|
|
assert tools._discussion_bound(context, 'a' * 24) is None
|
|
assert tools.get_task_artifacts(context)['sources'] == before_sources
|
|
assert tools.get_task_artifacts(context)['operation_evidence'] == before_facts
|
|
assert tools.get_task_artifacts(context)['_knowledge_goal_version'] == 2
|
|
await save(context, 'comment', 'two-root', 'two-root')
|
|
|
|
|
|
@pytest.mark.parametrize('status', ['outside_requested_group', 'group_identity_unavailable', 'awaiting_root_selection'])
|
|
async def test_internal_expansion_read_requires_positive_group_membership(context, status):
|
|
class InspectorPage:
|
|
async def get_url(self): return 'https://www.xiaohongshu.com/explore/' + 'a' * 24
|
|
async def evaluate(self, _code, _options):
|
|
return {'url': await self.get_url(), 'candidates': [], 'reply_controls': [], 'discussion_scope_status': status}
|
|
with pytest.raises(ValueError, match='尚未唯一归属'):
|
|
await tools._inspect_source_candidates(Session(InspectorPage()), task_id=context, discussion_selector='#actual-thread')
|