Added is_host_up utility function

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## Next version
+
+### 🚀 New
+
+* Added `is_host_up` utility function.
+
+
 ## 0.5.7 - January 13, 2025
 
 ### ✨ Improved

pyproject.toml

@@ -23,7 +23,8 @@ dependencies = [
     "aiohttp>=3.11.11",
     "jinja2>=3.1.5",
     "cachetools>=5.5.0",
-    "aiocache>=0.12.3"
+    "aiocache>=0.12.3",
+    "python3-nmap>=1.9.1",
 ]
 
 [project.optional-dependencies]

src/lvmopstools/utils.py

@@ -13,7 +13,10 @@
 
 from typing import Any, Coroutine, TypeVar
 
+from nmap3 import NmapHostDiscovery
+
 from clu import AMQPClient
+from sdsstools.utils import run_in_executor
 
 
 __all__ = [
@@ -23,6 +26,7 @@
     "with_timeout",
     "is_notebook",
     "Trigger",
+    "is_host_up",
 ]
 
 
@@ -202,3 +206,36 @@ def is_set(self):
         self._check()
 
         return self._triggered
+
+
+async def is_host_up(host: str) -> bool:
+    """Returns whether a host is up.
+
+    Parameters
+    ----------
+    host
+        The host to check.
+
+    Returns
+    -------
+    is_up
+        ``True`` if the host is up, ``False`` otherwise.
+
+    """
+
+    nmap = NmapHostDiscovery()
+    result = await run_in_executor(
+        nmap.nmap_no_portscan,
+        host,
+        args="--host-timeout=1 --max-retries=2",
+    )
+
+    if (
+        host not in result
+        or not isinstance(result[host], dict)
+        or "state" not in result[host]
+        or "state" not in result[host]["state"]
+    ):
+        return False
+
+    return result[host]["state"]["state"] == "up"

tests/test_utils.py

@@ -116,3 +116,23 @@ async def test_trigger_reset():
     trigger.reset()
     trigger.set()
     assert not trigger.is_set()
+
+
+async def test_host_is_up(mocker: pytest_mock.MockerFixture):
+    mocker.patch.object(
+        lvmopstools.utils.NmapHostDiscovery,
+        "nmap_no_portscan",
+        return_value={"host1": {"state": {"state": "up"}}},
+    )
+
+    assert await lvmopstools.utils.is_host_up("host1")
+
+
+async def test_host_is_up_bad_reply(mocker: pytest_mock.MockerFixture):
+    mocker.patch.object(
+        lvmopstools.utils.NmapHostDiscovery,
+        "nmap_no_portscan",
+        return_value={"host1": None},
+    )
+
+    assert (await lvmopstools.utils.is_host_up("host1")) is False