server: Create an interface for server event listeners

subiquity/server/controllers/reporting.py

@@ -27,6 +27,7 @@
 from curtin.reporter.handlers import LogHandler as CurtinLogHandler
 
 from subiquity.server.controller import NonInteractiveController
+from subiquity.server.event_listener import EventListener
 from subiquitycore.context import Context
 
 
@@ -46,7 +47,7 @@ def publish_event(self, event):
 NON_INTERACTIVE_CONFIG = {"builtin": {"type": "print"}}
 
 
-class ReportingController(NonInteractiveController):
+class ReportingController(EventListener, NonInteractiveController):
     autoinstall_key = "reporting"
     autoinstall_schema = {
         "type": "object",

subiquity/server/event_listener.py

@@ -0,0 +1,44 @@
+# Copyright 2024 Canonical, Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+from abc import ABC, abstractmethod
+from typing import Any
+
+from subiquitycore.context import Context
+
+
+class EventListener(ABC):
+    """Interface for SubiquitySever event listeners"""
+
+    @abstractmethod
+    def report_start_event(self, context: Context, description: str) -> None:
+        """Report a "start" event."""
+
+    @abstractmethod
+    def report_finish_event(
+        self, context: Context, description: str, result: Any
+    ) -> None:
+        """Report a "finish" event."""
+
+    @abstractmethod
+    def report_info_event(self, context: Context, message: str) -> None:
+        """Report an "info" event."""
+
+    @abstractmethod
+    def report_warning_event(self, context: Context, message: str) -> None:
+        """Report a "warning" event."""
+
+    @abstractmethod
+    def report_error_event(self, context: Context, message: str) -> None:
+        """Report an "error" event."""

subiquity/server/server.py

@@ -46,6 +46,7 @@
 from subiquity.server.controller import SubiquityController
 from subiquity.server.dryrun import DRConfig
 from subiquity.server.errors import ErrorController
+from subiquity.server.event_listener import EventListener
 from subiquity.server.geoip import DryRunGeoIPStrategy, GeoIP, HTTPGeoIPStrategy
 from subiquity.server.nonreportable import NonReportableException
 from subiquity.server.pkghelper import get_package_installer
@@ -326,7 +327,7 @@ def __init__(self, opts, block_log_dir):
             log.info("no snapd socket found. Snap support is disabled")
             self.snapd = None
         self.note_data_for_apport("SnapUpdated", str(self.updated))
-        self.event_listeners = []
+        self.event_listeners: list[EventListener] = []
         self.autoinstall_config = None
         self.hub.subscribe(InstallerChannels.NETWORK_UP, self._network_change)
         self.hub.subscribe(InstallerChannels.NETWORK_PROXY_SET, self._proxy_set)
@@ -344,7 +345,7 @@ def load_serialized_state(self):
         for controller in self.controllers.instances:
             controller.load_state()
 
-    def add_event_listener(self, listener):
+    def add_event_listener(self, listener: EventListener):
         self.event_listeners.append(listener)
 
     def _maybe_push_to_journal(