from __future__ import annotations

import tempfile
import unittest
from pathlib import Path

from src.agent_tools import build_tool_context, default_tool_registry, execute_tool
from src.agent_types import AgentPermissions, AgentRuntimeConfig
from src.lsp_runtime import LSPRuntime


class ExtendedToolTests(unittest.TestCase):
    def test_web_fetch_rejects_file_url(self) -> None:
        """Verify that file:// URLs are blocked to prevent SSRF attacks."""
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            target = workspace / 'page.txt'
            target.write_text('hello from web fetch\n', encoding='utf-8')
            context = build_tool_context(
                AgentRuntimeConfig(cwd=workspace),
                tool_registry=default_tool_registry(),
            )
            result = execute_tool(
                default_tool_registry(),
                'web_fetch',
                {'url': target.resolve().as_uri()},
                context,
            )

        self.assertFalse(result.ok)
        self.assertIn('http or https', result.content)

    def test_tool_search_lists_matching_tools(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as tmp_dir:
            context = build_tool_context(
                AgentRuntimeConfig(cwd=Path(tmp_dir)),
                tool_registry=registry,
            )
            result = execute_tool(
                registry,
                'tool_search',
                {'query': 'file'},
                context,
            )

        self.assertTrue(result.ok)
        self.assertIn('# Tool Search', result.content)
        self.assertIn('read_file', result.content)
        self.assertIn('write_file', result.content)

    def test_grep_search_skips_generated_dirs_by_default(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / '.next' / 'static').mkdir(parents=True)
            (workspace / '.next' / 'static' / 'bundle.js').write_text(
                'router_session_parquet should not be searched by default\n',
                encoding='utf-8',
            )
            (workspace / 'README.md').write_text(
                'router_session_parquet is documented here\n',
                encoding='utf-8',
            )
            context = build_tool_context(
                AgentRuntimeConfig(cwd=workspace),
                tool_registry=registry,
            )
            result = execute_tool(
                registry,
                'grep_search',
                {'pattern': 'router_session_parquet'},
                context,
            )

        self.assertTrue(result.ok)
        self.assertIn('README.md:1:', result.content)
        self.assertNotIn('.next/static/bundle.js', result.content)

    def test_grep_search_truncates_very_long_lines(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            long_line = 'prefix needle ' + ('x' * 5000)
            (workspace / 'large.txt').write_text(long_line, encoding='utf-8')
            context = build_tool_context(
                AgentRuntimeConfig(cwd=workspace, max_output_chars=1200),
                tool_registry=registry,
            )
            result = execute_tool(
                registry,
                'grep_search',
                {'pattern': 'needle'},
                context,
            )

        self.assertTrue(result.ok)
        self.assertIn('large.txt:1:', result.content)
        self.assertIn('[line truncated,', result.content)
        self.assertLess(len(result.content), 1200)

    def test_read_file_can_read_explicit_external_path(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as workspace_dir, tempfile.TemporaryDirectory() as external_dir:
            external = Path(external_dir) / 'reference.txt'
            external.write_text('external reference\n', encoding='utf-8')
            context = build_tool_context(
                AgentRuntimeConfig(cwd=Path(workspace_dir)),
                tool_registry=registry,
            )
            result = execute_tool(
                registry,
                'read_file',
                {'path': str(external)},
                context,
            )

        self.assertTrue(result.ok)
        self.assertIn('external reference', result.content)

    def test_write_file_blocks_platform_code_paths(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'src').mkdir()
            (workspace / 'src' / 'agent_tools.py').write_text('', encoding='utf-8')
            (workspace / 'backend' / 'api').mkdir(parents=True)
            (workspace / 'backend' / 'api' / 'server.py').write_text('', encoding='utf-8')
            (workspace / 'frontend' / 'app').mkdir(parents=True)
            scratchpad = (
                workspace
                / '.port_sessions'
                / 'accounts'
                / 'user'
                / 'sessions'
                / 'thread'
                / 'scratchpad'
            )
            scratchpad.mkdir(parents=True)
            context = build_tool_context(
                AgentRuntimeConfig(
                    cwd=workspace,
                    permissions=AgentPermissions(allow_file_write=True),
                ),
                scratchpad_directory=scratchpad,
                tool_registry=registry,
            )

            result = execute_tool(
                registry,
                'write_file',
                {'path': 'src/new_file.py', 'content': 'print(1)\n'},
                context,
            )
            output_result = execute_tool(
                registry,
                'write_file',
                {
                    'path': '.port_sessions/accounts/user/sessions/thread/output/report.txt',
                    'content': 'ok\n',
                },
                context,
            )

        self.assertFalse(result.ok)
        self.assertEqual(result.metadata.get('error_kind'), 'permission_denied')
        self.assertTrue(output_result.ok)

    def test_logical_session_paths_route_to_current_session(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            scratchpad = (
                workspace
                / '.port_sessions'
                / 'accounts'
                / 'user'
                / 'sessions'
                / 'thread'
                / 'scratchpad'
            )
            scratchpad.mkdir(parents=True)
            context = build_tool_context(
                AgentRuntimeConfig(
                    cwd=workspace,
                    permissions=AgentPermissions(allow_file_write=True),
                ),
                scratchpad_directory=scratchpad,
                tool_registry=registry,
            )
            write_result = execute_tool(
                registry,
                'write_file',
                {'path': 'output/report.md', 'content': 'session report\n'},
                context,
            )
            read_result = execute_tool(
                registry,
                'read_file',
                {'path': 'output/report.md'},
                context,
            )
            session_file_exists = (scratchpad.parent / 'output' / 'report.md').is_file()
            root_file_exists = (workspace / 'output' / 'report.md').exists()

        self.assertTrue(write_result.ok, write_result.content)
        self.assertTrue(read_result.ok, read_result.content)
        self.assertIn('session report', read_result.content)
        self.assertTrue(session_file_exists)
        self.assertFalse(root_file_exists)

    def test_write_file_serializes_structured_payloads(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            scratchpad = workspace / '.port_sessions' / 'accounts' / 'user' / 'sessions' / 'thread' / 'scratchpad'
            scratchpad.mkdir(parents=True)
            context = build_tool_context(
                AgentRuntimeConfig(
                    cwd=workspace,
                    permissions=AgentPermissions(allow_file_write=True),
                ),
                scratchpad_directory=scratchpad,
                tool_registry=registry,
            )
            json_result = execute_tool(
                registry,
                'write_file',
                {
                    'path': 'output/data.json',
                    'json_content': {'query': '导航到公司', 'target': 'Agent(tag="地图导航")'},
                    'newline_at_end': True,
                },
                context,
            )
            jsonl_result = execute_tool(
                registry,
                'write_file',
                {
                    'path': 'output/data.jsonl',
                    'jsonl_records': [
                        {'query': '附近的奶茶', 'target': 'Agent(tag="餐饮服务")'},
                        {'query': '导航去海底捞', 'target': 'Agent(tag="地图导航")'},
                    ],
                    'newline_at_end': True,
                },
                context,
            )
            csv_result = execute_tool(
                registry,
                'write_file',
                {
                    'path': 'output/data.csv',
                    'csv_headers': ['query', 'target'],
                    'csv_rows': [
                        ['附近的奶茶', 'Agent(tag="餐饮服务")'],
                        ['导航去海底捞', 'Agent(tag="地图导航")'],
                    ],
                    'newline_at_end': True,
                },
                context,
            )
            output_dir = scratchpad.parent / 'output'
            json_text = (output_dir / 'data.json').read_text(encoding='utf-8')
            jsonl_text = (output_dir / 'data.jsonl').read_text(encoding='utf-8')
            csv_text = (output_dir / 'data.csv').read_text(encoding='utf-8')

        self.assertTrue(json_result.ok, json_result.content)
        self.assertTrue(jsonl_result.ok, jsonl_result.content)
        self.assertTrue(csv_result.ok, csv_result.content)
        self.assertIn('"query": "导航到公司"', json_text)
        self.assertEqual(
            jsonl_text.splitlines()[0],
            '{"query":"附近的奶茶","target":"Agent(tag=\\"餐饮服务\\")"}',
        )
        self.assertIn('query,target', csv_text)

    def test_sleep_tool_waits_briefly_and_returns_metadata(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as tmp_dir:
            context = build_tool_context(
                AgentRuntimeConfig(cwd=Path(tmp_dir)),
                tool_registry=registry,
            )
            result = execute_tool(
                registry,
                'sleep',
                {'seconds': 0.01},
                context,
            )

        self.assertTrue(result.ok)
        self.assertIn('slept for', result.content)
        self.assertEqual(result.metadata.get('action'), 'sleep')

    def test_notebook_edit_updates_ipynb_cell(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            notebook = workspace / 'demo.ipynb'
            notebook.write_text(
                '{\n'
                ' "cells": [\n'
                '  {"cell_type": "code", "metadata": {}, "source": ["print(1)\\n"], "outputs": [], "execution_count": null}\n'
                ' ],\n'
                ' "metadata": {},\n'
                ' "nbformat": 4,\n'
                ' "nbformat_minor": 5\n'
                '}\n',
                encoding='utf-8',
            )
            context = build_tool_context(
                AgentRuntimeConfig(
                    cwd=workspace,
                    permissions=AgentPermissions(allow_file_write=True),
                ),
                tool_registry=registry,
            )
            result = execute_tool(
                registry,
                'notebook_edit',
                {'path': 'demo.ipynb', 'cell_index': 0, 'source': 'print(2)\n'},
                context,
            )
            updated = notebook.read_text(encoding='utf-8')

        self.assertTrue(result.ok)
        self.assertIn('updated notebook cell 0', result.content)
        self.assertIn('print(2)', updated)
        self.assertEqual(result.metadata.get('action'), 'notebook_edit')

    def test_lsp_tool_returns_definition_report(self) -> None:
        registry = default_tool_registry()
        with tempfile.TemporaryDirectory() as tmp_dir:
            workspace = Path(tmp_dir)
            (workspace / 'sample.py').write_text(
                'def helper(value):\n'
                '    return value * 2\n'
                '\n'
                'def run(item):\n'
                '    return helper(item)\n',
                encoding='utf-8',
            )
            context = build_tool_context(
                AgentRuntimeConfig(cwd=workspace),
                tool_registry=registry,
                lsp_runtime=LSPRuntime.from_workspace(workspace),
            )
            result = execute_tool(
                registry,
                'LSP',
                {
                    'operation': 'goToDefinition',
                    'file_path': 'sample.py',
                    'line': 5,
                    'character': 12,
                },
                context,
            )

        self.assertTrue(result.ok)
        self.assertIn('# LSP Definition', result.content)
        self.assertIn('function helper', result.content)
        self.assertEqual(result.metadata.get('action'), 'lsp_query')
        self.assertEqual(result.metadata.get('operation'), 'goToDefinition')
