#!/usr/bin/env python3

# Libervia: an XMPP client
# Copyright (C) 2009-2023 Jérôme Poisson (goffi@goffi.org)

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from unittest.mock import AsyncMock, MagicMock

from gi.repository import Gst
from libervia.backend.core import exceptions
from libervia.backend.tools.common import data_format
import pytest

from libervia.desktop_kivy import G
from libervia.desktop_kivy.plugins import plugin_wid_calls


@pytest.fixture
def host(monkeypatch):
    host = MagicMock()
    host.a_bridge = AsyncMock()
    host.app.expand = lambda s: s
    monkeypatch.setattr(G, "_host", host, raising=False)
    return host


@pytest.fixture(scope="function")
def webrtc():
    """Fixture for WebRTC instantiation."""
    host_mock = MagicMock()
    profile = "test_profile"
    instance = plugin_wid_calls.WebRTC(host_mock, profile)

    instance._set_media_types = MagicMock()
    instance.start_pipeline = MagicMock()
    instance.webrtc = MagicMock()
    instance.webrtc.emit = MagicMock()

    instance.GstSdp_SDPMessage_new_from_text = MagicMock()
    instance.GstWebRTC_WebRTCSessionDescription_new = MagicMock()
    instance.Gst_Promise_new_with_change_func = MagicMock()

    return instance


@pytest.fixture
def calls(monkeypatch, host):
    """Fixture for Call UI instantiation."""
    for attr in ("header_box", "local_video", "remote_video", "screen_manager"):
        monkeypatch.setattr(
            plugin_wid_calls.Calls,
            attr,
            MagicMock()
        )
    calls = plugin_wid_calls.Calls(
        host,
        "test_peer@example.org",
        ["test_profile"]
    )
    calls.jid_selector = MagicMock()
    calls.header_input = MagicMock()
    calls.header_input.text = "fake_jid@domain"
    calls.webrtc = MagicMock()
    calls.webrtc.setup_call = AsyncMock()
    calls.webrtc.start_pipeline = MagicMock()
    calls.end_call = AsyncMock()

    return calls


class TestWebRtc:
    def test_get_payload_types(self, webrtc):
        """The method can identify the correct payload types for video and audio."""
        fake_sdpmsg = MagicMock()
        fake_media = MagicMock()
        fake_caps = MagicMock()
        fake_structure = MagicMock()

        # This side effect will return 'fake_video_encoding' first, then
        # 'fake_audio_encoding'.
        fake_structure.__getitem__.side_effect = [
            "fake_video_encoding",
            "fake_audio_encoding",
        ]
        fake_caps.get_structure.return_value = fake_structure
        fake_media.get_format.side_effect = ["webrtc-datachannel", "10", "20"]
        fake_media.get_caps_from_media.return_value = fake_caps
        fake_sdpmsg.get_media.return_value = fake_media
        fake_sdpmsg.medias_len.return_value = 1
        fake_media.formats_len.return_value = 3

        result = webrtc.get_payload_types(
            fake_sdpmsg, "fake_video_encoding", "fake_audio_encoding"
        )
        expected_result = {"fake_video_encoding": 10, "fake_audio_encoding": 20}

        assert result == expected_result

    def test_on_accepted_call(self, webrtc):
        """The method correctly sets the remote SDP upon acceptance of an outgoing call."""
        sdp_str = "mock_sdp_string"
        profile_str = "test_profile"

        webrtc.on_accepted_call(sdp_str, profile_str)

        # remote description must be set
        assert webrtc.webrtc.emit.call_count == 1
        assert webrtc.webrtc.emit.call_args[0][0] == "set-remote-description"

    @pytest.mark.asyncio
    async def test_answer_call(self, webrtc, monkeypatch):
        """The method correctly answers an incoming call."""
        mock_setup_call = AsyncMock()

        def mock_get_payload_types(sdpmsg, video_encoding, audio_encoding):
            return {"VP8": 96, "OPUS": 97}

        monkeypatch.setattr(webrtc, "setup_call", mock_setup_call)
        monkeypatch.setattr(webrtc, "get_payload_types", mock_get_payload_types)

        sdp_str = "mock_sdp_string"
        profile_str = "mock_profile"

        await webrtc.answer_call(sdp_str, profile_str)

        mock_setup_call.assert_called_once_with("responder", audio_pt=97, video_pt=96)

        # remote description must be set
        assert webrtc.webrtc.emit.call_count == 1
        assert webrtc.webrtc.emit.call_args[0][0] == "set-remote-description"

    def test_on_remote_decodebin_stream_video(self, webrtc, monkeypatch):
        """The method correctly handles video streams from the remote decodebin."""
        mock_pipeline = MagicMock()
        monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)

        mock_pad = MagicMock()
        mock_caps = MagicMock()
        mock_structure = MagicMock()

        mock_pad.has_current_caps.return_value = True
        mock_pad.get_current_caps.return_value = mock_caps
        mock_caps.__len__.return_value = 1
        mock_caps.__getitem__.return_value = mock_structure
        mock_structure.get_name.return_value = "video/x-h264"
        # We use non-standard resolution as example to trigger the workaround
        mock_structure.get_int.side_effect = lambda x: MagicMock(
            value=990 if x == "width" else 557
        )

        webrtc.on_remote_decodebin_stream(None, mock_pad)

        assert webrtc._remote_video_pad == mock_pad
        mock_pipeline.add.assert_called()
        mock_pipeline.set_state.assert_called()
        mock_pad.link.assert_called()

    def test_on_remote_decodebin_stream_audio(self, webrtc, monkeypatch):
        """The method correctly handles audio streams from the remote decodebin."""
        mock_pipeline = MagicMock()
        monkeypatch.setattr(webrtc, "pipeline", mock_pipeline)

        mock_pad = MagicMock()
        mock_caps = MagicMock()
        mock_structure = MagicMock()

        mock_pad.has_current_caps.return_value = True
        mock_pad.get_current_caps.return_value = mock_caps
        mock_caps.__len__.return_value = 1
        mock_caps.__getitem__.return_value = mock_structure
        mock_structure.get_name.return_value = "audio/x-raw"

        webrtc.on_remote_decodebin_stream(None, mock_pad)

        mock_pipeline.add.assert_called()
        mock_pipeline.set_state.assert_called()
        mock_pad.link.assert_called()

    @pytest.mark.asyncio
    async def test_setup_call_correct_role(self, host, webrtc, monkeypatch):
        """Roles are set in setup_call."""
        monkeypatch.setattr(Gst, "parse_launch", MagicMock())
        monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))

        await webrtc.setup_call("initiator")
        assert webrtc.role == "initiator"

        await webrtc.setup_call("responder")
        assert webrtc.role == "responder"

        with pytest.raises(AssertionError):
            await webrtc.setup_call("invalid_role")

    @pytest.mark.asyncio
    async def test_setup_call_test_mode(self, host, webrtc, monkeypatch):
        """Test mode use fake video and audio in setup_call."""
        monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
        monkeypatch.setattr(webrtc, "test_mode", True)
        await webrtc.setup_call("initiator")
        assert "videotestsrc" in webrtc.gst_pipe_desc
        assert "audiotestsrc" in webrtc.gst_pipe_desc

    @pytest.mark.asyncio
    async def test_setup_call_normal_mode(self, host, webrtc, monkeypatch):
        """Normal mode use real video and audio in setup_call."""
        monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
        monkeypatch.setattr(webrtc, "test_mode", False)
        await webrtc.setup_call("initiator")
        assert "v4l2src" in webrtc.gst_pipe_desc
        assert "pulsesrc" in webrtc.gst_pipe_desc

    @pytest.mark.asyncio
    async def test_setup_call_with_stun_and_turn(self, host, webrtc, monkeypatch):
        """STUN and TURN server configurations are done in setup_call."""
        mock_pipeline = MagicMock()
        mock_parse_launch = MagicMock()
        mock_parse_launch.return_value = mock_pipeline
        monkeypatch.setattr(Gst, "parse_launch", mock_parse_launch)

        mock_pipeline.get_by_name.return_value = webrtc.webrtc

        mock_external_disco = [
            {"type": "stun", "transport": "udp", "host": "stun.host", "port": "3478"},
            {
                "type": "turn",
                "transport": "udp",
                "host": "turn.host",
                "port": "3478",
                "username": "user",
                "password": "pass",
            },
        ]

        monkeypatch.setattr(
            data_format, "deserialise", MagicMock(return_value=mock_external_disco)
        )

        mock_emit = AsyncMock()
        monkeypatch.setattr(webrtc.webrtc, "emit", mock_emit)

        mock_set_property = AsyncMock()
        monkeypatch.setattr(webrtc.webrtc, "set_property", mock_set_property)

        await webrtc.setup_call("initiator")

        G.host.a_bridge.external_disco_get.assert_called_once_with("", webrtc.profile)
        mock_set_property.assert_any_call("stun-server", "stun://stun.host:3478")
        mock_emit.assert_called_once_with(
            "add-turn-server", "turn://user:pass@turn.host:3478"
        )

    @pytest.mark.asyncio
    async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch):
        """Test setup_call method handling Gstreamer pipeline failure."""
        monkeypatch.setattr(Gst, "parse_launch", lambda _: None)
        with pytest.raises(exceptions.InternalError):
            await webrtc.setup_call("initiator")


class TestCalls:

    @pytest.mark.asyncio
    async def test_toggle_call_sid_none(self, monkeypatch, calls):
        """Call is started when there is not sid set."""
        monkeypatch.setattr(calls.webrtc, "sid", None)

        await calls.toggle_call()

        calls.webrtc.setup_call.assert_called_once_with("initiator")
        calls.webrtc.start_pipeline.assert_called_once()
        assert calls.in_call == True

    @pytest.mark.asyncio
    async def test_toggle_call_sid_set(self, monkeypatch, host, calls):
        """Call is ended when a sid is set"""
        monkeypatch.setattr(calls.webrtc, "sid", "test_sid")

        await calls.toggle_call()

        calls.end_call.assert_called_once_with({"reason": "terminated"}, calls.profile)
        host.a_bridge.call_end.assert_called_once_with("test_sid", "", calls.profile)
        assert calls.in_call == False


    @pytest.mark.asyncio
    async def test_on_incoming_call_sid_none(self, monkeypatch, host, calls):
        """Incoming call is accepted if no ongoing call."""
        monkeypatch.setattr(calls.webrtc, "sid", None)
        fake_action_id = "fake_action_id"
        fake_action_data = {"session_id": "test_sid"}
        fake_profile = "fake_profile"

        await calls.on_incoming_call(fake_action_data, fake_action_id, fake_profile)

        assert calls.in_call == True
        assert calls.webrtc.sid == "test_sid"
        host.a_bridge.action_launch.assert_called_once_with(
            fake_action_id,
            data_format.serialise({"cancelled": False}),
            fake_profile
        )

    @pytest.mark.asyncio
    async def test_on_incoming_call_sid_set(self, monkeypatch, host, calls):
        """Incoming call is ignored if there's an ongoing call."""
        monkeypatch.setattr(calls.webrtc, "sid", "fake_old_sid")
        fake_action_id = "fake_action_id"
        fake_action_data = {"session_id": "test_sid_new"}
        fake_profile = "fake_profile"

        await calls.on_incoming_call(fake_action_data, fake_action_id, fake_profile)

        # Ensuring the state hasn't been changed to True
        assert calls.in_call == False
        host.a_bridge.action_launch.assert_not_called()

    @pytest.mark.asyncio
    async def test_on_call_setup_initiator(self, calls):
        """Correct method called if role is 'initiator'."""
        setup_data = {
            "role": "initiator",
            "sdp": "fake_sdp"
        }
        profile = "fake_profile"

        await calls.on_call_setup(setup_data, profile)

        calls.webrtc.on_accepted_call.assert_called_once_with(setup_data["sdp"], profile)

    @pytest.mark.asyncio
    async def test_on_call_setup_responder(self, monkeypatch, calls):
        """Correct method called if role is 'responder'."""
        monkeypatch.setattr(
            calls.webrtc,
            "answer_call",
            AsyncMock()
        )
        setup_data = {
            "role": "responder",
            "sdp": "fake_sdp"
        }
        profile = "fake_profile"

        await calls.on_call_setup(setup_data, profile)

        calls.webrtc.answer_call.assert_called_once_with(setup_data["sdp"], profile)
        calls.webrtc.on_accepted_call.assert_not_called()

    @pytest.mark.asyncio
    async def test_on_call_setup_invalid_role(self, calls):
        """Nothing is called if role is neither 'initiator' nor 'responder'."""
        setup_data = {
            "role": "invalid_role",
            "sdp": "fake_sdp"
        }
        profile = "fake_profile"

        await calls.on_call_setup(setup_data, profile)

        calls.webrtc.answer_call.assert_not_called()
        calls.webrtc.on_accepted_call.assert_not_called()