A support of Python 3.8 [typing]

dmitry-lipetsk · dmitry-lipetsk · commit 1618d7cad2e7 · 2025-04-08T17:18:29.000+03:00
Python 3.8 does not support tuple[...], list[...], set[...] and so on.

We will use the analogues from typing package: typing.Tuple[...], typing.List[...] and typing.Set[...].
diff --git a/testgres/operations/remote_ops.py b/testgres/operations/remote_ops.py
@@ -5,6 +5,7 @@
 import tempfile
 import io
 import logging
+import typing
 
 from ..exceptions import ExecUtilException
 from ..exceptions import InvalidOperationException
@@ -669,8 +670,8 @@ def _is_port_free__process_1(error: str) -> bool:
         return True
 
     @staticmethod
-    def _make_exec_env_list() -> list[str]:
-        result = list[str]()
+    def _make_exec_env_list() -> typing.List[str]:
+        result: typing.List[str] = list()
         for envvar in os.environ.items():
             if not __class__._does_put_envvar_into_exec_cmd(envvar[0]):
                 continue
diff --git a/testgres/port_manager.py b/testgres/port_manager.py
@@ -6,6 +6,7 @@
 
 import threading
 import random
+import typing
 
 
 class PortManager:
@@ -50,16 +51,16 @@ class PortManager__Generic(PortManager):
     _os_ops: OsOperations
     _guard: object
     # TODO: is there better to use bitmap fot _available_ports?
-    _available_ports: set[int]
-    _reserved_ports: set[int]
+    _available_ports: typing.Set[int]
+    _reserved_ports: typing.Set[int]
 
     def __init__(self, os_ops: OsOperations):
         assert os_ops is not None
         assert isinstance(os_ops, OsOperations)
         self._os_ops = os_ops
         self._guard = threading.Lock()
-        self._available_ports = set[int](range(1024, 65535))
-        self._reserved_ports = set[int]()
+        self._available_ports: typing.Set[int] = set(range(1024, 65535))
+        self._reserved_ports: typing.Set[int] = set()
 
     def reserve_port(self) -> int:
         assert self._guard is not None
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -26,6 +26,10 @@ class TestConfigPropNames:
     TEST_CFG__LOG_DIR = "TEST_CFG__LOG_DIR"
 
 
+# /////////////////////////////////////////////////////////////////////////////
+
+T_TUPLE__str_int = typing.Tuple[str, int]
+
 # /////////////////////////////////////////////////////////////////////////////
 # TestStartupData__Helper
 
@@ -110,11 +114,11 @@ class TEST_PROCESS_STATS:
     cUnexpectedTests: int = 0
     cAchtungTests: int = 0
 
-    FailedTests = list[str, int]()
-    XFailedTests = list[str, int]()
-    NotXFailedTests = list[str]()
-    WarningTests = list[str, int]()
-    AchtungTests = list[str]()
+    FailedTests: typing.List[T_TUPLE__str_int] = list()
+    XFailedTests: typing.List[T_TUPLE__str_int] = list()
+    NotXFailedTests: typing.List[str] = list()
+    WarningTests: typing.List[T_TUPLE__str_int] = list()
+    AchtungTests: typing.List[str] = list()
 
     cTotalDuration: datetime.timedelta = datetime.timedelta()
 
@@ -769,7 +773,7 @@ def helper__calc_W(n: int) -> int:
 
 
 # ------------------------------------------------------------------------
-def helper__print_test_list(tests: list[str]) -> None:
+def helper__print_test_list(tests: typing.List[str]) -> None:
     assert type(tests) == list  # noqa: E721
 
     assert helper__calc_W(9) == 1
@@ -796,7 +800,7 @@ def helper__print_test_list(tests: list[str]) -> None:
 
 
 # ------------------------------------------------------------------------
-def helper__print_test_list2(tests: list[str, int]) -> None:
+def helper__print_test_list2(tests: typing.List[T_TUPLE__str_int]) -> None:
     assert type(tests) == list  # noqa: E721
 
     assert helper__calc_W(9) == 1
@@ -843,7 +847,7 @@ def LOCAL__print_line1_with_header(header: str):
         assert header != ""
         logging.info(C_LINE1 + " [" + header + "]")
 
-    def LOCAL__print_test_list(header: str, test_count: int, test_list: list[str]):
+    def LOCAL__print_test_list(header: str, test_count: int, test_list: typing.List[str]):
         assert type(header) == str  # noqa: E721
         assert type(test_count) == int  # noqa: E721
         assert type(test_list) == list  # noqa: E721
@@ -858,7 +862,7 @@ def LOCAL__print_test_list(header: str, test_count: int, test_list: list[str]):
             logging.info("")
 
     def LOCAL__print_test_list2(
-        header: str, test_count: int, test_list: list[str, int]
+        header: str, test_count: int, test_list: typing.List[T_TUPLE__str_int]
     ):
         assert type(header) == str  # noqa: E721
         assert type(test_count) == int  # noqa: E721
diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py
@@ -12,13 +12,14 @@
 import logging
 import socket
 import threading
+import typing
 
 from ..testgres import InvalidOperationException
 from ..testgres import ExecUtilException
 
 
 class TestOsOpsCommon:
-    sm_os_ops_descrs: list[OsOpsDescr] = [
+    sm_os_ops_descrs: typing.List[OsOpsDescr] = [
         OsOpsDescrs.sm_local_os_ops_descr,
         OsOpsDescrs.sm_remote_os_ops_descr
     ]
diff --git a/tests/test_testgres_common.py b/tests/test_testgres_common.py
@@ -56,7 +56,7 @@ def removing(os_ops: OsOperations, f):
 
 
 class TestTestgresCommon:
-    sm_node_svcs: list[PostgresNodeService] = [
+    sm_node_svcs: typing.List[PostgresNodeService] = [
         PostgresNodeServices.sm_local,
         PostgresNodeServices.sm_local2,
         PostgresNodeServices.sm_remote,
@@ -315,8 +315,8 @@ def test_child_pids(self, node_svc: PostgresNodeService):
 
         def LOCAL__test_auxiliary_pids(
             node: PostgresNode,
-            expectedTypes: list[ProcessType]
-        ) -> list[ProcessType]:
+            expectedTypes: typing.List[ProcessType]
+        ) -> typing.List[ProcessType]:
             # returns list of the absence processes
             assert node is not None
             assert type(node) == PostgresNode  # noqa: E721
@@ -327,15 +327,15 @@ def LOCAL__test_auxiliary_pids(
             assert pids is not None  # noqa: E721
             assert type(pids) == dict  # noqa: E721
 
-            result = list[ProcessType]()
+            result: typing.List[ProcessType] = list()
             for ptype in expectedTypes:
                 if not (ptype in pids):
                     result.append(ptype)
             return result
 
         def LOCAL__check_auxiliary_pids__multiple_attempts(
                 node: PostgresNode,
-                expectedTypes: list[ProcessType]):
+                expectedTypes: typing.List[ProcessType]):
             assert node is not None
             assert type(node) == PostgresNode  # noqa: E721
             assert expectedTypes is not None
diff --git a/tests/test_utils.py b/tests/test_utils.py
@@ -7,10 +7,11 @@
 from ..testgres import scoped_config
 
 import pytest
+import typing
 
 
 class TestUtils:
-    sm_os_ops_descrs: list[OsOpsDescr] = [
+    sm_os_ops_descrs: typing.List[OsOpsDescr] = [
         OsOpsDescrs.sm_local_os_ops_descr,
         OsOpsDescrs.sm_remote_os_ops_descr
     ]
