"""Unit tests for the conferences component (ICE restart paths)."""

from pathlib import Path
from types import SimpleNamespace

from pytest import fixture
from twisted.internet import defer
from twisted.words.protocols.jabber import jid

from libervia.backend.plugins.plugin_comp_conferences import Conference

# the SDP line endings are CRLF, as in real offers (RFC 4566)
UPLINK_OFFER_SDP = ("""v=0
o=- 0 0 IN IP4 127.0.0.1
s=-
t=0 0
a=group:BUNDLE 0 1
m=audio 9 UDP/TLS/RTP/SAVPF 111
c=IN IP4 0.0.0.0
a=mid:0
a=sendonly
a=ice-ufrag:uf1
a=ice-pwd:pw1
m=video 9 UDP/TLS/RTP/SAVPF 96
c=IN IP4 0.0.0.0
a=mid:1
a=sendonly
a=ice-ufrag:uf1
a=ice-pwd:pw1
""").replace("\n", "\r\n")

# answer of the renegotiated uplink offer (an ICE restart by the user): new ICE
# credentials, with inline candidates
UPLINK_ANSWER_SDP = ("""v=0
o=- 0 0 IN IP4 127.0.0.1
s=-
t=0 0
a=group:BUNDLE 0 1
m=audio 9 UDP/TLS/RTP/SAVPF 111
c=IN IP4 0.0.0.0
a=mid:0
a=recvonly
a=ice-ufrag:uf3
a=ice-pwd:pw3
a=candidate:1 1 udp 2122260223 192.168.1.1 30000 typ host
m=video 9 UDP/TLS/RTP/SAVPF 96
c=IN IP4 0.0.0.0
a=mid:1
a=recvonly
a=ice-ufrag:uf3
a=ice-pwd:pw3
a=candidate:2 1 udp 2122260223 192.168.1.1 30001 typ host
""").replace("\n", "\r\n")

# new offer for an existing stream (an ICE restart by the SFU): new ICE credentials, with
# inline candidates
DOWNLINK_REOFFER_SDP = ("""v=0
o=- 0 0 IN IP4 127.0.0.1
s=-
t=0 0
a=group:BUNDLE 0 1
m=audio 9 UDP/TLS/RTP/SAVPF 111
c=IN IP4 0.0.0.0
a=mid:0
a=sendonly
a=ice-ufrag:uf9
a=ice-pwd:pw9
a=candidate:1 1 udp 2122260223 192.168.1.1 30000 typ host
m=video 9 UDP/TLS/RTP/SAVPF 96
c=IN IP4 0.0.0.0
a=mid:1
a=sendonly
a=ice-ufrag:uf9
a=ice-pwd:pw9
a=candidate:2 1 udp 2122260223 192.168.1.1 30001 typ host
""").replace("\n", "\r\n")


class MockProtocol:
    """Records the data sent to the SFU."""

    client_id = "c0ffee"

    def __init__(self):
        self.sent = []

    def send_data(self, data):
        self.sent.append(data)

    def sendClose(self):
        pass


class MockIceUdp:
    """Records the calls to ice_candidates_add (the transport-info send path)."""

    def __init__(self, session: dict):
        self.session = session
        self.calls = []

    def ice_candidates_add(self, client, session_id, content_ice_data):
        # like XEP-0176: the session's local ICE data is updated with the new credentials
        for name, ice_data in content_ice_data.items():
            local_ice_data = (
                self.session["contents"]
                .get(name, {})
                .get("transport_data", {})
                .get("local_ice_data")
            )
            if local_ice_data is not None:
                local_ice_data.update(ufrag=ice_data["ufrag"], pwd=ice_data["pwd"])
        self.calls.append((client, session_id, content_ice_data))
        return defer.succeed(None)


@fixture
def conference():
    """A Conference with mocked parent (SFU protocol and XEP-0176)."""
    session = {
        "id": "sess123",
        "local_jid": jid.JID("conferences.test.example"),
        "peer_jid": jid.JID("user@test.example"),
        "role": "responder",
        "contents": {},
        "streams": {},
        "uplink_offer_sdp": UPLINK_OFFER_SDP,
    }
    parent = SimpleNamespace(
        client=SimpleNamespace(get_virtual_client=lambda jid_: object()),
        _ice_udp=MockIceUdp(session),
    )
    conf = Conference(
        parent,
        "testroom",
        Path("testroom.json"),
        "ws://localhost",
        "http://localhost",
        session,
    )
    conf._protocol = MockProtocol()
    return conf


def test_restart_uplink_same_credentials(conference):
    """No renegotiation if the uplink credentials are unchanged."""
    conference._restart_uplink(
        conference.session, {"ufrag": "uf1", "pwd": "pw1", "candidates": []}
    )
    assert conference._protocol.sent == []


def test_restart_uplink_new_credentials(conference):
    """New uplink credentials trigger a renegotiation offer to the SFU."""
    conference._restart_uplink(
        conference.session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []}
    )
    (offer,) = conference._protocol.sent
    assert offer["type"] == "offer"
    assert offer["id"] == conference.galene_id
    assert offer["sdp"].count("a=ice-ufrag:uf2") == 2
    assert offer["sdp"].count("a=ice-pwd:pw2") == 2
    # the CRLF line endings are preserved (a mangled SDP would not be parsable)
    assert offer["sdp"].count("\r\n") == UPLINK_OFFER_SDP.count("\r\n")
    # the offer is stored for the next restart
    assert conference.session["uplink_offer_sdp"] == offer["sdp"]
    assert isinstance(conference.session["answer_sdp_d"], defer.Deferred)
    # the renegotiation is pending: the answer will carry new credentials
    assert conference.session["uplink_renegotiating"] is True


def test_restart_uplink_no_stored_offer(conference):
    """Without the stored uplink offer, nothing is re-offered."""
    del conference.session["uplink_offer_sdp"]
    conference._restart_uplink(
        conference.session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []}
    )
    assert conference._protocol.sent == []


def test_restart_uplink_answer_consumed(conference):
    """The answer of the renegotiation offer is consumed by on_answer()."""
    conference._restart_uplink(
        conference.session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []}
    )
    conference.on_answer({"sdp": "v=0\n"})
    assert conference.session.pop("answer_sdp_d", None) is None


def test_restart_uplink_answer_credentials_forwarded(conference):
    """The answer's ICE credentials and inline candidates are forwarded to the user."""
    session = conference.session
    for name in ("0", "1"):
        session["contents"][name] = {
            "transport_data": {
                "local_ice_data": {
                    "ufrag": "uf_old",
                    "pwd": "pw_old",
                    "candidates": [],
                }
            }
        }
    conference._restart_uplink(session, {"ufrag": "uf2", "pwd": "pw2", "candidates": []})
    conference.on_answer({"sdp": UPLINK_ANSWER_SDP})
    assert "uplink_renegotiating" not in session
    calls = conference.parent._ice_udp.calls
    # two transport-info with the new credentials (one per content), then the two inline
    # candidates
    assert [next(iter(data)) for data in (c[2] for c in calls)] == [
        "0",
        "1",
        "0",
        "1",
    ]
    for _, session_id, data in calls:
        assert session_id == "sess123"
        ((name, ice_data),) = data.items()
        assert (ice_data["ufrag"], ice_data["pwd"]) == ("uf3", "pw3")
    assert calls[2][2]["0"]["candidates"]
    assert calls[3][2]["1"]["candidates"]


def test_on_answer_initial_no_credentials_forwarded(conference):
    """Without an ICE restart, the answer is not forwarded to the user."""
    session = conference.session
    for name in ("0", "1"):
        session["contents"][name] = {
            "transport_data": {
                "local_ice_data": {
                    "ufrag": "uf3",
                    "pwd": "pw3",
                    "candidates": [],
                }
            }
        }
    session["answer_sdp_d"] = defer.Deferred()
    conference.on_answer({"sdp": UPLINK_ANSWER_SDP})
    assert conference.parent._ice_udp.calls == []


async def test_renegotiation_downlink(conference):
    """A re-offered stream: new credentials to the user, stored answer to the SFU."""
    session = conference.session
    session["streams"] = {"down1": ["2-0", "2-1"]}
    session["stream_answer_sdp"] = {"down1": "ANSWER_SDP"}
    # the contents of the stream, with the new local ICE data (set in production by the
    # ice_candidates_add calls of _handle_renegotiation)
    for name, media in (("2-0", "audio"), ("2-1", "video")):
        session["contents"][name] = {
            "application_data": {"media": media},
            "transport_data": {
                "local_ice_data": {"ufrag": "uf9", "pwd": "pw9", "candidates": []}
            },
        }
    await conference._handle_renegotiation({"id": "down1", "sdp": DOWNLINK_REOFFER_SDP})
    # two transport-info with the new credentials (one per content), then the two inline
    # candidates
    calls = conference.parent._ice_udp.calls
    assert len(calls) == 4
    for _, session_id, data in calls:
        assert session_id == "sess123"
        ((name, ice_data),) = data.items()
        assert (ice_data["ufrag"], ice_data["pwd"]) == ("uf9", "pw9")
    assert [next(iter(data)) for data in (c[2] for c in calls)] == [
        "2-0",
        "2-1",
        "2-0",
        "2-1",
    ]
    assert calls[2][2]["2-0"]["candidates"]
    assert calls[3][2]["2-1"]["candidates"]
    answer = [m for m in conference._protocol.sent if m["type"] == "answer"]
    assert len(answer) == 1
    assert answer[0]["id"] == "down1"
    assert answer[0]["sdp"] == "ANSWER_SDP"


async def test_renegotiation_downlink_no_stored_answer(conference):
    """Without the stored answer, no answer is re-sent to the SFU."""
    session = conference.session
    session["streams"] = {"down1": ["2-0", "2-1"]}
    for name, media in (("2-0", "audio"), ("2-1", "video")):
        session["contents"][name] = {
            "application_data": {"media": media},
            "transport_data": {
                "local_ice_data": {"ufrag": "uf9", "pwd": "pw9", "candidates": []}
            },
        }
    await conference._handle_renegotiation({"id": "down1", "sdp": DOWNLINK_REOFFER_SDP})
    assert all(m["type"] != "answer" for m in conference._protocol.sent)


async def test_offer_mids_renamed(conference):
    """The mids of a new downlink stream are prefixed with the stream counter."""
    session = conference.session
    added = []

    class MockRtp:
        async def call_update(
            self, client, session_id, call_data, jingle_child_elts=None
        ):
            added.append(call_data)
            return ["1-0", "1-1"]

    class MockAvConf:
        def build_origin_elt(self, origin):
            return None

    conference.parent._rtp = MockRtp()
    conference.parent._av_conf = MockAvConf()
    data = {"id": "down1", "sdp": DOWNLINK_REOFFER_SDP, "username": "user@test.example"}
    await conference.on_offer(data)
    (call_data,) = added
    sdp = call_data["sdp"]
    assert "a=mid:1-0" in sdp
    assert "a=mid:1-1" in sdp
    # the CRLF line endings are preserved (a mangled SDP would not be parsable)
    assert sdp.count("\r\n") == DOWNLINK_REOFFER_SDP.count("\r\n")
    assert session["streams"] == {"down1": ["1-0", "1-1"]}