"""Bounded, read-only navigation on local Page observations; no browser/network.""" import json from copy import deepcopy from pathlib import Path from types import SimpleNamespace import pytest from backend.browser import tools from backend.prospecting.contracts import ProspectingBrief from backend.prospecting.prompts import browser_prompt NOTE = 'a' * 24 NOTE_URL = 'https://www.xiaohongshu.com/explore/' + NOTE PROFILE = 'https://www.xiaohongshu.com/user/profile/' + 'b' * 24 OTHER_PROFILE = 'https://www.xiaohongshu.com/user/profile/' + 'c' * 24 class NavigationPage: def __init__(self): self.url = NOTE_URL self.position = 0 self.comments = 0 self.clicks = [] self.scrolls = [] self.comment_entry = '#comment-entry' self.scroll_area = '#note-scroll' self.click_enters = True self.scroll_moves = True self.profile_navigates = True self.hover = False self.profile_header = False self.profile_href = PROFILE self.element_href = PROFILE self.element_count = 1 self.source_text = '本人正在找装修公司' self.source_author = '本机夹具作者' self.source_selector = '#body' self.profile_selector = '#actual-avatar' self.source_type = 'note' self.source_key = '' self.mouse_calls = 0 async def get_url(self): return self.url async def evaluate(self, code, options=None): if code == tools._OBSERVE_SOURCE: value = {'url': self.url, 'text': self.source_text, 'source_type': self.source_type, 'source_key': self.source_key, 'author': self.source_author, 'author_profile_url': self.profile_href, 'author_profile_selector': self.profile_selector} elif code == tools._OBSERVE_DISCUSSION_NAVIGATION: value = {'url': self.url, 'visible_comment_count': self.comments, 'comment_control_selector': self.comment_entry, 'scroll': {'selector': self.scroll_area, 'anchor_x': 210, 'anchor_y': 370, 'delta_y': 420} if self.scroll_area else None, 'signature': json.dumps([self.position, self.comments]), 'discussion_state': 'visible_comments' if self.comments else 'comments_not_in_view'} elif code == tools._OBSERVE_PROFILE_DESTINATION: value = {'url': self.url, 'profile_header_visible': self.profile_header, 'hover_card_visible': self.hover} elif code == tools._INSPECT_SOURCE_CANDIDATES: value = {'url': self.url, 'candidates': [], 'saved_count': 1, 'reply_controls': [], 'discussion_state': 'visible_comments' if self.comments else 'comments_not_in_view', 'visible_comment_count': self.comments} else: raise AssertionError('Unrecognised DOM observation, no arbitrary script allowed') return json.dumps(value) async def get_elements_by_css_selector(self, selector): assert selector in {self.comment_entry, self.scroll_area, self.profile_selector}, 'No guessed selector' async def attribute(name): assert name == 'href' return self.element_href async def click(): self.clicks.append(selector) if selector == self.comment_entry: if self.click_enters: self.comments, self.position = 2, self.position + 1 elif selector == self.profile_selector: if self.profile_navigates: self.url, self.profile_header = self.element_href, True else: self.hover = True else: raise AssertionError('Scroll region must never be clicked') return [SimpleNamespace(click=click, get_attribute=attribute) for _ in range(self.element_count)] @property async def mouse(self): async def scroll(**kwargs): self.scrolls.append(kwargs) if self.scroll_moves: self.position += 1 self.comments = 2 self.mouse_calls += 1 return SimpleNamespace(scroll=scroll) class Session: def __init__(self, page): self.page = page async def get_current_page(self): return self.page @pytest.fixture def context(monkeypatch, tmp_path): tid = 'navigation_d38' tools.clear_task_context(tid) monkeypatch.setattr(tools, 'task_doc_dir', lambda _: tmp_path) async def no_delay(_): pass monkeypatch.setattr(tools.asyncio, 'sleep', no_delay) source = {'source_id': 'source_fixture', 'note_id': NOTE, 'source_type': 'note', 'source_selector': '#body', 'content': '本人正在找装修公司', 'author': '本机夹具作者', 'author_profile_url': '', 'author_profile_selector': ''} tools.set_prospecting_context(tid, artifacts={'sources': [source]}) yield tid, tmp_path tools.clear_task_context(tid) def source(tid): return tools.get_task_artifacts(tid)['sources'][0] def test_navigation_tools_accept_object_or_mode_only_and_add_no_outreach_capability(context): actions = tools.build_prospecting_tools(context[0]).registry.registry.actions assert 'advance_comments' in actions and 'open_source_profile' in actions assert set(actions['advance_comments'].param_model.model_fields) == {'mode'} assert set(actions['open_source_profile'].param_model.model_fields) == {'source_id'} for action in ['perform_authorized_follow', 'send_authorized_message', 'evaluate', 'upload_file']: assert action not in actions prompt = browser_prompt(ProspectingBrief(target_candidates=1), []) assert 'comments_not_in_view' in prompt and 'open_source_profile' in prompt and '小红书号是可选项' in prompt async def test_saved_body_without_viewport_comments_is_unread_not_empty_or_completed(context): page = NavigationPage() value = await tools._inspect_source_candidates(Session(page), task_id=context[0]) assert value['empty_reason'] == 'comments_not_in_view' and not value['comments_complete'] assert value['saved_sources'] == [{'source_id':'source_fixture','source_type':'note','source_selector':'#body', 'parent_resolution':'','has_reply_target':False,'relations_checked':False,'author_link_observed':False}] assert '尚未进入当前视口' in value['next_action'] assert not page.clicks and not page.scrolls disk = json.loads((context[1] / 'evidence.json').read_text()) assert disk['source_observations'][0]['discussion_state'] == 'comments_not_in_view' async def test_enter_clicks_actual_comment_control_once_then_reports_visible_without_claiming_read(context): page = NavigationPage() result = await tools._advance_comments(context[0], Session(page)) assert result['status'] == 'comments_visible' and not result['comments_complete'] assert page.clicks == ['#comment-entry'] and not page.scrolls second = await tools._advance_comments(context[0], Session(page)) assert second['status'] == 'comments_visible' and page.clicks == ['#comment-entry'] records = json.loads((context[1] / 'evidence.json').read_text())['navigation_attempts'] assert len(records) == 1 and records[0]['id'] and records[0]['status'] == 'comments_visible' assert records[0]['pre_fingerprint'] != records[0]['post_fingerprint'] assert not tools.get_task_artifacts(context[0]).get('operation_evidence') async def test_next_scroll_uses_only_current_observed_bounds_and_not_script_click(context): page = NavigationPage() page.comment_entry = '' result = await tools._advance_comments(context[0], Session(page), 'next') assert result['status'] == 'scrolled' and not result['comments_complete'] assert not page.clicks assert page.scrolls == [{'x': 210, 'y': 370, 'delta_y': 420}] async def test_comment_entry_without_effect_does_not_loop_but_allows_distinct_scroll_fallback(context): page = NavigationPage() page.click_enters = False first = await tools._advance_comments(context[0], Session(page)) assert first['status'] == 'not_moved' with pytest.raises(ValueError, match='不要重复'): await tools._advance_comments(context[0], Session(page)) assert len(page.clicks) == 1 await tools._advance_comments(context[0], Session(page), 'next') assert len(page.scrolls) == 1 async def test_scroll_without_new_state_is_not_repeated_and_missing_entry_stays_unread(context): page = NavigationPage() page.scroll_moves = False assert (await tools._advance_comments(context[0], Session(page), 'next'))['status'] == 'not_moved' with pytest.raises(ValueError, match='不要重复'): await tools._advance_comments(context[0], Session(page), 'next') assert len(page.scrolls) == 1 page.position += 1 page.comment_entry = page.scroll_area = '' assert (await tools._advance_comments(context[0], Session(page)))['status'] == 'inaccessible' @pytest.mark.parametrize('bad', [-1, 0, 701, '420', None]) async def test_invalid_scroll_observation_never_becomes_guessed_gesture(context, bad): page = NavigationPage() native_evaluate = page.evaluate async def evaluate(code, options=None): raw = json.loads(await native_evaluate(code, options)) if code == tools._OBSERVE_DISCUSSION_NAVIGATION: raw['scroll']['delta_y'] = bad return json.dumps(raw) page.evaluate = evaluate with pytest.raises(ValueError, match='不能猜坐标'): await tools._advance_comments(context[0], Session(page), 'next') assert not page.scrolls and not page.clicks async def test_open_profile_refreshes_existing_source_link_without_rewriting_quote_or_old_fact(context): tid = context[0] page = NavigationPage() original = deepcopy(source(tid)) old_fact = {'id': 'original_fact', 'details': {'trace': {'run_id': 'old'}}} tools._task_artifacts[tid]['operation_evidence'] = [deepcopy(old_fact)] result = await tools._open_source_profile(tid, Session(page), original['source_id']) assert result['status'] == 'profile_opened' and page.clicks == ['#actual-avatar'] assert source(tid)['author_profile_url'] == PROFILE and source(tid)['author_profile_selector'] == '#actual-avatar' assert source(tid)['content'] == original['content'] and source(tid)['source_id'] == original['source_id'] assert tools.get_task_artifacts(tid)['operation_evidence'] == [old_fact] disk = json.loads((context[1] / 'evidence.json').read_text()) assert disk['sources'][0]['author_profile_url'] == PROFILE and disk['navigation_attempts'][0]['status'] == 'profile_opened' assert 'profile_url' not in result and PROFILE not in json.dumps(result) async def test_hover_card_stops_repeat_click_until_actual_view_state_changes(context): tid = context[0] page = NavigationPage() page.profile_navigates = False assert (await tools._open_source_profile(tid, Session(page), 'source_fixture'))['status'] == 'hover_only' with pytest.raises(ValueError, match='不要重复点击'): await tools._open_source_profile(tid, Session(page), 'source_fixture') assert page.clicks == ['#actual-avatar'] page.position += 1 # A genuinely different observed viewport can be retried. page.profile_navigates = True assert (await tools._open_source_profile(tid, Session(page), 'source_fixture'))['status'] == 'profile_opened' assert len(page.clicks) == 2 @pytest.mark.parametrize('mutate,match', [ (lambda p: setattr(p, 'url', 'https://www.xiaohongshu.com/explore/'+'d'*24), '原笔记'), (lambda p: setattr(p, 'source_author', '另一位作者'), '作者'), (lambda p: setattr(p, 'source_text', '他人的不同原话'), '原话'), (lambda p: setattr(p, 'source_type', 'comment'), '层级'), (lambda p: setattr(p, 'profile_selector', ''), '唯一可见'), (lambda p: setattr(p, 'profile_href', 'https://evil.example/user/profile/'+'b'*24), '小红书'), (lambda p: setattr(p, 'profile_href', ''), '小红书'), (lambda p: setattr(p, 'element_href', OTHER_PROFILE), '点击前'), (lambda p: setattr(p, 'element_count', 2), '入口已经变化'), ]) async def test_source_author_navigation_refuses_unproven_or_changed_target_before_click(context, mutate, match): page = NavigationPage(); mutate(page) with pytest.raises(ValueError, match=match): await tools._open_source_profile(context[0], Session(page), 'source_fixture') assert not page.clicks async def test_preexisting_source_identity_cannot_be_reassigned_or_arbitrary_profile_claimed(context): tid = context[0] source(tid)['author_profile_url'] = OTHER_PROFILE page = NavigationPage() with pytest.raises(ValueError, match='原关联'): await tools._open_source_profile(tid, Session(page), 'source_fixture') assert not page.clicks and source(tid)['author_profile_url'] == OTHER_PROFILE page.url = PROFILE with pytest.raises(ValueError, match='不能按昵称归属'): await tools._open_source_profile(tid, Session(page), 'source_fixture') @pytest.mark.parametrize('actual_key', ['different-comment', '']) async def test_same_quote_and_nickname_cannot_reassign_a_different_or_missing_stable_comment_key(context, actual_key): tid = context[0] source(tid).update(source_type='comment', source_key='original-comment-id') original = deepcopy(source(tid)) page = NavigationPage() page.source_type = 'comment' page.source_key = actual_key with pytest.raises(ValueError, match='稳定来源标识'): await tools._open_source_profile(tid, Session(page), 'source_fixture') assert source(tid) == original and not page.clicks async def test_matching_stable_comment_key_can_refresh_the_same_existing_source(context): tid = context[0] source(tid).update(source_type='comment', source_key='original-comment-id') page = NavigationPage() page.source_type, page.source_key = 'comment', 'original-comment-id' result = await tools._open_source_profile(tid, Session(page), 'source_fixture') assert result['status'] == 'profile_opened' assert source(tid)['source_key'] == 'original-comment-id' and source(tid)['author_profile_url'] == PROFILE async def test_post_click_wrong_profile_is_unknown_and_never_records_success(context): page = NavigationPage() original = page.get_elements_by_css_selector async def elements(selector): nodes = await original(selector) async def click(): page.clicks.append(selector); page.url = OTHER_PROFILE; page.profile_header = True nodes[0].click = click return nodes page.get_elements_by_css_selector = elements with pytest.raises(ValueError, match='实际打开的主页'): await tools._open_source_profile(context[0], Session(page), 'source_fixture') assert tools.get_task_artifacts(context[0])['navigation_attempts'][-1]['status'] == 'unknown' assert not tools.get_task_artifacts(context[0]).get('operation_evidence') def dom_observer(code, *, arg=None, scenario='note'): import subprocess fixture = Path(__file__).with_name('navigation_dom_fixture_d38.cjs') result = subprocess.run(['node', str(fixture)], input=json.dumps({'code': code, 'arg': arg, 'scenario': scenario}), text=True, capture_output=True, timeout=5, check=True) return json.loads(result.stdout) def test_actual_observer_reads_unique_note_header_anchor_without_inheriting_comment_person(): observed = dom_observer(tools._OBSERVE_SOURCE, arg='#detail-desc') assert observed['source_type'] == 'note' and observed['author'] == '正文作者' assert observed['author_profile_url'] == PROFILE and observed['author_profile_selector'] == '#author-avatar' assert '评论本人的需求' not in observed['text'] def test_actual_observer_preserves_comment_owner_and_excludes_nested_reply_author(): observed = dom_observer(tools._OBSERVE_SOURCE, arg='#comment-content') assert observed['source_type'] == 'comment' and observed['author'] == '评论作者' assert observed['author_profile_url'] == OTHER_PROFILE and observed['author_profile_selector'] == '#comment-author' assert observed['text'] == '评论本人的需求' def test_actual_observer_rejects_ambiguous_note_header_profile_identities(): observed = dom_observer(tools._OBSERVE_SOURCE, arg='#detail-desc', scenario='ambiguous') assert observed['author_profile_url'] == '' and observed['author_profile_selector'] == '' def test_actual_narrow_viewport_inspector_does_not_turn_comment_count_into_read_comments(): observed = dom_observer(tools._INSPECT_SOURCE_CANDIDATES, arg={'max_comments': 5, 'saved_selectors': [], 'note_saved': True}, scenario='below_viewport') assert observed['discussion_state'] == 'comments_not_in_view' and observed['visible_comment_count'] == 0 assert observed['candidates'] == [] and observed['saved_count'] == 1 navigation = dom_observer(tools._OBSERVE_DISCUSSION_NAVIGATION, scenario='below_viewport') assert navigation['visible_comment_count'] == 0 and navigation['comment_control_selector'] == '#comment-entry' assert navigation['scroll']['selector'] == 'html' def test_actual_visible_discussion_has_independent_comment_and_reply_containers(): observed = dom_observer(tools._INSPECT_SOURCE_CANDIDATES, arg={'max_comments': 5, 'saved_selectors': [], 'note_saved': True}) assert observed['discussion_state'] == 'visible_comments' assert [(row['source_type'],row['source_selector']) for row in observed['candidates']] == [ ('comment','#comment-content'),('reply','#reply-content')]