2.17.0 - Added net.get_host + AsyncIO versions of various net module functions

Someguy123 · Someguy123 · commit 804965e49e63 · 2020-07-20T05:22:08.000+01:00
**`privex.helpers.net` module**

 - Refactored `resolve_ips` to enable re-use of validation and other things between it and the AsyncIO version

 - Added `resolve_ips_async` - which is simply an AsyncIO version of `resolve_ips`
 - Added `resolve_ip_async` - again, an AsyncIO version of `resolve_ip`
 - Added `resolve_ips_multi_async` (AsyncIO `resolve_ips_multi`)
 - Added `get_rdns_async` (AsyncIO `get_rdns`)
 - Added new `check_host` function, which checks if a given service (port) on a host is up and working
 - Added `check_host_async` - same as `check_host` but uses AsyncIO
diff --git a/docs/source/helpers/net/privex.helpers.net.check_host.rst b/docs/source/helpers/net/privex.helpers.net.check_host.rst
@@ -0,0 +1,6 @@
+check\_host
+===========
+
+.. currentmodule:: privex.helpers.net
+
+.. autofunction:: check_host
diff --git a/docs/source/helpers/net/privex.helpers.net.check_host_async.rst b/docs/source/helpers/net/privex.helpers.net.check_host_async.rst
@@ -0,0 +1,6 @@
+check\_host\_async
+==================
+
+.. currentmodule:: privex.helpers.net
+
+.. autofunction:: check_host_async
diff --git a/docs/source/helpers/net/privex.helpers.net.get_rdns_async.rst b/docs/source/helpers/net/privex.helpers.net.get_rdns_async.rst
@@ -0,0 +1,6 @@
+get\_rdns\_async
+================
+
+.. currentmodule:: privex.helpers.net
+
+.. autofunction:: get_rdns_async
diff --git a/docs/source/helpers/net/privex.helpers.net.resolve_ip_async.rst b/docs/source/helpers/net/privex.helpers.net.resolve_ip_async.rst
@@ -0,0 +1,6 @@
+resolve\_ip\_async
+==================
+
+.. currentmodule:: privex.helpers.net
+
+.. autofunction:: resolve_ip_async
diff --git a/docs/source/helpers/net/privex.helpers.net.resolve_ips_async.rst b/docs/source/helpers/net/privex.helpers.net.resolve_ips_async.rst
@@ -0,0 +1,6 @@
+resolve\_ips\_async
+===================
+
+.. currentmodule:: privex.helpers.net
+
+.. autofunction:: resolve_ips_async
diff --git a/docs/source/helpers/net/privex.helpers.net.resolve_ips_multi_async.rst b/docs/source/helpers/net/privex.helpers.net.resolve_ips_multi_async.rst
@@ -0,0 +1,6 @@
+resolve\_ips\_multi\_async
+==========================
+
+.. currentmodule:: privex.helpers.net
+
+.. autofunction:: resolve_ips_multi_async
diff --git a/docs/source/helpers/privex.helpers.net.rst b/docs/source/helpers/privex.helpers.net.rst
@@ -15,11 +15,18 @@ privex.helpers.net
       :toctree: net
    
       asn_to_name
+      check_host
+      check_host_async
+
       get_rdns
+      get_rdns_async
       get_rdns_multi
       resolve_ip
+      resolve_ip_async
       resolve_ips
+      resolve_ips_async
       resolve_ips_multi
+      resolve_ips_multi_async
       ip4_to_rdns
       ip6_to_rdns
       ip_is_v4
diff --git a/privex/helpers/__init__.py b/privex/helpers/__init__.py
@@ -148,7 +148,7 @@ def _setup_logging(level=logging.WARNING):
 log = _setup_logging()
 name = 'helpers'
 
-VERSION = '2.16.0'
+VERSION = '2.17.0'
 
 
 
diff --git a/privex/helpers/net.py b/privex/helpers/net.py
diff --git a/tests/test_net.py b/tests/test_net.py
@@ -1,7 +1,10 @@
 """
 Test cases related to :py:mod:`privex.helpers.net` or generally network related functions such as :py:func:`.ping`
 """
+import socket
 import warnings
+
+from privex.helpers import loop_run
 from tests import PrivexBaseCase
 from privex import helpers
 
@@ -124,6 +127,19 @@ def test_get_rdns_multi_invalid(self):
         self.assertIsNone(hosts['fe80::5123'])
         self.assertEqual(hosts['2a07:e00::333'], 'se.dns.privex.io')
 
+    def test_check_host(self):
+        self.assertTrue(helpers.check_host('hiveseed-se.privex.io', 2001))
+        self.assertFalse(helpers.check_host('hiveseed-se.privex.io', 9991))
+
+    def test_check_host_send(self):
+        http_req = b"GET / HTTP/1.1\n\n"
+        self.assertTrue(helpers.check_host('files.privex.io', 80, send=http_req))
+        self.assertFalse(helpers.check_host('files.privex.io', 9991))
+
+    def test_check_host_throw(self):
+        with self.assertRaises(ConnectionRefusedError):
+            helpers.check_host('files.privex.io', 9991, throw=True)
+
 
 class TestNetResolveIP(PrivexBaseCase):
     """
@@ -246,3 +262,170 @@ def test_resolve_ips_multi_v6(self):
 
         self.assertIn('2a07:e00::333', ips['2a07:e00::333'])
         self.assertIsNone(ips['8.8.4.4'])
+
+
+class TestAsyncResolveIP(PrivexBaseCase):
+    # --- privex.helpers.net.resolve_ips ---
+    def test_resolve_ips_ipv4_addr(self):
+        """Test :func:`.resolve_ips` returns the same IPv4 address passed to it"""
+        ips = loop_run(helpers.resolve_ips_async('185.130.44.5'))
+        self.assertEqual(len(ips), 1)
+        self.assertEqual(ips[0], '185.130.44.5')
+    
+    def test_resolve_ips_ipv6_addr(self):
+        """Test :func:`.resolve_ips` returns the same IPv6 address passed to it"""
+        ips = loop_run(helpers.resolve_ips_async('2a07:e00::333'))
+        self.assertEqual(len(ips), 1)
+        self.assertEqual(ips[0], '2a07:e00::333')
+    
+    def test_resolve_ips_ipv4_addr_invalid(self):
+        """Test :func:`.resolve_ips` raises :class:`.AttributeError` when ``version`` is v4 but an IPv6 address was passed"""
+        with self.assertRaises(AttributeError):
+            loop_run(helpers.resolve_ips_async('2a07:e00::333', 'v4'))
+    
+    def test_resolve_ips_ipv6_addr_invalid(self):
+        """Test :func:`.resolve_ips` raises :class:`.AttributeError` when ``version`` is v6 but an IPv4 address was passed"""
+        with self.assertRaises(AttributeError):
+            loop_run(helpers.resolve_ips_async('185.130.44.5', 'v6'))
+    
+    def test_resolve_ips_hiveseed(self):
+        """Test :func:`.resolve_ips` returns expected v4 + v6 for ``hiveseed-fin.privex.io``"""
+        ips = helpers.resolve_ips('hiveseed-fin.privex.io')
+        self.assertEqual(len(ips), 2)
+        self.assertIn('2a01:4f9:2a:3d4::2', ips)
+        self.assertIn('95.216.3.171', ips)
+    
+    def test_resolve_ips_hiveseed_v4(self):
+        """Test :func:`.resolve_ips` returns only v4 for ``hiveseed-fin.privex.io`` when version is set to v4"""
+        ips = loop_run(helpers.resolve_ips_async('hiveseed-fin.privex.io', 'v4'))
+        self.assertEqual(len(ips), 1)
+        self.assertEqual(ips[0], '95.216.3.171')
+    
+    def test_resolve_ips_hiveseed_v6(self):
+        """Test :func:`.resolve_ips` returns only v6 for ``hiveseed-fin.privex.io`` when version is set to v6"""
+        ips = helpers.resolve_ips('hiveseed-fin.privex.io', 'v6')
+        self.assertEqual(len(ips), 1)
+        self.assertEqual(ips[0], '2a01:4f9:2a:3d4::2')
+    
+    def test_resolve_ips_v4_convert_false(self):
+        """Test :func:`.resolve_ips` returns an empty list for ``microsoft.com`` when v6 requested without v4_convert"""
+        ips = loop_run(helpers.resolve_ips_async('microsoft.com', 'v6', v4_convert=False))
+        self.assertEqual(len(ips), 0)
+    
+    def test_resolve_ips_v4_convert(self):
+        """Test :func:`.resolve_ips` returns IPv6-wrapped IPv4 addresses for ``microsoft.com`` when v4_convert is enabled + v6 version"""
+        ips = loop_run(helpers.resolve_ips_async('microsoft.com', 'v6', v4_convert=True))
+        if ips is None or len(ips) == 0:
+            return pytest.skip(
+                f"Skipping test TestNetResolveIP.test_resolve_ips_v4_convert as v6-wrapped IPv4 addresses "
+                f"aren't supported on this platform."
+            )
+        self.assertTrue(ips[0].startswith('::ffff:'))
+    
+    # --- privex.helpers.net.resolve_ip ---
+    
+    def test_resolve_ip_v4_convert(self):
+        """Test :func:`.resolve_ip` returns an IPv6-wrapped IPv4 address for ``microsoft.com`` when v4_convert is enabled + v6 version"""
+        ip = loop_run(helpers.resolve_ip_async('microsoft.com', 'v6', v4_convert=True))
+        if ip is None:
+            return pytest.skip(
+                f"Skipping test TestNetResolveIP.test_resolve_ip_v4_convert as v6-wrapped IPv4 addresses "
+                f"aren't supported on this platform."
+            )
+        self.assertTrue(ip.startswith('::ffff:'))
+    
+    def test_resolve_ip_hiveseed(self):
+        """Test :func:`.resolve_ip` returns expected either correct v4 or v6 for ``hiveseed-fin.privex.io``"""
+        self.assertIn(loop_run(helpers.resolve_ip_async('hiveseed-fin.privex.io')), ['95.216.3.171', '2a01:4f9:2a:3d4::2'])
+    
+    def test_resolve_ip_hiveseed_v4(self):
+        """Test :func:`.resolve_ip` returns only v4 for ``hiveseed-fin.privex.io`` when version is v4"""
+        self.assertEqual(loop_run(helpers.resolve_ip_async('hiveseed-fin.privex.io', 'v4')), '95.216.3.171')
+    
+    def test_resolve_ip_hiveseed_v6(self):
+        """Test :func:`.resolve_ip` returns only v6 for ``hiveseed-fin.privex.io`` when version is v6"""
+        self.assertEqual(loop_run(helpers.resolve_ip_async('hiveseed-fin.privex.io', 'v6')), '2a01:4f9:2a:3d4::2')
+    
+    @staticmethod
+    async def _resolve_multi_async(*addr, version='any', v4_convert=False):
+        res = []
+        async for x in helpers.resolve_ips_multi_async(*addr, version=version, v4_convert=v4_convert):
+            res.append(x)
+        return res
+
+    def _resolve_multi(self, *addr, version='any', v4_convert=False):
+        return loop_run(self._resolve_multi_async(*addr, version=version, v4_convert=v4_convert))
+
+    # --- privex.helpers.net.resolve_ips_multi ---
+    def test_resolve_ips_multi_any(self):
+        """Test :func:`.resolve_ips_multi` with 2 domains and an IPv4 address"""
+        ips = dict(self._resolve_multi('hiveseed-fin.privex.io', 'privex.io', '8.8.4.4'))
+        self.assertIn('2a01:4f9:2a:3d4::2', ips['hiveseed-fin.privex.io'])
+        self.assertIn('95.216.3.171', ips['hiveseed-fin.privex.io'])
+        
+        self.assertIn('2a07:e00::abc', ips['privex.io'])
+        self.assertIn('185.130.44.10', ips['privex.io'])
+        
+        self.assertIn('8.8.4.4', ips['8.8.4.4'])
+    
+    def test_resolve_ips_multi_v4(self):
+        """Test :func:`.resolve_ips_multi` with 2 domains, an IPv4 address, and an IPv6 address with version ``v4``"""
+        ips = dict(self._resolve_multi('hiveseed-fin.privex.io', 'privex.io', '8.8.4.4', '2a07:e00::333', version='v4'))
+        self.assertNotIn('2a01:4f9:2a:3d4::2', ips['hiveseed-fin.privex.io'])
+        self.assertIn('95.216.3.171', ips['hiveseed-fin.privex.io'])
+        
+        self.assertNotIn('2a07:e00::abc', ips['privex.io'])
+        self.assertIn('185.130.44.10', ips['privex.io'])
+        
+        self.assertIn('8.8.4.4', ips['8.8.4.4'])
+        self.assertIsNone(ips['2a07:e00::333'])
+    
+    def test_resolve_ips_multi_v6(self):
+        """Test :func:`.resolve_ips_multi` with 2 domains, an IPv4 address, and an IPv6 address with version ``v6``"""
+        ips = dict(self._resolve_multi('hiveseed-fin.privex.io', 'privex.io', '8.8.4.4', '2a07:e00::333', version='v6'))
+        self.assertIn('2a01:4f9:2a:3d4::2', ips['hiveseed-fin.privex.io'])
+        self.assertNotIn('95.216.3.171', ips['hiveseed-fin.privex.io'])
+        
+        self.assertIn('2a07:e00::abc', ips['privex.io'])
+        self.assertNotIn('185.130.44.10', ips['privex.io'])
+        
+        self.assertIn('2a07:e00::333', ips['2a07:e00::333'])
+        self.assertIsNone(ips['8.8.4.4'])
+
+
+class TestAsyncNet(PrivexBaseCase):
+    def test_get_rdns_privex_ns1_ip(self):
+        """Test resolving IPv4 and IPv6 addresses into ns1.privex.io"""
+        self.assertEqual(loop_run(helpers.get_rdns_async('2a07:e00::100')), 'ns1.privex.io')
+        self.assertEqual(loop_run(helpers.get_rdns_async('185.130.44.3')), 'ns1.privex.io')
+    
+    def test_get_rdns_privex_ns1_host(self):
+        """Test resolving rDNS for the domains ``steemseed-fin.privex.io`` and ``ns1.privex.io``"""
+        self.assertEqual(loop_run(helpers.get_rdns_async('ns1.privex.io')), 'ns1.privex.io')
+        self.assertEqual(loop_run(helpers.get_rdns_async('steemseed-fin.privex.io')), 'hiveseed-fin.privex.io')
+    
+    def test_get_rdns_invalid_domain(self):
+        """Test :func:`.get_rdns` raises :class:`.InvalidHost` when given a non-existent domain"""
+        with self.assertRaises(helpers.InvalidHost):
+            loop_run(helpers.get_rdns_async('non-existent.domain.example'))
+    
+    def test_get_rdns_no_rdns_records(self):
+        """Test :func:`.get_rdns` raises :class:`.ReverseDNSNotFound` when given a valid IP that has no rDNS records"""
+        with self.assertRaises(helpers.ReverseDNSNotFound):
+            loop_run(helpers.get_rdns_async('192.168.5.1'))
+    
+    def test_check_host_async(self):
+        self.assertTrue(loop_run(helpers.check_host_async('hiveseed-se.privex.io', 2001)))
+        self.assertFalse(loop_run(helpers.check_host_async('hiveseed-se.privex.io', 9991)))
+
+    def test_check_host_async_send(self):
+        http_req = b"GET / HTTP/1.1\n\n"
+        self.assertTrue(loop_run(helpers.check_host_async('files.privex.io', 80, send=http_req)))
+        # self.assertTrue(loop_run(helpers.check_host_async('files.privex.io', 443)))
+        self.assertFalse(loop_run(helpers.check_host_async('files.privex.io', 9991)))
+
+    def test_check_host_async_throw(self):
+        with self.assertRaises(ConnectionRefusedError):
+            loop_run(helpers.check_host_async('files.privex.io', 9991, throw=True))
+
+
