plesty.server.model.ports

TCP port rules for device servers on one bench.

Precedence for the probed port: an explicit port key, then --tcp-port/-tp in args, then a *_TCP_PORT value in env, then plesty-lib’s default 5555.

Attributes

`DEFAULT_TCP_PORT`
`AGENT_PORT`
`DEVICE_PORT_BASE`
`_PORT_FLAGS`

Functions

`port_from`(→ int)	Resolve the TCP port a server binds from its args and environment.
`declares_port`(→ bool)	Whether args or env name the port themselves.
`port_is_free`(→ bool)	Whether port can actually be bound right now.
`port_listening`(→ bool)	Whether something accepts TCP connections on port right now.
`allocate_port`(→ int)	The lowest free port from base up, skipping taken and the agent's.
`_int_port`(→ int)

Module Contents

plesty.server.model.ports.DEFAULT_TCP_PORT = 5555

plesty.server.model.ports.AGENT_PORT = 5550

plesty.server.model.ports.DEVICE_PORT_BASE = 5551

plesty.server.model.ports._PORT_FLAGS = ('--tcp-port', '-tp')

plesty.server.model.ports.port_from(args: tuple[str, Ellipsis], env: collections.abc.Mapping[str, str]) → int

Resolve the TCP port a server binds from its args and environment.

Parameters:

args (tuple[str, Ellipsis]) – Command-line arguments passed to python -m <module>.
env (collections.abc.Mapping[str, str]) – Environment values the fleet sets for the process.

Returns:

The port to probe, falling back to DEFAULT_TCP_PORT.

Return type:

int

plesty.server.model.ports.declares_port(args: tuple[str, Ellipsis], env: collections.abc.Mapping[str, str]) → bool

Whether args or env name the port themselves.

A declaration that names its own port is left alone — the bench allocates only for devices that would otherwise take the default and collide.

Parameters:

args (tuple[str, Ellipsis])
env (collections.abc.Mapping[str, str])

Return type:

bool

plesty.server.model.ports.port_is_free(port: int, host: str = '127.0.0.1') → bool

Whether port can actually be bound right now.

The fleet file and the process records say what this bench believes it is using. A server that outlived its record, or any other program on the machine, believes nothing — and the socket is the only thing that knows.

The socket option differs because the platforms disagree about what SO_REUSEADDR means. On POSIX it says “a port left in TIME_WAIT is still free”, which is the question. On Windows it says “bind even if another socket is listening here” — so the probe succeeded against a live server and every busy port reported free, which silently disabled everything built on this: the port a start moves away from, and the port allocate_port() hands out. SO_EXCLUSIVEADDRUSE is the Windows way to ask the question POSIX answers with SO_REUSEADDR.

Parameters:

port (int)
host (str)

Return type:

bool

plesty.server.model.ports.port_listening(port: int, host: str = '127.0.0.1', timeout: float = 0.3) → bool

Whether something accepts TCP connections on port right now.

The complement of port_is_free() for a port a server holds: a device server binds tcp://*, and on some hosts a specific-address bind still succeeds beside a wildcard listener — connecting is the one test that answers the question everywhere.

Parameters:

port (int)
host (str)
timeout (float)

Return type:

bool

plesty.server.model.ports.allocate_port(taken: collections.abc.Iterable[int], base: int = DEVICE_PORT_BASE, is_free: collections.abc.Callable[[int], bool] = port_is_free) → int

The lowest free port from base up, skipping taken and the agent’s.

Parameters:

taken (collections.abc.Iterable[int]) – Ports already spoken for on this bench.
base (int) – Where counting starts.
is_free (collections.abc.Callable[[int], bool]) – Asks the host whether a port can be bound; injected by tests.

Returns:

A port no other device on the bench holds and nothing on the host is listening on.

Raises:

FleetError – When counting runs past the end of the port range.

Return type:

int

plesty.server.model.ports._int_port(value: str, where: str) → int

Parameters:

value (str)
where (str)

Return type:

int