3.8 compatibility

Author: hategan

tests/installer/dialogs.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import cast
+from typing import cast, TypeVar, Generic
 
 from textual import on
 from textual.app import App, ComposeResult
@@ -11,7 +11,10 @@
 from installer.widgets import DottedLoadingIndicator, ShortcutButton
 
 
-class RunnableDialog[T](ModalScreen[None]):
+T = TypeVar('T')
+
+
+class RunnableDialog(ModalScreen[None], Generic[T]):
     def __init__(self) -> None:
         super().__init__()
         self.done = asyncio.get_running_loop().create_future()

tests/installer/install_methods.py

@@ -34,7 +34,7 @@ def _save_env() -> None:
 
 class InstallMethod(ABC):
     @abstractmethod
-    def is_available(self) -> Tuple[bool, str | None]:
+    def is_available(self) -> Tuple[bool, Optional[str]]:
         pass
 
     @abstractmethod
@@ -73,7 +73,7 @@ def __init__(self) -> None:
         minute = random.randint(0, 59)
         self.line = f'{minute} {hour} * * * "{cwd}/psij-ci-run" --log'
 
-    def is_available(self) -> Tuple[bool, str | None]:
+    def is_available(self) -> Tuple[bool, Optional[str]]:
         if not _succeeds("ps -eo command | awk '{print $1}' | grep cron"):
             return False, 'not found'
         if _succeeds('crontab -l 2>&1 | grep "not allowed"'):
@@ -126,7 +126,7 @@ def __init__(self) -> None:
         self.minute = random.randint(0, 59)
         self.cmd = f'psij-ci-run --reschedule {self.hour}:{self.minute} --log'
 
-    def is_available(self) -> Tuple[bool, str | None]:
+    def is_available(self) -> Tuple[bool, Optional[str]]:
         fd, path = mkstemp()
         os.close(fd)
 
@@ -187,7 +187,7 @@ class Screen(InstallMethod):
     def __init__(self) -> None:
         self.cmd = 'screen -d -m bash -c "./psij-ci-run --repeat --log"'
 
-    def is_available(self) -> Tuple[bool, str | None]:
+    def is_available(self) -> Tuple[bool, Optional[str]]:
         if _succeeds('which screen'):
             return True, None
         else:
@@ -219,7 +219,7 @@ def help_message(self) -> str:
 
 
 class Custom(InstallMethod):
-    def is_available(self) -> Tuple[bool, str | None]:
+    def is_available(self) -> Tuple[bool, Optional[str]]:
         return True, None
 
     def already_installed(self) -> bool:
@@ -255,7 +255,7 @@ def help_message(self) -> str:
 ]
 
 
-def existing() -> InstallMethod | None:
+def existing() -> Optional[InstallMethod]:
     for method in METHODS:
         if method.already_installed():
             return method

tests/installer/panels/schedule_panel.py

@@ -1,4 +1,4 @@
-from typing import cast
+from typing import cast, Optional
 
 from .panel import Panel
 
@@ -16,7 +16,8 @@ class MRadioSet(RadioSet):
         ('enter,space', 'select')
     ]
 
-    def __init__(self, panel: 'SchedulePanel', *radios: RadioButton, id: str | None = None) -> None:
+    def __init__(self, panel: 'SchedulePanel', *radios: RadioButton,
+                 id: Optional[str] = None) -> None:
         super().__init__(*radios, id=id)
         self.panel = panel
 
@@ -39,7 +40,7 @@ def action_select(self) -> None:
 
         self.action_toggle_button()
 
-    def get_selected_index(self) -> int | None:
+    def get_selected_index(self) -> Optional[int]:
         return self._selected
 
 
@@ -86,7 +87,7 @@ def label(self) -> str:
     def name(self) -> str:
         return 'install'
 
-    def _get_selected_method(self) -> InstallMethod | None:
+    def _get_selected_method(self) -> Optional[InstallMethod]:
         radio_set = self.get_widget_by_id('rs-method')
         assert isinstance(radio_set, MRadioSet)
         selected_index = radio_set.get_selected_index()
@@ -115,7 +116,7 @@ async def activate(self) -> None:
                 btn.value = True
                 break
 
-    def radio_focused(self, btn: RadioButton | None) -> None:
+    def radio_focused(self, btn: Optional[RadioButton]) -> None:
         log.write(f'focused {btn}\n')
         preview = self.get_widget_by_id('method-preview')
         assert isinstance(preview, TextArea)

tests/installer/state.py

@@ -1,12 +1,13 @@
 import asyncio
+import functools
 import json
 import os
 import pathlib
 import types
 
 import requests
 from collections import namedtuple
-from typing import List, Dict, Optional, Callable, Awaitable, Tuple, cast
+from typing import List, Dict, Optional, Callable, Awaitable, Tuple, cast, Union
 
 from .install_methods import InstallMethod
 from .log import log
@@ -18,21 +19,27 @@
 Attr = namedtuple('Attr', ['filter', 'name', 'value'])
 
 
+async def _to_thread(func, /, *args, **kwargs):  # type: ignore
+    loop = asyncio.get_running_loop()
+    func_call = functools.partial(func, *args, **kwargs)
+    return await loop.run_in_executor(None, func_call)
+
+
 class _Options:
     def __init__(self) -> None:
         self.custom_attributes = None
 
 
 class ConfWrapper:
-    def __init__(self, dict: Dict[str, str | int | bool | None]):
+    def __init__(self, dict: Dict[str, Union[str, int, bool, None]]):
         self.dict = dict
         self.option = _Options()
         dict['run_id'] = 'x'
         dict['save_results'] = False
         dict['upload_results'] = True
         dict['branch_name_override'] = None
 
-    def getoption(self, name: str) -> str | int | bool | None:
+    def getoption(self, name: str) -> Union[str, int, bool, None]:
         return self.dict[name]
 
 
@@ -43,16 +50,16 @@ def __init__(self, conftest: types.ModuleType, ci_runner: types.ModuleType):
         log.write('Conf: ' + str(self.conf) + '\n')
         log.write(str(self.env) + '\n')
         self.disable_install = False
-        self.install_method: InstallMethod | None = None
-        self.active_panel: int | None = None
-        self.scheduler: str | None = None
+        self.install_method: Optional[InstallMethod] = None
+        self.active_panel: Optional[int] = None
+        self.scheduler: Optional[str] = None
         self.attrs = self._parse_attributes()
         self.run_test_job = True
         self.has_key = KEY_PATH.exists()
         if self.has_key:
             with open(KEY_PATH, 'r') as f:
                 self.key = f.read().strip()
-        self.key_is_valid: bool | None = None
+        self.key_is_valid: Optional[bool] = None
         log.write(f'has key: {self.has_key}\n')
 
     def _parse_attributes(self) -> List[Attr]:
@@ -89,15 +96,15 @@ def update_conf(self, name: str, value: str) -> None:
             self._write_conf_value(name, value)
 
     async def request(self, query: str, data: Dict[str, object], title: str,
-                      error_cb: Callable[[str, str], Awaitable[Dict[str, object] | None]]) \
-            -> Dict[str, object] | None:
+                      error_cb: Callable[[str, str], Awaitable[Optional[Dict[str, object]]]]) \
+            -> Optional[Dict[str, object]]:
         baseUrl = self.conf['server_url']
-        response = await asyncio.to_thread(requests.post, baseUrl + query, data=data)
+        response = await _to_thread(requests.post, baseUrl + query, data=data)  # type: ignore
         return await self._check_error(response, title, error_cb)
 
     async def _check_error(self, response: requests.Response, title: str,
-                           error_cb: Callable[[str, str], Awaitable[Dict[str, object] | None]]) \
-            -> Dict[str, object] | None:
+                           error_cb: Callable[[str, str], Awaitable[Optional[Dict[str, object]]]]) \
+            -> Optional[Dict[str, object]]:
         log.write(f'Response: {response.text}\n')
         if response.status_code != 200:
             msg = self._extract_response_message(response.text)