fix: use context manager for selector and smtp session

src/ahriman/core/report/email.py

@@ -74,6 +74,18 @@ class Email(Report, JinjaTemplate):
         self.ssl = SmtpSSLSettings.from_option(configuration.get(section, "ssl", fallback="disabled"))
         self.user = configuration.get(section, "user", fallback=None)
 
+    @property
+    def _smtp_session(self) -> type[smtplib.SMTP]:
+        """
+        build SMTP session based on configuration settings
+
+        Returns:
+            type[smtplib.SMTP]: SMTP or SMTP_SSL session depending on whether SSL is enabled or not
+        """
+        if self.ssl == SmtpSSLSettings.SSL:
+            return smtplib.SMTP_SSL
+        return smtplib.SMTP
+
     def _send(self, text: str, attachment: dict[str, str]) -> None:
         """
         send email callback
@@ -93,16 +105,13 @@ class Email(Report, JinjaTemplate):
             attach.add_header("Content-Disposition", "attachment", filename=filename)
             message.attach(attach)
 
-        if self.ssl != SmtpSSLSettings.SSL:
-            session = smtplib.SMTP(self.host, self.port)
+        with self._smtp_session(self.host, self.port) as session:
             if self.ssl == SmtpSSLSettings.STARTTLS:
                 session.starttls()
-        else:
-            session = smtplib.SMTP_SSL(self.host, self.port)
-        if self.user is not None and self.password is not None:
-            session.login(self.user, self.password)
-        session.sendmail(self.sender, self.receivers, message.as_string())
-        session.quit()
+
+            if self.user is not None and self.password is not None:
+                session.login(self.user, self.password)
+            session.sendmail(self.sender, self.receivers, message.as_string())
 
     def generate(self, packages: list[Package], result: Result) -> None:
         """

src/ahriman/core/utils.py

@@ -164,6 +164,11 @@ def check_output(*args: str, exception: Exception | Callable[[int, list[str], st
         if key in ("PATH",)  # whitelisted variables only
     } | environment
 
+    result: dict[str, list[str]] = {
+        "stdout": [],
+        "stderr": [],
+    }
+
     with subprocess.Popen(args, cwd=cwd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                           user=user, env=full_environment, text=True, encoding="utf8", errors="backslashreplace",
                           bufsize=1) as process:
@@ -172,30 +177,27 @@ def check_output(*args: str, exception: Exception | Callable[[int, list[str], st
             input_channel.write(input_data)
             input_channel.close()
 
-        selector = selectors.DefaultSelector()
-        selector.register(get_io(process, "stdout"), selectors.EVENT_READ, data="stdout")
-        selector.register(get_io(process, "stderr"), selectors.EVENT_READ, data="stderr")
+        with selectors.DefaultSelector() as selector:
+            selector.register(get_io(process, "stdout"), selectors.EVENT_READ, data="stdout")
+            selector.register(get_io(process, "stderr"), selectors.EVENT_READ, data="stderr")
 
-        result: dict[str, list[str]] = {
-            "stdout": [],
-            "stderr": [],
-        }
-        while selector.get_map():  # while there are unread selectors, keep reading
-            for key_data, output in poll(selector):
-                result[key_data].append(output)
-
-        stdout = "\n".join(result["stdout"]).rstrip("\n")  # remove newline at the end of any
-        stderr = "\n".join(result["stderr"]).rstrip("\n")
+            while selector.get_map():  # while there are unread selectors, keep reading
+                for key_data, output in poll(selector):
+                    result[key_data].append(output)
 
         status_code = process.wait()
-        if status_code != 0:
-            if isinstance(exception, Exception):
-                raise exception
-            if callable(exception):
-                raise exception(status_code, list(args), stdout, stderr)
-            raise CalledProcessError(status_code, list(args), stderr)
 
-        return stdout
+    stdout = "\n".join(result["stdout"]).rstrip("\n")  # remove newline at the end of any
+    stderr = "\n".join(result["stderr"]).rstrip("\n")
+
+    if status_code != 0:
+        if isinstance(exception, Exception):
+            raise exception
+        if callable(exception):
+            raise exception(status_code, list(args), stdout, stderr)
+        raise CalledProcessError(status_code, list(args), stderr)
+
+    return stdout
 
 
 def check_user(root: Path, *, unsafe: bool) -> None: