Add tests for Config class

.coveragerc

@@ -8,4 +8,4 @@ omit =
     # omit abstract endpoints
     */abstracts/*
     # omit config due to base_config
-    */config.py
+    # */config.py

fio_wrapper/config.py

@@ -39,42 +39,36 @@ def data_merge(self, a, b):
         key = None
         # ## debug output
         # sys.stderr.write("DEBUG: %s to %s\n" %(b,a))
-        try:
-            if (
-                a is None
-                or isinstance(a, str)
-                or isinstance(a, int)
-                or isinstance(a, float)
-            ):
-                # border case for first run or if a is a primitive
-                a = b
-            elif isinstance(a, list):
-                # lists can be only appended
-                if isinstance(b, list):
-                    # merge lists
-                    a.extend(b)
-                else:
-                    # append to list
-                    a.append(b)
-            elif isinstance(a, dict):
-                # dicts must be merged
-                if isinstance(b, dict):
-                    for key in b:
-                        if key in a:
-                            a[key] = self.data_merge(a[key], b[key])
-                        else:
-                            a[key] = b[key]
-                else:
-                    raise Exception(
-                        'Cannot merge non-dict "%s" into dict "%s"' % (b, a)
-                    )
+
+        if (
+            a is None
+            or isinstance(a, str)
+            or isinstance(a, int)
+            or isinstance(a, float)
+        ):
+            # border case for first run or if a is a primitive
+            a = b
+        elif isinstance(a, list):
+            # lists can be only appended
+            if isinstance(b, list):
+                # merge lists
+                a.extend(b)
             else:
-                raise Exception('NOT IMPLEMENTED "%s" into "%s"' % (b, a))
-        except TypeError as e:
-            raise Exception(
-                'TypeError "%s" in key "%s" when merging "%s" into "%s"'
-                % (e, key, b, a)
-            )
+                # append to list
+                a.append(b)
+        elif isinstance(a, dict):
+            # dicts must be merged
+            if isinstance(b, dict):
+                for key in b:
+                    if key in a:
+                        a[key] = self.data_merge(a[key], b[key])
+                    else:
+                        a[key] = b[key]
+            else:
+                raise Exception('Cannot merge non-dict "%s" into dict "%s"' % (b, a))
+        else:
+            raise Exception('NOT IMPLEMENTED "%s" into "%s"' % (b, a))
+
         return a
 
     def __init__(
@@ -131,10 +125,7 @@ def versions(self) -> List[str]:
         Returns:
             List[str]: List of versions
         """
-        try:
-            return self.data["fio"]["versions"]
-        except Exception as exc:
-            raise SystemExit("No list of available FIO versions provided") from exc
+        return self.data["fio"]["versions"]
 
     @property
     def api_key(self) -> str:
@@ -156,92 +147,61 @@ def api_key(self) -> str:
     def version(self) -> str:
         """Gets the FIO version specified
 
-        Raises:
-            UnknownConfig: No version setting found
-
         Returns:
             str: FIO API version
         """
         if self._version is not None:
             return self._version
 
-        try:
-            return self.data["fio"]["version"]
-        except Exception as exc:
-            raise UnknownConfig("No version setting found") from exc
+        return self.data["fio"]["version"]
 
     @property
     def application(self) -> str:
         """Gets the application name
 
-        Raises:
-            UnknownConfig: No application setting found
-
         Returns:
             str: Application name
         """
         if self._application is not None:
             return self._application
 
-        try:
-            return self.get("fio", "application")
-        except KeyError as exc:
-            raise UnknownConfig("No application setting found") from exc
+        return self.data["fio"]["application"]
 
     @property
     def base_url(self) -> str:
         """Gets the FIO base url
 
-        Raises:
-            UnknownConfig: No base_url setting found
-
         Returns:
             str: FIO base url
         """
         if self._base_url is not None:
             return self._base_url
 
-        try:
-            return self.data["fio"]["base_url"]
-        except KeyError as exc:
-            raise UnknownConfig("No base_url setting found") from exc
+        return self.data["fio"]["base_url"]
 
     @property
     def timeout(self) -> float:
         """Gets the timeout parameter
 
-        Raises:
-            UnknownConfig: No timeout setting found
-
         Returns:
             float: Timeout parameter
         """
         if self._timeout is not None:
             return self._timeout
 
-        try:
-            # timeout value must be float
-            return self.data["fio"]["timeout"]
-        except KeyError as exc:
-            raise UnknownConfig("No timeout setting found") from exc
+        return self.data["fio"]["timeout"]
 
     @property
     def ssl_verify(self) -> float:
         """Gets the ssl verification parameter
 
-        Raises:
-            UnknownConfig: No ssl_verify setting found
-
         Returns:
             float: Seconds as float of request timeout
         """
         if self._ssl_verify is not None:
             return self._ssl_verify
 
-        try:
-            return self.data["fio"]["ssl_verify"]
-        except KeyError as exc:
-            raise UnknownConfig("No ssl_verify setting found") from exc
+        return self.data["fio"]["ssl_verify"]
 
     @property
     def cache(self) -> bool:
@@ -302,32 +262,23 @@ def cache_url_expirations(self) -> Dict[str, any]:
             Dict[str, any]: URL specific expiration settings
         """
         # check if requests-cache is installed
-        if self.cache and importlib.util.find_spec("requests_cache") is not None:
-            from requests_cache import DO_NOT_CACHE, NEVER_EXPIRE
-
-            try:
-                expiration_list: Dict[str, any] = {}
-
-                # walk through given urls
-                for url, expiration in self.data["cache"]["urls"].items():
-                    # decide on potential values coming as int or str:
-                    #   int = take as seconds
-                    if isinstance(expiration, int):
-                        expiration_list[url] = expiration
-                    #   NEVER_EXPIRE
-                    elif expiration == "NEVER_EXPIRE":
-                        expiration_list[url] = NEVER_EXPIRE
-                    #   DO_NOT_CACHE
-                    elif expiration == "DO_NOT_CACHE":
-                        expiration_list[url] = DO_NOT_CACHE
-                    else:
-                        logger.warn(
-                            "Unknown expiration configuration: %s | %s", url, expiration
-                        )
-                return expiration_list
-            # no cache urls defined, return empty list
-            except KeyError:
-                return {}
-
-        else:
+        if not self.cache or importlib.util.find_spec("requests_cache") is None:
             return {}
+
+        from requests_cache import DO_NOT_CACHE, NEVER_EXPIRE
+
+        expiration_list = {}
+
+        for url, expiration in self.data.get("cache", {}).get("urls", {}).items():
+            if isinstance(expiration, int):
+                expiration_list[url] = expiration
+            elif expiration == "NEVER_EXPIRE":
+                expiration_list[url] = NEVER_EXPIRE
+            elif expiration == "DO_NOT_CACHE":
+                expiration_list[url] = DO_NOT_CACHE
+            else:
+                logger.warning(
+                    "Unknown expiration configuration: %s | %s", url, expiration
+                )
+
+        return expiration_list

tests/test_cache.py

@@ -0,0 +1,141 @@
+import pytest
+from fio_wrapper import FIO, Config
+from fio_wrapper.exceptions import UnknownConfig
+
+
+@pytest.fixture(scope="class")
+def standard_config() -> Config:
+    fio = FIO()
+    return fio.adapter.config
+
+
+def test_cache_data_merge(standard_config: Config) -> None:
+    config = standard_config
+
+    # merge list
+    a = {"l": [1, 2, 3]}
+    b = {"l": [4]}
+
+    merged = config.data_merge(a, b)
+    assert merged["l"] == [1, 2, 3, 4]
+
+    # append list
+    a = {"l": [1, 2, 3]}
+    b = {"l": 4}
+    merged = config.data_merge(a, b)
+    assert merged["l"] == [1, 2, 3, 4]
+
+    # can't merge non-dict
+    a = {"l": {"foo": "moo"}}
+    b = {"l": 1}
+    with pytest.raises(Exception):
+        config.data_merge(a, b)
+
+    # can only merge Dict, None, str, int, float
+    a = standard_config
+    b = {"l": 1}
+    with pytest.raises(Exception):
+        config.data_merge(a, b)
+
+
+def test_cache_properties(standard_config: Config) -> None:
+    config = standard_config
+
+    config._application = "foo"
+    assert config.application == "foo"
+
+    config._timeout = 5
+    assert config.timeout == 5
+
+    config._ssl_verify = None
+    assert config.ssl_verify == True
+
+
+def test_cache_get(standard_config: Config) -> None:
+    with pytest.raises(UnknownConfig):
+        standard_config.get("foo", "moo")
+
+
+def test_cache_get_url(standard_config: Config) -> None:
+    config = standard_config
+
+    config.data["cache"]["urls"] = {
+        "*/material/*": 3600,
+        "*/exchange/all": "NEVER_EXPIRE",
+        "*/exchange/full": "DO_NOT_CACHE",
+    }
+
+    with pytest.raises(UnknownConfig):
+        config.get_url("foo")
+
+
+def test_cache_url_expirations_with_requests_cache_installed(
+    standard_config, monkeypatch
+):
+    monkeypatch.setattr("importlib.util.find_spec", lambda x: True)
+    monkeypatch.setattr("requests_cache.DO_NOT_CACHE", 1)
+    monkeypatch.setattr("requests_cache.NEVER_EXPIRE", 2)
+
+    standard_config.data = {
+        "cache": {
+            "enabled": True,
+            "urls": {
+                "url1": 10,
+                "url2": "NEVER_EXPIRE",
+                "url3": "DO_NOT_CACHE",
+                "url4": "unknown_value",
+            },
+        }
+    }
+
+    result = standard_config.cache_url_expirations()
+
+    expected_result = {
+        "url1": 10,
+        "url2": 2,  # Converted from "NEVER_EXPIRE"
+        "url3": 1,  # Converted from "DO_NOT_CACHE"
+    }
+
+    assert result == expected_result
+
+
+def test_cache_url_expirations_with_requests_cache_not_installed(
+    standard_config, monkeypatch
+):
+    monkeypatch.setattr("importlib.util.find_spec", lambda x: None)
+
+    standard_config.data = {
+        "cache": {
+            "enabled": True,
+            "urls": {
+                "url1": 10,
+                "url2": "NEVER_EXPIRE",
+                "url3": "DO_NOT_CACHE",
+                "url4": "unknown_value",
+            },
+        }
+    }
+
+    result = standard_config.cache_url_expirations()
+
+    assert result == {}
+
+
+def test_cache_url_expirations_with_cache_disabled(standard_config, monkeypatch):
+    monkeypatch.setattr("importlib.util.find_spec", lambda x: True)
+
+    standard_config.data = {
+        "cache": {
+            "enabled": False,
+            "urls": {
+                "url1": 10,
+                "url2": "NEVER_EXPIRE",
+                "url3": "DO_NOT_CACHE",
+                "url4": "unknown_value",
+            },
+        }
+    }
+
+    result = standard_config.cache_url_expirations()
+
+    assert result == {}