put sequential by itself

.github/workflows/ci.yml

@@ -4,11 +4,12 @@ on:
   push:
     paths:
       - "**.py"
+      - .github/workflows/ci.yml
 
 jobs:
 
   linux:
-    runs-on: ubuntu-latest
+    runs-on: windows-latest
     strategy:
       matrix:
         python-version: [3.7, 3.x]
@@ -30,7 +31,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [windows-latest, macos-latest]
+        os: [ubuntu-latest, macos-latest]
     steps:
     - uses: actions/checkout@v2
     - uses: actions/setup-python@v2

src/findssh/__init__.py

@@ -1,2 +1,8 @@
+import os
+import asyncio
+
 from .coro import get_hosts
 from .base import netfromaddress, getLANip
+
+if os.name == "nt":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

src/findssh/__main__.py

@@ -14,13 +14,12 @@
 Wifi timeout should be 1 second or more
 """
 
-import os
 import asyncio
 import logging
 import ipaddress as ip
 from argparse import ArgumentParser
 
-from .base import getLANip, netfromaddress
+from .base import getLANip, netfromaddress, get_hosts_seq
 from .coro import get_hosts as coro_get_hosts
 from .threadpool import get_hosts as threadpool_get_hosts
 
@@ -52,14 +51,12 @@ def main():
     net = netfromaddress(ownip)
     print("searching", net)
 
-    if P.threadpool or isinstance(net, ip.IPv6Network):
-        hosts = threadpool_get_hosts(net, P.port, P.service, P.timeout, debug=P.verbose)
+    if P.threadpool:
+        threadpool_get_hosts(net, P.port, P.service, P.timeout)
+    elif isinstance(net, ip.IPv6Network):
+        get_hosts_seq(net, P.port, P.service, P.timeout)
     else:
-        if os.name == "nt":
-            asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
-        hosts = asyncio.run(coro_get_hosts(net, P.port, P.service, P.timeout))
-
-    return hosts
+        asyncio.run(coro_get_hosts(net, P.port, P.service, P.timeout))
 
 
 if __name__ == "__main__":

src/findssh/base.py

@@ -1,5 +1,6 @@
 import ipaddress as ip
 import socket
+import typing as T
 
 
 def getLANip() -> ip.IPv4Address:
@@ -44,3 +45,42 @@ def netfromaddress(addr: ip.IPv4Address, mask: str = "24") -> ip.IPv4Network:
     else:
         raise TypeError(addr)
     return net
+
+
+def isportopen(
+    host: ip.IPv4Address, port: int, service: str, timeout: float
+) -> T.Tuple[ip.IPv4Address, str]:
+    """
+    is a port open? Without coroutines.
+    """
+
+    h = host.exploded
+
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.settimeout(timeout)  # seconds
+        try:
+            if s.connect_ex((h, port)):
+                return None
+        except socket.gaierror:
+            return None
+        # %% service decode (optional)
+        try:
+            svc_txt = validateservice(service, h, s.recv(32))
+        except (socket.timeout, ConnectionError):
+            return None
+    if svc_txt:
+        return host, svc_txt
+    return None
+
+
+def get_hosts_seq(
+    net: ip.IPv4Network, port: int, service: str, timeout: float
+) -> T.Iterable[T.Tuple[ip.IPv4Address, str]]:
+    """
+    find hosts sequentially (no parallelism or concurrency)
+    """
+
+    for host in net.hosts():
+        res = isportopen(host, port, service, timeout)
+        if res:
+            yield res

src/findssh/tests/test_coro.py

@@ -11,7 +11,7 @@
 def test_coroutine():
     net = findssh.netfromaddress(findssh.getLANip())
     hosts = asyncio.run(findssh.get_hosts(net, PORT, SERVICE, TIMEOUT))
-    if len(hosts) > 0:
-        host = hosts[0]
-        assert isinstance(host[0], ipaddress.IPv4Address)
-        assert isinstance(host[1], str)
+    for host, svc in hosts:
+        assert isinstance(host, ipaddress.IPv4Address)
+        assert isinstance(svc, str)
+        break

src/findssh/tests/test_threadpool.py

@@ -12,8 +12,8 @@
 
 def test_threadpool():
     net = findssh.netfromaddress(findssh.getLANip())
-    host_res = findssh.threadpool.get_hosts(net, PORT, SERVICE, TIMEOUT)
-    for host, svc in host_res:
+    hosts = findssh.threadpool.get_hosts(net, PORT, SERVICE, TIMEOUT)
+    for host, svc in hosts:
         assert isinstance(host, ipaddress.IPv4Address)
         assert isinstance(svc, str)
         break

src/findssh/threadpool.py

@@ -4,55 +4,22 @@
 """
 import concurrent.futures
 import ipaddress as ip
-import typing
-import socket
-import logging
-from .base import validateservice
+import typing as T
 
-
-def isportopen(
-    host: ip.IPv4Address, port: int, service: str, timeout: float
-) -> typing.Tuple[ip.IPv4Address, str]:
-    h = host.exploded
-
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.settimeout(timeout)  # seconds
-        try:
-            if s.connect_ex((h, port)):
-                return None
-        except socket.gaierror:
-            return None
-        # %% service decode (optional)
-        try:
-            svc_txt = validateservice(service, h, s.recv(32))
-        except (socket.timeout, ConnectionError):
-            return None
-    if svc_txt:
-        return host, svc_txt
-    return None
+from .base import isportopen
 
 
 def get_hosts(
-    net: ip.IPv4Network, port: int, service: str, timeout: float, debug: bool = False
-) -> typing.Iterable[typing.Tuple[ip.IPv4Address, str]]:
+    net: ip.IPv4Network, port: int, service: str, timeout: float
+) -> T.Iterable[T.Tuple[ip.IPv4Address, str]]:
     """
     loops over hosts in network
     One thread per address.
-
-    IPv6 is not well supported, it will overwhelm RAM except by a plain for loop.
-    A different approach is needed to handle IPv6 scale, but it's fine for IPv4.
     """
 
-    if debug or isinstance(net, ip.IPv6Network):
-        for host in net.hosts():
-            logging.debug(host)
-            res = isportopen(host, port, service, timeout)
+    with concurrent.futures.ThreadPoolExecutor(max_workers=100) as exc:
+        futures = (exc.submit(isportopen, host, port, service, timeout) for host in net.hosts())
+        for future in concurrent.futures.as_completed(futures):
+            res = future.result()
             if res:
                 yield res
-    else:
-        with concurrent.futures.ThreadPoolExecutor(max_workers=100) as exc:
-            futures = (exc.submit(isportopen, host, port, service, timeout) for host in net.hosts())
-            for future in concurrent.futures.as_completed(futures):
-                res = future.result()
-                if res:
-                    yield res