Skip to content

Commit

Permalink
Lint with ruff and black
Browse files Browse the repository at this point in the history
  • Loading branch information
fizyk committed Oct 10, 2023
1 parent ca3a175 commit 3302529
Show file tree
Hide file tree
Showing 10 changed files with 81 additions and 53 deletions.
3 changes: 3 additions & 0 deletions newsfragments/170.misc.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
Reformatted code with ruff and black.

Ruff skips docopt and _download_ranges module for the time being.
6 changes: 3 additions & 3 deletions port_for/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,14 @@
from .api import (
available_good_ports,
available_ports,
is_available,
get_port,
good_port_ranges,
is_available,
port_is_used,
select_random,
get_port,
)
from .store import PortStore
from .exceptions import PortForException
from .store import PortStore

__all__ = (
"UNASSIGNED_RANGES",
Expand Down
45 changes: 24 additions & 21 deletions port_for/api.py
Original file line number Diff line number Diff line change
@@ -1,25 +1,24 @@
# -*- coding: utf-8 -*-
"""main port-for functionality."""
import contextlib
import socket
import errno
import random
import socket
from itertools import chain
from typing import Optional, Set, List, Tuple, Iterable, TypeVar, Type, Union
from typing import Iterable, List, Optional, Set, Tuple, Type, TypeVar, Union

from port_for import ephemeral, utils

from ._ranges import UNASSIGNED_RANGES
from .exceptions import PortForException


SYSTEM_PORT_RANGE = (0, 1024)


def select_random(
ports: Optional[Set[int]] = None,
exclude_ports: Optional[Iterable[int]] = None,
) -> int:
"""
Returns random unused port number.
"""
"""Return random unused port number."""
if ports is None:
ports = available_good_ports()

Expand All @@ -35,9 +34,7 @@ def select_random(


def is_available(port: int) -> bool:
"""
Returns if port is good to choose.
"""
"""Return if port is good to choose."""
return port in available_ports() and not port_is_used(port)


Expand All @@ -46,9 +43,12 @@ def available_ports(
high: int = 65535,
exclude_ranges: Optional[List[Tuple[int, int]]] = None,
) -> Set[int]:
"""
Returns a set of possible ports (excluding system,
ephemeral and well-known ports).
"""Return a set of possible ports.
.. note::
Excluding system, ephemeral and well-known ports.
Pass ``high`` and/or ``low`` to limit the port range.
"""
if exclude_ranges is None:
Expand All @@ -74,8 +74,8 @@ def available_ports(
def good_port_ranges(
ports: Optional[Set[int]] = None, min_range_len: int = 20, border: int = 3
) -> List[Tuple[int, int]]:
"""
Returns a list of 'good' port ranges.
"""Return a list of 'good' port ranges.
Such ranges are large and don't contain ephemeral or well-known ports.
Ranges borders are also excluded.
"""
Expand All @@ -94,14 +94,16 @@ def good_port_ranges(


def available_good_ports(min_range_len: int = 20, border: int = 3) -> Set[int]:
"""List available good ports."""
return utils.ranges_to_set(
good_port_ranges(min_range_len=min_range_len, border=border)
)


def port_is_used(port: int, host: str = "127.0.0.1") -> bool:
"""
Returns if port is used. Port is considered used if the current process
"""Return if port is used.
Port is considered used if the current process
can't bind to it or the port doesn't refuse connections.
"""
unused = _can_bind(port, host) and _refuses_connection(port, host)
Expand Down Expand Up @@ -130,7 +132,7 @@ def _refuses_connection(port: int, host: str) -> bool:


def filter_by_type(lst: Iterable, type_of: Type[T]) -> List[T]:
"""Returns a list of elements with given type."""
"""Return a list of elements with given type."""
return [e for e in lst if isinstance(e, type_of)]


Expand All @@ -152,9 +154,10 @@ def get_port(
ports: Optional[PortType],
exclude_ports: Optional[Iterable[int]] = None,
) -> Optional[int]:
"""
Retuns a random available port. If there's only one port passed
(e.g. 5000 or '5000') function does not check if port is available.
"""Retun a random available port.
If there's only one port passed (e.g. 5000 or '5000') function
does not check if port is available.
If there's -1 passed as an argument, function returns None.
:param ports:
Expand Down
4 changes: 2 additions & 2 deletions port_for/cmd.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
#!/usr/bin/env python
"""
cmd.py is a command-line utility that helps with local TCP ports management.
"""cmd.py is a command-line utility that helps with local TCP ports management.
It finds 'good' unused TCP localhost port and remembers the association.
Expand All @@ -26,6 +25,7 @@

import sys
from typing import Optional

import port_for
from port_for.docopt import docopt

Expand Down
10 changes: 4 additions & 6 deletions port_for/ephemeral.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,19 @@
# -*- coding: utf-8 -*-
"""
This module provide utilities to find ephemeral port ranges for the current OS.
"""Module provide utilities to find ephemeral port ranges for the current OS.
See http://www.ncftp.com/ncftpd/doc/misc/ephemeral_ports.html for more info
about ephemeral port ranges.
Currently only Linux and BSD (including OS X) are supported.
"""
import subprocess
from typing import List, Tuple, Dict
from typing import Dict, List, Tuple

DEFAULT_EPHEMERAL_PORT_RANGE = (32768, 65535)


def port_ranges() -> List[Tuple[int, int]]:
"""
Returns a list of ephemeral port ranges for current machine.
"""
"""Return a list of ephemeral port ranges for current machine."""
try:
return _linux_ranges()
except (OSError, IOError): # not linux, try BSD
Expand Down
6 changes: 5 additions & 1 deletion port_for/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
# -*- coding: utf-8 -*-
"""Port-for exceptions."""


class PortForException(Exception):
"""Main port-for exception class."""

pass
13 changes: 9 additions & 4 deletions port_for/store.py
Original file line number Diff line number Diff line change
@@ -1,22 +1,25 @@
# -*- coding: utf-8 -*-
"""PortStore implementation."""
import os
from configparser import ConfigParser, DEFAULTSECT
from typing import Optional, List, Tuple, Union
from configparser import DEFAULTSECT, ConfigParser
from typing import List, Optional, Tuple, Union

from .api import select_random
from .exceptions import PortForException


DEFAULT_CONFIG_PATH = "/etc/port-for.conf"


class PortStore(object):
"""PortStore binds, reads and stores bound ports in config."""

def __init__(self, config_filename: str = DEFAULT_CONFIG_PATH):
"""Initialize PortStore."""
self._config = config_filename

def bind_port(
self, app: str, port: Optional[Union[int, str]] = None
) -> int:
"""Binds port to app in the config."""
if "=" in app or ":" in app:
raise Exception('invalid app name: "%s"' % app)

Expand Down Expand Up @@ -61,11 +64,13 @@ def bind_port(
return int(requested_port)

def unbind_port(self, app: str) -> None:
"""Remove port assignement to application."""
parser = self._get_parser()
parser.remove_option(DEFAULTSECT, app)
self._save(parser)

def bound_ports(self) -> List[Tuple[str, int]]:
"""List all bound ports."""
return [
(app, int(port))
for app, port in self._get_parser().items(DEFAULTSECT)
Expand Down
10 changes: 4 additions & 6 deletions port_for/utils.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,10 @@
# -*- coding: utf-8 -*-
"""Port for utils."""
import itertools
from typing import Iterable, Iterator, Tuple, Set
from typing import Iterable, Iterator, Set, Tuple


def ranges_to_set(lst: Iterable[Tuple[int, int]]) -> Set[int]:
"""
Convert a list of ranges to a set of numbers::
"""Convert a list of ranges to a set of numbers.
>>> ranges = [(1,3), (5,6)]
>>> sorted(list(ranges_to_set(ranges)))
Expand All @@ -16,8 +15,7 @@ def ranges_to_set(lst: Iterable[Tuple[int, int]]) -> Set[int]:


def to_ranges(lst: Iterable[int]) -> Iterator[Tuple[int, int]]:
"""
Convert a list of numbers to a list of ranges::
"""Convert a list of numbers to a list of ranges.
>>> numbers = [1,2,3,5,6]
>>> list(to_ranges(numbers))
Expand Down
4 changes: 4 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,10 @@ select = [
"I", # isort
"D", # pydocstyle
]
exclude = [
"port_for/docopt.py",
"port_for/_download_ranges.py"
]


[tool.tbump]
Expand Down
33 changes: 23 additions & 10 deletions tests/test_cases.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
# -*- coding: utf-8 -*-
import unittest
"""Test cases."""
import os
import socket
import tempfile
import unittest
from typing import List, Set, Tuple, Union
from unittest import mock
import socket
import os
from typing import Union, List, Set, Tuple

import pytest

Expand All @@ -14,11 +14,13 @@


def test_common_ports() -> None:
"""Check common ports (not available)."""
assert not port_for.is_available(80)
assert not port_for.is_available(11211)


def test_good_port_ranges() -> None:
"""Select good ranges of ports aout of provided."""
ranges = [
(10, 15), # too short
(100, 200), # good
Expand All @@ -33,16 +35,18 @@ def test_good_port_ranges() -> None:


def test_something_works() -> None:
"""Test default behaviour of good_port_ranges and available_good_ports."""
assert len(port_for.good_port_ranges()) > 10
assert len(port_for.available_good_ports()) > 1000


def test_binding() -> None:
# low ports are not available
"""Low ports are not available."""
assert port_for.port_is_used(10)


def test_binding_high() -> None:
"""Test ports that are not used."""
s = socket.socket()
s.bind(("", 0))
port = s.getsockname()[1]
Expand Down Expand Up @@ -115,13 +119,17 @@ def test_port_mix() -> None:


class SelectPortTest(unittest.TestCase):
"""Port selecting tests."""

@mock.patch("port_for.api.port_is_used")
def test_all_used(self, port_is_used: mock.MagicMock) -> None:
"""Check behaviour if there are no ports to use."""
port_is_used.return_value = True
self.assertRaises(port_for.PortForException, port_for.select_random)

@mock.patch("port_for.api.port_is_used")
def test_random_port(self, port_is_used: mock.MagicMock) -> None:
"""Test random ports."""
ports = set([1, 2, 3])
used = {1: True, 2: False, 3: True}
port_is_used.side_effect = lambda port: used[port]
Expand All @@ -131,14 +139,19 @@ def test_random_port(self, port_is_used: mock.MagicMock) -> None:


class StoreTest(unittest.TestCase):
"""Port Store test suite."""

def setUp(self) -> None:
"""Set up tests."""
fd, self.fname = tempfile.mkstemp()
self.store = port_for.PortStore(self.fname)

def tearDown(self) -> None:
"""Tear down tests."""
os.remove(self.fname)

def test_store(self) -> None:
"""Test port store."""
assert self.store.bound_ports() == []

port = self.store.bind_port("foo")
Expand All @@ -156,14 +169,14 @@ def test_store(self) -> None:
self.assertEqual(self.store.bound_ports(), [("foo", port)])

def test_rebind(self) -> None:
# try to rebind an used port for an another app
"""Try to rebind an used port for an another app."""
port = self.store.bind_port("foo")
self.assertRaises(
port_for.PortForException, self.store.bind_port, "baz", port
)

def test_change_port(self) -> None:
# changing app ports is not supported.
"""Changing app ports is not supported."""
port = self.store.bind_port("foo")
another_port = port_for.select_random()
assert port != another_port
Expand All @@ -172,13 +185,13 @@ def test_change_port(self) -> None:
)

def test_bind_unavailable(self) -> None:
# it is possible to explicitly bind currently unavailable port
"""It is possible to explicitly bind currently unavailable port."""
port = self.store.bind_port("foo", 80)
self.assertEqual(port, 80)
self.assertEqual(self.store.bound_ports(), [("foo", 80)])

def test_bind_non_auto(self) -> None:
# it is possible to pass a port
"""It is possible to pass a port."""
port = port_for.select_random()
res_port = self.store.bind_port("foo", port)
self.assertEqual(res_port, port)

0 comments on commit 3302529

Please sign in to comment.