Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 1 addition & 19 deletions src/whoseport/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,22 +5,20 @@
import argparse
import json
import os
import re
import shutil
import sys
from collections import Counter

import whoseport
from whoseport import core, kill
from whoseport.core import BackendError, PortUser, UnsupportedPlatformError
from whoseport.tokens import parse_port_tokens

EXIT_OK = 0
EXIT_NOTHING_FOUND = 1
EXIT_USAGE = 2
EXIT_KILL_FAILED = 3

_PORT_TOKEN = re.compile(r"^(\d{1,5})(?:-(\d{1,5}))?$")


# ---------------------------------------------------------------- colors ---

Expand Down Expand Up @@ -93,22 +91,6 @@ def sym(fancy: str, plain: str) -> str:
# ---------------------------------------------------------------- parsing ---


def parse_port_tokens(tokens: list) -> set:
ports = set()
for token in tokens:
m = _PORT_TOKEN.match(token)
if not m:
raise ValueError(f"invalid port or range: {token!r} (examples: 8080, 3000-3005)")
start = int(m.group(1))
end = int(m.group(2)) if m.group(2) else start
if not (1 <= start <= 65535 and 1 <= end <= 65535):
raise ValueError(f"port out of range 1-65535: {token!r}")
if end < start:
raise ValueError(f"range end before start: {token!r}")
ports.update(range(start, end + 1))
return ports


def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="whoseport",
Expand Down
61 changes: 61 additions & 0 deletions src/whoseport/tokens.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
"""Port token parsing: ranges, host:port, and URLs."""

from __future__ import annotations

import re
from urllib.parse import urlparse

_PORT_TOKEN = re.compile(r"^(\d{1,5})(?:-(\d{1,5}))?$")


def _strip_port_suffix(port_s: str) -> str:
"""Drop path / query / fragment so ``host:8080/health#x`` still parses."""
return port_s.split("/", 1)[0].split("?", 1)[0].split("#", 1)[0]


def port_from_endpoint(token: str) -> int | None:
"""Pull a port out of ``host:port``, ``[ipv6]:port`` or a URL.

All-digit left-hand sides such as ``8080:8090`` stay rejected so a
mistyped range is not silently turned into a single port.
"""
text = token.strip()
if "://" in text:
try:
port = urlparse(text).port
except ValueError:
return None
return int(port) if port else None
if text.startswith("[") and "]:" in text:
port_s = _strip_port_suffix(text.rsplit("]:", 1)[-1])
return int(port_s) if port_s.isdigit() else None
if ":" in text and not text.startswith(":"):
host, _, port_s = text.rpartition(":")
port_s = _strip_port_suffix(port_s)
if host and not host.isdigit() and port_s.isdigit():
return int(port_s)
return None


def parse_port_tokens(tokens: list) -> set:
ports = set()
for token in tokens:
extracted = port_from_endpoint(token)
if extracted is not None:
if not 1 <= extracted <= 65535:
raise ValueError(f"port out of range 1-65535: {token!r}")
ports.add(extracted)
continue
m = _PORT_TOKEN.match(token)
if not m:
raise ValueError(
f"invalid port or range: {token!r} (examples: 8080, 3000-3005, localhost:8080)"
)
start = int(m.group(1))
end = int(m.group(2)) if m.group(2) else start
if not (1 <= start <= 65535 and 1 <= end <= 65535):
raise ValueError(f"port out of range 1-65535: {token!r}")
if end < start:
raise ValueError(f"range end before start: {token!r}")
ports.update(range(start, end + 1))
return ports
38 changes: 38 additions & 0 deletions tests/test_host_port.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
import json

from whoseport import cli, core
from whoseport.tokens import parse_port_tokens


def test_parse_host_port_and_url():
assert parse_port_tokens(["localhost:8080"]) == {8080}
assert parse_port_tokens(["127.0.0.1:3000"]) == {3000}
assert parse_port_tokens(["[::1]:5173"]) == {5173}
assert parse_port_tokens(["http://127.0.0.1:4173/app"]) == {4173}
assert parse_port_tokens(["localhost:8080/health"]) == {8080}
assert parse_port_tokens(["[::1]:5173/app"]) == {5173}
assert parse_port_tokens(["localhost:8080#ready"]) == {8080}
assert parse_port_tokens(["[::1]:5173/app?x=1#top"]) == {5173}


def test_numeric_colon_range_still_rejected():
try:
parse_port_tokens(["8080:8090"])
except ValueError as exc:
assert "invalid" in str(exc)
else:
raise AssertionError("8080:8090 must stay rejected")


def test_cli_reexports_token_parser():
assert cli.parse_port_tokens is parse_port_tokens
assert cli.parse_port_tokens(["localhost:8080", "3000-3001"]) == {8080, 3000, 3001}


def test_cli_accepts_host_port(monkeypatch, capsys, make_row):
monkeypatch.setattr(core, "collect", lambda **kw: [make_row(local_port=8080)])
rc = cli.main(["localhost:8080", "--json"])
doc = json.loads(capsys.readouterr().out)
assert rc == 0
assert doc["query"] == [8080]
assert doc["sockets"][0]["local_port"] == 8080
Loading