diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-23 05:06:51 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-23 05:06:51 +0000 |
commit | 57f704845136d9d0f0266c880cfa1dd7b902e880 (patch) | |
tree | 1c7ffc0d0bf371b2e3d425efbb1c055a24340dc7 /asynceapi/aio_portcheck.py | |
parent | Releasing debian version 0.14.0-2. (diff) | |
download | anta-57f704845136d9d0f0266c880cfa1dd7b902e880.tar.xz anta-57f704845136d9d0f0266c880cfa1dd7b902e880.zip |
Merging upstream version 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 |