Use a validation_complete event rather then looking for bus events

Author: mdegat01

supervisor/addons/manager.py

@@ -184,7 +184,9 @@ async def shutdown(self, stage: AddonStartup) -> None:
         on_condition=AddonsJobError,
         concurrency=JobConcurrency.QUEUE,
     )
-    async def install(self, slug: str) -> None:
+    async def install(
+        self, slug: str, *, validation_complete: asyncio.Event | None = None
+    ) -> None:
         """Install an add-on."""
         self.sys_jobs.current.reference = slug
 
@@ -197,6 +199,10 @@ async def install(self, slug: str) -> None:
 
         store.validate_availability()
 
+        # If being run in the background, notify caller that validation has completed
+        if validation_complete:
+            validation_complete.set()
+
         await Addon(self.coresys, slug).install()
 
         _LOGGER.info("Add-on '%s' successfully installed", slug)
@@ -226,7 +232,11 @@ async def uninstall(self, slug: str, *, remove_config: bool = False) -> None:
         on_condition=AddonsJobError,
     )
     async def update(
-        self, slug: str, backup: bool | None = False
+        self,
+        slug: str,
+        backup: bool | None = False,
+        *,
+        validation_complete: asyncio.Event | None = None,
     ) -> asyncio.Task | None:
         """Update add-on.
 
@@ -251,6 +261,10 @@ async def update(
         # Check if available, Maybe something have changed
         store.validate_availability()
 
+        # If being run in the background, notify caller that validation has completed
+        if validation_complete:
+            validation_complete.set()
+
         if backup:
             await self.sys_backups.do_backup_partial(
                 name=f"addon_{addon.slug}_{addon.version}",

supervisor/api/backups.py

@@ -45,8 +45,6 @@
     ATTR_TYPE,
     ATTR_VERSION,
     REQUEST_FROM,
-    BusEvent,
-    CoreState,
 )
 from ..coresys import CoreSysAttributes
 from ..exceptions import APIError, APIForbidden, APINotFound
@@ -74,8 +72,6 @@
 # Remove: 2022.08
 _ALL_FOLDERS = ALL_FOLDERS + [FOLDER_HOMEASSISTANT]
 
-_freeze_state_filter = lambda new_state: new_state == CoreState.FREEZE
-
 
 def _ensure_list(item: Any) -> list:
     """Ensure value is a list."""
@@ -308,11 +304,7 @@ async def backup_full(self, request: web.Request):
 
         background = body.pop(ATTR_BACKGROUND)
         backup_task, job_id = await background_task(
-            self,
-            self.sys_backups.do_backup_full,
-            bus_event=BusEvent.SUPERVISOR_STATE_CHANGE,
-            event_filter=_freeze_state_filter,
-            **body,
+            self, self.sys_backups.do_backup_full, **body
         )
 
         if background and not backup_task.done():
@@ -348,11 +340,7 @@ async def backup_partial(self, request: web.Request):
 
         background = body.pop(ATTR_BACKGROUND)
         backup_task, job_id = await background_task(
-            self,
-            self.sys_backups.do_backup_partial,
-            bus_event=BusEvent.SUPERVISOR_STATE_CHANGE,
-            event_filter=_freeze_state_filter,
-            **body,
+            self, self.sys_backups.do_backup_partial, **body
         )
 
         if background and not backup_task.done():
@@ -376,12 +364,7 @@ async def restore_full(self, request: web.Request):
         )
         background = body.pop(ATTR_BACKGROUND)
         restore_task, job_id = await background_task(
-            self,
-            self.sys_backups.do_restore_full,
-            backup,
-            bus_event=BusEvent.SUPERVISOR_STATE_CHANGE,
-            event_filter=_freeze_state_filter,
-            **body,
+            self, self.sys_backups.do_restore_full, backup, **body
         )
 
         if background and not restore_task.done() or await restore_task:
@@ -401,12 +384,7 @@ async def restore_partial(self, request: web.Request):
         )
         background = body.pop(ATTR_BACKGROUND)
         restore_task, job_id = await background_task(
-            self,
-            self.sys_backups.do_restore_partial,
-            backup,
-            bus_event=BusEvent.SUPERVISOR_STATE_CHANGE,
-            event_filter=_freeze_state_filter,
-            **body,
+            self, self.sys_backups.do_restore_partial, backup, **body
         )
 
         if background and not restore_task.done() or await restore_task:

supervisor/api/homeassistant.py

@@ -181,7 +181,6 @@ async def update(self, request: web.Request) -> dict[str, str] | None:
         update_task, job_id = await background_task(
             self,
             self.sys_homeassistant.core.update,
-            job_names={"docker_interface_update", "backup_manager_partial_backup"},
             version=body.get(ATTR_VERSION, self.sys_homeassistant.latest_version),
             backup=body.get(ATTR_BACKUP),
         )

supervisor/api/store.py

@@ -233,7 +233,7 @@ async def addons_addon_install(self, request: web.Request) -> dict[str, str] | N
         background = body[ATTR_BACKGROUND]
 
         install_task, job_id = await background_task(
-            self, self.sys_addons.install, addon.slug, job_names={"addon_install"}
+            self, self.sys_addons.install, addon.slug
         )
 
         if background and not install_task.done():
@@ -255,7 +255,6 @@ async def addons_addon_update(self, request: web.Request) -> dict[str, str] | No
             self,
             self.sys_addons.update,
             addon.slug,
-            job_names={"backup_manager_partial_backup", "addon_update"},
             backup=body.get(ATTR_BACKUP),
         )
 

supervisor/api/utils.py

@@ -22,7 +22,6 @@
     REQUEST_FROM,
     RESULT_ERROR,
     RESULT_OK,
-    BusEvent,
 )
 from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import APIError, BackupFileNotFoundError, DockerAPIError, HassioError
@@ -208,19 +207,13 @@ async def background_task(
     coresys_obj: CoreSysAttributes,
     task_method: Callable,
     *args,
-    bus_event: BusEvent = BusEvent.SUPERVISOR_JOB_START,
-    event_filter: Callable[[Any], bool] | None = None,
-    job_names: set[str] | None = None,
     **kwargs,
 ) -> tuple[asyncio.Task, str]:
     """Start task in background and return task and job ID.
 
     Args:
         coresys_obj: Instance that accesses coresys data using CoreSysAttributes
-        task_method: The method to execute in the background
-        bus_event: Event type to listen for which any initial validation has passed
-        event_filter: Function to determine if the event is the one specific to the job by the data
-        job_names: Alternative bus_event and event_filter. Validation considered passed when a named child job of primary one starts
+        task_method: The method to execute in the background. Must include a keyword arg 'validation_complete' of type asyncio.Event. Should set it after any initial validation has completed
         *args: Arguments to pass to task_method
         **kwargs: Keyword arguments to pass to task_method
 
@@ -232,36 +225,23 @@ async def background_task(
     job, task = cast(
         tuple[SupervisorJob, asyncio.Task],
         coresys_obj.sys_jobs.schedule_job(
-            task_method, JobSchedulerOptions(), *args, **kwargs
+            task_method,
+            JobSchedulerOptions(),
+            *args,
+            validation_complete=event,
+            **kwargs,
         ),
     )
 
-    if job_names:
-
-        def child_job_filter(job_event: SupervisorJob) -> bool:
-            """Return true if job is a child of main job and name is in set."""
-            return job_event.parent_id == job.uuid and job_event.name in job_names
-
-        event_filter = child_job_filter
-
-    async def release_on_job_start(event_data: Any):
-        """Release if filter passes or no filter is provided."""
-        if not event_filter or event_filter(event_data):
-            event.set()
-
     # Wait for provided event before returning
     # If the task fails validation it should raise before getting there
-    listener = coresys_obj.sys_bus.register_event(bus_event, release_on_job_start)
-    try:
-        event_task = coresys_obj.sys_create_task(event.wait())
-        _, pending = await asyncio.wait(
-            (task, event_task),
-            return_when=asyncio.FIRST_COMPLETED,
-        )
-        # It seems task returned early (error or something), make sure to cancel
-        # the event task to avoid "Task was destroyed but it is pending!" errors.
-        if event_task in pending:
-            event_task.cancel()
-        return (task, job.uuid)
-    finally:
-        coresys_obj.sys_bus.remove_listener(listener)
+    event_task = coresys_obj.sys_create_task(event.wait())
+    _, pending = await asyncio.wait(
+        (task, event_task),
+        return_when=asyncio.FIRST_COMPLETED,
+    )
+    # It seems task returned early (error or something), make sure to cancel
+    # the event task to avoid "Task was destroyed but it is pending!" errors.
+    if event_task in pending:
+        event_task.cancel()
+    return (task, job.uuid)

supervisor/backups/manager.py

@@ -598,6 +598,7 @@ async def do_backup_full(
         homeassistant_exclude_database: bool | None = None,
         extra: dict | None = None,
         additional_locations: list[LOCATION_TYPE] | None = None,
+        validation_complete: asyncio.Event | None = None,
     ) -> Backup | None:
         """Create a full backup."""
         await self._check_location(location)
@@ -614,6 +615,10 @@ async def do_backup_full(
             name, filename, BackupType.FULL, password, compressed, location, extra
         )
 
+        # If being run in the background, notify caller that validation has completed
+        if validation_complete:
+            validation_complete.set()
+
         _LOGGER.info("Creating new full backup with slug %s", new_backup.slug)
         backup = await self._do_backup(
             new_backup,
@@ -648,6 +653,7 @@ async def do_backup_partial(
         homeassistant_exclude_database: bool | None = None,
         extra: dict | None = None,
         additional_locations: list[LOCATION_TYPE] | None = None,
+        validation_complete: asyncio.Event | None = None,
     ) -> Backup | None:
         """Create a partial backup."""
         await self._check_location(location)
@@ -684,6 +690,10 @@ async def do_backup_partial(
                 continue
             _LOGGER.warning("Add-on %s not found/installed", addon_slug)
 
+        # If being run in the background, notify caller that validation has completed
+        if validation_complete:
+            validation_complete.set()
+
         backup = await self._do_backup(
             new_backup,
             addon_list,
@@ -817,8 +827,10 @@ async def _validate_backup_location(
     async def do_restore_full(
         self,
         backup: Backup,
+        *,
         password: str | None = None,
         location: str | None | type[DEFAULT] = DEFAULT,
+        validation_complete: asyncio.Event | None = None,
     ) -> bool:
         """Restore a backup."""
         # Add backup ID to job
@@ -838,6 +850,10 @@ async def do_restore_full(
                 _LOGGER.error,
             )
 
+        # If being run in the background, notify caller that validation has completed
+        if validation_complete:
+            validation_complete.set()
+
         _LOGGER.info("Full-Restore %s start", backup.slug)
         await self.sys_core.set_state(CoreState.FREEZE)
 
@@ -876,11 +892,13 @@ async def do_restore_full(
     async def do_restore_partial(
         self,
         backup: Backup,
+        *,
         homeassistant: bool = False,
         addons: list[str] | None = None,
         folders: list[str] | None = None,
         password: str | None = None,
         location: str | None | type[DEFAULT] = DEFAULT,
+        validation_complete: asyncio.Event | None = None,
     ) -> bool:
         """Restore a backup."""
         # Add backup ID to job
@@ -908,6 +926,10 @@ async def do_restore_partial(
                 _LOGGER.error,
             )
 
+        # If being run in the background, notify caller that validation has completed
+        if validation_complete:
+            validation_complete.set()
+
         _LOGGER.info("Partial-Restore %s start", backup.slug)
         await self.sys_core.set_state(CoreState.FREEZE)
 

supervisor/homeassistant/core.py

@@ -229,6 +229,7 @@ async def update(
         self,
         version: AwesomeVersion | None = None,
         backup: bool | None = False,
+        validation_complete: asyncio.Event | None = None,
     ) -> None:
         """Update HomeAssistant version."""
         to_version = version or self.sys_homeassistant.latest_version
@@ -248,6 +249,10 @@ async def update(
                 f"Version {to_version!s} is already installed", _LOGGER.warning
             )
 
+        # If being run in the background, notify caller that validation has completed
+        if validation_complete:
+            validation_complete.set()
+
         if backup:
             await self.sys_backups.do_backup_partial(
                 name=f"core_{self.instance.version}",

tests/api/test_homeassistant.py

@@ -14,7 +14,6 @@
 from supervisor.homeassistant.api import APIState
 from supervisor.homeassistant.core import HomeAssistantCore
 from supervisor.homeassistant.module import HomeAssistant
-from supervisor.jobs.decorator import _JOB_NAMES, Job
 
 from tests.api import common_test_api_advanced_logs
 from tests.common import load_json_fixture
@@ -207,18 +206,15 @@ async def test_home_assistant_background_update(
 ):
     """Test background update of Home Assistant."""
     coresys.hardware.disk.get_disk_free_space = lambda x: 5000
-    _JOB_NAMES.remove("docker_interface_update")
-    _JOB_NAMES.remove("backup_manager_partial_backup")
     event = asyncio.Event()
     mock_update_called = mock_backup_called = False
 
-    @Job(name="docker_interface_update")
+    # Mock backup/update as long-running tasks
     async def mock_docker_interface_update(*args, **kwargs):
         nonlocal mock_update_called
         mock_update_called = True
         await event.wait()
 
-    @Job(name="backup_manager_partial_backup")
     async def mock_partial_backup(*args, **kwargs):
         nonlocal mock_backup_called
         mock_backup_called = True

tests/api/test_store.py

@@ -18,7 +18,6 @@
 from supervisor.docker.const import ContainerState
 from supervisor.docker.interface import DockerInterface
 from supervisor.docker.monitor import DockerContainerStateEvent
-from supervisor.jobs.decorator import _JOB_NAMES, Job
 from supervisor.store.addon import AddonStore
 from supervisor.store.repository import Repository
 
@@ -398,10 +397,9 @@ async def test_api_store_addons_changelog_corrupted(
 async def test_addon_install_in_background(api_client: TestClient, coresys: CoreSys):
     """Test installing an addon in the background."""
     coresys.hardware.disk.get_disk_free_space = lambda x: 5000
-    _JOB_NAMES.remove("addon_install")
     event = asyncio.Event()
 
-    @Job(name="addon_install")
+    # Mock a long-running install task
     async def mock_addon_install(*args, **kwargs):
         await event.wait()
 
@@ -448,18 +446,15 @@ async def test_addon_update_in_background(
     """Test updating an addon in the background."""
     coresys.hardware.disk.get_disk_free_space = lambda x: 5000
     install_addon_ssh.data_store["version"] = "10.0.0"
-    _JOB_NAMES.remove("addon_update")
-    _JOB_NAMES.remove("backup_manager_partial_backup")
     event = asyncio.Event()
     mock_update_called = mock_backup_called = False
 
-    @Job(name="addon_update")
+    # Mock backup/update as long-running tasks
     async def mock_addon_update(*args, **kwargs):
         nonlocal mock_update_called
         mock_update_called = True
         await event.wait()
 
-    @Job(name="backup_manager_partial_backup")
     async def mock_partial_backup(*args, **kwargs):
         nonlocal mock_backup_called
         mock_backup_called = True