from __future__ import annotations

import json
import unittest
from unittest.mock import patch

from src.openai_compat import (
    OpenAICompatClient,
    OpenAICompatError,
    _anthropic_base_url,
    _build_response_format,
    _join_url,
    _normalize_content,
    _optional_int,
    _parse_tool_arguments,
    _parse_usage,
    _temperature_for_model,
    _uses_anthropic_messages_api,
)
from src.agent_types import ModelConfig, OutputSchemaConfig, UsageStats


class FakeHTTPResponse:
    def __init__(self, payload: dict[str, object]) -> None:
        self.payload = payload

    def read(self) -> bytes:
        return json.dumps(self.payload).encode('utf-8')

    def __enter__(self) -> 'FakeHTTPResponse':
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        return None


class FakeStreamingHTTPResponse:
    def __init__(self, payloads: list[dict[str, object]]) -> None:
        self.lines: list[bytes] = []
        for payload in payloads:
            self.lines.append(b'event: message\n')
            self.lines.append(f'data: {json.dumps(payload)}\n'.encode('utf-8'))
            self.lines.append(b'\n')

    def readline(self) -> bytes:
        if not self.lines:
            return b''
        return self.lines.pop(0)

    def __enter__(self) -> 'FakeStreamingHTTPResponse':
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        return None


class TestJoinUrl(unittest.TestCase):
    def test_base_with_trailing_slash(self):
        self.assertEqual(_join_url('http://localhost:8000/', 'v1/chat'), 'http://localhost:8000/v1/chat')

    def test_base_without_trailing_slash(self):
        self.assertEqual(_join_url('http://localhost:8000', 'v1/chat'), 'http://localhost:8000/v1/chat')

    def test_suffix_with_leading_slash(self):
        self.assertEqual(_join_url('http://localhost:8000', '/v1/chat'), 'http://localhost:8000/v1/chat')


class TestNormalizeContent(unittest.TestCase):
    def test_string_passthrough(self):
        self.assertEqual(_normalize_content('hello'), 'hello')

    def test_none_returns_empty(self):
        self.assertEqual(_normalize_content(None), '')

    def test_list_of_strings_joined(self):
        self.assertEqual(_normalize_content(['hello', ' ', 'world']), 'hello world')

    def test_list_of_text_dicts(self):
        items = [{'type': 'text', 'text': 'hello'}, {'type': 'text', 'text': ' world'}]
        self.assertEqual(_normalize_content(items), 'hello world')

    def test_list_of_mixed_items(self):
        items = ['start ', {'type': 'text', 'text': 'middle'}, ' end']
        self.assertEqual(_normalize_content(items), 'start middle end')

    def test_non_string_non_list_returns_str(self):
        self.assertEqual(_normalize_content(42), '42')


class TestParseToolArguments(unittest.TestCase):
    def test_dict_passthrough(self):
        d = {'key': 'value'}
        self.assertIs(_parse_tool_arguments(d), d)

    def test_valid_json_string(self):
        self.assertEqual(_parse_tool_arguments('{"a": 1}'), {'a': 1})

    def test_empty_string_returns_empty_dict(self):
        self.assertEqual(_parse_tool_arguments(''), {})

    def test_none_returns_empty_dict(self):
        self.assertEqual(_parse_tool_arguments(None), {})

    def test_invalid_json_raises(self):
        with self.assertRaises(OpenAICompatError):
            _parse_tool_arguments('{bad json}')

    def test_json_non_dict_raises(self):
        with self.assertRaises(OpenAICompatError):
            _parse_tool_arguments('[1, 2, 3]')

    def test_unsupported_type_raises(self):
        with self.assertRaises(OpenAICompatError):
            _parse_tool_arguments(12345)


class TestParseUsage(unittest.TestCase):
    def test_standard_fields(self):
        usage = _parse_usage({'input_tokens': 10, 'output_tokens': 20})
        self.assertEqual(usage.input_tokens, 10)
        self.assertEqual(usage.output_tokens, 20)

    def test_prompt_completion_aliases(self):
        usage = _parse_usage({'prompt_tokens': 15, 'completion_tokens': 25})
        self.assertEqual(usage.input_tokens, 15)
        self.assertEqual(usage.output_tokens, 25)

    def test_ollama_aliases(self):
        usage = _parse_usage({'prompt_eval_count': 12, 'eval_count': 18})
        self.assertEqual(usage.input_tokens, 12)
        self.assertEqual(usage.output_tokens, 18)

    def test_cache_tokens(self):
        usage = _parse_usage({
            'input_tokens': 1,
            'output_tokens': 1,
            'cache_creation_input_tokens': 100,
            'cache_read_input_tokens': 200,
        })
        self.assertEqual(usage.cache_creation_input_tokens, 100)
        self.assertEqual(usage.cache_read_input_tokens, 200)

    def test_reasoning_tokens_top_level_and_details(self):
        usage_top = _parse_usage({'input_tokens': 1, 'output_tokens': 1, 'reasoning_tokens': 50})
        self.assertEqual(usage_top.reasoning_tokens, 50)

        usage_details = _parse_usage({
            'input_tokens': 1,
            'output_tokens': 1,
            'completion_tokens_details': {'reasoning_tokens': 75},
        })
        self.assertEqual(usage_details.reasoning_tokens, 75)

    def test_non_dict_returns_empty(self):
        usage = _parse_usage('not a dict')
        self.assertEqual(usage, UsageStats())

    def test_string_number_coercion(self):
        usage = _parse_usage({'input_tokens': '10', 'output_tokens': '20'})
        self.assertEqual(usage.input_tokens, 10)
        self.assertEqual(usage.output_tokens, 20)


class TestBuildResponseFormat(unittest.TestCase):
    def test_none_returns_none(self):
        self.assertIsNone(_build_response_format(None))

    def test_valid_schema(self):
        schema = OutputSchemaConfig(
            name='test_schema',
            schema={'type': 'object', 'properties': {'x': {'type': 'integer'}}},
            strict=True,
        )
        result = _build_response_format(schema)
        self.assertEqual(result, {
            'type': 'json_schema',
            'json_schema': {
                'name': 'test_schema',
                'schema': {'type': 'object', 'properties': {'x': {'type': 'integer'}}},
                'strict': True,
            },
        })


class TestOptionalInt(unittest.TestCase):
    def test_int_passthrough(self):
        self.assertEqual(_optional_int(42), 42)

    def test_float_truncated(self):
        self.assertEqual(_optional_int(3.9), 3)

    def test_string_parsed(self):
        self.assertEqual(_optional_int('7'), 7)

    def test_bool_returns_zero(self):
        self.assertEqual(_optional_int(True), 0)
        self.assertEqual(_optional_int(False), 0)

    def test_none_returns_zero(self):
        self.assertEqual(_optional_int(None), 0)

    def test_invalid_string_returns_zero(self):
        self.assertEqual(_optional_int('abc'), 0)


class TestProviderTemperatureFloor(unittest.TestCase):
    def test_minimax_temperature_is_clamped_to_provider_minimum(self):
        self.assertEqual(_temperature_for_model('minimax/MiniMax-M2.5', 0.0), 0.01)

    def test_wenxin_temperature_is_clamped_to_provider_minimum(self):
        self.assertEqual(_temperature_for_model('wenxin/ernie-4.0-turbo-128k', 0.0), 0.1)

    def test_regular_models_keep_configured_temperature(self):
        self.assertEqual(_temperature_for_model('xiaomi/mimo-v2-flash', 0.0), 0.0)

    def test_payload_uses_temperature_floor(self):
        client = OpenAICompatClient(
            ModelConfig(model='minimax/MiniMax-M2.5', temperature=0.0)
        )
        payload = client._build_payload(  # noqa: SLF001 - verify payload compatibility.
            messages=[{'role': 'user', 'content': 'hi'}],
            tools=[],
            stream=False,
            output_schema=None,
        )
        self.assertEqual(payload['temperature'], 0.01)

    def test_payload_omits_tool_choice_when_no_tools_are_supplied(self):
        client = OpenAICompatClient(ModelConfig(model='azure_openai/gpt-4o-mini'))
        payload = client._build_payload(  # noqa: SLF001 - verify provider compatibility.
            messages=[{'role': 'user', 'content': 'hi'}],
            tools=[],
            stream=False,
            output_schema=None,
        )
        self.assertNotIn('tools', payload)
        self.assertNotIn('tool_choice', payload)

    def test_payload_includes_tool_choice_when_tools_are_supplied(self):
        client = OpenAICompatClient(ModelConfig(model='azure_openai/gpt-4o-mini'))
        payload = client._build_payload(  # noqa: SLF001 - verify provider compatibility.
            messages=[{'role': 'user', 'content': 'hi'}],
            tools=[{'type': 'function', 'function': {'name': 'noop', 'parameters': {}}}],
            stream=False,
            output_schema=None,
        )
        self.assertIn('tools', payload)
        self.assertEqual(payload['tool_choice'], 'auto')


class TestAnthropicMessagesRouting(unittest.TestCase):
    def test_ppio_pa_claude_uses_anthropic_messages_api(self):
        self.assertTrue(
            _uses_anthropic_messages_api(
                'ppio/pa/claude-opus-4-7',
                'http://model.mify.ai.srv/v1',
            )
        )
        self.assertFalse(
            _uses_anthropic_messages_api(
                'ppio/gemini-2.5-pro',
                'http://model.mify.ai.srv/v1',
            )
        )

    def test_anthropic_base_url_is_derived_from_openai_base(self):
        self.assertEqual(
            _anthropic_base_url('http://model.mify.ai.srv/v1'),
            'http://model.mify.ai.srv/anthropic',
        )
        self.assertEqual(
            _anthropic_base_url('http://model.mify.ai.srv/anthropic'),
            'http://model.mify.ai.srv/anthropic',
        )

    def test_anthropic_complete_converts_tools_and_parses_tool_use(self):
        recorded: dict[str, object] = {}

        def fake_urlopen(request_obj, timeout=None):  # noqa: ANN001
            recorded['url'] = request_obj.full_url
            recorded['payload'] = json.loads(request_obj.data.decode('utf-8'))
            return FakeHTTPResponse(
                {
                    'id': 'msg_1',
                    'type': 'message',
                    'role': 'assistant',
                    'content': [
                        {'type': 'text', 'text': '我来读取文件。'},
                        {
                            'type': 'tool_use',
                            'id': 'toolu_1',
                            'name': 'read_file',
                            'input': {'path': 'hello.txt'},
                        },
                    ],
                    'stop_reason': 'tool_use',
                    'usage': {'input_tokens': 10, 'output_tokens': 4},
                }
            )

        client = OpenAICompatClient(
            ModelConfig(
                model='ppio/pa/claude-opus-4-7',
                base_url='http://model.mify.ai.srv/v1',
                api_key='token',
            )
        )
        with patch('src.openai_compat.request.urlopen', side_effect=fake_urlopen):
            turn = client.complete(
                messages=[
                    {'role': 'system', 'content': '你是工具型助手。'},
                    {'role': 'user', 'content': '读取 hello.txt'},
                ],
                tools=[
                    {
                        'type': 'function',
                        'function': {
                            'name': 'read_file',
                            'description': '读取文件',
                            'parameters': {
                                'type': 'object',
                                'properties': {'path': {'type': 'string'}},
                                'required': ['path'],
                            },
                        },
                    }
                ],
            )

        self.assertEqual(
            recorded['url'],
            'http://model.mify.ai.srv/anthropic/v1/messages',
        )
        payload = recorded['payload']
        self.assertEqual(payload['model'], 'ppio/pa/claude-opus-4-7')
        self.assertEqual(payload['system'], '你是工具型助手。')
        self.assertEqual(payload['tools'][0]['name'], 'read_file')
        self.assertEqual(payload['tools'][0]['input_schema']['required'], ['path'])
        self.assertEqual(turn.content, '我来读取文件。')
        self.assertEqual(turn.finish_reason, 'tool_use')
        self.assertEqual(turn.tool_calls[0].id, 'toolu_1')
        self.assertEqual(turn.tool_calls[0].arguments, {'path': 'hello.txt'})
        self.assertEqual(turn.usage.input_tokens, 10)

    def test_anthropic_stream_parses_text_usage_and_tool_use(self):
        payloads = [
            {
                'type': 'message_start',
                'message': {'usage': {'input_tokens': 7, 'output_tokens': 1}},
            },
            {
                'type': 'content_block_start',
                'index': 0,
                'content_block': {'type': 'text', 'text': ''},
            },
            {
                'type': 'content_block_delta',
                'index': 0,
                'delta': {'type': 'text_delta', 'text': '读取'},
            },
            {
                'type': 'content_block_start',
                'index': 1,
                'content_block': {
                    'type': 'tool_use',
                    'id': 'toolu_1',
                    'name': 'read_file',
                    'input': {},
                },
            },
            {
                'type': 'content_block_delta',
                'index': 1,
                'delta': {'type': 'input_json_delta', 'partial_json': '{"path":'},
            },
            {
                'type': 'content_block_delta',
                'index': 1,
                'delta': {'type': 'input_json_delta', 'partial_json': '"hello.txt"}'},
            },
            {
                'type': 'message_delta',
                'delta': {'stop_reason': 'tool_use'},
                'usage': {'output_tokens': 5},
            },
            {'type': 'message_stop'},
        ]

        def fake_urlopen(request_obj, timeout=None):  # noqa: ANN001
            return FakeStreamingHTTPResponse(payloads)

        client = OpenAICompatClient(
            ModelConfig(
                model='ppio/pa/claude-opus-4-7',
                base_url='http://model.mify.ai.srv/v1',
                api_key='token',
            )
        )
        with patch('src.openai_compat.request.urlopen', side_effect=fake_urlopen):
            events = list(
                client.stream(
                    messages=[{'role': 'user', 'content': '读取 hello.txt'}],
                    tools=[
                        {
                            'type': 'function',
                            'function': {
                                'name': 'read_file',
                                'parameters': {'type': 'object'},
                            },
                        }
                    ],
                )
            )

        self.assertEqual(events[0].type, 'message_start')
        self.assertEqual(
            ''.join(event.delta for event in events if event.type == 'content_delta'),
            '读取',
        )
        tool_events = [event for event in events if event.type == 'tool_call_delta']
        self.assertEqual(tool_events[0].tool_call_index, 0)
        self.assertEqual(tool_events[0].tool_call_id, 'toolu_1')
        self.assertEqual(tool_events[0].tool_name, 'read_file')
        self.assertEqual(
            ''.join(event.arguments_delta for event in tool_events),
            '{"path":"hello.txt"}',
        )
        self.assertTrue(any(event.type == 'usage' for event in events))
        self.assertEqual(events[-1].type, 'message_stop')
        self.assertEqual(events[-1].finish_reason, 'tool_use')


if __name__ == '__main__':
    unittest.main()
