diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-23 05:06:46 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-23 05:06:46 +0000 |
commit | 92240acb5cc600eec60624ece9ed4b9ec43b386f (patch) | |
tree | 13328d0de4f37030731a96e25749563742fce0cb /asynceapi/aio_portcheck.py | |
parent | Adding upstream version 0.14.0. (diff) | |
download | anta-92240acb5cc600eec60624ece9ed4b9ec43b386f.tar.xz anta-92240acb5cc600eec60624ece9ed4b9ec43b386f.zip |
Adding upstream version 0.15.0.upstream/0.15.0
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'asynceapi/aio_portcheck.py')
-rw-r--r-- | asynceapi/aio_portcheck.py | 58 |
1 files changed, 58 insertions, 0 deletions
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 |