"""Transport tests against mocked HTTP serving the DMTF public-rackmount1 mockup."""

import base64
import json
import logging
import os

import pytest
import requests
from conftest import BASE_URL, SESSION_PATH, SESSIONS, TOKEN

from napalm_redfish.transport import (
    RedfishAuthError,
    RedfishError,
    RedfishHTTPError,
    RedfishNotFound,
    RedfishTransport,
)

BASIC = "Basic " + base64.b64encode(b"netork:s3cret").decode()


def make(**kwargs) -> RedfishTransport:
    kwargs.setdefault("verify", False)
    kwargs.setdefault("retries", 0)
    return RedfishTransport("bmc.example.net", "netork", "s3cret", **kwargs)


# --- base URL ------------------------------------------------------------------------------


@pytest.mark.parametrize(
    "hostname, port, expected",
    [
        ("bmc.example.net", None, "https://bmc.example.net"),
        ("bmc.example.net", 8443, "https://bmc.example.net:8443"),
        ("192.0.2.10", None, "https://192.0.2.10"),
        ("2001:db8::10", None, "https://[2001:db8::10]"),
        ("2001:db8::10", 8443, "https://[2001:db8::10]:8443"),
    ],
)
def test_base_url(hostname, port, expected):
    assert RedfishTransport(hostname, "u", "p", port=port).base_url == expected


def test_rejects_unknown_auth_method():
    with pytest.raises(ValueError):
        RedfishTransport("bmc.example.net", "u", "p", auth="digest")


# --- session login -------------------------------------------------------------------------


def test_open_reads_service_root_before_login(server):
    t = make()
    t.open()
    first = server.rsps.calls[0].request
    assert first.method == "GET" and first.url.rstrip("/") == BASE_URL + "/redfish/v1"
    assert "X-Auth-Token" not in first.headers and "Authorization" not in first.headers
    assert t.root["RedfishVersion"] == "1.15.0"
    t.close()


def test_session_login_posts_credentials_and_uses_token(server):
    t = make()
    t.open()
    assert t.auth_method == "session"
    (login,) = server.calls("POST", SESSIONS)
    assert json.loads(login.request.body) == {"UserName": "netork", "Password": "s3cret"}

    t.get("/redfish/v1/Systems")
    get = server.calls("GET", "/redfish/v1/Systems")[0].request
    assert get.headers["X-Auth-Token"] == TOKEN
    assert "Authorization" not in get.headers
    t.close()


def test_close_deletes_the_session(server):
    t = make()
    t.open()
    t.close()
    (delete,) = server.calls("DELETE", SESSION_PATH)
    assert delete.request.headers["X-Auth-Token"] == TOKEN
    assert t.auth_method is None


def test_open_twice_logs_in_once(server):
    t = make()
    t.open()
    t.open()
    t.close()
    assert len(server.calls("POST", SESSIONS)) == 1
    assert len(server.calls("DELETE", SESSION_PATH)) == 1


def test_close_twice_deletes_once(server):
    t = make()
    t.open()
    t.close()
    t.close()
    assert len(server.calls("DELETE")) == 1


def test_context_manager_deletes_session_on_error(server):
    with pytest.raises(RuntimeError):
        with make() as t:
            t.get("/redfish/v1/Systems")
            raise RuntimeError("boom")
    assert len(server.calls("DELETE", SESSION_PATH)) == 1


def test_close_survives_failed_logout(server, caplog):
    server.override("DELETE", SESSION_PATH, 500, {"error": {"message": "nope"}})
    t = make()
    t.open()
    with caplog.at_level(logging.WARNING):
        t.close()
    assert t.auth_method is None
    assert "session" in caplog.text.lower()


@pytest.mark.filterwarnings("ignore:Service incorrectly did not provide the 'Location'")
def test_session_uri_from_body_when_location_missing(server):
    server.override("POST", SESSIONS, 201, {"@odata.id": SESSION_PATH}, {"X-Auth-Token": TOKEN})
    t = make()
    t.open()
    t.close()
    assert len(server.calls("DELETE", SESSION_PATH)) == 1


def test_wrong_password_raises_without_basic_retry(server):
    server.override("POST", SESSIONS, 401, {"error": {"message": "bad credentials"}})
    t = make()
    with pytest.raises(RedfishAuthError):
        t.open()
    # One failed login only: a second attempt with basic auth would count twice against
    # the BMC's lockout threshold.
    assert len(server.calls("POST")) == 1
    assert not [c for c in server.rsps.calls if "Authorization" in c.request.headers]


@pytest.mark.parametrize("status", [404, 405, 501])
def test_falls_back_to_basic_when_sessions_unsupported(server, status):
    server.override("POST", SESSIONS, status, {"error": {"message": "not supported"}})
    t = make()
    t.open()
    assert t.auth_method == "basic"
    t.get("/redfish/v1/Systems")
    get = server.calls("GET", "/redfish/v1/Systems")[0].request
    assert get.headers["Authorization"] == BASIC
    assert "X-Auth-Token" not in get.headers
    t.close()
    assert not server.calls("DELETE")


@pytest.mark.filterwarnings("ignore:Service incorrectly did not provide the 'Location'")
def test_falls_back_to_basic_when_no_token_returned(server):
    server.override("POST", SESSIONS, 201, {"@odata.id": SESSION_PATH})
    t = make()
    t.open()
    assert t.auth_method == "basic"
    t.close()


def test_no_fallback_when_session_auth_is_forced(server):
    server.override("POST", SESSIONS, 405, {})
    t = make(auth="session")
    with pytest.raises(RedfishAuthError):
        t.open()


def test_basic_auth_never_creates_a_session(server):
    t = make(auth="basic")
    t.open()
    assert t.auth_method == "basic"
    t.get("/redfish/v1/Systems")
    assert not server.calls("POST")
    assert server.calls("GET", "/redfish/v1/Systems")[0].request.headers["Authorization"] == BASIC
    t.close()


def test_basic_auth_wrong_password(server):
    server.override("GET", SESSIONS, 401, {})
    with pytest.raises(RedfishAuthError):
        make(auth="basic").open()


def test_login_url_from_service_root(server):
    custom = dict(_root(), Links={"Sessions": {"@odata.id": "/redfish/v1/Custom/Sessions"}})
    server.override("GET", "/redfish/v1/", 200, custom)
    server.override("GET", "/redfish/v1", 200, custom)
    server.override("POST", "/redfish/v1/Custom/Sessions", 201, {},
                    {"X-Auth-Token": TOKEN, "Location": SESSION_PATH})
    t = make()
    t.open()
    assert server.calls("POST", "/redfish/v1/Custom/Sessions")
    t.close()


def _root():
    from conftest import load

    return load("dmtf-public-rackmount1", "/redfish/v1")


def test_unreachable_service_raises_redfish_error(http):
    t = make()
    with pytest.raises(RedfishError):
        t.open()  # nothing registered on the mock: connection refused


def test_requests_before_open_fail(server):
    with pytest.raises(RedfishError):
        make().get("/redfish/v1/Systems")


# --- certificate verification --------------------------------------------------------------


def _verify_of(server):
    return server.rsps.calls[0].request.req_kwargs["verify"]


def test_verify_true_uses_ca_bundle(server):
    t = make(verify=True)
    t.open()
    verify = _verify_of(server)
    # The redfish library turns verification off unless it gets a CA file, so True must
    # become a real bundle, not None/False.
    assert isinstance(verify, str) and os.path.isfile(verify)
    assert verify == requests.certs.where()
    t.close()


def test_verify_false_disables_verification(server):
    t = make(verify=False)
    t.open()
    assert _verify_of(server) is False
    t.close()


def test_verify_path_is_passed_through(server, tmp_path):
    ca = tmp_path / "bmc-ca.pem"
    ca.write_text("dummy")
    t = make(verify=str(ca))
    t.open()
    assert _verify_of(server) == str(ca)
    t.close()


def test_verify_path_must_exist():
    with pytest.raises(ValueError):
        RedfishTransport("bmc.example.net", "u", "p", verify="/nonexistent/ca.pem")


def test_verify_is_the_default():
    assert RedfishTransport("bmc.example.net", "u", "p").verify is True


# --- GET and errors ------------------------------------------------------------------------


@pytest.fixture
def transport(server):
    t = make()
    t.open()
    yield t
    t.close()


def test_get_returns_json(transport):
    assert transport.get("/redfish/v1/Systems/437XR1138R2")["Model"] == "3500"


def test_get_accepts_odata_reference(transport):
    ref = {"@odata.id": "/redfish/v1/Systems/437XR1138R2"}
    assert transport.get(ref)["Id"] == "437XR1138R2"


def test_get_strips_fragment(transport, server):
    data = transport.get("/redfish/v1/Chassis/1U/Thermal#/Fans/0")
    assert data["Id"] == "Thermal"
    assert server.calls("GET", "/redfish/v1/Chassis/1U/Thermal")[0].request.url.endswith("/Thermal")


def test_get_accepts_same_host_absolute_url(transport):
    assert transport.get(BASE_URL + "/redfish/v1/Systems")["Members@odata.count"] == 1


def test_get_refuses_links_to_other_hosts(transport, server):
    before = len(server.rsps.calls)
    # PowerSupply.Links.Outlet in the mockup points at a PDU on another host. Following it
    # would send this BMC's session token there.
    with pytest.raises(RedfishError):
        transport.get("https://redfishpdu.contoso.com/redfish/v1/PowerEquipment/RackPDUs/1")
    assert len(server.rsps.calls) == before


def test_get_refuses_relative_paths(transport):
    with pytest.raises(RedfishError):
        transport.get("Systems/1")


def test_get_404_raises_not_found(transport):
    with pytest.raises(RedfishNotFound) as exc:
        transport.get("/redfish/v1/Systems/1")
    assert exc.value.status == 404
    assert exc.value.path == "/redfish/v1/Systems/1"


def test_get_optional_returns_none_on_404(transport):
    assert transport.get_optional("/redfish/v1/Chassis/1U/Missing") is None
    assert transport.get_optional(None) is None


def test_get_error_carries_redfish_message(transport, server):
    server.override("GET", "/redfish/v1/Managers", 500, {
        "error": {"code": "Base.1.0.InternalError", "message": "General error",
                  "@Message.ExtendedInfo": [{"Message": "The BMC is busy."}]}})
    with pytest.raises(RedfishHTTPError) as exc:
        transport.get("/redfish/v1/Managers")
    assert exc.value.status == 500
    assert "The BMC is busy." in str(exc.value)


@pytest.mark.parametrize("status", [401, 403])
def test_get_auth_errors(transport, server, status):
    server.override("GET", "/redfish/v1/AccountService", status, {})
    with pytest.raises(RedfishAuthError):
        transport.get("/redfish/v1/AccountService")


def test_get_invalid_json(transport, server):
    server.override("GET", "/redfish/v1/Managers", 200, "<html>login</html>")
    with pytest.raises(RedfishError):
        transport.get("/redfish/v1/Managers")


# --- collections ---------------------------------------------------------------------------


def test_members_follows_odata_ids(transport, server):
    systems = transport.members("/redfish/v1/Systems")
    # The mockup's system ID is its serial number, not "1".
    assert [s["Id"] for s in systems] == ["437XR1138R2"]
    assert server.calls("GET", "/redfish/v1/Systems/437XR1138R2")


def test_members_ignores_member_count(transport):
    # Members@odata.count says 2, but three members are listed: trust the list.
    fw = transport.members("/redfish/v1/UpdateService/FirmwareInventory")
    assert [f["Id"] for f in fw] == ["BMC", "SS", "BIOS"]


def test_members_from_link_on_resource(transport):
    system = transport.get("/redfish/v1/Systems/437XR1138R2")
    cpus = transport.members(system["Processors"])
    assert [c["Id"] for c in cpus] == ["CPU1", "CPU2", "FPGA1"]


def test_members_uses_expanded_members_without_extra_requests(transport, server):
    entries = transport.members("/redfish/v1/Managers/BMC/LogServices/Log/Entries")
    assert entries[0]["Message"] == "System May be Melting"
    assert not server.calls("GET", "/redfish/v1/Managers/BMC/LogServices/Log/Entries/1")


def test_members_follows_next_link(transport, server):
    path = "/redfish/v1/Systems/437XR1138R2/LogServices/Log1/Entries"
    server.override("GET", path + "?$skiptoken=2", 200, {
        "Members": [{"@odata.id": path + "/3", "Id": "3", "Message": "third"}],
        "@odata.id": path,
    })
    entries = transport.members(path)
    assert [e["Id"] for e in entries] == ["1", "2", "3"]


def test_members_next_link_loop_terminates(transport, server):
    path = "/redfish/v1/Managers"
    server.override("GET", path, 200, {
        "Members": [{"@odata.id": "/redfish/v1/Managers/BMC"}],
        "Members@odata.nextLink": path,
    })
    assert [m["Id"] for m in transport.members(path)] == ["BMC"]


def test_members_limit(transport, server):
    path = "/redfish/v1/Systems/437XR1138R2/LogServices/Log1/Entries"
    entries = transport.members(path, limit=1)
    assert [e["Id"] for e in entries] == ["1"]
    assert len(server.calls("GET", path)) == 1


def test_members_of_missing_collection(transport):
    with pytest.raises(RedfishNotFound):
        transport.members("/redfish/v1/Systems/437XR1138R2/Storage")


def test_follow(transport):
    manager = transport.get("/redfish/v1/Managers/BMC")
    assert transport.follow(manager, "NetworkProtocol")["Id"] == "NetworkProtocol"
    assert transport.follow(manager, "NoSuchLink") is None


def test_follow_missing_target_is_none(transport, server):
    system = transport.get("/redfish/v1/Systems/437XR1138R2")
    # Storage is not linked from this system; Bios is linked but not in the fixtures.
    assert transport.follow(system, "Storage") is None
    assert transport.follow(system, "Bios") is None


# --- POST ----------------------------------------------------------------------------------


def test_post_sends_json_and_returns_body(transport, server):
    target = "/redfish/v1/Systems/437XR1138R2/Actions/ComputerSystem.Reset"
    server.override("POST", target, 204, {})
    assert transport.post(target, {"ResetType": "GracefulRestart"}) == {}
    (call,) = server.calls("POST", target)
    assert json.loads(call.request.body) == {"ResetType": "GracefulRestart"}
    assert call.request.headers["X-Auth-Token"] == TOKEN


def test_post_error(transport, server):
    target = "/redfish/v1/Managers/BMC/Actions/Manager.Reset"
    server.override("POST", target, 400, {"error": {"message": "ActionParameterValueNotInList"}})
    with pytest.raises(RedfishHTTPError) as exc:
        transport.post(target, {"ResetType": "PowerCycle"})
    assert exc.value.status == 400