|
|
|
@@ -0,0 +1,160 @@
|
|
|
|
|
"""Tests for store.py of NestQuest."""
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
import ast
|
|
|
|
|
import asyncio
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
import sys
|
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
|
|
|
|
|
|
# Mock homeassistant modules if not installed in environment
|
|
|
|
|
for _mod in (
|
|
|
|
|
"homeassistant",
|
|
|
|
|
"homeassistant.core",
|
|
|
|
|
"homeassistant.config_entries",
|
|
|
|
|
):
|
|
|
|
|
sys.modules.setdefault(_mod, MagicMock())
|
|
|
|
|
|
|
|
|
|
from custom_components.nestquest import store
|
|
|
|
|
from custom_components.nestquest.const import SQLITE_DB_FILENAME
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _run(coro):
|
|
|
|
|
return asyncio.new_event_loop().run_until_complete(coro)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _make_hass(config_dir: Path, execute: bool = True):
|
|
|
|
|
hass = MagicMock()
|
|
|
|
|
hass.config.path = MagicMock(return_value=str(config_dir / SQLITE_DB_FILENAME))
|
|
|
|
|
hass.async_add_executor_job = AsyncMock(
|
|
|
|
|
side_effect=(lambda fn, *a: fn(*a)) if execute else None
|
|
|
|
|
)
|
|
|
|
|
return hass
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_returns_absolute_path_from_config_path() -> None:
|
|
|
|
|
hass = _make_hass(Path("/config"), execute=False)
|
|
|
|
|
db_path = _run(store.async_get_db_path(hass))
|
|
|
|
|
hass.config.path.assert_called_once_with(SQLITE_DB_FILENAME)
|
|
|
|
|
assert db_path == Path("/config") / SQLITE_DB_FILENAME
|
|
|
|
|
assert db_path.is_absolute()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_relative_config_path_is_normalized_to_absolute(tmp_path, monkeypatch) -> None:
|
|
|
|
|
monkeypatch.chdir(tmp_path)
|
|
|
|
|
hass = MagicMock()
|
|
|
|
|
hass.config.path = MagicMock(return_value="relative-config/" + SQLITE_DB_FILENAME)
|
|
|
|
|
hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
|
|
|
|
|
db_path = _run(store.async_get_db_path(hass))
|
|
|
|
|
assert db_path.is_absolute()
|
|
|
|
|
assert db_path.name == SQLITE_DB_FILENAME
|
|
|
|
|
assert str(db_path).endswith("relative-config/" + SQLITE_DB_FILENAME)
|
|
|
|
|
assert db_path.parent.is_dir()
|
|
|
|
|
assert (tmp_path / "relative-config").is_dir()
|
|
|
|
|
assert not db_path.exists()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_creates_missing_parent_directory(tmp_path) -> None:
|
|
|
|
|
nested = tmp_path / "homeassistant" / "config"
|
|
|
|
|
hass = _make_hass(nested)
|
|
|
|
|
db_path = _run(store.async_get_db_path(hass))
|
|
|
|
|
assert db_path.parent == nested
|
|
|
|
|
assert nested.is_dir()
|
|
|
|
|
assert db_path.parent.is_dir()
|
|
|
|
|
assert not db_path.exists()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_two_calls_return_same_path(tmp_path) -> None:
|
|
|
|
|
hass = _make_hass(tmp_path)
|
|
|
|
|
first = _run(store.async_get_db_path(hass))
|
|
|
|
|
second = _run(store.async_get_db_path(hass))
|
|
|
|
|
assert first == second
|
|
|
|
|
assert first.parent.is_dir()
|
|
|
|
|
assert not first.exists()
|
|
|
|
|
assert not second.exists()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_existing_directory_does_not_fail(tmp_path) -> None:
|
|
|
|
|
nested = tmp_path / "config"
|
|
|
|
|
nested.mkdir()
|
|
|
|
|
hass = _make_hass(nested)
|
|
|
|
|
db_path = _run(store.async_get_db_path(hass))
|
|
|
|
|
assert db_path == nested / SQLITE_DB_FILENAME
|
|
|
|
|
assert db_path.parent.is_dir()
|
|
|
|
|
assert not db_path.exists()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_mkdir_runs_via_executor(tmp_path) -> None:
|
|
|
|
|
nested = tmp_path / "deep" / "nested"
|
|
|
|
|
hass = _make_hass(nested)
|
|
|
|
|
db_path = _run(store.async_get_db_path(hass))
|
|
|
|
|
assert nested.is_dir()
|
|
|
|
|
hass.async_add_executor_job.assert_called_once()
|
|
|
|
|
args, kwargs = hass.async_add_executor_job.call_args
|
|
|
|
|
fn, fn_args = args[0], args[1:]
|
|
|
|
|
assert callable(fn)
|
|
|
|
|
assert fn is store._ensure_dir
|
|
|
|
|
assert fn_args == (nested,)
|
|
|
|
|
assert kwargs == {}
|
|
|
|
|
assert db_path.parent == nested
|
|
|
|
|
assert db_path.parent.is_dir()
|
|
|
|
|
assert not db_path.exists()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_executor_helper_creates_directory(tmp_path) -> None:
|
|
|
|
|
nested = tmp_path / "executor" / "made"
|
|
|
|
|
store._ensure_dir(nested)
|
|
|
|
|
assert nested.is_dir()
|
|
|
|
|
store._ensure_dir(nested)
|
|
|
|
|
assert nested.is_dir()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_no_sqlite3_import_in_package() -> None:
|
|
|
|
|
pkg_dir = Path(store.__file__).parent
|
|
|
|
|
py_files = list(pkg_dir.glob("*.py"))
|
|
|
|
|
assert py_files
|
|
|
|
|
for py_file in py_files:
|
|
|
|
|
tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
|
|
|
|
|
for node in ast.walk(tree):
|
|
|
|
|
if isinstance(node, ast.Import):
|
|
|
|
|
for alias in node.names:
|
|
|
|
|
assert alias.name != "sqlite3", (
|
|
|
|
|
f"Found sqlite3 import in {py_file.name}:{node.lineno}"
|
|
|
|
|
)
|
|
|
|
|
elif isinstance(node, ast.ImportFrom) and node.module == "sqlite3":
|
|
|
|
|
raise AssertionError(
|
|
|
|
|
f"Found sqlite3 import in {py_file.name}:{node.lineno}"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_no_file_creation_calls_in_store() -> None:
|
|
|
|
|
source = Path(store.__file__).read_text(encoding="utf-8")
|
|
|
|
|
tree = ast.parse(source, filename=str(store.__file__))
|
|
|
|
|
for node in ast.walk(tree):
|
|
|
|
|
if isinstance(node, ast.Call):
|
|
|
|
|
func = node.func
|
|
|
|
|
if isinstance(func, ast.Name) and func.id == "open":
|
|
|
|
|
raise AssertionError(
|
|
|
|
|
f"Found open() call in store.py:{node.lineno}"
|
|
|
|
|
)
|
|
|
|
|
if (
|
|
|
|
|
isinstance(func, ast.Attribute)
|
|
|
|
|
and func.attr == "touch"
|
|
|
|
|
):
|
|
|
|
|
raise AssertionError(
|
|
|
|
|
f"Found touch() call in store.py:{node.lineno}"
|
|
|
|
|
)
|
|
|
|
|
assert "touch(" not in source
|
|
|
|
|
assert "open(" not in source
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_container_style_path_resolves() -> None:
|
|
|
|
|
hass = _make_hass(Path("/config"), execute=False)
|
|
|
|
|
assert _run(store.async_get_db_path(hass)) == Path("/config") / SQLITE_DB_FILENAME
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_core_style_path_resolves(tmp_path) -> None:
|
|
|
|
|
config_dir = tmp_path / ".homeassistant"
|
|
|
|
|
hass = _make_hass(config_dir)
|
|
|
|
|
assert _run(store.async_get_db_path(hass)) == config_dir / SQLITE_DB_FILENAME
|
|
|
|
|
assert (config_dir / SQLITE_DB_FILENAME).parent.is_dir()
|