diff --git a/newsfragments/170.misc.rst b/newsfragments/170.misc.rst new file mode 100644 index 0000000..a5fbb33 --- /dev/null +++ b/newsfragments/170.misc.rst @@ -0,0 +1,3 @@ +Reformatted code with ruff and black. + +Ruff skips docopt and _download_ranges module for the time being. \ No newline at end of file diff --git a/port_for/__init__.py b/port_for/__init__.py index 15c664d..6d1f935 100644 --- a/port_for/__init__.py +++ b/port_for/__init__.py @@ -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", diff --git a/port_for/api.py b/port_for/api.py index 4ecf9f8..c4e15e6 100644 --- a/port_for/api.py +++ b/port_for/api.py @@ -1,15 +1,16 @@ -# -*- 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) @@ -17,9 +18,7 @@ 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() @@ -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) @@ -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: @@ -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. """ @@ -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) @@ -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)] @@ -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: diff --git a/port_for/cmd.py b/port_for/cmd.py index be28902..ba9add7 100644 --- a/port_for/cmd.py +++ b/port_for/cmd.py @@ -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. @@ -26,6 +25,7 @@ import sys from typing import Optional + import port_for from port_for.docopt import docopt diff --git a/port_for/ephemeral.py b/port_for/ephemeral.py index a3c03ec..41a4685 100644 --- a/port_for/ephemeral.py +++ b/port_for/ephemeral.py @@ -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 diff --git a/port_for/exceptions.py b/port_for/exceptions.py index 256a255..dc03591 100644 --- a/port_for/exceptions.py +++ b/port_for/exceptions.py @@ -1,3 +1,7 @@ -# -*- coding: utf-8 -*- +"""Port-for exceptions.""" + + class PortForException(Exception): + """Main port-for exception class.""" + pass diff --git a/port_for/store.py b/port_for/store.py index a07b737..685753c 100644 --- a/port_for/store.py +++ b/port_for/store.py @@ -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) @@ -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) diff --git a/port_for/utils.py b/port_for/utils.py index 361d5c0..997f63a 100644 --- a/port_for/utils.py +++ b/port_for/utils.py @@ -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))) @@ -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)) diff --git a/pyproject.toml b/pyproject.toml index a0fdd0c..3945a52 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -98,6 +98,10 @@ select = [ "I", # isort "D", # pydocstyle ] +exclude = [ + "port_for/docopt.py", + "port_for/_download_ranges.py" +] [tool.tbump] diff --git a/tests/test_cases.py b/tests/test_cases.py index 02772a8..8f5f3f1 100644 --- a/tests/test_cases.py +++ b/tests/test_cases.py @@ -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 @@ -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 @@ -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] @@ -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] @@ -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") @@ -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 @@ -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)