[#27] In a Github workflow, run unit tests inside a HA DevContainer

.devcontainer/devcontainer.json

@@ -24,5 +24,15 @@
         }
       }
     },
-    "mounts": [ "type=volume,target=/var/lib/docker" ]
+    "mounts": [
+      "type=volume,target=/var/lib/docker",
+
+      // Custom configuration directory
+      //"source=C:\\Users\\steph\\Qsync\\Workspace\\Python\\gazpar2haws\\mount\\config,target=${containerWorkspaceFolder}/config,type=bind"
+      // Custom component
+      //"source=${localEnv:HOME}/custom_repo/custom_components/component_name,target=${containerWorkspaceFolder}/config/custom_components/component_name,type=bind",
+      //Custom library
+      //"source=${localEnv:HOME}/path/to/custom_repo/custom_libraries/library_name,target=${containerWorkspaceFolder}/config/custom_libraries/library_name,type=bind",
+
+    ]
   }

gazpar2haws/haws.py

@@ -190,3 +190,35 @@ async def clear_statistics(self, entity_ids: list[str]):
         await self.send_message(clear_statistics_message)
 
         Logger.debug(f"Cleared {entity_ids} statistics")
+
+    # ----------------------------------
+    async def create_token(self, url: str, key: str) -> str:
+
+        socket_connection = await websockets.connect(url)
+
+        # socket_connection = websockets.create_connection("ws://localhost:YOUR-PORT/api/websocket".format(url), sslopt={'cert_reqs': ssl.CERT_NONE})
+        socket_response = await socket_connection.recv()
+        result = json.loads(socket_response)
+
+        # Check if auth required, if so send password
+        auth = None
+        if result["type"] == "auth_required":
+            if key is not None:
+                auth = json.dumps({
+                    "type": "auth",
+                    "api_password": key
+                })
+                await socket_connection.send(auth)
+                auth_repsonse = await socket_connection.recv()
+
+        create_token = json.dumps({
+            "id": 11,
+            "type": "auth/long_lived_access_token",
+            "client_name": "appdaemon",
+            "lifespan": 365
+            })
+        await socket_connection.send(create_token)
+        token_response = await socket_connection.recv()
+        q = json.loads(token_response)
+
+        return q["result"]

tests/test_haws.py

@@ -1,172 +1,148 @@
 """Test haws module."""
 import pytest
 from gazpar2haws.haws import HomeAssistantWS
-from gazpar2haws import config_utils
 
 # See WebSocket source code here: https://git.informatik.uni-kl.de/s_menne19/hassio-core/-/blob/fix-tests-assist/homeassistant/components/recorder/websocket_api.py
 
 
-# ----------------------------------
-@pytest.mark.skip(reason="Requires Home Assistant server")
-@pytest.mark.asyncio
-async def test_connect():
+class TestHomeAssistantWS:
 
-    # Load configuration
-    config = config_utils.ConfigLoader("config/configuration.yaml", "config/secrets.yaml")
-    config.load_secrets()
-    config.load_config()
+    @classmethod
+    def setup_class(cls):
+        """ setup any state specific to the execution of the given class (which
+        usually contains tests).
+        """
 
-    ha_host = config.get("homeassistant.host")
-    ha_port = config.get("homeassistant.port")
-    ha_endpoint = config.get("homeassistant.endpoint")
-    ha_token = config.get("homeassistant.token")
+    @classmethod
+    def teardown_class(cls):
+        """ teardown any state that was previously setup with a call to
+        setup_class.
+        """
 
-    haws = HomeAssistantWS(ha_host, ha_port, ha_endpoint, ha_token)
+    def setup_method(self):
+        """ setup any state tied to the execution of the given method in a
+        class.  setup_method is invoked for every test method of a class.
+        """
+        self._ha_host = "localhost"
+        self._ha_port = 7123
+        self._ha_endpoint = "/api/websocket"
+        self._ha_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJjNTI5N2JmN2U1ZjE0MWRmYWVmNzE4NWRiOTQyYmM3NyIsImlhdCI6MTczNjAxMzA3NiwiZXhwIjoyMDUxMzczMDc2fQ.zsoLmKM1e7CzHvUxbYNTcMYmafTxY9699PFMnOMR0rY"
 
-    await haws.connect()
+    def teardown_method(self):
+        """ teardown any state that was previously setup with a setup_method
+        call.
+        """
 
-    await haws.disconnect()
+    # ----------------------------------
+    # @pytest.mark.skip(reason="Requires Home Assistant server")
+    @pytest.mark.asyncio
+    async def test_connect(self):
 
+        haws = HomeAssistantWS(self._ha_host, self._ha_port, self._ha_endpoint, self._ha_token)
 
-# ----------------------------------
-@pytest.mark.skip(reason="Requires Home Assistant server")
-@pytest.mark.asyncio
-async def test_list_statistic_ids():
+        await haws.connect()
 
-    # Load configuration
-    config = config_utils.ConfigLoader("config/configuration.yaml", "config/secrets.yaml")
-    config.load_secrets()
-    config.load_config()
+        await haws.disconnect()
 
-    ha_host = config.get("homeassistant.host")
-    ha_port = config.get("homeassistant.port")
-    ha_endpoint = config.get("homeassistant.endpoint")
-    ha_token = config.get("homeassistant.token")
+    # ----------------------------------
+    # @pytest.mark.skip(reason="Requires Home Assistant server")
+    @pytest.mark.asyncio
+    async def test_list_statistic_ids(self):
 
-    haws = HomeAssistantWS(ha_host, ha_port, ha_endpoint, ha_token)
+        haws = HomeAssistantWS(self._ha_host, self._ha_port, self._ha_endpoint, self._ha_token)
 
-    await haws.connect()
+        await haws.connect()
 
-    statistics = await haws.list_statistic_ids("sum")
+        statistics = await haws.list_statistic_ids("sum")
 
-    assert statistics is not None
+        assert statistics is not None
 
-    await haws.disconnect()
+        await haws.disconnect()
 
+    # ----------------------------------
+    # @pytest.mark.skip(reason="Requires Home Assistant server")
+    @pytest.mark.asyncio
+    async def test_exists_statistic_id(self):
 
-# ----------------------------------
-@pytest.mark.skip(reason="Requires Home Assistant server")
-@pytest.mark.asyncio
-async def test_exists_statistic_id():
+        haws = HomeAssistantWS(self._ha_host, self._ha_port, self._ha_endpoint, self._ha_token)
 
-    # Load configuration
-    config = config_utils.ConfigLoader("config/configuration.yaml", "config/secrets.yaml")
-    config.load_secrets()
-    config.load_config()
+        await haws.connect()
 
-    ha_host = config.get("homeassistant.host")
-    ha_port = config.get("homeassistant.port")
-    ha_endpoint = config.get("homeassistant.endpoint")
-    ha_token = config.get("homeassistant.token")
+        exists_statistic_id = await haws.exists_statistic_id("sensor.gazpar2haws_volume")
 
-    haws = HomeAssistantWS(ha_host, ha_port, ha_endpoint, ha_token)
+        assert exists_statistic_id is not None
 
-    await haws.connect()
+        await haws.disconnect()
 
-    exists_statistic_id = await haws.exists_statistic_id("sensor.gazpar2haws_volume")
+    # ----------------------------------
+    # @pytest.mark.skip(reason="Requires Home Assistant server")
+    @pytest.mark.asyncio
+    async def test_get_last_statistic(self):
 
-    assert exists_statistic_id is not None
+        haws = HomeAssistantWS(self._ha_host, self._ha_port, self._ha_endpoint, self._ha_token)
 
-    await haws.disconnect()
+        await haws.connect()
 
+        statistics = await haws.get_last_statistic("sensor.gazpar2haws_volume")
 
-# ----------------------------------
-@pytest.mark.skip(reason="Requires Home Assistant server")
-@pytest.mark.asyncio
-async def test_get_last_statistic():
+        assert statistics is not None
 
-    # Load configuration
-    config = config_utils.ConfigLoader("config/configuration.yaml", "config/secrets.yaml")
-    config.load_secrets()
-    config.load_config()
+        await haws.disconnect()
 
-    ha_host = config.get("homeassistant.host")
-    ha_port = config.get("homeassistant.port")
-    ha_endpoint = config.get("homeassistant.endpoint")
-    ha_token = config.get("homeassistant.token")
+    # ----------------------------------
+    # @pytest.mark.skip(reason="Requires Home Assistant server")
+    @pytest.mark.asyncio
+    async def test_import_statistics(self):
 
-    haws = HomeAssistantWS(ha_host, ha_port, ha_endpoint, ha_token)
+        haws = HomeAssistantWS(self._ha_host, self._ha_port, self._ha_endpoint, self._ha_token)
 
-    await haws.connect()
+        await haws.connect()
 
-    statistics = await haws.get_last_statistic("sensor.gazpar2haws_volume")
+        statistics = [
+            {
+                "start": "2024-12-14T00:00:00+00:00",
+                "state": 100.0,
+                "sum": 100.0
+            },
+            {
+                "start": "2024-12-15T00:00:00+00:00",
+                "state": 200.0,
+                "sum": 200.0
+            },
+            {
+                "start": "2024-12-16T00:00:00+00:00",
+                "state": 300.0,
+                "sum": 300.0
+            }
+        ]
 
-    assert statistics is not None
+        await haws.import_statistics("sensor.gazpar2haws_volume", "recorder", "test", "m³", statistics)
 
-    await haws.disconnect()
+        await haws.disconnect()
 
+    # ----------------------------------
+    # @pytest.mark.skip(reason="Requires Home Assistant server")
+    @pytest.mark.asyncio
+    async def test_clear_statistics(self):
 
-# ----------------------------------
-@pytest.mark.skip(reason="Requires Home Assistant server")
-@pytest.mark.asyncio
-async def test_import_statistics():
+        haws = HomeAssistantWS(self._ha_host, self._ha_port, self._ha_endpoint, self._ha_token)
 
-    # Load configuration
-    config = config_utils.ConfigLoader("config/configuration.yaml", "config/secrets.yaml")
-    config.load_secrets()
-    config.load_config()
+        await haws.connect()
 
-    ha_host = config.get("homeassistant.host")
-    ha_port = config.get("homeassistant.port")
-    ha_endpoint = config.get("homeassistant.endpoint")
-    ha_token = config.get("homeassistant.token")
+        await haws.clear_statistics(["sensor.gazpar2haws_volume"])
 
-    haws = HomeAssistantWS(ha_host, ha_port, ha_endpoint, ha_token)
+        await haws.disconnect()
 
-    await haws.connect()
+    # ----------------------------------
+    @pytest.mark.asyncio
+    async def test_create_token(self):
 
-    statistics = [
-        {
-            "start": "2024-12-14T00:00:00+00:00",
-            "state": 100.0,
-            "sum": 100.0
-        },
-        {
-            "start": "2024-12-15T00:00:00+00:00",
-            "state": 200.0,
-            "sum": 200.0
-        },
-        {
-            "start": "2024-12-16T00:00:00+00:00",
-            "state": 300.0,
-            "sum": 300.0
-        }
-    ]
+        # Mount HA config: https://community.home-assistant.io/t/developing-home-assistant-core-in-a-vscode-devcontainer/235650
 
-    await haws.import_statistics("sensor.gazpar2haws_volume", "recorder", "test", "m³", statistics)
+        # Configure API passwordin config: https://github.com/home-assistant/core/issues/25952
 
-    await haws.disconnect()
+        haws = HomeAssistantWS(self._ha_host, self._ha_port, self._ha_endpoint, self._ha_token)
 
+        token = await haws.create_token("ws://localhost:7123/api/websocket", "abc123")        
 
-# ----------------------------------
-@pytest.mark.skip(reason="Requires Home Assistant server")
-@pytest.mark.asyncio
-async def test_clear_statistics():
-
-    # Load configuration
-    config = config_utils.ConfigLoader("config/configuration.yaml", "config/secrets.yaml")
-    config.load_secrets()
-    config.load_config()
-
-    ha_host = config.get("homeassistant.host")
-    ha_port = config.get("homeassistant.port")
-    ha_endpoint = config.get("homeassistant.endpoint")
-    ha_token = config.get("homeassistant.token")
-
-    haws = HomeAssistantWS(ha_host, ha_port, ha_endpoint, ha_token)
-
-    await haws.connect()
-
-    await haws.clear_statistics(["sensor.gazpar2haws_volume"])
-
-    await haws.disconnect()
+        assert token is not None