import asyncio import logging from pathlib import Path from typing import List import pytest from appdaemon.appdaemon import AppDaemon from appdaemon.logging import Logging from appdaemon.models.config import AppDaemonConfig CONFIG_DIR = Path(__file__).parent / 'conf' @pytest.fixture def base_config() -> AppDaemonConfig: return AppDaemonConfig( latitude=0.0, longitude=0.0, elevation=0, time_zone='America/Chicago', config_dir=CONFIG_DIR, config_file='appdaemon.yaml', stop_function=lambda: print('Stopping'), ) @pytest.fixture def ad_system(base_config) -> AppDaemon: loop = asyncio.new_event_loop() ad = AppDaemon(Logging(), loop, base_config) return ad # def test_dependency_graph(): # app_dir = Path('/home/john/conf/apps') # files = app_dir.rglob('*.yaml') # files = (f for f in files if f.stem != 'secrets' and f.stem != 'appdaemon') # cfg = AllAppConfig.from_config_files(files) # graph = cfg.depedency_graph() # rev = cfg.reversed_dependency_graph() # assert 'App1Foo' in graph['App2'], 'App2 is not dependent on App1Foo' # assert 'App2' in rev['App1Foo'], 'App1 is not a predicate of App1Foo' # print(graph) async def delayed_stop(ad: AppDaemon, delay: float): await asyncio.sleep(delay) ad.stop() def get_load_order(caplog: pytest.LogCaptureFixture) -> List[str]: for record in caplog.records: if record.name == 'AppDaemon._app_management': if 'Determined module load order' in record.msg: return record.args[0] def validate_module_load_order(ad: AppDaemon, module_load_order: List[str]): dependency_graph = ad.app_management.module_dependencies for node, deps in dependency_graph.items(): # skip parts that if not node.startswith('appdaemon'): continue node_idx = module_load_order.index(node) for dep in deps: dep_idx = module_load_order.index(dep) assert dep_idx < node_idx def test_startup(ad_system: AppDaemon, caplog: pytest.LogCaptureFixture): assert isinstance(ad_system, AppDaemon) # logger = logging.getLogger('AppDaemon._app_management') logging.getLogger('AppDaemon').propagate = True stop_coro = delayed_stop(ad_system, 2.0) with caplog.at_level(logging.DEBUG, logger='AppDaemon._app_management'): ad_system.loop.run_until_complete(stop_coro) assert "Started 'hello_world'" in caplog.text assert 'App initialization complete' in caplog.text module_load_order = get_load_order(caplog) validate_module_load_order(ad_system, module_load_order) print()