91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
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
|
|
|
|
|
|
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_app_dependencies(ad: AppDaemon):
|
|
graph = ad.app_management.app_config.depedency_graph()
|
|
assert 'hello_world' in graph['food_app']
|
|
|
|
|
|
def validate_module_dependencies(ad: AppDaemon):
|
|
graph = ad.app_management.module_dependencies
|
|
assert 'food' in graph['restaurant']
|
|
assert 'food.meal' in graph['restaurant']
|
|
|
|
|
|
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
|
|
|
|
validate_app_dependencies(ad_system)
|
|
validate_module_dependencies(ad_system)
|
|
|
|
module_load_order = get_load_order(caplog)
|
|
validate_module_load_order(ad_system, module_load_order)
|
|
|
|
print()
|