signal-bot/tests/test_portcheck.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Refusing to start on a taken port, and naming what holds it."""

from __future__ import annotations

import socket
from typing import TYPE_CHECKING

import pytest

from signal_bot.portcheck import PortInUseError, assert_port_free, describe_listener

if TYPE_CHECKING:
    from pathlib import Path

LISTEN = "0A"
ESTABLISHED = "01"


def _tcp_line(port: int, inode: str, state: str = LISTEN) -> str:
    """One /proc/net/tcp row for a socket on the given port."""
    return (
        f"   0: 0100007F:{port:04X} 00000000:0000 {state} "
        f"00000000:00000000 00:00000000 00000000  1000        0 {inode} 1"
    )


def _proc(tmp_path: Path, rows: list[str]) -> Path:
    """Build a fake /proc containing just a net/tcp table."""
    net = tmp_path / "net"
    net.mkdir(parents=True, exist_ok=True)
    (net / "tcp").write_text("  sl  local_address ...\n" + "\n".join(rows) + "\n")
    return tmp_path


def _process(
    root: Path, pid: str, inode: str, cmdline: bytes = b"/usr/bin/python\0-m\0bot\0"
) -> Path:
    """Add a process to the fake /proc holding the given socket inode."""
    pid_dir = root / pid
    (pid_dir / "fd").mkdir(parents=True)
    (pid_dir / "fd" / "3").symlink_to(f"socket:[{inode}]")
    (pid_dir / "cmdline").write_bytes(cmdline)
    return pid_dir


def test_names_the_pid_and_command_holding_the_port(tmp_path: Path) -> None:
    root = _proc(tmp_path, [_tcp_line(8137, "25863")])
    _process(root, "4242", "25863")
    assert describe_listener(8137, root) == "held by pid 4242: /usr/bin/python -m bot"


def test_reports_nothing_when_no_socket_listens_on_the_port(tmp_path: Path) -> None:
    root = _proc(tmp_path, [_tcp_line(9999, "25863")])
    assert describe_listener(8137, root) == ""


def test_ignores_sockets_that_are_not_listening(tmp_path: Path) -> None:
    root = _proc(tmp_path, [_tcp_line(8137, "25863", state=ESTABLISHED)])
    assert describe_listener(8137, root) == ""


def test_survives_a_proc_with_no_tcp_tables_at_all(tmp_path: Path) -> None:
    assert describe_listener(8137, tmp_path) == ""


def test_skips_short_and_unparseable_rows(tmp_path: Path) -> None:
    root = _proc(
        tmp_path,
        [
            "   1: truncated row",
            "   2: NOTHEX:NOTHEX 00000000:0000 0A a b c  1000  0 99 1",
            _tcp_line(8137, "25863"),
        ],
    )
    _process(root, "4242", "25863")
    assert "pid 4242" in describe_listener(8137, root)


def test_reports_another_user_when_the_owner_is_invisible(tmp_path: Path) -> None:
    """Scanning /proc/<pid>/fd fails for processes that are not ours."""
    root = _proc(tmp_path, [_tcp_line(8137, "25863")])
    (root / "4242").mkdir()  # no fd/ directory at all
    (root / "not-a-pid").mkdir()
    assert describe_listener(8137, root) == "held by a process belonging to another user"


def test_skips_descriptors_that_are_not_symlinks(tmp_path: Path) -> None:
    root = _proc(tmp_path, [_tcp_line(8137, "25863")])
    pid_dir = root / "4242"
    (pid_dir / "fd").mkdir(parents=True)
    (pid_dir / "fd" / "3").write_text("not a link")
    assert describe_listener(8137, root) == "held by a process belonging to another user"


def test_falls_back_when_the_command_line_is_unreadable(tmp_path: Path) -> None:
    root = _proc(tmp_path, [_tcp_line(8137, "25863")])
    pid_dir = root / "4242"
    (pid_dir / "fd").mkdir(parents=True)
    (pid_dir / "fd" / "3").symlink_to("socket:[25863]")
    # No cmdline file: a process that exited between the two reads.
    assert describe_listener(8137, root) == "held by pid 4242: (command unavailable)"


def test_falls_back_when_the_command_line_is_empty(tmp_path: Path) -> None:
    root = _proc(tmp_path, [_tcp_line(8137, "25863")])
    _process(root, "4242", "25863", cmdline=b"")
    assert describe_listener(8137, root) == "held by pid 4242: (command unavailable)"


def test_a_free_port_is_accepted(tmp_path: Path) -> None:
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        free = probe.getsockname()[1]
    # Must simply not raise; a bare comparison here would assert nothing.
    assert_port_free("127.0.0.1", free, tmp_path)


def test_a_taken_port_raises_and_names_the_holder(tmp_path: Path) -> None:
    with socket.socket() as held:
        held.bind(("127.0.0.1", 0))
        held.listen(1)
        port = held.getsockname()[1]
        root = _proc(tmp_path, [_tcp_line(port, "25863")])
        _process(root, "4242", "25863")
        with pytest.raises(PortInUseError, match="held by pid 4242") as caught:
            assert_port_free("127.0.0.1", port, root)
    assert "Set BIND_PORT" in str(caught.value)


def test_a_taken_port_still_raises_when_the_holder_is_unknown(tmp_path: Path) -> None:
    with socket.socket() as held:
        held.bind(("127.0.0.1", 0))
        held.listen(1)
        port = held.getsockname()[1]
        with pytest.raises(PortInUseError, match=f"cannot bind 127.0.0.1:{port}"):
            assert_port_free("127.0.0.1", port, tmp_path)


def test_scans_past_processes_holding_other_sockets(tmp_path: Path) -> None:
    """The first process examined rarely owns the port; the scan must go on."""
    root = _proc(tmp_path, [_tcp_line(8137, "25863")])
    _process(root, "1111", "99999")  # a different socket entirely
    _process(root, "4242", "25863")
    assert describe_listener(8137, root) == "held by pid 4242: /usr/bin/python -m bot"