Merge pull request #1 from bt-cryptomancer/ping

Ping function

Author: Someguy123

docs/source/helpers/net/privex.helpers.net.ping.rst

@@ -0,0 +1,6 @@
+ping
+====
+
+.. currentmodule:: privex.helpers.net
+
+.. autofunction:: ping

docs/source/helpers/privex.helpers.net.rst

@@ -16,7 +16,7 @@ privex.helpers.net
       ip_is_v4
       ip_is_v6
       ip_to_rdns
-   
+      ping 
 
 
 
@@ -25,4 +25,4 @@ privex.helpers.net
 
 
 
-   
+   

docs/source/helpers/tests/tests.TestGeneral.rst

@@ -81,6 +81,7 @@ TestGeneral
       ~TestGeneral.test_async_decorator
       ~TestGeneral.test_async_decorator_return
       ~TestGeneral.test_chunked
+      ~TestGeneral.test_ping
       ~TestGeneral.test_retry_on_err
       ~TestGeneral.test_retry_on_err_return
       ~TestGeneral.test_run_sync
@@ -102,4 +103,4 @@ TestGeneral
       ~TestGeneral.maxDiff
       ~TestGeneral.truthy
 
-   
+   

privex/helpers/__init__.py

@@ -64,7 +64,7 @@ def _setup_logging(level=logging.WARNING):
 log = _setup_logging()
 name = 'helpers'
 
-VERSION = '1.3.3'
+VERSION = '1.3.4'
 
 
 class ImproperlyConfigured(Exception):

privex/helpers/net.py

@@ -38,6 +38,7 @@
 
 """
 import logging
+import subprocess
 from privex.helpers.exceptions import BoundaryException
 from ipaddress import ip_address, IPv4Address, IPv6Address
 from typing import Union
@@ -212,3 +213,18 @@ def ip_is_v6(ip: str) -> bool:
     :return bool: True if IPv6, False if not (i.e. probably IPv4)
     """
     return type(ip_address(ip)) == IPv6Address
+
+def ping(ip: str, timeout: int = 30) -> bool:
+    """
+    Sends a ping to a given IP
+
+    :param str ip: An IP address as a string, e.g. 192.168.1.1
+    :param int timeout: Number of seconds to wait for a response from the ping before timing out
+    :raises ValueError: When the given IP address ``ip`` is invalid or ``timeout`` < 1
+    :return bool: True if ping got a response from the given IP, False if not
+    """
+    ip_obj = ip_address(ip)   # verify IP is valid (this will throw if it isn't)
+    if timeout < 1:
+        raise ValueError('timeout value cannot be less than 1 second')
+    with subprocess.Popen(["/bin/ping", "-c1", "-w{}".format(timeout), ip], stdout=subprocess.PIPE) as proc:
+        return 'bytes from {}'.format(ip) in proc.stdout.read().decode('utf-8')

tests.py

@@ -82,7 +82,7 @@
 
 from privex import helpers
 from privex.loghelper import LogHelper
-from privex.helpers import ip_to_rdns, BoundaryException, plugin, r_cache, random_str, CacheAdapter, env_bool
+from privex.helpers import ping, ip_to_rdns, BoundaryException, plugin, r_cache, random_str, CacheAdapter, env_bool
 
 if env_bool('DEBUG', False) is True:
     LogHelper('privex.helpers', level=logging.DEBUG).add_console_handler(logging.DEBUG)
@@ -426,7 +426,7 @@ class TestCacheDecoratorRedis(TestCacheDecoratorMemory):
     @classmethod
     def setUpClass(cls):
         if not plugin.HAS_REDIS:
-            print('The package "redis" is not installed, skipping Redis dependent tests.')
+            print(f'The package "redis" is not installed, skipping Redis dependent tests ({cls.__name__}).')
             return cls.tearDownClass()
         helpers.cache.adapter_set(helpers.RedisCache())
 
@@ -437,6 +437,17 @@ class TestGeneral(PrivexBaseCase):
     def setUp(self):
         self.tries = 0
 
+    def test_ping(self):
+        """Test success & failure cases for ping function, as well as input validation"""
+        with self.assertRaises(ValueError):
+            ping('127.0.0.1', -1)
+        with self.assertRaises(ValueError):
+            ping('127.0.0.1', 0)
+        with self.assertRaises(ValueError):
+            ping('notavalidip', 1)
+        self.assertTrue(ping('127.0.0.1', 3))
+        self.assertFalse(ping('192.0.2.0', 3))
+
     def test_chunked(self):
         """Create a 20 element long list, split it into 4 chunks, and verify the chunks are correctly made"""
         x = list(range(0, 20))
@@ -602,6 +613,9 @@ class TestRedisCache(TestMemoryCache):
     @classmethod
     def setUpClass(cls):
         """Set the current cache adapter to an instance of RedisCache() and make it available through ``self.cache``"""
+        if not plugin.HAS_REDIS:
+            print(f'The package "redis" is not installed, skipping Redis dependent tests ({cls.__name__}).')
+            return cls.tearDownClass()
         helpers.cache.adapter_set(helpers.RedisCache())
         cls.cache = helpers.cache
 
@@ -642,4 +656,4 @@ def setUpClass(cls):
 HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION 
 OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE 
 SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
-"""
+"""