feat: add basefiles

tests/test_base.py

@@ -0,0 +1,55 @@
+import ipaddress
+import random
+import subprocess
+
+from ward import test, fixture
+
+from ports.commands.base import BaseCommand
+
+
+def _get_random_number_between_1_and_254() -> int:
+    return random.randint(1, 254)
+
+
+@fixture
+def dev_null_pid():
+    p = subprocess.Popen(["tail", "-f", "/dev/null"])
+    yield p.pid
+    p.kill()
+
+
+@test(
+    "`_get_name_of_pid` of non-existing process should return empty string instead of "
+    "raising `NoSuchProcess`"
+)
+def _():
+    assert not BaseCommand._get_name_of_pid(10000000)
+
+
+@test("`_get_name_of_pid` should return `tail` ")
+def _(dnp: int = dev_null_pid):
+    assert BaseCommand._get_name_of_pid(dnp) == "tail"
+
+
+@test("`_normalize_interface` of `::`, `::1` and `0.0.0.0` should return `localhost`")
+def _():
+    assert (
+        BaseCommand._normalize_interface("::")
+        == BaseCommand._normalize_interface("::1")
+        == BaseCommand._normalize_interface("0.0.0.0")
+        == BaseCommand._normalize_interface("localhost")
+        == "localhost"
+    )
+
+
+@test("`_normalize_interface` of any `127.0.0.0/8` ip should return `localhost`")
+def _():
+    # iterating over the real /8 range would take a bit too much
+    # so let's randomize 32 IPs
+
+    random_second_block = _get_random_number_between_1_and_254()
+    random_third_block = _get_random_number_between_1_and_254()
+    for ip in ipaddress.IPv4Network(
+        f"127.{random_second_block}.{random_third_block}.0/24"
+    ):
+        assert BaseCommand._normalize_interface(str(ip)) == "localhost"

tests/test_open.py

@@ -0,0 +1,60 @@
+import subprocess
+from time import sleep
+
+from ward import test, fixture, raises
+
+from ports.commands.base import PortInfo
+from ports.commands.open import OpenCommand
+
+
+@fixture
+def closed_portinfs_8000_to_8020():
+    """
+    can't yield this because `_fill_portinfos` consumes it for every port
+    """
+    return [PortInfo(is_open=False, port=i) for i in range(8000, 8021)]
+
+
+@fixture
+def http_server_pid():
+    p = subprocess.Popen(["python3", "-m", "http.server", "14520"])
+    sleep(1)  # the server may take a bit to be up-n-running
+    yield p.pid
+    p.kill()
+
+
+@fixture
+def open_command_8000_to_8021():
+    yield OpenCommand(start_port=7990, end_port=8025)
+
+
+@test("initializing with start port 80 and end port 79 should raise `ValueError`")
+def _():
+    with raises(ValueError):
+        OpenCommand(start_port=80, end_port=79)
+
+
+@test(
+    "`closed_portinfos` returns a list with the port 14520 closed by the "
+    "`http.server` process"
+)
+def _(hpd: int = http_server_pid):
+    closed_portinfos = OpenCommand().closed_portinfos
+    portinfo_14520 = [
+        portinfo for portinfo in closed_portinfos if portinfo.port == 14520
+    ][0]
+    assert portinfo_14520
+    assert portinfo_14520.pid == hpd
+    assert portinfo_14520.interface == "localhost"
+    assert portinfo_14520.process_name == "python3"
+
+
+@test("`_fill_portinfos` of [7990..8025] should return [7990..7999],[8022...8025]")
+def _(
+    cp: list[PortInfo] = closed_portinfs_8000_to_8020,
+    op: OpenCommand = open_command_8000_to_8021,
+):
+    filled_portinfos = op._fill_portinfos(cp)
+    assert len(filled_portinfos) == 8025 - 8000 - 20 + 8000 - 7990
+    for portinfo in filled_portinfos:
+        assert portinfo.port in range(7990, 8000) or portinfo.port in range(8021, 8026)