"""An SSH connection pointed at the machine LLeMbas itself runs on.

The whole design of agent chats is one sentence -- nothing runs on this host --
and a profile pointed at 127.0.0.1 walks past it while looking, from the SSH
layer down, exactly like a container on the network. So it is an administrator's
decision in three positions, and what is pinned here is that every one of them is
enforced where it has to be rather than only in the form.

Note the conftest: the suite runs with the switch **open**, because the tests
that stand up a real asyncssh server can only listen on loopback. Every test in
this file closes it explicitly, which is also why they are together in one file.
"""

from __future__ import annotations

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from lembas.db.models import KIND_AGENT, Chat, SshProfile, User
from lembas.services import settings_store
from lembas.services.agent import hosts
from lembas.services.agent import session as agent_session
from lembas.services.crypto import encrypt


def _set(db, mode: str, port: int = 0) -> None:
    settings_store.update(
        db, {"enabled": True, "loopback": mode, "loopback_port": port}, key=settings_store.AGENTS
    )
    db.commit()


def _profile(db, host: str = "127.0.0.1", port: int = 22) -> SshProfile:
    profile = SshProfile(
        owner_id=db.scalars(select(User)).first().id,
        name=f"box-{host}-{port}",
        host=host,
        port=port,
        username="deploy",
        auth="password",
        password_encrypted=encrypt("hunter2"),
        host_key="ssh-ed25519 AAAA",
        host_fingerprint="SHA256:x",
        enabled=True,
    )
    db.add(profile)
    db.commit()
    return profile


# --- Recognising the machine --------------------------------------------------
@pytest.mark.parametrize(
    "host",
    ["127.0.0.1", "127.0.0.53", "127.1.2.3", "::1", "[::1]", "localhost", "LOCALHOST"],
)
def test_the_obvious_spellings_are_recognised(host: str):
    """All of them, and from the string alone. `127.0.0.53` is the one worth
    listing: the whole of 127.0.0.0/8 is loopback, and a check written against
    the single address `127.0.0.1` misses every other way of writing it."""
    assert hosts.is_loopback(host)


@pytest.mark.parametrize("host", ["10.0.0.9", "192.168.1.5", "example.test", ""])
def test_anything_else_is_not(host: str):
    """Including a private address. This guard is about *this machine*, not
    about the network it is on -- refusing 192.168.x would refuse the container
    on the shelf, which is the intended use of the whole feature."""
    assert not hosts.is_loopback(host)


def test_recognising_a_host_never_makes_a_network_call(monkeypatch):
    """`getaddrinfo` blocks. `is_loopback` is called several times per page
    render -- the composer's picker, `resolve_tools`, the canvas -- so a lookup
    behind it makes an agent page wait out a DNS timeout for a host nobody is
    talking to. The first version of this did exactly that and the suite went
    from two minutes to not finishing."""
    import socket

    def explode(*args, **kwargs):  # pragma: no cover - the point is it is unused
        raise AssertionError("is_loopback resolved a name on the request path")

    monkeypatch.setattr(socket, "getaddrinfo", explode)

    assert hosts.is_loopback("127.0.0.1")
    assert not hosts.is_loopback("build.example")


def test_a_name_is_settled_by_resolving_it(monkeypatch):
    """Which is where a hosts-file entry gets caught. Blocking, and therefore
    called only where a network call is already expected."""
    import socket

    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *a, **k: [(2, 1, 6, "", ("127.0.0.1", 22))],
    )
    assert hosts.resolves_here("build.example")

    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *a, **k: [(2, 1, 6, "", ("93.184.216.34", 22))],
    )
    assert not hosts.resolves_here("build.example")


def test_a_name_that_does_not_resolve_is_not_this_machine(monkeypatch):
    """Refusing it would turn every DNS hiccup into "your connection is on this
    machine", which is both wrong and unactionable. The connection fails on its
    own terms a moment later."""
    import socket

    def fail(*args, **kwargs):
        raise OSError("no such host")

    monkeypatch.setattr(socket, "getaddrinfo", fail)
    assert not hosts.resolves_here("nowhere.invalid")


# --- The three positions ------------------------------------------------------
def test_off_refuses_every_port(db, registered):
    _set(db, hosts.MODE_OFF)
    assert hosts.refusal(db, "127.0.0.1", 22)
    assert hosts.refusal(db, "127.0.0.1", 2222)


def test_on_allows_every_port(db, registered):
    _set(db, hosts.MODE_ON)
    assert hosts.refusal(db, "127.0.0.1", 22) == ""
    assert hosts.refusal(db, "127.0.0.1", 2222) == ""


def test_one_port_allows_that_one_and_nothing_else(db, registered):
    _set(db, hosts.MODE_PORT, 2222)
    assert hosts.refusal(db, "127.0.0.1", 2222) == ""
    assert hosts.refusal(db, "127.0.0.1", 2200)
    # And a host that is not this machine is not this switch's business.
    assert hosts.refusal(db, "10.0.0.9", 22) == ""


def test_port_22_is_refused_even_when_it_is_the_named_one(db, registered):
    """The position exists for a container that published its SSH port on the
    loopback interface. Port 22 is not that -- it is this host's own sshd, and
    naming it is either a mistake or the thing the switch exists to stop."""
    _set(db, hosts.MODE_PORT, 22)
    assert hosts.refusal(db, "127.0.0.1", 22)


def test_naming_no_port_refuses_rather_than_allows(db, registered):
    """`port` with nothing named is an unfinished setting, and the safe reading
    of an unfinished setting is the one that refuses."""
    _set(db, hosts.MODE_PORT, 0)
    assert hosts.refusal(db, "127.0.0.1", 2222)


def test_an_unrecognised_stored_position_means_off(db, registered):
    """Fail closed. The direction that is safe to get wrong is refusing a
    connection somebody has to re-allow; the other direction is a shell on the
    host holding the database and the encryption key."""
    settings_store.update(db, {"enabled": True, "loopback": "sometimes"}, key=settings_store.AGENTS)
    db.commit()
    assert hosts.refusal(db, "127.0.0.1", 2222)


# --- Where it is enforced -----------------------------------------------------
def test_a_refused_connection_cannot_be_saved(client: TestClient, db, registered):
    _set(db, hosts.MODE_OFF)

    response = client.post(
        "/api/agents",
        data={
            "name": "Here",
            "host": "127.0.0.1",
            "port": "22",
            "username": "deploy",
            "auth": "password",
            "password": "hunter2",
            "enabled": "on",
        },
        follow_redirects=False,
    )

    assert response.status_code == 200  # the form again, with the reason on it
    assert "points at the machine LLeMbas itself runs on" in response.text
    assert db.scalar(select(SshProfile).where(SshProfile.name == "Here")) is None


def test_a_connection_saved_before_the_switch_moved_stops_working(db, registered):
    """The one that matters. A row can predate a setting, so refusing only at
    save time would leave every existing loopback profile working exactly as it
    did -- which is the whole of the flaw, untouched.

    `session.resolve` is the control: it is what every agent tool, the terminal
    and the canvas go through.
    """
    _set(db, hosts.MODE_ON)
    profile = _profile(db)
    chat = Chat(
        user_id=profile.owner_id, kind=KIND_AGENT, ssh_profile_id=profile.id, title="t"
    )
    db.add(chat)
    db.commit()

    owner = db.get(User, profile.owner_id)
    assert agent_session.resolve(db, chat, owner) is not None

    _set(db, hosts.MODE_OFF)
    assert agent_session.resolve(db, chat, owner) is None


def test_a_refused_connection_is_not_offered_in_the_composer(
    client: TestClient, db, registered
):
    """Or the reader picks it, starts an agent chat, and finds a chat with no
    tools and nothing said about why."""
    _set(db, hosts.MODE_ON)
    profile = _profile(db)

    assert profile.name in client.get("/chat").text

    _set(db, hosts.MODE_OFF)
    assert profile.name not in client.get("/chat").text


def test_the_listing_says_why_rather_than_hiding_it(client: TestClient, db, registered):
    """The connection is still the owner's and still editable -- pointing it at
    another host is the way out. A row that silently stopped working would be
    indistinguishable from a fault."""
    _set(db, hosts.MODE_OFF)
    profile = _profile(db)

    body = client.get("/agents").text

    assert profile.name in body
    assert "not allowed" in body
    assert "points at the machine LLeMbas itself runs on" in body


def test_check_opens_no_socket_for_a_refused_connection(
    client: TestClient, db, registered, monkeypatch
):
    """Check is the one button here that connects, so it must refuse before it
    does rather than after."""
    _set(db, hosts.MODE_OFF)
    profile = _profile(db)

    async def explode(*args, **kwargs):  # pragma: no cover - the point is it is unused
        raise AssertionError("a refused connection was contacted")

    from lembas.services.agent import ssh as ssh_service

    monkeypatch.setattr(ssh_service, "capture_host_key", explode)

    body = client.post(f"/api/agents/{profile.id}/check").text

    assert "points at the machine LLeMbas itself runs on" in body


def test_a_draft_cannot_be_opened_against_a_refused_connection(
    client: TestClient, db, registered
):
    """The draft is what the terminal and the canvas open against before a chat
    exists, so this is the new-chat path's own gate."""
    _set(db, hosts.MODE_OFF)
    profile = _profile(db)

    assert client.get(f"/api/agents/{profile.id}/draft").status_code == 403


def test_the_unspecified_address_is_this_machine_too():
    """`0.0.0.0` and `::` are neither a real destination nor a refused one:
    connect() to either goes to loopback on Linux, so a profile pointed at
    `0.0.0.0` reached this host's own sshd.

    `is_loopback` alone answered a decided **False**, which also short-circuited
    `resolves_here` -- so the DNS half never ran either, and the one spelling of
    "this machine" that mattered walked past a guard whose whole job is that
    sentence.
    """
    for spelling in ("0.0.0.0", "::", "[::]", "0.0.0.0 "):
        assert hosts.is_loopback(spelling), spelling


def test_a_real_address_is_still_not_this_machine():
    for spelling in ("192.168.1.5", "10.0.0.1", "example.com", "203.0.113.9"):
        assert not hosts.is_loopback(spelling), spelling
