"""
A device that refuses the local network is named, not reported as "DataChannel closed".

Android 17 makes reaching the local network a permission of the application
(ACCESS_LOCAL_NETWORK), enforced for any application that targets it. Chrome
does from 155, and holds the permission only once the person grants it. Until
then the browser drops the node's addresses on the phone's own subnet: they are
absent from getStats, no check goes to them, the node's own checks get no
answer, ICE fails and the channel closes. On mobile data the same group opens,
over the node's public address. Measured on a Pixel, Android 17, Chrome 155,
2026-10-16; the addresses below are that session's.

These run the shipped transport: harness/local_network_harness.mjs.
"""

import json
import shutil
import subprocess
from pathlib import Path

import pytest
from spa_source import transport_argv

HARNESS = Path(__file__).parent / "harness" / "local_network_harness.mjs"

pytestmark = pytest.mark.skipif(shutil.which("node") is None, reason="node is not available")


def _run(cfg: dict):
    proc = subprocess.run(["node", str(HARNESS), transport_argv(), json.dumps(cfg)],
                          capture_output=True, text=True, encoding="utf-8", timeout=120)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def _sdp(*candidates: tuple[str, str]) -> str:
    lines = ["v=0", "m=application 9 UDP/DTLS/SCTP webrtc-datachannel"]
    for i, (address, kind) in enumerate(candidates):
        lines.append(f"a=candidate:{i} 1 udp 2130706431 {address} {50000 + i} typ {kind}")
    return "\r\n".join(lines) + "\r\n"


PHONE_WIFI = _sdp(("192.168.1.13", "host"), ("2a01:cb15:80e8:de00:c601:3ad3:2b17:e6a6", "host"),
                  ("90.112.206.172", "srflx"))
DESKTOP = _sdp(("64f602b7-7667-411f-b8ae-910c4d1de6fb.local", "host"),
               ("07b5e6a0-2b57-45ce-8748-ce2918cc6145.local", "host"),
               ("90.112.206.172", "srflx"))
PHONE_MOBILE = _sdp(("10.161.7.20", "host"), ("2a04:cec0:1000:2c41::1", "host"),
                    ("92.184.100.12", "srflx"))
PHONE_V6_ONLY = _sdp(("2a04:cec0:1000:2c41::1", "host"))
NODE = _sdp(("192.168.1.22", "host"), ("2a01:cb15:80e8:de00:d455:bcab:c324:4d72", "host"),
            ("192.168.200.254", "host"), ("192.168.100.254", "host"),
            ("90.112.206.172", "srflx"))
NODE_NO_STUN = _sdp(("192.168.1.22", "host"))
EVERY_NODE_ADDRESS = ["192.168.1.22", "2a01:cb15:80e8:de00:d455:bcab:c324:4d72",
                      "192.168.200.254", "192.168.100.254", "90.112.206.172"]
# What Chrome 155 kept, refused: the node's virtual adapters, off the phone's
# subnet, and its public address; not 192.168.1.22 nor the IPv6 of the same /64.
KEPT_WHEN_REFUSED = ["192.168.200.254", "192.168.100.254", "90.112.206.172"]


def test_the_measured_refusal_is_recognised():
    assert _run({"classify": [[PHONE_WIFI, NODE, KEPT_WHEN_REFUSED]]}) == [True]


def test_a_browser_that_pairs_every_address_is_not_refusing():
    """Desktop Chrome 152, 155 and 157 listed all six of the node's addresses."""
    assert _run({"classify": [[DESKTOP, NODE, EVERY_NODE_ADDRESS],
                              [PHONE_WIFI, NODE, EVERY_NODE_ADDRESS]]}) == [False, False]


def test_an_unreachable_node_is_not_blamed_on_the_device():
    """On mobile data the node's LAN address is paired and simply never answers:
    a node that is down, not a permission."""
    assert _run({"classify": [[PHONE_MOBILE, NODE, EVERY_NODE_ADDRESS]]}) == [False]


def test_without_an_ipv4_of_our_own_no_absence_means_anything():
    """No IPv4 candidate here, so no IPv4 address of the node could pair."""
    assert _run({"classify": [[PHONE_V6_ONLY, NODE, ["2a01:cb15:80e8:de00:d455:bcab:c324:4d72"]]]}) \
        == [False]


def test_a_node_with_no_public_address_is_judged_by_the_subnet():
    """With nothing paired at all, only an address on our own subnet tells."""
    other_lan = _sdp(("10.0.0.5", "host"))
    assert _run({"classify": [[PHONE_WIFI, NODE_NO_STUN, []],
                              [other_lan, NODE_NO_STUN, []]]}) == [True, False]


def test_connect_fails_with_the_reason():
    out = _run({"offer": PHONE_WIFI, "answer": NODE, "remote": KEPT_WHEN_REFUSED})
    assert out["reason"] == "local_network_blocked", out


def test_a_connection_that_did_get_through_is_not_a_refusal():
    """ICE connected once: whatever failed after, the local network was reached."""
    out = _run({"offer": PHONE_WIFI, "answer": NODE, "remote": KEPT_WHEN_REFUSED,
                "iceConnects": True})
    assert out["reason"] is None and out["statsCalls"] == 0, out


def test_stats_that_cannot_be_read_leave_the_old_error():
    out = _run({"offer": PHONE_WIFI, "answer": NODE, "remote": KEPT_WHEN_REFUSED,
                "statsThrow": True})
    assert out["reason"] is None and out["outcome"] == "DataChannel closed", out