"""Native reply-target verification must return to and recheck the same source."""
import json
import subprocess
from pathlib import Path
from types import SimpleNamespace

import pytest

from backend.browser import tools


NOTE = 'a' * 24
NOTE_URL = 'https://www.xiaohongshu.com/explore/' + NOTE
PARENT = 'c' * 24
PROFILE = 'https://www.xiaohongshu.com/user/profile/' + PARENT


def observed_point(*, occluded=False, source_key='child-key', outside=False, label='目标用户'):
    target = {'tag':'span','attrs':{'id':'target'},'text':label}
    child = {'attrs':{'id':'child','class':'comment-item-sub','data-comment-id':'child-key'},
             'children':[{'attrs':{'id':'content','class':'content'},'text':'回复 ',
                          'children':[] if outside else [target]}]}
    tree = {'attrs':{'id':'noteContainer'}, 'children':[child], 'hit_target_id':'target', 'occluded':occluded}
    if outside:
        tree['children'].append({'attrs':{'class':'comment-item'},'children':[target]})
    request = {'code':tools._OBSERVE_REPLY_TARGET_POINT,
               'arg':{'source_selector':'#content','target_selector':'#target','source_key':source_key},'tree':tree}
    result = subprocess.run(['node', str(Path(__file__).with_name('reply_dom_fixture_d41.cjs'))],
                            input=json.dumps(request, ensure_ascii=False), text=True, capture_output=True, check=True, timeout=5)
    return json.loads(result.stdout)


def test_target_point_observer_requires_actual_painted_same_reply_element():
    result = observed_point()
    assert result['source_key'] == 'child-key'
    assert 0 <= result['click_point']['x'] < result['click_point']['viewport_width']
    assert 0 <= result['click_point']['y'] < result['click_point']['viewport_height']


@pytest.mark.parametrize('options,error', [
    ({'occluded':True}, 'reply_target_not_visible'),
    ({'outside':True}, 'target_outside_reply'),
    ({'source_key':'a-different-key'}, 'source_key_changed'),
    ({'label':'关注'}, 'not_reply_target'),
])
def test_target_point_observer_rejects_occlusion_other_reply_changed_key_and_action_buttons(options, error):
    result = observed_point(**options)
    assert result['error'] == error and 'click_point' not in result


class Page:
    def __init__(self):
        self.url = NOTE_URL
        self.clicks = []
        self.pointer_moves = []
        self.backs = 0
        self.target_visible = True
        self.moves = True
        self.header = True
        self.hover = False
        self.back_fails = False
        self.return_changed = False
        self.parent_changed = False
        self.ambiguous = False
        self.profile_id = PARENT
        self.child_quote = '回复 目标用户：新的独立需求'
        self.after_back = None
        self.after_move = None
        self.point_shift = False

    async def get_url(self): return self.url

    async def evaluate(self, code, arg=None):
        if code == tools._OBSERVE_PROFILE_DESTINATION:
            return {'url': self.url, 'profile_header_visible': self.header and '/user/profile/' in self.url,
                    'hover_card_visible': self.hover}
        if code == tools._OBSERVE_REPLY_TARGET_POINT:
            assert arg == {'source_selector': '#child', 'target_selector': '#target', 'source_key': 'child-key'}
            if not self.target_visible:
                return {'error': 'reply_target_not_visible'}
            return {'url': self.url, 'source_key': 'child-key', 'target_text': '目标用户',
                    'click_point': {'x': 141 if self.point_shift else 140, 'y': 240, 'viewport_width': 400, 'viewport_height': 600}}
        assert code == tools._OBSERVE_SOURCE
        selector = arg['selector'] if isinstance(arg, dict) else arg
        verified = arg.get('verified_target_identity') if isinstance(arg, dict) else None
        values = {
            '#body': {'source_type': 'note', 'text': '正文'},
            '#root': {'source_type': 'comment', 'text': '根评论', 'source_key': 'root-key', 'author_profile_url': 'https://www.xiaohongshu.com/user/profile/' + 'b' * 24},
            '#parent': {'source_type': 'reply', 'text': '父回复', 'source_key': 'parent-key',
                        'parent_source_selector': '#root', 'root_source_selector': '#root',
                        'parent_source_key': 'root-key', 'root_source_key': 'root-key', 'parent_resolution': 'thread_root'},
            '#child': {'source_type': 'reply', 'text': self.child_quote, 'source_key': 'child-key',
                       'root_source_selector': '#root', 'root_source_key': 'root-key',
                       'reply_to': '目标用户', 'reply_to_selector': '#target', 'parent_resolution': 'target_identity_missing'},
        }
        value = dict(values[selector])
        if selector == '#parent' and self.parent_changed and self.backs:
            value['source_key'] = 'a-different-parent'
        if selector == '#child' and self.return_changed and self.backs:
            value['source_key'] = 'a-different-child'
        if selector == '#child' and verified:
            if verified == PARENT and not self.ambiguous:
                value.update(parent_source_selector='#parent', parent_source_key='a-different-parent' if self.parent_changed and self.backs else 'parent-key', parent_resolution='native_target_profile')
            else:
                value['parent_resolution'] = 'ambiguous_or_missing_target_comment'
        return {'url': self.url, **value}

    @property
    async def mouse(self):
        async def move(**point):
            self.pointer_moves.append(point)
            if self.after_move:
                self.after_move()
        async def click(**point):
            self.clicks.append(point)
            if self.moves:
                self.url = 'https://www.xiaohongshu.com/user/profile/' + self.profile_id
        return SimpleNamespace(move=move, click=click)

    async def go_back(self):
        self.backs += 1
        if self.back_fails:
            raise RuntimeError('native history unavailable')
        self.url = NOTE_URL
        if self.after_back:
            self.after_back()

    async def get_elements_by_css_selector(self, _selector):
        raise AssertionError('Reply targets must use fresh painted Mouse.click, never actor pre-scroll coordinates')


class Session:
    def __init__(self, page): self.page = page
    async def get_current_page(self): return self.page


@pytest.fixture
async def context(tmp_path, monkeypatch):
    task_id = 'native_reply_target_d42'
    monkeypatch.setattr(tools, 'task_doc_dir', lambda _: tmp_path)
    async def no_delay(_): pass
    monkeypatch.setattr(tools.asyncio, 'sleep', no_delay)
    tools.set_prospecting_context(task_id)
    tools.set_prospecting_run_context(task_id, 'first-run', goal_version=1)
    page, session = Page(), Session(None)
    session.page = page
    saved = {}
    for kind, selector, quote in [('note', '#body', '正文'), ('comment', '#root', '根评论'),
                                  ('reply', '#parent', '父回复'), ('reply', '#child', '新的独立需求')]:
        saved[selector], _ = await tools._save_source(task_id, session, source_type=kind, source_selector=selector, content=quote)
    yield task_id, page, session, saved
    tools.clear_task_context(task_id)


async def test_native_target_round_trip_refreshes_parent_and_child_for_the_current_goal(context):
    task_id, page, session, saved = context
    old_saved = tools.get_task_artifacts(task_id)['operation_evidence'][-1]
    tools.set_prospecting_run_context(task_id, 'second-run', goal_version=2)
    result = await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert result['status'] == 'resolved' and result['native_target_rechecked']
    assert page.clicks == [{'x':140,'y':240}] and page.backs == 1 and page.url == NOTE_URL
    assert page.pointer_moves == [{'x':140,'y':240}]
    assert saved['#child']['parent_source_id'] == saved['#parent']['source_id']
    assert saved['#child']['parent_resolution'] == 'native_target_profile'
    assert old_saved['outcome'] == 'success' and old_saved['details']['post_snapshot']['parent_source_id'] is None
    assert old_saved['details']['post_snapshot']['relations_checked'] is False
    current = tools.get_task_artifacts(task_id)['operation_evidence']
    current_ids = {row['details']['post_snapshot']['source_id'] for row in current
                   if row['outcome'] == 'success' and row['details'].get('trace', {}).get('goal_version') == 2}
    assert {saved['#parent']['source_id'], saved['#child']['source_id']} <= current_ids
    before_count = len(current)
    again = await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert again['status'] == 'resolved' and len(page.clicks) == 1
    assert len(page.pointer_moves) == 1
    assert len(tools.get_task_artifacts(task_id)['operation_evidence']) == before_count
    assert len(tools.get_task_artifacts(task_id)['sources']) == 4


@pytest.mark.parametrize('option,status', [
    ('ambiguous', 'parent_source_unresolved'),
    ('wrong_profile', 'parent_source_unresolved'),
    ('back_fails', 'return_failed'),
    ('return_changed', 'returned_source_unconfirmed'),
    ('parent_changed', 'parent_source_unresolved'),
])
async def test_uncertain_profile_parent_or_return_never_creates_a_successful_child_relation(context, option, status):
    task_id, page, session, saved = context
    if option == 'wrong_profile': page.profile_id = 'e' * 24
    else: setattr(page, option, True)
    result = await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert result['status'] == status
    successes = [row for row in tools.get_task_artifacts(task_id)['operation_evidence']
                 if row['outcome'] == 'success' and row['details']['post_snapshot']['relations_checked']
                 and row['details']['post_snapshot']['source_id'] == saved['#child']['source_id']]
    if option != 'parent_changed':
        assert successes == []
    else:
        assert result['parent_refresh']['status'] == 'missing_parent_refresh'


async def test_offscreen_target_is_not_clicked_or_treated_as_a_profile(context):
    task_id, page, session, saved = context
    page.target_visible = False
    result = await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert result['status'] == 'reply_target_not_visible' and page.clicks == [] and page.backs == 0
    assert saved['#child']['parent_source_id'] is None


async def test_hover_or_no_navigation_does_not_go_back_or_repeat_the_same_target(context):
    task_id, page, session, saved = context
    page.moves, page.hover = False, True
    result = await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert result['status'] == 'hover_only' and len(page.clicks) == 1 and page.backs == 0
    assert result['target_verified'] is False
    with pytest.raises(ValueError, match='不重复点击'):
        await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert len(page.clicks) == 1


@pytest.mark.parametrize('changed', ['point', 'occluded', 'source'])
async def test_hover_changes_require_new_observation_instead_of_clicking_stale_coordinates(context, changed):
    task_id, page, session, saved = context
    def change():
        if changed == 'point': page.point_shift = True
        elif changed == 'occluded': page.target_visible = False
        else: page.child_quote += ' 页面已经变化'
    page.after_move = change
    result = await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert result['status'] == 'target_changed_after_move' and result['target_verified'] is False
    assert page.pointer_moves == [{'x':140,'y':240}] and page.clicks == [] and page.backs == 0
    assert saved['#child']['parent_source_id'] is None


async def test_run_change_after_native_move_prevents_the_click(context):
    task_id, page, session, saved = context
    page.after_move = lambda: tools.set_prospecting_run_context(task_id, 'changed-run', goal_version=2)
    result = await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert result['status'] == 'run_changed' and result['target_verified'] is False
    assert page.clicks == [] and len(page.pointer_moves) == 1


async def test_profile_url_without_real_profile_header_does_not_create_proof_or_go_back(context):
    task_id, page, session, saved = context
    page.header = False
    result = await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert result['status'] == 'not_moved' and page.backs == 0
    assert tools._reply_target_proofs.get(task_id, {}) == {}


async def test_a_new_run_cannot_reuse_previous_native_target_proof(context):
    task_id, page, session, saved = context
    await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert tools._reply_target_proofs[task_id]
    tools.set_prospecting_run_context(task_id, 'new-run', goal_version=2)
    assert tools._reply_target_proofs.get(task_id, {}) == {}


async def test_execution_change_during_native_trip_cannot_bind_the_old_observation_to_the_new_run(context):
    task_id, page, session, saved = context
    page.after_back = lambda: tools.set_prospecting_run_context(task_id, 'another-run', goal_version=2)
    result = await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    assert result['status'] == 'run_changed'
    assert saved['#child']['parent_source_id'] is None
    assert tools._reply_target_proofs.get(task_id, {}) == {}


async def test_changed_container_invalidates_cached_proof_instead_of_reusing_the_nickname(context):
    task_id, page, session, saved = context
    await tools._resolve_reply_target(task_id, session, saved['#child']['source_id'])
    page.child_quote += ' 已有新变化'
    raw = await tools._observe_source(session, '#child')
    observed = await tools._apply_reply_target_proof(task_id, session, '#child', raw)
    assert observed['parent_resolution'] == 'target_identity_missing'
    assert tools._reply_target_proofs.get(task_id, {}) == {}


def test_registered_target_tool_exposes_only_the_saved_source_id_not_identity_or_coordinates(monkeypatch):
    monkeypatch.setattr(tools.settings, 'reply_relation_preflight_enabled', True)
    action = tools.build_prospecting_tools('schema_only').registry.registry.actions['resolve_reply_target']
    assert set(action.param_model.model_fields) == {'source_id'}