84 lines
2.2 KiB
Python
84 lines
2.2 KiB
Python
import asyncio
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Generator, List, Literal
|
|
|
|
import pytest
|
|
from appdaemon.appdaemon import AppDaemon
|
|
from git import Repo
|
|
|
|
|
|
async def delayed_stop(ad: AppDaemon, delay: float):
|
|
await asyncio.sleep(delay)
|
|
ad.stop()
|
|
|
|
|
|
def get_load_order(
|
|
caplog: pytest.LogCaptureFixture, ordered: bool = True
|
|
) -> Generator[str, None, None]:
|
|
records = (
|
|
record.args[0]
|
|
for record in get_logger_records(caplog, 'AppDaemon._app_management')
|
|
if 'Determined module load order' in record.msg
|
|
)
|
|
try:
|
|
result = list(records)[-1]
|
|
if not ordered:
|
|
result = set(result)
|
|
return result
|
|
except Exception:
|
|
return
|
|
|
|
|
|
def get_logger_records(caplog: pytest.LogCaptureFixture, logger_name: str):
|
|
for record in caplog.records:
|
|
if record.name == logger_name:
|
|
yield record
|
|
|
|
|
|
def get_app_orders(
|
|
caplog: pytest.LogCaptureFixture,
|
|
phase: Literal['start', 'stop'],
|
|
ordered: bool = True,
|
|
) -> list[str]:
|
|
"""Extracts the app start/stop order from the captured log lines"""
|
|
records = (
|
|
record.args[0]
|
|
for record in get_logger_records(caplog, 'AppDaemon._app_management')
|
|
if re.search(f'App {phase} order', record.msg)
|
|
)
|
|
try:
|
|
result = list(records)[-1]
|
|
if not ordered:
|
|
result = set(result)
|
|
return result
|
|
except Exception:
|
|
return
|
|
|
|
|
|
def get_loaded_apps(caplog: pytest.LogCaptureFixture):
|
|
for record in get_logger_records(caplog, 'AppDaemon._app_management'):
|
|
if re.search('Apps to be loaded', record.msg):
|
|
return record.args[0]
|
|
|
|
|
|
def count_error_lines(caplog: pytest.LogCaptureFixture) -> int:
|
|
error_log_lines = [
|
|
msg for name, lvl, msg in caplog.record_tuples if name.startswith('Error')
|
|
]
|
|
return len(error_log_lines)
|
|
|
|
|
|
def reset_file(repo: Repo, changed: Path):
|
|
if not changed.is_absolute():
|
|
changed = Path(repo.working_tree_dir) / changed
|
|
repo.git.checkout('HEAD', '--', changed)
|
|
|
|
|
|
def inject_into_file(file: Path, pos: int, line: str):
|
|
content = file.read_text()
|
|
lines = content.splitlines()
|
|
lines.insert(pos, line)
|
|
new_content = '\n'.join(lines)
|
|
file.write_text(new_content)
|