import threading
import unittest
from unittest import mock

import pykka

from mopidy import audio
from mopidy.audio.constants import PlaybackState
from mopidy.internal import path
from mopidy.internal.gi import Gst

from tests import dummy_audio, path_to_data_dir

# We want to make sure both our real audio class and the fake one behave
# correctly. So each test is first run against the real class, then repeated
# against our dummy.


class BaseTest(unittest.TestCase):
    uris = [
        path.path_to_uri(path_to_data_dir("song1.wav")),
        path.path_to_uri(path_to_data_dir("song2.wav")),
    ]

    audio_class = audio.Audio

    def setUp(self):  # noqa: N802
        config = {
            "audio": {
                "buffer_time": None,
                "mixer": "foomixer",
                "mixer_volume": None,
                "output": "testoutput",
                "visualizer": None,
            },
            "proxy": {"hostname": ""},
        }
        self.song_uri = path.path_to_uri(path_to_data_dir("song1.wav"))
        self.audio = self.audio_class.start(config=config, mixer=None).proxy()

    def tearDown(self):  # noqa
        pykka.ActorRegistry.stop_all()

    def possibly_trigger_fake_playback_error(self, uri):
        pass

    def possibly_trigger_fake_about_to_finish(self):
        pass

    def possibly_trigger_fake_source_setup(self):
        pass


class DummyMixin:
    audio_class = dummy_audio.DummyAudio

    def possibly_trigger_fake_playback_error(self, uri):
        self.audio.trigger_fake_playback_failure(uri)

    def possibly_trigger_fake_about_to_finish(self):
        callback = self.audio.get_about_to_finish_callback().get()
        if callback:
            callback()

    def possibly_trigger_fake_source_setup(self):
        callback = self.audio.get_source_setup_callback().get()
        if callback:
            callback()


class AudioTest(BaseTest):
    def test_start_playback_existing_file(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        assert self.audio.start_playback().get()

    def test_start_playback_non_existing_file(self):
        self.possibly_trigger_fake_playback_error(self.uris[0] + "bogus")

        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0] + "bogus")
        assert not self.audio.start_playback().get()

    def test_pause_playback_while_playing(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()
        assert self.audio.pause_playback().get()

    def test_stop_playback_while_playing(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()
        assert self.audio.stop_playback().get()

    @unittest.SkipTest
    def test_deliver_data(self):
        pass  # TODO

    @unittest.SkipTest
    def test_end_of_data_stream(self):
        pass  # TODO

    @unittest.SkipTest
    def test_set_mute(self):
        pass  # TODO Probably needs a fakemixer with a mixer track

    @unittest.SkipTest
    def test_set_state_encapsulation(self):
        pass  # TODO

    @unittest.SkipTest
    def test_set_position(self):
        pass  # TODO

    @unittest.SkipTest
    def test_invalid_output_raises_error(self):
        pass  # TODO


class AudioDummyTest(DummyMixin, AudioTest):
    pass


class DummyAudioListener(pykka.ThreadingActor, audio.AudioListener):
    def __init__(self):
        super().__init__()
        self.events = []
        self.waiters = {}

    def on_event(self, event, **kwargs):
        self.events.append((event, kwargs))
        if event in self.waiters:
            self.waiters[event].set()

    def wait(self, event):
        self.waiters[event] = threading.Event()
        return self.waiters[event]

    def get_events(self):
        return self.events

    def clear_events(self):
        self.events = []


class AudioEventTest(BaseTest):
    def setUp(self):  # noqa: N802
        super().setUp()
        self.audio.enable_sync_handler().get()
        self.listener = DummyAudioListener.start().proxy()

    def tearDown(self):  # noqa: N802
        super().tearDown()

    def assertEvent(self, event, **kwargs):  # noqa: N802
        assert (event, kwargs) in self.listener.get_events().get()

    def assertNotEvent(self, event, **kwargs):  # noqa: N802
        assert (event, kwargs) not in self.listener.get_events().get()

    # TODO: test without uri set, with bad uri and gapless...
    # TODO: playing->playing triggered by seek should be removed
    # TODO: codify expected state after EOS
    # TODO: consider returning a future or a threading event?

    def test_state_change_stopped_to_playing_event(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent(
            "state_changed",
            old_state=PlaybackState.STOPPED,
            new_state=PlaybackState.PLAYING,
            target_state=None,
        )

    def test_state_change_stopped_to_paused_event(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.pause_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent(
            "state_changed",
            old_state=PlaybackState.STOPPED,
            new_state=PlaybackState.PAUSED,
            target_state=None,
        )

    def test_state_change_paused_to_playing_event(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.pause_playback()

        self.audio.wait_for_state_change()
        self.listener.clear_events()
        self.audio.start_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent(
            "state_changed",
            old_state=PlaybackState.PAUSED,
            new_state=PlaybackState.PLAYING,
            target_state=None,
        )

    def test_state_change_paused_to_stopped_event(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.pause_playback()

        self.audio.wait_for_state_change()
        self.listener.clear_events()
        self.audio.stop_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent(
            "state_changed",
            old_state=PlaybackState.PAUSED,
            new_state=PlaybackState.STOPPED,
            target_state=None,
        )

    def test_state_change_playing_to_paused_event(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()

        self.audio.wait_for_state_change()
        self.listener.clear_events()
        self.audio.pause_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent(
            "state_changed",
            old_state=PlaybackState.PLAYING,
            new_state=PlaybackState.PAUSED,
            target_state=None,
        )

    def test_state_change_playing_to_stopped_event(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()

        self.audio.wait_for_state_change()
        self.listener.clear_events()
        self.audio.stop_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent(
            "state_changed",
            old_state=PlaybackState.PLAYING,
            new_state=PlaybackState.STOPPED,
            target_state=None,
        )

    def test_stream_changed_event_on_playing(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.listener.clear_events()
        self.audio.start_playback()

        # Since we are going from stopped to playing, the state change is
        # enough to ensure the stream changed.
        self.audio.wait_for_state_change().get()
        self.assertEvent("stream_changed", uri=self.uris[0])

    def test_stream_changed_event_on_multiple_changes(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.listener.clear_events()
        self.audio.start_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent("stream_changed", uri=self.uris[0])

        self.audio.prepare_change()
        self.audio.set_uri(self.uris[1])
        self.audio.pause_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent("stream_changed", uri=self.uris[1])

    def test_stream_changed_event_on_playing_to_paused(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.listener.clear_events()
        self.audio.start_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent("stream_changed", uri=self.uris[0])

        self.listener.clear_events()
        self.audio.pause_playback()

        self.audio.wait_for_state_change().get()
        self.assertNotEvent("stream_changed", uri=self.uris[0])

    def test_stream_changed_event_on_paused_to_stopped(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.pause_playback()

        self.audio.wait_for_state_change()
        self.listener.clear_events()
        self.audio.stop_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent("stream_changed", uri=None)

    def test_position_changed_on_pause(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.pause_playback()
        self.audio.wait_for_state_change()

        self.audio.wait_for_state_change().get()
        self.assertEvent("position_changed", position=0)

    def test_stream_changed_event_on_paused_to_playing(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.listener.clear_events()
        self.audio.pause_playback()

        self.audio.wait_for_state_change().get()
        self.assertEvent("stream_changed", uri=self.uris[0])

        self.listener.clear_events()
        self.audio.start_playback()

        self.audio.wait_for_state_change().get()
        self.assertNotEvent("stream_changed", uri=self.uris[0])

    def test_position_changed_on_play(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()
        self.audio.wait_for_state_change()

        self.audio.wait_for_state_change().get()
        self.assertEvent("position_changed", position=0)

    def test_position_changed_on_seek_while_stopped(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.set_position(2000)

        self.audio.wait_for_state_change().get()
        self.assertNotEvent("position_changed", position=0)

    def test_position_changed_on_seek_after_play(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()

        self.audio.wait_for_state_change()
        self.listener.clear_events()
        self.audio.set_position(2000)

        self.audio.wait_for_state_change().get()
        self.assertEvent("position_changed", position=2000)

    def test_position_changed_on_seek_after_pause(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.pause_playback()

        self.audio.wait_for_state_change()
        self.listener.clear_events()
        self.audio.set_position(2000)

        self.audio.wait_for_state_change().get()
        self.assertEvent("position_changed", position=2000)

    def test_tags_changed_on_playback(self):
        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()
        self.audio.wait_for_state_change().get()

        self.assertEvent("tags_changed", tags=mock.ANY)

    # Unlike the other events, having the state changed done is not
    # enough to ensure our event is called. So we setup a threading
    # event that we can wait for with a timeout while the track playback
    # completes.

    def test_stream_changed_event_on_paused(self):
        event = self.listener.wait("stream_changed").get()

        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.pause_playback().get()
        self.audio.wait_for_state_change().get()

        if not event.wait(timeout=1.0):
            self.fail("Stream changed not reached within deadline")

        self.assertEvent("stream_changed", uri=self.uris[0])

    def test_reached_end_of_stream_event(self):
        event = self.listener.wait("reached_end_of_stream").get()

        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()
        self.audio.wait_for_state_change().get()

        self.possibly_trigger_fake_about_to_finish()
        if not event.wait(timeout=1.0):
            self.fail("End of stream not reached within deadline")

        assert not self.audio.get_current_tags().get()

    def test_gapless(self):
        uris = self.uris[1:]
        event = self.listener.wait("reached_end_of_stream").get()

        def callback():
            if uris:
                self.audio.set_uri(uris.pop()).get()

        self.audio.set_about_to_finish_callback(callback).get()

        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()

        self.possibly_trigger_fake_about_to_finish()
        self.audio.wait_for_state_change().get()

        self.possibly_trigger_fake_about_to_finish()
        self.audio.wait_for_state_change().get()
        if not event.wait(timeout=1.0):
            self.fail("EOS not received")

        # Check that both uris got played
        self.assertEvent("stream_changed", uri=self.uris[0])
        self.assertEvent("stream_changed", uri=self.uris[1])

        # Check that events counts check out.
        keys = [k for k, v in self.listener.get_events().get()]
        assert keys.count("stream_changed") == 2
        assert keys.count("position_changed") == 2
        assert keys.count("state_changed") == 1
        assert keys.count("reached_end_of_stream") == 1

        # TODO: test tag states within gaples

    def test_source_setup(self):
        mock_callback = mock.Mock()

        self.audio.prepare_change()
        self.audio.set_source_setup_callback(mock_callback).get()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()

        self.possibly_trigger_fake_source_setup()
        self.audio.wait_for_state_change().get()

        mock_callback.assert_called_once()

    # TODO: this does not belong in this testcase
    def test_current_tags_are_blank_to_begin_with(self):
        assert not self.audio.get_current_tags().get()

    def test_current_tags_blank_after_end_of_stream(self):
        event = self.listener.wait("reached_end_of_stream").get()

        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()

        self.possibly_trigger_fake_about_to_finish()
        self.audio.wait_for_state_change().get()

        if not event.wait(timeout=1.0):
            self.fail("EOS not received")

        assert not self.audio.get_current_tags().get()

    def test_current_tags_stored(self):
        event = self.listener.wait("reached_end_of_stream").get()
        tags = []

        def callback():
            tags.append(self.audio.get_current_tags().get())

        self.audio.set_about_to_finish_callback(callback).get()

        self.audio.prepare_change()
        self.audio.set_uri(self.uris[0])
        self.audio.start_playback()

        self.possibly_trigger_fake_about_to_finish()
        self.audio.wait_for_state_change().get()

        if not event.wait(timeout=1.0):
            self.fail("EOS not received")

        assert tags[0]

    # TODO: test that we reset when we expect between songs


class AudioDummyEventTest(DummyMixin, AudioEventTest):

    """Exercise the AudioEventTest against our mock audio classes."""


# TODO: move to mixer tests...
class MixerTest(BaseTest):
    @unittest.SkipTest
    def test_set_mute(self):
        for value in (True, False):
            assert self.audio.set_mute(value).get()
            assert self.audio.get_mute().get() == value

    @unittest.SkipTest
    def test_set_state_encapsulation(self):
        pass  # TODO

    @unittest.SkipTest
    def test_set_position(self):
        pass  # TODO

    @unittest.SkipTest
    def test_invalid_output_raises_error(self):
        pass  # TODO


class AudioStateTest(unittest.TestCase):
    def setUp(self):  # noqa: N802
        self.audio = audio.Audio(config=None, mixer=None)

    def test_state_starts_as_stopped(self):
        assert audio.PlaybackState.STOPPED == self.audio.state

    def test_state_does_not_change_when_in_gst_ready_state(self):
        self.audio._handler.on_playbin_state_changed(
            Gst.State.NULL, Gst.State.READY, Gst.State.VOID_PENDING
        )

        assert audio.PlaybackState.STOPPED == self.audio.state

    def test_state_changes_from_stopped_to_playing_on_play(self):
        self.audio._handler.on_playbin_state_changed(
            Gst.State.NULL, Gst.State.READY, Gst.State.PLAYING
        )
        self.audio._handler.on_playbin_state_changed(
            Gst.State.READY, Gst.State.PAUSED, Gst.State.PLAYING
        )
        self.audio._handler.on_playbin_state_changed(
            Gst.State.PAUSED, Gst.State.PLAYING, Gst.State.VOID_PENDING
        )

        assert audio.PlaybackState.PLAYING == self.audio.state

    def test_state_changes_from_playing_to_paused_on_pause(self):
        self.audio.state = audio.PlaybackState.PLAYING

        self.audio._handler.on_playbin_state_changed(
            Gst.State.PLAYING, Gst.State.PAUSED, Gst.State.VOID_PENDING
        )

        assert audio.PlaybackState.PAUSED == self.audio.state

    def test_state_changes_from_playing_to_stopped_on_stop(self):
        self.audio.state = audio.PlaybackState.PLAYING

        self.audio._handler.on_playbin_state_changed(
            Gst.State.PLAYING, Gst.State.PAUSED, Gst.State.NULL
        )
        self.audio._handler.on_playbin_state_changed(
            Gst.State.PAUSED, Gst.State.READY, Gst.State.NULL
        )
        # We never get the following call, so the logic must work without it
        # self.audio._handler.on_playbin_state_changed(
        #     Gst.State.READY, Gst.State.NULL, Gst.State.VOID_PENDING)

        assert audio.PlaybackState.STOPPED == self.audio.state


class AudioBufferingTest(unittest.TestCase):
    def setUp(self):  # noqa: N802
        self.audio = audio.Audio(config=None, mixer=None)
        self.audio._playbin = mock.Mock(spec=["set_state"])

    def test_pause_when_buffer_empty(self):
        playbin = self.audio._playbin
        self.audio.start_playback()
        playbin.set_state.assert_called_with(Gst.State.PLAYING)
        playbin.set_state.reset_mock()

        self.audio._handler.on_buffering(0)
        playbin.set_state.assert_called_with(Gst.State.PAUSED)
        assert self.audio._buffering

    def test_stay_paused_when_buffering_finished(self):
        playbin = self.audio._playbin
        self.audio.pause_playback()
        playbin.set_state.assert_called_with(Gst.State.PAUSED)
        playbin.set_state.reset_mock()

        self.audio._handler.on_buffering(100)
        assert playbin.set_state.call_count == 0
        assert not self.audio._buffering

    def test_change_to_paused_while_buffering(self):
        playbin = self.audio._playbin
        self.audio.start_playback()
        playbin.set_state.assert_called_with(Gst.State.PLAYING)
        playbin.set_state.reset_mock()

        self.audio._handler.on_buffering(0)
        playbin.set_state.assert_called_with(Gst.State.PAUSED)
        self.audio.pause_playback()
        playbin.set_state.reset_mock()

        self.audio._handler.on_buffering(100)
        assert playbin.set_state.call_count == 0
        assert not self.audio._buffering

    def test_change_to_stopped_while_buffering(self):
        playbin = self.audio._playbin
        self.audio.start_playback()
        playbin.set_state.assert_called_with(Gst.State.PLAYING)
        playbin.set_state.reset_mock()

        self.audio._handler.on_buffering(0)
        playbin.set_state.assert_called_with(Gst.State.PAUSED)
        playbin.set_state.reset_mock()

        self.audio.stop_playback()
        playbin.set_state.assert_called_with(Gst.State.NULL)
        assert not self.audio._buffering


class AudioLiveTest(unittest.TestCase):
    def setUp(self):  # noqa: N802
        config = {"proxy": {}}
        self.audio = audio.Audio(config=config, mixer=None)
        self.audio._playbin = mock.Mock(spec=["set_property"])

        self.source = mock.MagicMock()
        # Avoid appsrc.configure()
        self.source.get_factory.get_name = mock.Mock(return_value="not_appsrc")
        self.source.props = mock.Mock(spec=["is_live"])

    def test_not_live_mode(self):
        self.audio._live_stream = False

        self.audio._on_source_setup("dummy", self.source)

        self.source.set_live.assert_not_called()

    def test_live_mode(self):
        self.audio._live_stream = True

        self.audio._on_source_setup("dummy", self.source)

        self.source.set_live.assert_called_with(True)

    def test_not_live_mode_after_set_appsrc(self):
        self.audio._live_stream = True

        # Embrace appsrc.configure()
        self.source.get_factory.get_name.return_value = "appsrc"

        self.audio.set_appsrc("")
        self.audio._on_source_setup("dummy", self.source)

        self.source.set_live.assert_not_called()


class DownloadBufferingTest(unittest.TestCase):
    def setUp(self):  # noqa: N802
        self.audio = audio.Audio(config=None, mixer=None)
        self.audio._playbin = mock.Mock(spec=["set_property"])

    def test_download_flag_is_passed_to_playbin_if_download_buffering_is_enabled(
        self,
    ):
        playbin = self.audio._playbin

        self.audio.set_uri("some:uri", False, True)

        playbin.set_property.assert_has_calls([mock.call("flags", 0x02 | 0x80)])

    def test_download_flag_is_not_passed_to_playbin_if_download_buffering_is_not_enabled(  # noqa: B950
        self,
    ):
        playbin = self.audio._playbin

        self.audio.set_uri("some:uri", False, False)

        playbin.set_property.assert_has_calls([mock.call("flags", 0x02)])

    def test_download_flag_is_not_passed_to_playbin_if_set_appsrc(  # noqa: B950
        self,
    ):
        playbin = self.audio._playbin

        self.audio.set_appsrc("")

        playbin.set_property.assert_has_calls([mock.call("flags", 0x02)])


class SourceSetupCallbackTest(unittest.TestCase):
    def setUp(self):  # noqa: N802
        config = {"proxy": {}}
        self.audio = audio.Audio(config=config, mixer=None)
        self.audio._playbin = mock.Mock(spec=["set_property"])

        self.source = mock.MagicMock()
        # Avoid appsrc.configure()
        self.source.get_factory.get_name = mock.Mock(return_value="not_appsrc")

    def test_source_setup_callback(self):
        mock_callback = mock.MagicMock()
        self.audio.set_source_setup_callback(mock_callback)

        self.audio._on_source_setup("dummy", self.source)

        mock_callback.assert_called_once_with(self.source)

        self.audio.set_source_setup_callback(None)

        self.audio._on_source_setup("dummy", self.source)

        mock_callback.assert_called_once()