update types to fix tests, and propagate errors

Author: Carreau

Diff for: ipykernel/kernelapp.py

@@ -16,6 +16,7 @@
 from io import FileIO, TextIOWrapper
 from logging import StreamHandler
 from pathlib import Path
+from typing import Optional
 
 import zmq
 import zmq.asyncio
@@ -144,9 +145,9 @@ class IPKernelApp(BaseIPythonApplication, InteractiveShellApp, ConnectionFileMix
     stdin_socket = Any()
     iopub_socket = Any()
 
-    iopub_thread: BaseThread
-    control_thread: BaseThread
-    shell_channel_thread: BaseThread
+    iopub_thread: Optional[IOPubThread] = Instance(IOPubThread, allow_none=True)  # type:ignore[assignment]
+    control_thread: Optional[BaseThread] = Instance(BaseThread, allow_none=True)  # type:ignore[assignment]
+    shell_channel_thread: Optional[BaseThread] = Instance(BaseThread, allow_none=True)  # type:ignore[assignment]
 
     _ports = Dict()
 
@@ -263,7 +264,7 @@ def _bind_socket(self, s, port):
                     raise
         return None
 
-    def write_connection_file(self):
+    def write_connection_file(self, **kwargs: t.Any) -> None:
         """write connection info to JSON file"""
         cf = self.abs_connection_file
         connection_info = dict(
@@ -403,15 +404,15 @@ def close(self):
         if self.heartbeat:
             self.log.debug("Closing heartbeat channel")
             self.heartbeat.context.term()
-        if self.iopub_thread:
+        if self.iopub_thread is not None:
             self.log.debug("Closing iopub channel")
             self.iopub_thread.stop()
             self.iopub_thread.close()
-        if self.control_thread and self.control_thread.is_alive():
+        if self.control_thread is not None and self.control_thread.is_alive():
             self.log.debug("Closing control thread")
             self.control_thread.stop()
             self.control_thread.join()
-        if self.shell_channel_thread and self.shell_channel_thread.is_alive():
+        if self.shell_channel_thread is not None and self.shell_channel_thread.is_alive():
             self.log.debug("Closing shell channel thread")
             self.shell_channel_thread.stop()
             self.shell_channel_thread.join()