"""Unit tests for the PSRP transport.

pypsrp is patched out at the module boundary: these tests pin down how the
transport configures it and how its failures reach the driver, not pypsrp itself.
"""

from __future__ import annotations

from unittest.mock import patch

import pytest
import requests
from napalm.base.exceptions import ConnectionClosedException, ConnectionException
from pypsrp.exceptions import AuthenticationError, WinRMTransportError

from napalm_windows.transport import PowerShellError, PsrpTransport


def _transport(**overrides):
    kwargs = {
        "port": 5986,
        "ssl": True,
        "cert_validation": True,
        "auth": "negotiate",
        "timeout": 60,
    }
    kwargs.update(overrides)
    return PsrpTransport("win01", "admin", "secret", **kwargs)


@pytest.fixture()
def psrp():
    """Patch WSMan, RunspacePool and PowerShell as the transport sees them."""
    with (
        patch("napalm_windows.transport.WSMan") as wsman,
        patch("napalm_windows.transport.RunspacePool") as pool,
        patch("napalm_windows.transport.PowerShell") as ps,
    ):
        yield {"wsman": wsman, "pool": pool, "ps": ps}


class TestOpen:
    def test_passes_connection_settings_to_wsman(self, psrp):
        _transport(port=5985, ssl=False, cert_validation=False, auth="ntlm").open()

        kwargs = psrp["wsman"].call_args.kwargs
        assert psrp["wsman"].call_args.args == ("win01",)
        assert kwargs["port"] == 5985
        assert kwargs["ssl"] is False
        assert kwargs["cert_validation"] is False
        assert kwargs["auth"] == "ntlm"
        assert kwargs["username"] == "admin"
        assert kwargs["password"] == "secret"  # noqa: S105

    def test_opens_one_runspace_pool_on_the_connection(self, psrp):
        _transport().open()

        psrp["pool"].assert_called_once_with(psrp["wsman"].return_value)
        psrp["pool"].return_value.open.assert_called_once_with()

    def test_authentication_failure_is_a_connection_exception(self, psrp):
        psrp["pool"].return_value.open.side_effect = AuthenticationError("bad creds")

        with pytest.raises(ConnectionException, match="Authentication failed"):
            _transport().open()

    def test_http_401_is_an_authentication_failure(self, psrp):
        psrp["pool"].return_value.open.side_effect = WinRMTransportError(
            "http", 401, "Unauthorized"
        )

        with pytest.raises(ConnectionException, match="Authentication failed"):
            _transport().open()

    def test_unreachable_host_is_a_connection_exception(self, psrp):
        psrp["pool"].return_value.open.side_effect = requests.ConnectionError("refused")

        with pytest.raises(ConnectionException, match="win01:5986"):
            _transport().open()

    def test_is_open_only_after_open(self, psrp):
        t = _transport()
        assert t.is_open is False
        t.open()
        assert t.is_open is True


class TestRun:
    def test_returns_output_objects_joined_by_newline(self, psrp):
        psrp["ps"].return_value.invoke.return_value = ['{"a":', "1}"]
        psrp["ps"].return_value.had_errors = False
        t = _transport()
        t.open()

        assert t.run("Get-Thing") == '{"a":\n1}'
        psrp["ps"].return_value.add_script.assert_called_once_with("Get-Thing")

    def test_none_objects_are_skipped(self, psrp):
        psrp["ps"].return_value.invoke.return_value = [None, "x", None]
        psrp["ps"].return_value.had_errors = False
        t = _transport()
        t.open()

        assert t.run("Get-Thing") == "x"

    def test_error_stream_raises_powershell_error_with_its_text(self, psrp):
        psrp["ps"].return_value.invoke.return_value = []
        psrp["ps"].return_value.had_errors = True
        psrp["ps"].return_value.streams.error = [
            "Cannot find any service with service name 'nope'."
        ]
        t = _transport()
        t.open()

        with pytest.raises(PowerShellError, match="service name 'nope'"):
            t.run("Start-Service nope")

    def test_run_before_open_raises_connection_closed(self):
        with pytest.raises(ConnectionClosedException):
            _transport().run("Get-Thing")


class TestClose:
    def test_close_closes_pool_and_connection(self, psrp):
        t = _transport()
        t.open()
        t.close()

        psrp["pool"].return_value.close.assert_called_once_with()
        psrp["wsman"].return_value.close.assert_called_once_with()
        assert t.is_open is False

    def test_close_swallows_errors_from_a_dead_connection(self, psrp):
        psrp["pool"].return_value.close.side_effect = requests.ConnectionError("gone")
        t = _transport()
        t.open()

        t.close()

        assert t.is_open is False

    def test_close_without_open_is_a_no_op(self):
        _transport().close()


def test_transport_does_not_print_password_in_repr():
    assert "secret" not in repr(_transport())