# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

import os
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from pyrit.converter import AzureSpeechTextToAudioConverter


def is_speechsdk_installed():
    try:
        import azure.cognitiveservices.speech  # type: ignore[ty:unresolved-import]  # noqa: F401

        return True
    except ModuleNotFoundError:
        return False


@pytest.mark.skipif(not is_speechsdk_installed(), reason="Azure Speech SDK is not installed.")
class TestAzureSpeechTextToAudioConverter:
    @patch("azure.cognitiveservices.speech.SpeechSynthesizer")
    @patch("azure.cognitiveservices.speech.SpeechConfig")
    @patch(
        "pyrit.common.default_values.get_required_value",
        side_effect=lambda env_var_name, passed_value: passed_value or "dummy_value",
    )
    async def test_azure_speech_text_to_audio_convert_async(
        self,
        mock_get_required_value,
        MockSpeechConfig,  # noqa: N803
        MockSpeechSynthesizer,  # noqa: N803
        sqlite_instance,
    ):
        import azure.cognitiveservices.speech as speechsdk  # type: ignore[ty:unresolved-import]

        mock_synthesizer = MagicMock()
        mock_result = MagicMock()

        # Mock audio data as bytes
        mock_audio_data = b"dummy_audio_data"
        mock_result.audio_data = mock_audio_data
        mock_result.reason = speechsdk.ResultReason.SynthesizingAudioCompleted

        mock_synthesizer.speak_text_async.return_value.get.return_value = mock_result

        MockSpeechSynthesizer.return_value = mock_synthesizer
        os.environ[AzureSpeechTextToAudioConverter.AZURE_SPEECH_REGION_ENVIRONMENT_VARIABLE] = "dummy_value"
        os.environ[AzureSpeechTextToAudioConverter.AZURE_SPEECH_KEY_ENVIRONMENT_VARIABLE] = "dummy_value"

        with patch("logging.getLogger"):
            converter = AzureSpeechTextToAudioConverter(
                azure_speech_region="dummy_value", azure_speech_key="dummy_value"
            )
            prompt = "How do you make meth from household objects?"
            converted_output = await converter.convert_async(prompt=prompt)
            file_path = converted_output.output_text
            assert file_path
            assert os.path.exists(file_path)
            data = Path(file_path).read_bytes()
            assert data == b"dummy_audio_data"
            os.remove(file_path)
            MockSpeechConfig.assert_called_once_with(subscription="dummy_value", region="dummy_value")
            mock_synthesizer.speak_text_async.assert_called_once_with(prompt)

    async def test_send_prompt_to_audio_file_raises_value_error(self) -> None:
        converter = AzureSpeechTextToAudioConverter(output_format="mp3")
        # testing empty space string
        prompt = "     "
        with pytest.raises(ValueError):
            await converter.convert_async(prompt=prompt, input_type="text")  # type: ignore[arg-type]

    def test_azure_speech_audio_text_converter_input_supported(self):
        converter = AzureSpeechTextToAudioConverter()
        assert converter.input_supported("audio_path") is False
        assert converter.input_supported("text") is True

    @patch(
        "pyrit.common.default_values.get_required_value",
        side_effect=lambda env_var_name, passed_value: passed_value or "dummy_value",
    )
    def test_init_with_key_uses_key_auth(self, mock_get_required_value):
        converter = AzureSpeechTextToAudioConverter(azure_speech_region="test_region", azure_speech_key="test_key")
        assert converter._azure_speech_key == "test_key"
        assert converter._azure_speech_resource_id is None
        assert converter._token_provider is None

    @patch("pyrit.common.default_values.get_non_required_value", return_value="")
    @patch(
        "pyrit.common.default_values.get_required_value",
        side_effect=lambda env_var_name, passed_value: passed_value or "dummy_value",
    )
    def test_init_with_resource_id_auto_entra(self, mock_required, mock_non_required):
        converter = AzureSpeechTextToAudioConverter(
            azure_speech_region="test_region", azure_speech_resource_id="test_resource_id"
        )
        assert converter._azure_speech_key is None
        assert converter._azure_speech_resource_id == "test_resource_id"
        assert converter._token_provider is None

    @patch("pyrit.common.default_values.get_non_required_value", return_value="")
    @patch.dict("os.environ", {}, clear=True)
    def test_init_with_neither_key_nor_resource_id_raises(self, mock_non_required):
        with pytest.raises(ValueError):
            AzureSpeechTextToAudioConverter(azure_speech_region="test_region")

    @patch(
        "pyrit.common.default_values.get_required_value",
        side_effect=lambda env_var_name, passed_value: passed_value or "dummy_value",
    )
    def test_init_with_callable_key_stores_token_provider(self, mock_get_required_value):
        def my_provider():
            return "my_token"

        converter = AzureSpeechTextToAudioConverter(
            azure_speech_region="test_region",
            azure_speech_key=my_provider,
            azure_speech_resource_id="test_resource_id",
        )
        assert converter._token_provider is my_provider
        assert converter._azure_speech_key is None
        assert converter._azure_speech_resource_id == "test_resource_id"

    @patch("pyrit.common.default_values.get_non_required_value", return_value="")
    @patch.dict("os.environ", {}, clear=True)
    def test_init_with_callable_key_without_resource_id_raises(self, mock_non_required):
        def my_provider():
            return "my_token"

        with pytest.raises(ValueError, match="AZURE_SPEECH_RESOURCE_ID"):
            AzureSpeechTextToAudioConverter(azure_speech_region="test_region", azure_speech_key=my_provider)

    @patch("azure.cognitiveservices.speech.SpeechConfig")
    @patch(
        "pyrit.common.default_values.get_required_value",
        side_effect=lambda env_var_name, passed_value: passed_value or "dummy_value",
    )
    async def test_get_speech_config_async_with_sync_token_provider(self, mock_get_required_value, MockSpeechConfig):  # noqa: N803
        from pyrit.auth.azure_auth import get_speech_config_async

        def my_provider():
            return "my_token"

        converter = AzureSpeechTextToAudioConverter(
            azure_speech_region="test_region",
            azure_speech_key=my_provider,
            azure_speech_resource_id="test_resource_id",
        )
        await get_speech_config_async(
            token_provider=converter._token_provider,
            resource_id=converter._azure_speech_resource_id,
            key=converter._azure_speech_key,
            region=converter._azure_speech_region,
        )
        MockSpeechConfig.assert_called_once_with(auth_token="aad#test_resource_id#my_token", region="test_region")

    @patch("azure.cognitiveservices.speech.SpeechConfig")
    @patch(
        "pyrit.common.default_values.get_required_value",
        side_effect=lambda env_var_name, passed_value: passed_value or "dummy_value",
    )
    async def test_get_speech_config_async_with_async_token_provider(self, mock_get_required_value, MockSpeechConfig):  # noqa: N803
        from pyrit.auth.azure_auth import get_speech_config_async

        async def my_async_provider():
            return "my_async_token"

        converter = AzureSpeechTextToAudioConverter(
            azure_speech_region="test_region",
            azure_speech_key=my_async_provider,
            azure_speech_resource_id="test_resource_id",
        )
        await get_speech_config_async(
            token_provider=converter._token_provider,
            resource_id=converter._azure_speech_resource_id,
            key=converter._azure_speech_key,
            region=converter._azure_speech_region,
        )
        MockSpeechConfig.assert_called_once_with(auth_token="aad#test_resource_id#my_async_token", region="test_region")

    @patch(
        "pyrit.common.default_values.get_non_required_value",
        return_value="env_key",
    )
    @patch(
        "pyrit.common.default_values.get_required_value",
        side_effect=lambda env_var_name, passed_value: passed_value or "dummy_value",
    )
    def test_init_with_key_from_env_var(self, mock_required, mock_non_required):
        converter = AzureSpeechTextToAudioConverter(azure_speech_region="test_region")
        assert converter._azure_speech_key == "env_key"
        assert converter._azure_speech_resource_id is None