"""Test NEW_ENVIRON, rfc-1572_."""

# std imports
import asyncio

# 3rd party
import pytest

# local
import telnetlib3
import telnetlib3.stream_writer
from telnetlib3.telopt import DO, IS, SB, SE, IAC, VAR, WILL, TTYPE, USERVAR, NEW_ENVIRON
from telnetlib3.tests.accessories import create_server, open_connection, asyncio_connection


async def test_telnet_server_on_environ(bind_host, unused_tcp_port):
    """Test Server's callback method on_environ()."""
    _waiter = asyncio.Future()

    class ServerTestEnviron(telnetlib3.TelnetServer):
        def on_environ(self, mapping):
            super().on_environ(mapping)
            _waiter.set_result(self)

    async with create_server(
        protocol_factory=ServerTestEnviron, host=bind_host, port=unused_tcp_port
    ):
        async with asyncio_connection(bind_host, unused_tcp_port) as (reader, writer):
            writer.write(IAC + WILL + NEW_ENVIRON)
            writer.write(
                IAC
                + SB
                + NEW_ENVIRON
                + IS
                + telnetlib3.stream_writer._encode_env_buf(
                    {"aLpHa": "oMeGa", "beta": "b", "gamma": "".join(chr(n) for n in range(0, 128))}
                )
                + IAC
                + SE
            )

            srv_instance = await asyncio.wait_for(_waiter, 0.5)
            assert srv_instance.get_extra_info("ALPHA") == "oMeGa"
            assert srv_instance.get_extra_info("BETA") == "b"
            assert srv_instance.get_extra_info("GAMMA") == ("".join(chr(n) for n in range(0, 128)))


async def test_telnet_client_send_environ(bind_host, unused_tcp_port):
    """Test Client's callback method send_environ() for specific requests."""
    _waiter = asyncio.Future()
    given_cols = 19
    given_rows = 84
    given_encoding = "cp437"
    given_term = "vt220"

    class ServerTestEnviron(telnetlib3.TelnetServer):
        def on_environ(self, mapping):
            super().on_environ(mapping)
            _waiter.set_result(mapping)

    async with create_server(
        protocol_factory=ServerTestEnviron, host=bind_host, port=unused_tcp_port
    ):
        async with open_connection(
            host=bind_host,
            port=unused_tcp_port,
            cols=given_cols,
            rows=given_rows,
            encoding=given_encoding,
            term=given_term,
        ) as (reader, writer):
            mapping = await asyncio.wait_for(_waiter, 0.5)
            assert mapping["COLUMNS"] == str(given_cols)
            assert mapping["LANG"] == "en_US." + given_encoding
            assert mapping["LINES"] == str(given_rows)
            assert mapping["TERM"] == "vt220"


async def test_telnet_client_send_var_uservar_environ(bind_host, unused_tcp_port):
    """Test Client's callback method send_environ() for VAR/USERVAR request."""
    _waiter = asyncio.Future()
    given_cols = 19
    given_rows = 84
    given_encoding = "cp437"
    given_term = "vt220"

    class ServerTestEnviron(telnetlib3.TelnetServer):
        def on_environ(self, mapping):
            super().on_environ(mapping)
            _waiter.set_result(mapping)

        def on_request_environ(self):
            return [VAR, USERVAR]

    async with create_server(
        protocol_factory=ServerTestEnviron, host=bind_host, port=unused_tcp_port
    ):
        async with open_connection(
            host=bind_host,
            port=unused_tcp_port,
            cols=given_cols,
            rows=given_rows,
            encoding=given_encoding,
            term=given_term,
            connect_maxwait=0.5,
        ) as (reader, writer):
            mapping = await asyncio.wait_for(_waiter, 0.5)
            assert mapping == {}


async def test_telnet_server_reject_environ(bind_host, unused_tcp_port):
    """Test Client's callback method send_environ() for specific requests."""
    given_cols = 19
    given_rows = 84
    given_term = "vt220"

    class ServerTestEnviron(telnetlib3.TelnetServer):
        def on_request_environ(self):
            return None

    async with create_server(
        protocol_factory=ServerTestEnviron,
        host=bind_host,
        port=unused_tcp_port,
        encoding=False,
        connect_maxwait=0.15,
    ):
        async with open_connection(
            host=bind_host,
            port=unused_tcp_port,
            cols=given_cols,
            rows=given_rows,
            encoding=False,
            term=given_term,
            connect_minwait=0.1,
            connect_maxwait=0.15,
        ) as (reader, writer):
            _failed = {key: val for key, val in writer.pending_option.items() if val}
            assert _failed == {SB + NEW_ENVIRON: True}


class _MockTransport:
    def get_extra_info(self, key, default=None):
        return default

    def write(self, data):
        pass

    def is_closing(self):
        return False


def _make_server():
    server = telnetlib3.TelnetServer()
    server.connection_made(_MockTransport())
    return server


@pytest.mark.parametrize(
    "ttype1,ttype2",
    [
        ("ANSI", "VT100"),
        ("ANSI", "ANSI"),
        ("ansi", "vt100"),
        ("xterm", "xterm"),
        ("xterm", "xterm-256color"),
    ],
)
async def test_negotiate_environ_always_sent(ttype1, ttype2):
    """DO NEW_ENVIRON is always sent regardless of client identity."""
    server = _make_server()
    server._extra["ttype1"] = ttype1
    server._extra["ttype2"] = ttype2
    server._negotiate_environ()
    assert server.writer.pending_option.get(DO + NEW_ENVIRON)


@pytest.mark.parametrize(
    "ttype1,ttype2,expect_user",
    [
        ("ANSI", "VT100", False),
        ("ANSI", "ANSI", True),
        ("ansi", "vt100", True),
        ("xterm", "xterm", True),
        ("xterm", "xterm-256color", True),
    ],
)
async def test_on_request_environ_user_excluded_for_ms_telnet(ttype1, ttype2, expect_user):
    """USER is excluded from NEW_ENVIRON request for Microsoft telnet."""
    server = _make_server()
    server._extra["ttype1"] = ttype1
    server._extra["ttype2"] = ttype2
    result = server.on_request_environ()
    assert ("USER" in result) is expect_user
    assert "LOGNAME" in result


@pytest.mark.parametrize("ttype_refused,final", [(True, False), (False, True)])
async def test_check_negotiation_triggers_environ(ttype_refused, final):
    """check_negotiation sends DO NEW_ENVIRON on TTYPE refusal or final."""
    server = _make_server()
    server._advanced = True
    if ttype_refused:
        server.writer.remote_option[TTYPE] = False
    server.check_negotiation(final=final)
    assert server._environ_requested
    assert server.writer.pending_option.get(DO + NEW_ENVIRON)


async def test_check_negotiation_no_advanced_skips_environ():
    """check_negotiation does not send DO NEW_ENVIRON without advanced."""
    server = _make_server()
    server.writer.remote_option[TTYPE] = False
    server.check_negotiation(final=True)
    assert not server._environ_requested
    assert not server.writer.pending_option.get(DO + NEW_ENVIRON)


@pytest.mark.parametrize("ttype,expect_requested", [("xterm", True), ("ANSI", False)])
async def test_on_ttype_environ_behavior(ttype, expect_requested):
    """on_ttype sends DO NEW_ENVIRON for non-ANSI, defers for ANSI."""
    server = _make_server()
    server.on_ttype(ttype)
    assert server._environ_requested is expect_requested
    assert bool(server.writer.pending_option.get(DO + NEW_ENVIRON)) is expect_requested


@pytest.mark.parametrize(
    "env,expect_force",
    [
        ({"LANG": "en_US.UTF-8"}, True),
        ({"LANG": "ja_JP.EUC-JP"}, True),
        ({"CHARSET": "UTF-8"}, True),
        ({"CHARSET": "ISO-8859-1", "USER": "test"}, True),
        ({"LANG": "en_US"}, False),
        ({"LANG": "C"}, False),
        ({"USER": "test", "TERM": "xterm"}, False),
        ({}, False),
    ],
)
async def test_on_environ_force_binary(env, expect_force):
    """on_environ sets force_binary when LANG has encoding or CHARSET is present."""
    server = _make_server()
    assert server.force_binary is False
    server.on_environ(dict(env))
    assert server.force_binary is expect_force