moved get_free_port and updated import

Signed-off-by: Lanuti_emanuele <[email protected]>
Signed-off-by: Emanuele Lanuti <[email protected]>

Author: LanutiEmanuele

tests/conftest.py

@@ -23,6 +23,8 @@
 from suzieq.shared.schema import Schema
 from suzieq.shared.utils import load_sq_config
 from suzieq.sqobjects import get_sqobject, get_tables
+import socket
+from contextlib import closing
 
 suzieq_cli_path = './suzieq/cli/sq_cli.py'
 suzieq_rest_server_path = './suzieq/restServer/sq_rest_server.py'
@@ -286,3 +288,15 @@ def validate_host_shape(df: pd.DataFrame, ns_dict: dict):
         if ns in df.namespace.unique():
             assert df.query(
                 f'namespace == "{ns}"').hostname.nunique() == ns_dict[ns]
+
+
+def get_free_port() -> int:
+    """Return a free port
+
+    Returns:
+        int: free port
+    """
+    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+        s.bind(('', 0))
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        return s.getsockname()[1]

tests/unit/poller/controller/sources/netbox/test_netbox.py

@@ -14,8 +14,8 @@
     NetboxFaker
 from tests.unit.poller.controller.sources.netbox.netbox_rest_server import \
     NetboxRestApp
-from tests.unit.poller.shared.utils import (get_free_port,
-                                            get_src_sample_config)
+from tests.unit.poller.shared.utils import (get_src_sample_config)
+from tests.conftest import get_free_port
 
 # pylint: disable=redefined-outer-name
 

tests/unit/poller/shared/utils.py

@@ -1,6 +1,4 @@
 import random
-import socket
-from contextlib import closing
 from pathlib import Path
 from typing import Any, Dict, Tuple
 
@@ -119,15 +117,3 @@ def read_yaml_file(path: str) -> Any:
     if not file_path.is_file():
         raise RuntimeError(f'Invalid file to read {path}')
     return yaml.safe_load(open(file_path, 'r'))
-
-
-def get_free_port() -> int:
-    """Return a free port
-
-    Returns:
-        int: free port
-    """
-    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
-        s.bind(('', 0))
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        return s.getsockname()[1]