summaryrefslogtreecommitdiff
path: root/kvmd/apps/ipmi
diff options
context:
space:
mode:
authorMaxim Devaev <[email protected]>2022-09-04 18:08:40 +0300
committerMaxim Devaev <[email protected]>2022-09-04 18:08:40 +0300
commitee3e224e396494cd0d69bb6167087a071a20349c (patch)
tree5becd28570e58a03c6e1e231d0db24c264a73f88 /kvmd/apps/ipmi
parent4b75221e9470b4a009955d7677f16adf8e23e302 (diff)
new typing style
Diffstat (limited to 'kvmd/apps/ipmi')
-rw-r--r--kvmd/apps/ipmi/__init__.py5
-rw-r--r--kvmd/apps/ipmi/auth.py7
-rw-r--r--kvmd/apps/ipmi/server.py21
3 files changed, 12 insertions, 21 deletions
diff --git a/kvmd/apps/ipmi/__init__.py b/kvmd/apps/ipmi/__init__.py
index fc450228..13aa9730 100644
--- a/kvmd/apps/ipmi/__init__.py
+++ b/kvmd/apps/ipmi/__init__.py
@@ -20,9 +20,6 @@
# ========================================================================== #
-from typing import List
-from typing import Optional
-
from ...clients.kvmd import KvmdClient
from ... import htclient
@@ -34,7 +31,7 @@ from .server import IpmiServer
# =====
-def main(argv: Optional[List[str]]=None) -> None:
+def main(argv: (list[str] | None)=None) -> None:
config = init(
prog="kvmd-ipmi",
description="IPMI to KVMD proxy",
diff --git a/kvmd/apps/ipmi/auth.py b/kvmd/apps/ipmi/auth.py
index 0ad3d208..2e495af8 100644
--- a/kvmd/apps/ipmi/auth.py
+++ b/kvmd/apps/ipmi/auth.py
@@ -22,9 +22,6 @@
import dataclasses
-from typing import List
-from typing import Dict
-
# =====
class IpmiPasswdError(Exception):
@@ -55,8 +52,8 @@ class IpmiAuthManager:
def get_credentials(self, ipmi_user: str) -> IpmiUserCredentials:
return self.__credentials[ipmi_user]
- def __parse_passwd_file(self, lines: List[str]) -> Dict[str, IpmiUserCredentials]:
- credentials: Dict[str, IpmiUserCredentials] = {}
+ def __parse_passwd_file(self, lines: list[str]) -> dict[str, IpmiUserCredentials]:
+ credentials: dict[str, IpmiUserCredentials] = {}
for (lineno, line) in enumerate(lines):
if len(line.strip()) == 0 or line.lstrip().startswith("#"):
continue
diff --git a/kvmd/apps/ipmi/server.py b/kvmd/apps/ipmi/server.py
index 597e6dce..87265607 100644
--- a/kvmd/apps/ipmi/server.py
+++ b/kvmd/apps/ipmi/server.py
@@ -28,9 +28,6 @@ import multiprocessing
import functools
import queue
-from typing import Dict
-from typing import Optional
-
import aiohttp
import serial
@@ -83,8 +80,8 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute
self.__sol_proxy_port = (sol_proxy_port or port)
self.__sol_lock = threading.Lock()
- self.__sol_console: Optional[IpmiConsole] = None
- self.__sol_thread: Optional[threading.Thread] = None
+ self.__sol_console: (IpmiConsole | None) = None
+ self.__sol_thread: (threading.Thread | None) = None
self.__sol_stop = False
def run(self) -> None:
@@ -100,7 +97,7 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute
# =====
- def handle_raw_request(self, request: Dict, session: IpmiServerSession) -> None:
+ def handle_raw_request(self, request: dict, session: IpmiServerSession) -> None:
handler = {
(6, 1): (lambda _, session: self.send_device_id(session)), # Get device ID
(6, 7): self.__get_power_state_handler, # Power state
@@ -125,13 +122,13 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute
# =====
- def __get_power_state_handler(self, _: Dict, session: IpmiServerSession) -> None:
+ def __get_power_state_handler(self, _: dict, session: IpmiServerSession) -> None:
# https://github.com/arcress0/ipmiutil/blob/e2f6e95127d22e555f959f136d9bb9543c763896/util/ireset.c#L654
result = self.__make_request(session, "atx.get_state() [power]", "atx.get_state")
data = [(0 if result["leds"]["power"] else 5)]
session.send_ipmi_response(data=data)
- def __get_selftest_status_handler(self, _: Dict, session: IpmiServerSession) -> None:
+ def __get_selftest_status_handler(self, _: dict, session: IpmiServerSession) -> None:
# https://github.com/arcress0/ipmiutil/blob/e2f6e95127d22e555f959f136d9bb9543c763896/util/ihealth.c#L858
data = [0x0055]
try:
@@ -140,12 +137,12 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute
data = [0]
session.send_ipmi_response(data=data)
- def __get_chassis_status_handler(self, _: Dict, session: IpmiServerSession) -> None:
+ def __get_chassis_status_handler(self, _: dict, session: IpmiServerSession) -> None:
result = self.__make_request(session, "atx.get_state() [chassis]", "atx.get_state")
data = [int(result["leds"]["power"]), 0, 0]
session.send_ipmi_response(data=data)
- def __chassis_control_handler(self, request: Dict, session: IpmiServerSession) -> None:
+ def __chassis_control_handler(self, request: dict, session: IpmiServerSession) -> None:
action = {
0: "off_hard",
1: "on",
@@ -179,7 +176,7 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute
# =====
- def __activate_sol_handler(self, _: Dict, session: IpmiServerSession) -> None:
+ def __activate_sol_handler(self, _: dict, session: IpmiServerSession) -> None:
with self.__sol_lock:
if not self.__sol_device_path:
session.send_ipmi_response(code=0x81) # SOL disabled
@@ -198,7 +195,7 @@ class IpmiServer(BaseIpmiServer): # pylint: disable=too-many-instance-attribute
])
self.__start_sol_worker(session)
- def __deactivate_sol_handler(self, _: Dict, session: IpmiServerSession) -> None:
+ def __deactivate_sol_handler(self, _: dict, session: IpmiServerSession) -> None:
with self.__sol_lock:
if not self.__sol_device_path:
session.send_ipmi_response(code=0x81)