Make sure cache directory exists for miotcloud

miio/miot_cloud.py

@@ -10,6 +10,7 @@
 from micloud.miotspec import MiotSpec
 from pydantic import BaseModel, Field
 
+from miio import CloudException
 from miio.miot_models import DeviceModel
 
 _LOGGER = logging.getLogger(__name__)
@@ -37,7 +38,9 @@ def info_for_model(self, model: str, *, status_filter="released") -> ReleaseInfo
         releases = [inst for inst in self.releases if inst.model == model]
 
         if not releases:
-            raise Exception(f"No releases found for {model=} with {status_filter=}")
+            raise CloudException(
+                f"No releases found for {model=} with {status_filter=}"
+            )
         elif len(releases) > 1:
             _LOGGER.warning(
                 "%s versions found for model %s: %s, using the newest one",
@@ -55,9 +58,25 @@ def info_for_model(self, model: str, *, status_filter="released") -> ReleaseInfo
 class MiotCloud:
     """Interface for miotspec data."""
 
+    MODEL_MAPPING_FILE = "model-to-urn.json"
+
     def __init__(self):
         self._cache_dir = Path(appdirs.user_cache_dir("python-miio"))
 
+    def get_release_list(self) -> ReleaseList:
+        """Fetch a list of available releases."""
+        cache_file = self._cache_dir / MiotCloud.MODEL_MAPPING_FILE
+        try:
+            mapping = self._file_from_cache(cache_file)
+            return ReleaseList.parse_obj(mapping)
+        except FileNotFoundError:
+            _LOGGER.debug("Did not found non-stale %s, trying to fetch", cache_file)
+
+        specs = MiotSpec.get_specs()
+        self._write_to_cache(cache_file, specs)
+
+        return ReleaseList.parse_obj(specs)
+
     def get_device_model(self, model: str) -> DeviceModel:
         """Get device model for model name."""
         file = self._cache_dir / f"{model}.json"
@@ -84,11 +103,11 @@ def get_model_schema(self, model: str) -> Dict:
 
     def _write_to_cache(self, file: Path, data: Dict):
         """Write given *data* to cache file *file*."""
-        file.parent.mkdir(exist_ok=True)
+        file.parent.mkdir(parents=True, exist_ok=True)
         written = file.write_text(json.dumps(data))
         _LOGGER.debug("Written %s bytes to %s", written, file)
 
-    def _file_from_cache(self, file, cache_hours=6) -> Optional[Dict]:
+    def _file_from_cache(self, file, cache_hours=6) -> Dict:
         def _valid_cache():
             expiration = timedelta(hours=cache_hours)
             if (
@@ -100,22 +119,7 @@ def _valid_cache():
             return False
 
         if file.exists() and _valid_cache():
-            _LOGGER.debug("Returning data from cache file %s", file)
+            _LOGGER.debug("Cache hit, returning contents of %s", file)
             return json.loads(file.read_text())
 
-        _LOGGER.debug("Cache file %s not found or it is stale", file)
-        return None
-
-    def get_release_list(self) -> ReleaseList:
-        """Fetch a list of available releases."""
-        mapping_file = "model-to-urn.json"
-
-        cache_file = self._cache_dir / mapping_file
-        mapping = self._file_from_cache(cache_file)
-        if mapping is not None:
-            return ReleaseList.parse_obj(mapping)
-
-        specs = MiotSpec.get_specs()
-        self._write_to_cache(cache_file, specs)
-
-        return ReleaseList.parse_obj(specs)
+        raise FileNotFoundError("Cache file %s not found or it is stale" % file)

miio/tests/fixtures/micloud_miotspec_releases.json

@@ -0,0 +1,24 @@
+{
+  "instances": [
+    {
+      "model": "vendor.plug.single_release",
+      "version": 1,
+      "type": "urn:miot-spec-v2:device:outlet:0000xxxx:vendor-single-release:1",
+      "status": "released",
+      "ts": 1234
+    },
+    {
+      "model": "vendor.plug.two_releases",
+      "version": 1,
+      "type": "urn:miot-spec-v2:device:outlet:0000xxxx:vendor-two-releases:1",
+      "ts": 12345
+    }
+    ,
+    {
+      "model": "vendor.plug.two_releases",
+      "version": 2,
+      "type": "urn:miot-spec-v2:device:outlet:0000xxxx:vendor-two-releases:2",
+      "ts": 123456
+    }
+  ]
+}

miio/tests/test_miot_cloud.py

@@ -0,0 +1,95 @@
+import json
+import logging
+from pathlib import Path
+
+import pytest
+from pytest_mock import MockerFixture
+
+from miio import CloudException
+from miio.miot_cloud import MiotCloud, ReleaseInfo, ReleaseList
+
+
+def load_fixture(filename: str) -> str:
+    """Load a fixture."""
+    # TODO: refactor to avoid code duplication
+    file = Path(__file__).parent.absolute() / "fixtures" / filename
+    with file.open() as f:
+        return json.load(f)
+
+
+@pytest.fixture(scope="module")
+def miotspec_releases() -> ReleaseList:
+    return ReleaseList.parse_obj(load_fixture("micloud_miotspec_releases.json"))
+
+
+def test_releaselist(miotspec_releases: ReleaseList):
+    assert len(miotspec_releases.releases) == 3
+
+
+def test_releaselist_single_release(miotspec_releases: ReleaseList):
+    wanted_model = "vendor.plug.single_release"
+    info: ReleaseInfo = miotspec_releases.info_for_model(wanted_model)
+    assert info.model == wanted_model
+    assert (
+        info.type == "urn:miot-spec-v2:device:outlet:0000xxxx:vendor-single-release:1"
+    )
+
+
+def test_releaselist_multiple_releases(miotspec_releases: ReleaseList):
+    """Test that the newest version gets picked."""
+    two_releases = miotspec_releases.info_for_model("vendor.plug.two_releases")
+    assert two_releases.version == 2
+    assert (
+        two_releases.type
+        == "urn:miot-spec-v2:device:outlet:0000xxxx:vendor-two-releases:2"
+    )
+
+
+def test_releaselist_missing_model(miotspec_releases: ReleaseList):
+    """Test that missing release causes an expected exception."""
+    with pytest.raises(CloudException):
+        miotspec_releases.info_for_model("foo.bar")
+
+
+def test_get_release_list(
+    tmp_path: Path, mocker: MockerFixture, caplog: pytest.LogCaptureFixture
+):
+    """Test that release list parsing works."""
+    caplog.set_level(logging.DEBUG)
+    ci = MiotCloud()
+    ci._cache_dir = tmp_path
+
+    get_specs = mocker.patch("micloud.miotspec.MiotSpec.get_specs", autospec=True)
+    get_specs.return_value = load_fixture("micloud_miotspec_releases.json")
+
+    # Initial call should download the file, and log the cache miss
+    releases = ci.get_release_list()
+    assert len(releases.releases) == 3
+    assert get_specs.called
+    assert "Did not found non-stale" in caplog.text
+
+    # Second call should return the data from cache
+    caplog.clear()
+    get_specs.reset_mock()
+
+    releases = ci.get_release_list()
+    assert len(releases.releases) == 3
+    assert not get_specs.called
+    assert "Did not found non-stale" not in caplog.text
+
+
+def test_write_to_cache(tmp_path: Path):
+    """Test that cache writes and reads function."""
+    file_path = tmp_path / "long" / "path" / "example.json"
+    ci = MiotCloud()
+    ci._write_to_cache(file_path, {"example": "data"})
+    data = ci._file_from_cache(file_path)
+    assert data["example"] == "data"
+
+
+def test_read_nonexisting_cache_file(tmp_path: Path):
+    """Test that cache reads return None if the file does not exist."""
+    file_path = tmp_path / "long" / "path" / "example.json"
+    ci = MiotCloud()
+    with pytest.raises(FileNotFoundError):
+        ci._file_from_cache(file_path)