From 92240acb5cc600eec60624ece9ed4b9ec43b386f Mon Sep 17 00:00:00 2001
From: Daniel Baumann <daniel.baumann@progress-linux.org>
Date: Thu, 23 May 2024 07:06:46 +0200
Subject: Adding upstream version 0.15.0.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
---
 asynceapi/aio_portcheck.py | 58 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 58 insertions(+)
 create mode 100644 asynceapi/aio_portcheck.py

(limited to 'asynceapi/aio_portcheck.py')

diff --git a/asynceapi/aio_portcheck.py b/asynceapi/aio_portcheck.py
new file mode 100644
index 0000000..79f4562
--- /dev/null
+++ b/asynceapi/aio_portcheck.py
@@ -0,0 +1,58 @@
+# Copyright (c) 2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+# Initially written by Jeremy Schulman at https://github.com/jeremyschulman/aio-eapi
+"""Utility function to check if a port is open."""
+# -----------------------------------------------------------------------------
+# System Imports
+# -----------------------------------------------------------------------------
+
+from __future__ import annotations
+
+import asyncio
+import socket
+from typing import TYPE_CHECKING
+
+# -----------------------------------------------------------------------------
+# Public Imports
+# -----------------------------------------------------------------------------
+
+if TYPE_CHECKING:
+    from httpx import URL
+
+# -----------------------------------------------------------------------------
+# Exports
+# -----------------------------------------------------------------------------
+
+__all__ = ["port_check_url"]
+
+# -----------------------------------------------------------------------------
+#
+#                                 CODE BEGINS
+#
+# -----------------------------------------------------------------------------
+
+
+async def port_check_url(url: URL, timeout: int = 5) -> bool:
+    """
+    Open the port designated by the URL given the timeout in seconds.
+
+    If the port is available then return True; False otherwise.
+
+    Parameters
+    ----------
+        url: The URL that provides the target system
+        timeout: Time to await for the port to open in seconds
+    """
+    port = url.port or socket.getservbyname(url.scheme)
+
+    try:
+        wr: asyncio.StreamWriter
+        _, wr = await asyncio.wait_for(asyncio.open_connection(host=url.host, port=port), timeout=timeout)
+
+        # MUST close if opened!
+        wr.close()
+
+    except TimeoutError:
+        return False
+    return True
-- 
cgit v1.2.3