feat: add executor-backed SQLite path resolver helper (#8)

custom_components/nestquest/store.py

@@ -0,0 +1,20 @@
+"""SQLite storage path resolution for the NestQuest integration."""
+from __future__ import annotations
+
+from pathlib import Path
+
+from homeassistant.core import HomeAssistant
+
+from .const import SQLITE_DB_FILENAME
+
+
+def _ensure_dir(path: Path) -> None:
+    """Create the given directory and any missing parents."""
+    path.mkdir(parents=True, exist_ok=True)
+
+
+async def async_get_db_path(hass: HomeAssistant) -> Path:
+    """Return the absolute path to the NestQuest SQLite database file."""
+    db_path = Path(hass.config.path(SQLITE_DB_FILENAME)).absolute()
+    await hass.async_add_executor_job(_ensure_dir, db_path.parent)
+    return db_path

tests/test_store.py

@@ -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()