try:
    from django.utils.unittest import TestCase
except ImportError:
    from django.test import TestCase

try:
    from django.utils import unittest
except ImportError:
    import unittest

import sys
import string
import mock
import pickle
import json

from mock import Mock, MagicMock
from evennia.server.portal import irc
from evennia.utils.test_resources import EvenniaTest

from twisted.conch.telnet import IAC, WILL, DONT, SB, SE, NAWS, DO
from twisted.test import proto_helpers
from twisted.trial.unittest import TestCase as TwistedTestCase
from twisted.internet.base import DelayedCall

from .telnet import TelnetServerFactory, TelnetProtocol
from .portal import PORTAL_SESSIONS
from .suppress_ga import SUPPRESS_GA
from .naws import DEFAULT_HEIGHT, DEFAULT_WIDTH
from .ttype import TTYPE, IS
from .mccp import MCCP
from .mssp import MSSP
from .mxp import MXP
from .telnet_oob import MSDP, MSDP_VAL, MSDP_VAR

from .amp import AMPMultiConnectionProtocol, MsgServer2Portal, MsgPortal2Server, AMP_MAXLEN
from .amp_server import AMPServerFactory

from autobahn.twisted.websocket import WebSocketServerFactory
from .webclient import WebSocketClient


class TestAMPServer(TwistedTestCase):
    """
    Test AMP communication
    """

    def setUp(self):
        super(TestAMPServer, self).setUp()
        portal = Mock()
        factory = AMPServerFactory(portal)
        self.proto = factory.buildProtocol(("localhost", 0))
        self.transport = MagicMock()  # proto_helpers.StringTransport()
        self.transport.client = ["localhost"]
        self.transport.write = MagicMock()

    def test_amp_out(self):
        self.proto.makeConnection(self.transport)

        self.proto.data_to_server(MsgServer2Portal, 1, test=2)

        if pickle.HIGHEST_PROTOCOL == 5:
            # Python 3.8+
            byte_out = (
                b"\x00\x04_ask\x00\x011\x00\x08_command\x00\x10MsgServer2Portal\x00\x0b"
                b"packed_data\x00 x\xdak`\x9d*\xc8\x00\x01\xde\x8c\xb5SzXJR"
                b"\x8bK\xa6x3\x15\xb7M\xd1\x03\x00VU\x07u\x00\x00"
            )
        elif pickle.HIGHEST_PROTOCOL == 4:
            # Python 3.7
            byte_out = (
                b"\x00\x04_ask\x00\x011\x00\x08_command\x00\x10MsgServer2Portal\x00\x0b"
                b"packed_data\x00 x\xdak`\x99*\xc8\x00\x01\xde\x8c\xb5SzXJR"
                b"\x8bK\xa6x3\x15\xb7M\xd1\x03\x00V:\x07t\x00\x00"
            )
        self.transport.write.assert_called_with(byte_out)
        with mock.patch("evennia.server.portal.amp.amp.AMP.dataReceived") as mocked_amprecv:
            self.proto.dataReceived(byte_out)
            mocked_amprecv.assert_called_with(byte_out)

    def test_amp_in(self):
        self.proto.makeConnection(self.transport)

        self.proto.data_to_server(MsgPortal2Server, 1, test=2)
        if pickle.HIGHEST_PROTOCOL == 5:
            # Python 3.8+
            byte_out = (
                b"\x00\x04_ask\x00\x011\x00\x08_command\x00\x10MsgPortal2Server\x00\x0b"
                b"packed_data\x00 x\xdak`\x9d*\xc8\x00\x01\xde\x8c\xb5SzXJR"
                b"\x8bK\xa6x3\x15\xb7M\xd1\x03\x00VU\x07u\x00\x00"
            )
        elif pickle.HIGHEST_PROTOCOL == 4:
            # Python 3.7
            byte_out = (
                b"\x00\x04_ask\x00\x011\x00\x08_command\x00\x10MsgPortal2Server\x00\x0b"
                b"packed_data\x00 x\xdak`\x99*\xc8\x00\x01\xde\x8c\xb5SzXJR"
                b"\x8bK\xa6x3\x15\xb7M\xd1\x03\x00V:\x07t\x00\x00"
            )
        self.transport.write.assert_called_with(byte_out)
        with mock.patch("evennia.server.portal.amp.amp.AMP.dataReceived") as mocked_amprecv:
            self.proto.dataReceived(byte_out)
            mocked_amprecv.assert_called_with(byte_out)

    def test_large_msg(self):
        """
        Send message larger than AMP_MAXLEN - should be split into several
        """
        self.proto.makeConnection(self.transport)
        outstr = "test" * AMP_MAXLEN
        self.proto.data_to_server(MsgServer2Portal, 1, test=outstr)

        if pickle.HIGHEST_PROTOCOL == 5:
            # Python 3.8+
            self.transport.write.assert_called_with(
                b"\x00\x04_ask\x00\x011\x00\x08_command\x00\x10MsgServer2Portal\x00\x0bpacked_data"
                b"\x00wx\xda\xed\xc6\xc1\t\x80 \x00@Q#=5Z\x0b\xb8\x80\x13\xe85h\x80\x8e\xbam`Dc\xf4><\xf8g"
                b"\x1a[\xf8\xda\x97\xa3_\xb1\x95\xdaz\xbe\xe7\x1a\xde\x03\x00\x00\x00\x00\x00\x00\x00"
                b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
                b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
                b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xe0\x1f\x1eP\x1d\x02\r\x00\rpacked_data.2"
                b"\x00Zx\xda\xed\xc3\x01\r\x00\x00\x08\xc0\xa0\xb4&\xf0\xfdg\x10a\xa3"
                b"\xd9RUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUU\xf5\xfb\x03m\xe0\x06"
                b"\x1d\x00\rpacked_data.3\x00Zx\xda\xed\xc3\x01\r\x00\x00\x08\xc0\xa0\xb4&\xf0\xfdg\x10a"
                b"\xa3fSUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUU\xf5\xfb\x03n\x1c"
                b"\x06\x1e\x00\rpacked_data.4\x00Zx\xda\xed\xc3\x01\t\x00\x00\x0c\x03\xa0\xb4O\xb0\xf5gA"
                b"\xae`\xda\x8b\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa"
                b"\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa"
                b"\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa"
                b"\xaa\xaa\xaa\xdf\x0fnI\x06,\x00\rpacked_data.5\x00\x18x\xdaK-.)I\xc5\x8e\xa7\xb22@\xc0"
                b"\x94\xe2\xb6)z\x00Z\x1e\x0e\xb6\x00\x00"
            )
        elif pickle.HIGHEST_PROTOCOL == 4:
            # Python 3.7
            self.transport.write.assert_called_with(
                b"\x00\x04_ask\x00\x011\x00\x08_command\x00\x10MsgServer2Portal\x00\x0bpacked_data"
                b"\x00wx\xda\xed\xc6\xc1\t\x80 \x00@Q#o\x8e\xd6\x02-\xe0\x04z\r\x1a\xa0\xa3m+$\xd2"
                b"\x18\xbe\x0f\x0f\xfe\x1d\xdf\x14\xfe\x8e\xedjO\xac\xb9\xd4v\xf6o\x0f\xf3\x00\x00"
                b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
                b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
                b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
                b"\x00X\xc3\x00P\x10\x02\x0c\x00\rpacked_data.2\x00Zx\xda\xed\xc3\x01\r\x00\x00\x08"
                b"\xc0\xa0\xb4&\xf0\xfdg\x10a\xa3\xd9RUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUU"
                b"\xf5\xfb\x03m\xe0\x06\x1d\x00\rpacked_data.3\x00Zx\xda\xed\xc3\x01\r\x00\x00\x08"
                b"\xc0\xa0\xb4&\xf0\xfdg\x10a\xa3fSUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUU"
                b"\xf5\xfb\x03n\x1c\x06\x1e\x00\rpacked_data.4\x00Zx\xda\xed\xc3\x01\t\x00\x00\x0c"
                b"\x03\xa0\xb4O\xb0\xf5gA\xae`\xda\x8b\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa"
                b"\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa"
                b"\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa"
                b"\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xaa\xdf\x0fnI\x06,\x00\rpacked_data.5"
                b"\x00\x18x\xdaK-.)I\xc5\x8e\xa7\xb22@\xc0\x94\xe2\xb6)z\x00Z\x1e\x0e\xb6\x00\x00"
            )


class TestIRC(TestCase):
    def test_plain_ansi(self):
        """
        Test that printable characters do not get mangled.
        """
        irc_ansi = irc.parse_ansi_to_irc(string.printable)
        ansi_irc = irc.parse_irc_to_ansi(string.printable)
        self.assertEqual(irc_ansi, string.printable)
        self.assertEqual(ansi_irc, string.printable)

    def test_bold(self):
        s_irc = "\x02thisisatest"
        s_eve = r"|hthisisatest"
        self.assertEqual(irc.parse_ansi_to_irc(s_eve), s_irc)
        self.assertEqual(s_eve, irc.parse_irc_to_ansi(s_irc))

    def test_italic(self):
        s_irc = "\x02thisisatest"
        s_eve = r"|hthisisatest"
        self.assertEqual(irc.parse_ansi_to_irc(s_eve), s_irc)

    def test_colors(self):
        color_map = (
            ("\0030", r"|w"),
            ("\0031", r"|X"),
            ("\0032", r"|B"),
            ("\0033", r"|G"),
            ("\0034", r"|r"),
            ("\0035", r"|R"),
            ("\0036", r"|M"),
            ("\0037", r"|Y"),
            ("\0038", r"|y"),
            ("\0039", r"|g"),
            ("\00310", r"|C"),
            ("\00311", r"|c"),
            ("\00312", r"|b"),
            ("\00313", r"|m"),
            ("\00314", r"|x"),
            ("\00315", r"|W"),
            ("\00399,5", r"|[r"),
            ("\00399,3", r"|[g"),
            ("\00399,7", r"|[y"),
            ("\00399,2", r"|[b"),
            ("\00399,6", r"|[m"),
            ("\00399,10", r"|[c"),
            ("\00399,15", r"|[w"),
            ("\00399,1", r"|[x"),
        )

        for m in color_map:
            self.assertEqual(irc.parse_irc_to_ansi(m[0]), m[1])
            self.assertEqual(m[0], irc.parse_ansi_to_irc(m[1]))

    def test_identity(self):
        """
        Test that the composition of the function and
        its inverse gives the correct string.
        """

        s = r"|wthis|Xis|gis|Ma|C|complex|*string"

        self.assertEqual(irc.parse_irc_to_ansi(irc.parse_ansi_to_irc(s)), s)


class TestTelnet(TwistedTestCase):
    def setUp(self):
        super(TestTelnet, self).setUp()
        factory = TelnetServerFactory()
        factory.protocol = TelnetProtocol
        factory.sessionhandler = PORTAL_SESSIONS
        factory.sessionhandler.portal = Mock()
        self.proto = factory.buildProtocol(("localhost", 0))
        self.transport = proto_helpers.StringTransport()
        self.addCleanup(factory.sessionhandler.disconnect_all)

    @mock.patch("evennia.server.portal.portalsessionhandler.reactor", new=MagicMock())
    def test_mudlet_ttype(self):
        self.transport.client = ["localhost"]
        self.transport.setTcpKeepAlive = Mock()
        d = self.proto.makeConnection(self.transport)
        # test suppress_ga
        self.assertTrue(self.proto.protocol_flags["NOGOAHEAD"])
        self.proto.dataReceived(IAC + DONT + SUPPRESS_GA)
        self.assertFalse(self.proto.protocol_flags["NOGOAHEAD"])
        self.assertEqual(self.proto.handshakes, 7)
        # test naws
        self.assertEqual(self.proto.protocol_flags["SCREENWIDTH"], {0: DEFAULT_WIDTH})
        self.assertEqual(self.proto.protocol_flags["SCREENHEIGHT"], {0: DEFAULT_HEIGHT})
        self.proto.dataReceived(IAC + WILL + NAWS)
        self.proto.dataReceived(b"".join([IAC, SB, NAWS, b"", b"x", b"", b"d", IAC, SE]))
        self.assertEqual(self.proto.protocol_flags["SCREENWIDTH"][0], 78)
        self.assertEqual(self.proto.protocol_flags["SCREENHEIGHT"][0], 45)
        self.assertEqual(self.proto.handshakes, 6)
        # test ttype
        self.assertFalse(self.proto.protocol_flags["TTYPE"])
        self.assertTrue(self.proto.protocol_flags["ANSI"])
        self.proto.dataReceived(IAC + WILL + TTYPE)
        self.proto.dataReceived(b"".join([IAC, SB, TTYPE, IS, b"MUDLET", IAC, SE]))
        self.assertTrue(self.proto.protocol_flags["XTERM256"])
        self.assertEqual(self.proto.protocol_flags["CLIENTNAME"], "MUDLET")
        self.assertTrue(self.proto.protocol_flags["FORCEDENDLINE"])
        self.assertTrue(self.proto.protocol_flags["NOGOAHEAD"])
        self.assertFalse(self.proto.protocol_flags["NOPROMPTGOAHEAD"])
        self.proto.dataReceived(b"".join([IAC, SB, TTYPE, IS, b"XTERM", IAC, SE]))
        self.proto.dataReceived(b"".join([IAC, SB, TTYPE, IS, b"MTTS 137", IAC, SE]))
        self.assertEqual(self.proto.handshakes, 5)
        # test mccp
        self.proto.dataReceived(IAC + DONT + MCCP)
        self.assertFalse(self.proto.protocol_flags["MCCP"])
        self.assertEqual(self.proto.handshakes, 4)
        # test mssp
        self.proto.dataReceived(IAC + DONT + MSSP)
        self.assertEqual(self.proto.handshakes, 3)
        # test oob
        self.proto.dataReceived(IAC + DO + MSDP)
        self.proto.dataReceived(
            b"".join([IAC, SB, MSDP, MSDP_VAR, b"LIST", MSDP_VAL, b"COMMANDS", IAC, SE])
        )
        self.assertTrue(self.proto.protocol_flags["OOB"])
        self.assertEqual(self.proto.handshakes, 2)
        # test mxp
        self.proto.dataReceived(IAC + DONT + MXP)
        self.assertFalse(self.proto.protocol_flags["MXP"])
        self.assertEqual(self.proto.handshakes, 1)
        # clean up to prevent Unclean reactor
        self.proto.nop_keep_alive.stop()
        self.proto._handshake_delay.cancel()
        return d


class TestWebSocket(EvenniaTest):
    def setUp(self):
        super().setUp()
        self.proto = WebSocketClient()
        self.proto.factory = WebSocketServerFactory()
        self.proto.factory.sessionhandler = PORTAL_SESSIONS
        self.proto.sessionhandler = PORTAL_SESSIONS
        self.proto.sessionhandler.portal = Mock()
        self.proto.transport = proto_helpers.StringTransport()
        # self.proto.transport = proto_helpers.FakeDatagramTransport()
        self.proto.transport.client = ["localhost"]
        self.proto.transport.setTcpKeepAlive = Mock()
        self.proto.state = MagicMock()
        self.addCleanup(self.proto.factory.sessionhandler.disconnect_all)
        DelayedCall.debug = True

    def tearDown(self):
        super().tearDown()

    @mock.patch("evennia.server.portal.portalsessionhandler.reactor", new=MagicMock())
    def test_data_in(self):
        self.proto.sessionhandler.data_in = MagicMock()
        self.proto.onOpen()
        msg = json.dumps(["logged_in", (), {}]).encode()
        self.proto.onMessage(msg, isBinary=False)
        self.proto.sessionhandler.data_in.assert_called_with(self.proto, logged_in=[[], {}])
        sendStr = "You can get anything you want at Alice's Restaurant."
        msg = json.dumps(["text", (sendStr,), {}]).encode()
        self.proto.onMessage(msg, isBinary=False)
        self.proto.sessionhandler.data_in.assert_called_with(self.proto, text=[[sendStr], {}])

    @mock.patch("evennia.server.portal.portalsessionhandler.reactor", new=MagicMock())
    def test_data_out(self):
        self.proto.onOpen()
        self.proto.sendLine = MagicMock()
        msg = json.dumps(["logged_in", (), {}])
        self.proto.sessionhandler.data_out(self.proto, text=[["Excepting Alice"], {}])
        self.proto.sendLine.assert_called_with(json.dumps(["text", ["Excepting Alice"], {}]))
