lint fixes

kvmd/apps/kvmd/api/log.py

@@ -20,8 +20,6 @@
 # ========================================================================== #
 
 
-from typing import Optional
-
 import aiohttp.web
 
 from ....validators.basic import valid_bool
@@ -44,14 +42,15 @@ class LogApi:
     async def __log_handler(self, request: aiohttp.web.Request) -> aiohttp.web.StreamResponse:
         seek = valid_log_seek(request.query.get("seek", "0"))
         follow = valid_bool(request.query.get("follow", "false"))
-        response: Optional[aiohttp.web.StreamResponse] = None
+
+        response = aiohttp.web.StreamResponse(status=200, reason="OK", headers={"Content-Type": "text/plain"})
+        await response.prepare(request)
+
         async for record in self.__log_reader.poll_log(seek, follow):
-            if response is None:
-                response = aiohttp.web.StreamResponse(status=200, reason="OK", headers={"Content-Type": "text/plain"})
-                await response.prepare(request)
             await response.write(("[%s %s] --- %s" % (
                 record["dt"].strftime("%Y-%m-%d %H:%M:%S"),
                 record["service"],
                 record["msg"],
             )).encode("utf-8") + b"\r\n")
+
         return response

kvmd/apps/kvmd/http.py

@@ -138,6 +138,8 @@ def make_json_exception(err: Exception, status: int) -> aiohttp.web.Response:
 # =====
 async def get_multipart_field(reader: aiohttp.MultipartReader, name: str) -> aiohttp.BodyPartReader:
     field = await reader.next()
+    if not isinstance(field, aiohttp.BodyPartReader):
+        raise ValidatorError(f"Expected body part as {name!r} field")
     if not field or field.name != name:
         raise ValidatorError(f"Missing {name!r} field")
     return field

kvmd/apps/vnc/kvmd.py

@@ -23,6 +23,7 @@
 import contextlib
 
 from typing import Dict
+from typing import AsyncGenerator
 
 import aiohttp
 
@@ -64,7 +65,7 @@ class KvmdClient:
             raise
 
     @contextlib.asynccontextmanager
-    async def ws(self, user: str, passwd: str) -> aiohttp.ClientWebSocketResponse:  # pylint: disable=invalid-name
+    async def ws(self, user: str, passwd: str) -> AsyncGenerator[aiohttp.ClientWebSocketResponse, None]:
         async with self.__make_session(user, passwd) as session:
             async with session.ws_connect(
                 url=f"http://{self.__host}:{self.__port}/ws",

kvmd/apps/vnc/streamer.py

@@ -61,7 +61,9 @@ class StreamerClient:
                     response.raise_for_status()
                     reader = aiohttp.MultipartReader.from_response(response)
                     while True:
-                        frame = await reader.next()
+                        frame = await reader.next()  # pylint: disable=not-callable
+                        if not isinstance(frame, aiohttp.BodyPartReader):
+                            raise RuntimeError("Expected body part")
                         yield (
                             (frame.headers["X-UStreamer-Online"] == "true"),
                             int(frame.headers["X-UStreamer-Width"]),

kvmd/logging.py

@@ -21,12 +21,16 @@
 
 
 import sys
+import types
 import logging
 
+from typing import Optional
+
 
 # =====
 def get_logger(depth: int=1) -> logging.Logger:
-    frame = sys._getframe(1)  # pylint: disable=protected-access
+    frame: Optional[types.FrameType] = sys._getframe(1)  # pylint: disable=protected-access
+    assert frame
     frames = []
     while frame:
         frames.append(frame)

kvmd/plugins/msd/__init__.py

@@ -94,8 +94,7 @@ class BaseMsd(BasePlugin):
         raise NotImplementedError()
 
     async def poll_state(self) -> AsyncGenerator[Dict, None]:
-        if True:  # pylint: disable=using-constant-test
+        if self is not None:  # XXX: Vulture and pylint hack
-            # XXX: Vulture hack
             raise NotImplementedError()
         yield
 
@@ -117,9 +116,8 @@ class BaseMsd(BasePlugin):
         raise NotImplementedError()
 
     @contextlib.asynccontextmanager
-    async def write_image(self, name: str) -> AsyncGenerator[None, None]:
+    async def write_image(self, name: str) -> AsyncGenerator[None, None]:  # pylint: disable=unused-argument
-        if True:  # pylint: disable=using-constant-test
+        if self is not None:  # XXX: Vulture and pylint hack
-            # XXX: Vulture hack
             raise NotImplementedError()
         yield
 

kvmd/plugins/msd/disabled.py

@@ -74,8 +74,7 @@ class Plugin(BaseMsd):
 
     @contextlib.asynccontextmanager
     async def write_image(self, name: str) -> AsyncGenerator[None, None]:
-        if True:  # pylint: disable=using-constant-test
+        if self is not None:  # XXX: Vulture and pylint hack
-            # XXX: Vulture hack
             raise MsdDisabledError()
         yield
 

kvmd/plugins/msd/relay.py

@@ -127,7 +127,7 @@ def _parse_image_info_bytes(data: bytes) -> Optional[_ImageInfo]:
 
 def _ioctl_uint32(device_file: IO, request: int) -> int:
     buf = b"\0" * 4
-    buf = fcntl.ioctl(device_file.fileno(), request, buf)
+    buf = fcntl.ioctl(device_file.fileno(), request, buf)  # type: ignore
     result = struct.unpack("I", buf)[0]
     assert result > 0, (device_file, request, buf)
     return result

testenv/linters/pylint.ini

@@ -44,7 +44,7 @@ max-line-length = 160
 bad-functions =
 
 # Good variable names which should always be accepted, separated by a comma
-good-names = _, __, x, y
+good-names = _, __, x, y, ws
 
 # Regular expression matching correct method names
 method-rgx = [a-z_][a-z0-9_]{2,50}$