chore: extend test set (#35)

* chore: add a few tests for light command

* chore: test all light commands

* chore: test unkown light mac

* chore: add battery:level tests

* chore: add light:level tests

* chore: add sensor:presence tests

* chore: add sensor:temperature tests

* chore: add system:* tests

* chore: cover all off api_factory

* chore: add tests for cached api

pyproject.toml

@@ -61,7 +61,7 @@ log_cli_date_format = "%Y-%m-%d %H:%M:%S"
 
 [tool.coverage.report]
 show_missing = true
-fail_under = 60
+fail_under = 85
 
 [tool.coverage.html]
 directory = "reports/coverage/html"

src/huemon/api/cached_api.py

@@ -7,18 +7,31 @@
 import os
 import tempfile
 import time
-from fcntl import LOCK_EX, LOCK_NB, flock
 from os.path import exists
 
 from huemon.api.api_interface import ApiInterface
 from huemon.infrastructure.logger_factory import create_logger
+from huemon.util import run_locked
 
 LOG = create_logger()
 
 DEFAULT_MAX_CACHE_AGE_SECONDS = 10
 DEFAULT_CACHE_PATH = tempfile.gettempdir()
 
 
+def cache_output_to_temp(cache_file_path, fn_call):
+    tmp_fd, tmp_file_path = tempfile.mkstemp()
+    with open(tmp_file_path, "w") as f_tmp:
+        f_tmp.write(json.dumps(fn_call()))
+
+    os.close(tmp_fd)
+
+    os.rename(tmp_file_path, cache_file_path)
+
+    with open(cache_file_path) as f_json:
+        return json.loads(f_json.read())
+
+
 class CachedApi(ApiInterface):
     def __init__(
         self,
@@ -59,23 +72,12 @@ def __cache(self, resource_type: str, fn_call):
                 )
                 return json.loads(f_json.read())
 
-        with open(lock_file, "w") as f_lock:
-            try:
-                flock(f_lock.fileno(), LOCK_EX | LOCK_NB)
-                LOG.debug("Acquired lock successfully (file=%s)", lock_file)
-
-                tmp_fd, tmp_file_path = tempfile.mkstemp()
-                with open(tmp_file_path, "w") as f_tmp:
-                    f_tmp.write(json.dumps(fn_call()))
-
-                os.close(tmp_fd)
-
-                os.rename(tmp_file_path, cache_file_path)
+        cached_result = run_locked(
+            lock_file, lambda: cache_output_to_temp(cache_file_path, fn_call)
+        )
 
-                with open(cache_file_path) as f_json:
-                    return json.loads(f_json.read())
-            except:  # pylint: disable=bare-except
-                LOG.debug("Failed to acquire lock, cache hit (file=%s)", lock_file)
+        if cached_result:
+            return cached_result
 
         if not does_cache_file_exist:
             return []

src/huemon/util.py

@@ -3,8 +3,11 @@
 # This source code is licensed under the MPL-2.0 license found in the
 # LICENSE file in the root directory of this source tree.
 
+import json
 import os
 import sys
+import tempfile
+from fcntl import LOCK_EX, LOCK_NB, flock
 from pathlib import Path
 
 from huemon.const import EXIT_FAIL
@@ -75,3 +78,29 @@ def assert_num_args(expected_number_of_arguments: int, arguments: list, context:
 def assert_exists(expected_values: list, value: str):
     if value not in expected_values:
         exit_fail("Received unknown value `%s` (expected=%s)", value, expected_values)
+
+
+def cache_output_to_temp(cache_file_path, fn_call):
+    tmp_fd, tmp_file_path = tempfile.mkstemp()
+    with open(tmp_file_path, "w") as f_tmp:
+        f_tmp.write(json.dumps(fn_call()))
+
+    os.close(tmp_fd)
+
+    os.rename(tmp_file_path, cache_file_path)
+
+    with open(cache_file_path) as f_json:
+        return json.loads(f_json.read())
+
+
+def run_locked(lock_file, fn_call):
+    with open(lock_file, "w") as f_lock:
+        try:
+            flock(f_lock.fileno(), LOCK_EX | LOCK_NB)
+            LOG.debug("Acquired lock successfully (file=%s)", lock_file)
+
+            return fn_call()
+        except:  # pylint: disable=bare-except
+            LOG.debug("Failed to acquire lock, cache hit (file=%s)", lock_file)
+
+    return None

src/tests/fixtures.py

@@ -20,6 +20,12 @@ def __generate_version():
     return str(time.process_time())
 
 
+def read_result(mock_print):
+    (printed_result, *_) = mock_print.call_args.args
+
+    return printed_result
+
+
 def create_system_config(version: str = None, is_update_available: bool = False):
     return {
         FIELD_SYSTEM_SWUPDATE2: {

src/tests/test_api_configuration.py

@@ -0,0 +1,31 @@
+# Copyright (c) Ely Deckers.
+#
+# This source code is licensed under the MPL-2.0 license found in the
+# LICENSE file in the root directory of this source tree.
+
+import unittest
+
+from huemon.api.api import Api
+from huemon.api.api_factory import create_api, create_hue_hub_url
+from huemon.api.cached_api import CachedApi
+
+
+class TestApiConfiguration(unittest.TestCase):
+    def test_when_no_api_configuration_available_throw(self):
+        with self.assertRaises(Exception):
+            create_hue_hub_url({})
+
+    def test_when_cache_enabled_return_cached(self):
+        api = create_api(
+            {"ip": "IRRELEVANT_IP", "key": "IRRELEVANT_KEY", "cache": {"enable": True}}
+        )
+
+        self.assertIsInstance(api, CachedApi)
+
+    def test_when_cache_enabled_return_regular(self):
+        api = create_api(
+            {"ip": "IRRELEVANT_IP", "key": "IRRELEVANT_KEY", "cache": {"enable": False}}
+        )
+
+        self.assertNotIsInstance(api, CachedApi)
+        self.assertIsInstance(api, Api)

src/tests/test_command_handler.py

@@ -12,7 +12,7 @@
 )
 from huemon.commands_available.system_command import SystemCommand
 from huemon.const import EXIT_FAIL
-from tests.fixtures import MutableApi, create_system_config
+from tests.fixtures import MutableApi, create_system_config, read_result
 
 CACHE_VALIDITY_INFINITE_SECONDS = 1_000_000
 CACHE_VALIDITY_ZERO_SECONDS = 0
@@ -55,7 +55,9 @@ def test_when_cache_not_expired_return_cache(self, mock_print: MagicMock):
 
         command_handler.exec("system", ["version"])
 
-        self.assertTrue(mock_print.call_args.__eq__(some_version))
+        system_version = read_result(mock_print)
+
+        self.assertEqual(some_version, system_version)
 
     def test_when_unknown_command_received_system_exit_is_called(self):
         command_handler = CommandHandler([])

src/tests/test_light_command.py

@@ -0,0 +1,156 @@
+# Copyright (c) Ely Deckers.
+#
+# This source code is licensed under the MPL-2.0 license found in the
+# LICENSE file in the root directory of this source tree.
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from huemon.commands.command_handler import (
+    CommandHandler,
+    create_name_to_command_mapping,
+)
+from huemon.commands_available.light_command import LightCommand
+from tests.fixtures import MutableApi, read_result
+
+CACHE_VALIDITY_INFINITE_SECONDS = 1_000_000
+CACHE_VALIDITY_ZERO_SECONDS = 0
+
+
+SOME_LIGHT_MAC_0 = "SO:ME:LI:GH:TM:AC:00"
+SOME_LIGHT_MAC_1 = "SO:ME:LI:GH:TM:AC:01"
+
+
+class TestLightCommand(unittest.TestCase):
+    def test_when_light_doesnt_exist(self):
+        mutable_api = MutableApi()
+        mutable_api.set_lights([])
+
+        command_handler = CommandHandler(
+            create_name_to_command_mapping({}, mutable_api, [LightCommand])
+        )
+
+        with self.assertRaises(Exception):
+            command_handler.exec("light", [SOME_LIGHT_MAC_0, "status"])
+
+    @patch("builtins.print")
+    def test_when_light_exists_return_status(self, mock_print: MagicMock):
+        mutable_api = MutableApi()
+        mutable_api.set_lights(
+            [
+                {
+                    "uniqueid": SOME_LIGHT_MAC_0,
+                    "state": {
+                        "on": 1,
+                    },
+                },
+                {
+                    "uniqueid": SOME_LIGHT_MAC_1,
+                    "state": {
+                        "on": 0,
+                    },
+                },
+            ]
+        )
+
+        command_handler = CommandHandler(
+            create_name_to_command_mapping({}, mutable_api, [LightCommand])
+        )
+
+        command_handler.exec("light", [SOME_LIGHT_MAC_0, "status"])
+        state_light_0 = read_result(mock_print)
+        command_handler.exec("light", [SOME_LIGHT_MAC_1, "status"])
+        state_light_1 = read_result(mock_print)
+
+        self.assertEqual(1, state_light_0)
+        self.assertEqual(0, state_light_1)
+
+    @patch("builtins.print")
+    def test_when_light_exists_return_is_upgrade_available(self, mock_print: MagicMock):
+        mutable_api = MutableApi()
+        mutable_api.set_lights(
+            [
+                {
+                    "uniqueid": SOME_LIGHT_MAC_0,
+                    "swupdate": {
+                        "state": "noupdates",
+                    },
+                },
+                {
+                    "uniqueid": SOME_LIGHT_MAC_1,
+                    "swupdate": {
+                        "state": "update-available",
+                    },
+                },
+            ]
+        )
+
+        command_handler = CommandHandler(
+            create_name_to_command_mapping({}, mutable_api, [LightCommand])
+        )
+
+        command_handler.exec("light", [SOME_LIGHT_MAC_0, "is_upgrade_available"])
+        state_light_0 = read_result(mock_print)
+        command_handler.exec("light", [SOME_LIGHT_MAC_1, "is_upgrade_available"])
+        state_light_1 = read_result(mock_print)
+
+        self.assertEqual(0, state_light_0)
+        self.assertEqual(1, state_light_1)
+
+    @patch("builtins.print")
+    def test_when_light_exists_return_is_reachable(self, mock_print: MagicMock):
+        mutable_api = MutableApi()
+        mutable_api.set_lights(
+            [
+                {"uniqueid": SOME_LIGHT_MAC_0, "state": {"reachable": 0}},
+                {
+                    "uniqueid": SOME_LIGHT_MAC_1,
+                    "state": {
+                        "reachable": 1,
+                    },
+                },
+            ]
+        )
+
+        command_handler = CommandHandler(
+            create_name_to_command_mapping({}, mutable_api, [LightCommand])
+        )
+
+        command_handler.exec("light", [SOME_LIGHT_MAC_0, "reachable"])
+        state_light_0 = read_result(mock_print)
+        command_handler.exec("light", [SOME_LIGHT_MAC_1, "reachable"])
+        state_light_1 = read_result(mock_print)
+
+        self.assertEqual(0, state_light_0)
+        self.assertEqual(1, state_light_1)
+
+    @patch("builtins.print")
+    def test_when_light_exists_return_version(self, mock_print: MagicMock):
+        some_version_0 = "some_version_0"
+        some_version_1 = "some_version_1"
+
+        mutable_api = MutableApi()
+        mutable_api.set_lights(
+            [
+                {
+                    "uniqueid": SOME_LIGHT_MAC_0,
+                    "swversion": some_version_0,
+                },
+                {
+                    "uniqueid": SOME_LIGHT_MAC_1,
+                    "swversion": some_version_1,
+                },
+            ]
+        )
+
+        command_handler = CommandHandler(
+            create_name_to_command_mapping({}, mutable_api, [LightCommand])
+        )
+
+        command_handler.exec("light", [SOME_LIGHT_MAC_0, "version"])
+        state_light_0 = read_result(mock_print)
+        command_handler.exec("light", [SOME_LIGHT_MAC_1, "version"])
+        state_light_1 = read_result(mock_print)
+
+        self.assertEqual(some_version_0, state_light_0)
+        self.assertEqual(some_version_1, state_light_1)