Merge branch 'multi-tests'

Author: jsl12

.vscode/settings.json

@@ -8,5 +8,5 @@
     ],
     "python.testing.unittestEnabled": false,
     "python.testing.pytestEnabled": true,
-    "python.testing.autoTestDiscoverOnSavePattern": "tests/**/*.py"
+    "python.testing.autoTestDiscoverOnSavePattern": "tests/**/*.py",
 }

appdaemon/__main__.py

@@ -412,7 +412,8 @@ def signal_handlers(self, loop: asyncio.AbstractEventLoop):
     def stop(self):
         """Called by the signal handler to shut AD down."""
         self.stop_time = perf_counter()
-        self.loop.create_task(self.AD.stop())
+        task = self.loop.create_task(self.AD.stop())
+        task.add_done_callback(lambda _: self.loop.stop())
 
     def run(self) -> None:
         """Start AppDaemon up after initial argument parsing."""

appdaemon/app_management.py

@@ -166,6 +166,8 @@ def valid_apps(self) -> set[str]:
     def start(self) -> None:
         self.logger.debug("Starting the app management subsystem")
         if self.AD.apps_enabled:
+            self.AD.loop.create_task(self.init_admin_entities())
+
             task = self.AD.loop.create_task(
                 self.check_app_updates(mode=UpdateMode.INIT),
                 name="check_app_updates",
@@ -202,6 +204,21 @@ async def get_state(self, name: str, **kwargs):
 
         return await self.AD.state.get_state("_app_management", "admin", entity_id, **kwargs)
 
+    async def init_admin_entities(self):
+        for app_name, cfg in self.app_config.root.items():
+            match cfg:
+                case AppConfig() as app_cfg:
+                    await self.add_entity(
+                        app_name,
+                        state="loaded",
+                        attributes={
+                            "totalcallbacks": 0,
+                            "instancecallbacks": 0,
+                            "args": app_cfg.args,
+                            "config_path": app_cfg.config_path,
+                        },
+                    )
+
     async def add_entity(self, name: str, state, attributes):
         # not a fully qualified entity name
         if "." not in name:
@@ -715,7 +732,7 @@ async def check_app_config_files(self, update_actions: UpdateActions):
                 threads_to_add = active_apps - self.AD.threading.thread_count
                 self.logger.debug(f"Adding {threads_to_add} threads based on the active app count")
                 for _ in range(threads_to_add):
-                    await self.AD.threading.add_thread(silent=False, pinthread=True)
+                    await self.AD.threading.add_thread(silent=False)
 
     @utils.executor_decorator
     def read_config_file(self, file: Path) -> AllAppConfig:
@@ -814,7 +831,6 @@ async def wrapper(*args, **kwargs):
 
         return wrapper
 
-    # @utils.timeit
     async def check_app_updates(
         self,
         plugin_ns: str | None = None,
@@ -1371,7 +1387,12 @@ def enable_app(self, app: str):
 
     @contextlib.asynccontextmanager
     async def app_run_context(self, app: str, **kwargs):
-        """Context manager for running an app."""
+        """Context manager for running an app to help during testing.
+
+        Args:
+            app (str): The name of the app to run. Must have an entry in the app_config root.
+            **kwargs: Arbitrary keyword arguments representing configuration fields to temporarily update the app with.
+        """
         match self.app_config.root.get(app):
             case AppConfig() as app_cfg:
                 # Store the complete original configuration
@@ -1382,14 +1403,21 @@ async def app_run_context(self, app: str, **kwargs):
                 return
 
         try:
-            self.update_app(app, **kwargs)
+            if kwargs:
+                self.update_app(app, **kwargs)
+                self.logger.debug("Temporarily updated app '%s' with: %s", app, kwargs)
+
+            # Ensure there's at least one thread available
+            if not self.AD.threading.thread_count:
+                await self.AD.threading.create_initial_threads()
 
+            created_app_object = False
             if app not in self.objects:
                 self.logger.debug("Creating ManagedObject for app '%s'", app)
                 await self.create_app_object(app)
                 await self.AD.threading.calculate_pin_threads()
+                created_app_object = True
 
-            self.logger.debug("Temporarily updated app '%s' with: %s", app, kwargs)
             await self.start_app(app)
             yield
         finally:
@@ -1399,6 +1427,8 @@ async def app_run_context(self, app: str, **kwargs):
                 self.logger.debug("Restored app '%s' to original state", app)
             except ValidationError as e:
                 self.logger.warning("Failed to restore app '%s' to original state: %s", app, e)
+            if created_app_object:
+                self.objects.pop(app)
 
     @utils.executor_decorator
     def remove_app(self, app: str, **kwargs):

appdaemon/appdaemon.py

@@ -23,14 +23,13 @@
 from appdaemon.threads import Threading
 from appdaemon.utility_loop import Utility
 
-from . import utils
 
 if TYPE_CHECKING:
     from appdaemon.http import HTTP
     from appdaemon.logging import Logging
 
 
-class AppDaemon(metaclass=utils.Singleton):
+class AppDaemon:
     """Top-level container for the subsystem objects. This gets passed to the subsystem objects and stored in them as
     the ``self.AD`` attribute.
 
@@ -382,6 +381,7 @@ def utility_delay(self):
         return self.config.utility_delay
 
     def start(self) -> None:
+        self.logger.debug("Starting AppDaemon")
         self.thread_async.start()
         self.sched.start()
         self.utility.start()
@@ -400,7 +400,7 @@ async def stop(self) -> None:
         - :class:`~.scheduler.Scheduler`
         - :class:`~.state.State`
         """
-        self.logger.info("Shutting down AppDaemon")
+        self.logger.info("Stopping AppDaemon")
         self.stopping = True
 
         # Subsystems are able to create tasks during their stop methods
@@ -419,13 +419,17 @@ async def stop(self) -> None:
         self.sched.stop()
         self.state.stop()
 
+        self.executor.shutdown(wait=True)
+
         # This creates a task that will wait for all the ones that were running when stop() was called to finish
         # before stopping the event loop. This allows subsystems to create tasks during their own stop methods
-        running_tasks = asyncio.all_tasks()
+        current_task = asyncio.current_task()
+        running_tasks = [task for task in asyncio.all_tasks() if task is not current_task]
         self.logger.debug(f"Waiting for {len(running_tasks)} tasks to finish before shutting down")
         all_coro = asyncio.wait(running_tasks, return_when=asyncio.ALL_COMPLETED, timeout=3)
         gather_task = asyncio.create_task(all_coro, name="appdaemon_stop_tasks")
-        gather_task.add_done_callback(lambda _: self.loop.stop())
+        gather_task.add_done_callback(lambda _: self.logger.debug("All tasks finished"))
+        await gather_task
 
     #
     # Utilities

appdaemon/exceptions.py

@@ -69,13 +69,22 @@ def user_exception_block(logger: Logger, exception: AppDaemonException, app_dir:
         indent = ' ' * i * 2
 
         match exc:
+            case AssertionError():
+                logger.error(f"{indent}{exc.__class__.__name__}: {exc}")
+                continue
             case ValidationError():
-                errors = exc.errors()
-                if errors[0]['type'] == 'missing':
-                    app_name = errors[0]['loc'][0]
-                    field = errors[0]['loc'][-1]
-                    logger.error(f"{indent}App '{app_name}' is missing required field: {field}")
-                    continue
+                for error in exc.errors():
+                    match error:
+                        case {"type": "missing", "msg": msg, "loc": loc}:
+                            app_name = loc[0]
+                            field = loc[-1]
+                            logger.error(f"{indent}App '{app_name}' has an assertion error in field '{field}': {msg}")
+                        case {"type": "assertion_error", "msg": msg, "loc": loc}:
+                            app_name = loc[0]
+                            field = loc[-1]
+                            logger.error(f"{indent}Assertion error in app '{app_name}' field '{field}': {msg}")
+                        case _:
+                            pass
             case AppDaemonException():
                 for i, line in enumerate(str(exc).splitlines()):
                     if i == 0:
@@ -239,6 +248,7 @@ def __str__(self):
             f"Services that exist in {self.domain}: {', '.join(self.domain_services)}"
         )
 
+
 @dataclass
 class DomainNotSpecified(AppDaemonException):
     namespace: str
@@ -389,6 +399,20 @@ class PinOutofRange(AppDaemonException):
     def __str__(self):
         return f"Pin thread {self.pin_thread} out of range. Must be between 0 and {self.total_threads - 1}"
 
+
+@dataclass
+class InvalidThreadConfiguration(AppDaemonException):
+    total_threads: int | None
+    pin_apps: bool
+    pin_threads: int | None
+
+    def __str__(self):
+        res = "Invalid thread configuration:\n"
+        res += f"  total_threads: {self.total_threads}\n"
+        res += f"  pin_apps:      {self.pin_apps}\n"
+        res += f"  pin_threads:   {self.pin_threads}\n"
+        return res
+
 @dataclass
 class BadClassSignature(AppDaemonException):
     class_name: str

appdaemon/models/config/appdaemon.py

@@ -55,7 +55,7 @@ class AppDaemonConfig(BaseModel, extra="allow"):
     max_clock_skew: int = 1
 
     loglevel: str = "INFO"
-    module_debug: ModuleLoggingLevels = Field(default_factory=dict)
+    module_debug: ModuleLoggingLevels = Field(default_factory=ModuleLoggingLevels)
 
     api_port: int | None = None
     api_key: SecretStr | None = None
@@ -159,6 +159,14 @@ def validate_plugins(cls, v: Any):
             v[n]["name"] = n
         return v
 
+    @model_validator(mode="before")
+    @classmethod
+    def validate_ad_cfg(cls, data: Any) -> Any:
+        if isinstance(data, dict):
+            if (file := data.get("config_file")) and not data.get("config_dir"):
+                data["config_dir"] = Path(file).parent
+        return data
+
     def model_post_init(self, __context: Any):
         # Convert app_dir to Path object
         self.app_dir = Path(self.app_dir) if not isinstance(self.app_dir, Path) else self.app_dir
@@ -169,10 +177,11 @@ def model_post_init(self, __context: Any):
 
         self.ext = ".toml" if self.write_toml else ".yaml"
 
-    @model_validator(mode="before")
-    @classmethod
-    def validate_ad_cfg(cls, data: Any) -> Any:
-        if isinstance(data, dict):
-            if (file := data.get("config_file")) and not data.get("config_dir"):
-                data["config_dir"] = Path(file).parent
-        return data
+        if self.total_threads is not None:
+            self.pin_apps = False
+
+        if self.pin_threads is not None and self.total_threads is not None:
+            # assert self.total_threads is not None, "Using pin_threads requires total_threads to be set."
+            assert self.pin_threads <= self.total_threads, (
+                "Number of pin threads has to be less than or equal to total threads."
+            )