diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json index e7cf06f..b710cbb 100644 --- a/.devcontainer/devcontainer.json +++ b/.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", + + ] } \ No newline at end of file diff --git a/gazpar2haws/haws.py b/gazpar2haws/haws.py index 57c3d5d..5b4f4a8 100644 --- a/gazpar2haws/haws.py +++ b/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"] diff --git a/tests/test_haws.py b/tests/test_haws.py index b90ab46..b7f19ab 100644 --- a/tests/test_haws.py +++ b/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 \ No newline at end of file