267 lines
14 KiB
Python
267 lines
14 KiB
Python
"""Read the shipped DOM observers without navigation, persistence or proofs."""
|
||
from copy import deepcopy
|
||
import json
|
||
from pathlib import Path
|
||
import subprocess
|
||
|
||
import pytest
|
||
|
||
from backend.browser import tools
|
||
|
||
|
||
URL = 'https://www.xiaohongshu.com/explore/' + 'a' * 24
|
||
FIXTURE = Path(__file__).with_name('reply_dom_fixture_d41.cjs')
|
||
|
||
|
||
def node(identity='', classes='', text='', *children, tag='div', **attrs):
|
||
return {'tag': tag, 'attrs': {'id': identity, 'class': classes, **attrs}, 'text': text, 'children': list(children)}
|
||
|
||
|
||
def content(identity, text='PRIVATE-QUOTE', *children):
|
||
return node(identity+'-content', 'content', '', node(identity+'-text', 'note-text', text, *children))
|
||
|
||
|
||
def speaker(identity, person, own=None, *, kind='reply', extra=(), **attrs):
|
||
author = node(identity+'-author', 'name', 'PRIVATE-SAME-NAME', tag='a',
|
||
href='https://www.xiaohongshu.com/user/profile/' + person*24)
|
||
return node(identity, 'comment-item' if kind == 'comment' else 'comment-item-sub', '',
|
||
node('', 'author-wrapper', '', author), own or content(identity), *extra,
|
||
**{'data-comment-id': identity+'-key', **attrs})
|
||
|
||
|
||
def target(*, href=True, identity='c', user_attribute=False):
|
||
attrs = {'href': 'https://www.xiaohongshu.com/user/profile/'+identity*24} if href else {}
|
||
if user_attribute:
|
||
attrs['data-target-user-id'] = identity*24
|
||
return content('child', '回复 ', node('target', 'user-name', 'PRIVATE-SAME-NAME',
|
||
tag='a' if href else 'span', **attrs), node('', '', ': PRIVATE-CHILD-QUOTE'))
|
||
|
||
|
||
def thread(*replies, identity='thread'):
|
||
return node(identity, 'parent-comment', '', speaker(identity+'-root', 'b', kind='comment'),
|
||
node(identity+'-replies', 'reply-container', '', *replies))
|
||
|
||
|
||
def tree(*groups):
|
||
return node('noteContainer', '', '', node('detail-desc', '', 'PRIVATE-NOTE'), *groups)
|
||
|
||
|
||
def find(value, identity):
|
||
if value.get('attrs', {}).get('id') == identity:
|
||
return value
|
||
for child in value.get('children', []):
|
||
found = find(child, identity)
|
||
if found:
|
||
return found
|
||
return None
|
||
|
||
|
||
class Page:
|
||
def __init__(self, value, mutate=None):
|
||
self.tree, self.mutate = deepcopy(value), mutate
|
||
self.url = URL
|
||
self.calls = []
|
||
self.source_calls = {}
|
||
|
||
async def get_url(self):
|
||
return self.url
|
||
|
||
async def evaluate(self, code, arg):
|
||
assert code in {tools._INSPECT_SOURCE_CANDIDATES, tools._OBSERVE_SOURCE, tools._OBSERVE_RELATION_RECHECK}
|
||
self.calls.append((code, arg))
|
||
if code == tools._OBSERVE_SOURCE:
|
||
assert isinstance(arg, str), 'preflight must never supply a native proof identity'
|
||
self.source_calls[arg] = self.source_calls.get(arg, 0)+1
|
||
if self.mutate:
|
||
self.mutate(self, code, arg)
|
||
response = subprocess.run(['node', str(FIXTURE)], text=True, capture_output=True, check=True, timeout=5,
|
||
input=json.dumps({'code': code, 'arg': arg, 'tree': self.tree}, ensure_ascii=False))
|
||
return response.stdout
|
||
|
||
|
||
class Session:
|
||
def __init__(self, page):
|
||
self.page = page
|
||
|
||
async def get_current_page(self):
|
||
return self.page
|
||
|
||
|
||
@pytest.fixture
|
||
def isolated(monkeypatch, tmp_path):
|
||
task_id = 'reply_preflight_d44'
|
||
tools.set_prospecting_context(task_id, artifacts={
|
||
'messages': [{'role': 'user', 'text': '只预检第一组讨论，不保存、不点击'}],
|
||
'sources': [{'source_id': 'old-note', 'source_type': 'note', 'note_id': 'a'*24,
|
||
'source_selector': '#detail-desc', 'content': 'PRIVATE-NOTE'},
|
||
{'source_id': 'old-child', 'source_type': 'reply', 'note_id': 'a'*24,
|
||
'source_selector': '#child-content', 'content': 'PRIVATE-CHILD-QUOTE', 'parent_source_id': None}],
|
||
'source_observations': [{'operation': 'old-observation'}], 'operation_evidence': [{'id': 'old-fact'}],
|
||
'knowledge_event_ids': ['old-knowledge'],
|
||
})
|
||
tools._reply_target_proofs[task_id] = {'child-key': {'run_id': 'old-run', 'signature': 'old', 'profile_id': 'c'*24}}
|
||
tools._prospecting_run_context[task_id] = {'run_id': 'new-run', 'goal_version': 2}
|
||
def forbidden(*_args, **_kwargs):
|
||
raise AssertionError('read-only preflight must not write facts, navigate or use proofs')
|
||
for name in ('_persist_evidence', '_operation_evidence', '_save_source', '_write_json_atomic', '_apply_reply_target_proof', 'task_doc_dir'):
|
||
monkeypatch.setattr(tools, name, forbidden)
|
||
before = deepcopy((tools._task_artifacts, tools._reply_target_proofs, tools._prospecting_run_context,
|
||
tools._pending_blocker, tools._pending_question))
|
||
yield task_id
|
||
assert before == (tools._task_artifacts, tools._reply_target_proofs, tools._prospecting_run_context,
|
||
tools._pending_blocker, tools._pending_question)
|
||
assert list(tmp_path.iterdir()) == []
|
||
tools.clear_task_context(task_id)
|
||
|
||
|
||
async def inspect(value, task_id, mutate=None, maximum=5):
|
||
page = Page(value, mutate)
|
||
result = await tools._inspect_reply_relations(Session(page), maximum, task_id)
|
||
serialized = json.dumps(result)
|
||
assert 'PRIVATE' not in serialized and 'https://' not in serialized and 'profile_id' not in serialized
|
||
assert result['note_id'] == 'a'*24
|
||
for identity in ('a'*24, 'b'*24, 'c'*24, 'd'*24, 'e'*24):
|
||
assert identity not in json.dumps({key: value for key, value in result.items() if key != 'note_id'})
|
||
return result, page
|
||
|
||
|
||
def child_row(result):
|
||
return next(row for row in result['candidates'] if row['source_selector'] == '#child-content')
|
||
|
||
|
||
@pytest.mark.parametrize('mode', ['href', 'user_attribute', 'comment_id'])
|
||
async def test_real_unique_identity_or_explicit_comment_ref_qualifies_without_saving(mode, isolated):
|
||
own = target(href=mode == 'href', user_attribute=mode == 'user_attribute')
|
||
child = speaker('child', 'd', own, **({'data-parent-comment-id': 'parent-key'} if mode == 'comment_id' else {}))
|
||
result, _ = await inspect(tree(thread(speaker('parent', 'c'), child)), isolated)
|
||
row = child_row(result)
|
||
assert row['classification'] == 'unique_dom_deep_reply' and row['eligible'] is True
|
||
assert row['parent_is_reply'] is True and row['parent_relation_observed'] is True
|
||
assert row['parent_source_selector'] == '#parent-content' and row['root_source_selector'] == '#thread-root-content'
|
||
assert row['native_navigation_confirmed'] is False
|
||
assert row['parent_resolution'] == ('explicit_comment_id' if mode == 'comment_id' else 'unique_target_profile')
|
||
|
||
|
||
async def test_actual_nested_parent_reply_has_structural_evidence(isolated):
|
||
child = speaker('child', 'd')
|
||
parent = speaker('parent', 'c', extra=[child])
|
||
root = speaker('root', 'b', kind='comment', extra=[parent])
|
||
result, _ = await inspect(tree(root), isolated)
|
||
row = child_row(result)
|
||
assert row['classification'] == 'unique_dom_deep_reply'
|
||
assert row['parent_resolution'] == 'structural_parent' and row['parent_is_reply'] is True
|
||
|
||
|
||
async def test_identical_names_and_two_actual_comments_by_target_identity_remain_ambiguous(isolated):
|
||
result, _ = await inspect(tree(thread(speaker('parent', 'c'), speaker('another', 'c'), speaker('child', 'd', target()))), isolated)
|
||
row = child_row(result)
|
||
assert row['classification'] == 'ambiguous' and not row['eligible']
|
||
assert row['relation_diagnostics']['matching_parent_comments'] == 2
|
||
assert row['parent_source_selector'] == ''
|
||
|
||
|
||
async def test_plain_span_is_unproved_even_with_an_existing_native_proof_cache(isolated):
|
||
result, _ = await inspect(tree(thread(speaker('parent', 'c'), speaker('child', 'd', target(href=False)))), isolated)
|
||
row = child_row(result)
|
||
assert row['classification'] == 'needs_native_identity' and row['observation_stable'] is True
|
||
assert row['relation_diagnostics']['target_identities'] == 0
|
||
assert row['relation_diagnostics']['target_profile_links'] == 0
|
||
assert not row['eligible'] and not row['native_navigation_confirmed']
|
||
|
||
|
||
@pytest.mark.parametrize('change', ['source_key', 'quote', 'rendering', 'identity', 'url'])
|
||
async def test_source_key_quote_rendering_identity_or_url_change_is_unknown(change, isolated):
|
||
def mutate(page, code, selector):
|
||
if code != tools._OBSERVE_SOURCE or selector != '#child-content' or page.source_calls[selector] != 2:
|
||
return
|
||
if change == 'source_key':
|
||
find(page.tree, 'child')['attrs']['data-comment-id'] = 'changed-key'
|
||
elif change == 'quote':
|
||
find(page.tree, 'child-text')['text'] = 'CHANGED-QUOTE'
|
||
elif change == 'rendering':
|
||
find(page.tree, 'child-content')['rect'] = {'left': 0, 'right': 300, 'top': 700, 'bottom': 735, 'width': 300, 'height': 35}
|
||
elif change == 'identity':
|
||
# Same nickname, target counts, parent key and selector; the actual
|
||
# target data-* identity changes together with its observed author.
|
||
find(page.tree, 'target')['attrs']['data-target-user-id'] = 'e'*24
|
||
find(page.tree, 'parent-author')['attrs']['href'] = 'https://www.xiaohongshu.com/user/profile/'+'e'*24
|
||
else:
|
||
page.url = URL.replace('a'*24, 'f'*24)
|
||
result, _ = await inspect(tree(thread(speaker('parent', 'c'), speaker('child', 'd', target(href=False, user_attribute=True)))), isolated, mutate)
|
||
row = child_row(result)
|
||
assert row['classification'] == 'unknown' and row['observation_stable'] is False and not row['eligible']
|
||
|
||
|
||
async def test_changed_candidate_excerpt_before_first_observation_is_unknown(isolated):
|
||
def mutate(page, code, selector):
|
||
if code == tools._OBSERVE_SOURCE and selector == '#child-content' and page.source_calls[selector] == 1:
|
||
find(page.tree, 'child-text')['text'] = 'DIFFERENT-QUOTE'
|
||
result, _ = await inspect(tree(thread(speaker('parent', 'c'), speaker('child', 'd', target()))), isolated, mutate)
|
||
assert child_row(result)['classification'] == 'unknown'
|
||
|
||
|
||
async def test_parent_moved_to_another_thread_does_not_qualify_old_child_observation(isolated):
|
||
def mutate(page, code, selector):
|
||
if (code == tools._OBSERVE_SOURCE and selector == '#parent-content'
|
||
and page.source_calls.get('#child-content') == 1):
|
||
old = find(page.tree, 'thread-replies')['children']
|
||
parent = old.pop(next(index for index, row in enumerate(old) if row['attrs']['id'] == 'parent'))
|
||
find(page.tree, 'other-replies')['children'].append(parent)
|
||
value = tree(thread(speaker('parent', 'c'), speaker('child', 'd', target())), thread(identity='other'))
|
||
result, _ = await inspect(value, isolated, mutate)
|
||
assert child_row(result)['classification'] == 'unknown' and not child_row(result)['eligible']
|
||
|
||
|
||
async def test_direct_root_reply_is_not_a_qualified_reply_to_reply_sample(isolated):
|
||
result, _ = await inspect(tree(thread(speaker('child', 'd'))), isolated)
|
||
row = child_row(result)
|
||
assert row['parent_is_reply'] is False and row['parent_resolution'] == 'thread_root'
|
||
assert not row['eligible'] and row['classification'] == 'unknown'
|
||
|
||
|
||
async def test_later_candidate_mutation_invalidates_an_earlier_qualified_row(isolated):
|
||
def mutate(page, code, selector):
|
||
if code == tools._OBSERVE_SOURCE and selector == '#later-content' and page.source_calls[selector] == 1:
|
||
find(page.tree, 'child')['attrs']['data-comment-id'] = 'new-child-key'
|
||
value = tree(thread(speaker('parent', 'c'), speaker('child', 'd', target()), speaker('later', 'e')))
|
||
result, _ = await inspect(value, isolated, mutate)
|
||
assert child_row(result)['classification'] == 'unknown' and not child_row(result)['eligible']
|
||
|
||
|
||
async def test_explicit_first_group_does_not_preflight_another_group(isolated):
|
||
value = tree(thread(speaker('first', 'c')), thread(speaker('parent', 'c'), speaker('child', 'd', target()), identity='other'))
|
||
result, _ = await inspect(value, isolated)
|
||
assert all(row['source_selector'] != '#child-content' for row in result['candidates'])
|
||
assert result['eligible_count'] == 0
|
||
|
||
|
||
@pytest.mark.parametrize('maximum', [0, 6, True, '2'])
|
||
async def test_batch_bound_is_strict_before_reading(maximum):
|
||
with pytest.raises(ValueError, match='1–5'):
|
||
await tools._inspect_reply_relations(object(), maximum)
|
||
|
||
|
||
async def test_ordinary_inspector_still_records_by_default_but_internal_no_record_does_not(monkeypatch):
|
||
tid = 'preflight_inspector_compatibility'
|
||
tools.set_prospecting_context(tid)
|
||
writes = []
|
||
monkeypatch.setattr(tools, '_persist_evidence', lambda task: writes.append(task))
|
||
page = Page(tree(thread(speaker('child', 'd'))))
|
||
try:
|
||
pure = await tools._inspect_source_candidates(Session(page), task_id=tid, record_observation=False)
|
||
assert 'source_observations' not in tools._task_artifacts[tid] and not writes
|
||
normal = await tools._inspect_source_candidates(Session(page), task_id=tid)
|
||
assert normal == pure
|
||
assert writes == [tid] and tools._task_artifacts[tid]['source_observations'][0]['operation'] == 'inspect_source_candidates'
|
||
finally:
|
||
tools.clear_task_context(tid)
|
||
|
||
|
||
def test_registered_preflight_accepts_only_a_bounded_count_not_selectors_or_identity(monkeypatch):
|
||
monkeypatch.setattr(tools.settings, 'reply_relation_preflight_enabled', True)
|
||
registry = tools.build_prospecting_tools('preflight_registration').registry.registry.actions
|
||
action = registry['inspect_reply_relations']
|
||
assert set(action.param_model.model_fields) == {'max_replies'}
|
||
assert '不自动调用resolver' in action.description
|
||
assert 'unique_dom_deep_reply' in action.description
|