first working tests

src/ad_test/__init__.py

@@ -0,0 +1,2 @@
+def hello() -> str:
+    return "Hello from ad-test!"

src/ad_test/conftest.py

@@ -0,0 +1,12 @@
+from typing import List
+
+from pytest import Function
+
+
+def pytest_collection_modifyitems(session, config, items: List[Function]):
+    for i, item in enumerate(items):
+        if isinstance(item, Function) and "startup" in item.name:
+            items.insert(0, items.pop(i))
+            break
+
+    return items

src/ad_test/fixtures.py

@@ -0,0 +1,50 @@
+import asyncio
+from pathlib import Path
+
+import pytest
+from appdaemon.appdaemon import AppDaemon
+from appdaemon.logging import Logging
+from appdaemon.models.config import AppDaemonConfig
+from git import Repo
+
+from .utils import delayed_stop
+
+CONFIG_DIR = Path(__file__).parents[2] / "conf"
+
+
+@pytest.fixture
+def config_repo() -> Repo:
+    repo = Repo("/home/john/ad-test")
+    return repo
+
+
+@pytest.fixture(scope="session")
+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(scope="session")
+def ad(base_config: AppDaemonConfig):
+    loop = asyncio.new_event_loop()
+
+    ad = AppDaemon(Logging(), loop, base_config)
+    yield ad
+    ad.stop()
+
+    tasks = asyncio.all_tasks(loop)
+    for t in tasks:
+        t.cancel()
+
+    with pytest.raises(asyncio.exceptions.CancelledError):
+        loop.run_until_complete(asyncio.gather(*tasks))
+
+    loop.close()
+    print("Cleaned up running AD")

src/ad_test/test_file_change.py

@@ -0,0 +1,51 @@
+import asyncio
+import logging
+import re
+from pathlib import Path
+
+import food.meal
+import pytest
+from appdaemon.appdaemon import AppDaemon
+from git import Repo
+
+from .fixtures import ad, base_config
+from .utils import get_load_order
+
+
+def reset_file(changed: Path):
+    for p in Path(__file__).parents:
+        if p.with_name(".git").exists():
+            root = p.parent
+            break
+
+    repo = Repo(root)
+    if not changed.is_absolute():
+        changed = root / changed
+    repo.git.checkout("HEAD", "--", changed)
+
+
+def modify_file(path: Path):
+    file_content = path.read_text()
+    modified_content = re.sub(r"Ham", r"Spam", file_content, flags=re.MULTILINE)
+    modified_content = re.sub(r"ham", r"spam", modified_content, flags=re.MULTILINE)
+    path.write_text(modified_content)
+
+
+def test_file(ad: AppDaemon, caplog: pytest.LogCaptureFixture):
+    assert isinstance(ad, AppDaemon)
+
+    # ad.loop.run_until_complete(asyncio.sleep(2.0))
+
+    module_file = Path(food.meal.__file__)
+    reset_file(module_file)
+    modify_file(module_file)
+
+    logging.getLogger("AppDaemon").propagate = True
+    with caplog.at_level(logging.DEBUG, logger="AppDaemon._app_management"):
+        # running_ad.loop.run_until_complete(delayed_stop(running_ad, 2.0))
+        print("waiting in test")
+        ad.loop.run_until_complete(asyncio.sleep(2.0))
+
+    module_load_order = get_load_order(caplog)
+    assert module_load_order == ["food.meal", "restaurant"]
+    print("done")

src/ad_test/test_startup.py

@@ -0,0 +1,56 @@
+import asyncio
+import logging
+from typing import List
+
+import pytest
+from appdaemon.appdaemon import AppDaemon
+
+from .fixtures import ad, base_config
+from .utils import get_load_order
+
+
+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: AppDaemon, caplog: pytest.LogCaptureFixture):
+    assert isinstance(ad, AppDaemon)
+
+    # logger = logging.getLogger('AppDaemon._app_management')
+    ad_logger = logging.getLogger("AppDaemon")
+    ad_logger.propagate = True
+
+    with caplog.at_level(logging.DEBUG, logger="AppDaemon._app_management"):
+        # ad_system.loop.run_until_complete(stop_coro)
+        ad.loop.run_until_complete(asyncio.sleep(2.0))
+
+    assert "Started 'hello_world'" in caplog.text
+    assert "App initialization complete" in caplog.text
+
+    validate_app_dependencies(ad)
+    validate_module_dependencies(ad)
+
+    module_load_order = get_load_order(caplog)
+    validate_module_load_order(ad, module_load_order)
+
+    # ad_system.loop.close()

src/ad_test/utils.py

@@ -0,0 +1,17 @@
+import asyncio
+from typing import List
+
+import pytest
+from appdaemon.appdaemon import AppDaemon
+
+
+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]