core/tests/components/anova/conftest.py

"""Common fixtures for Anova."""

from __future__ import annotations

import asyncio
from dataclasses import dataclass
import json
from typing import Any
from unittest.mock import AsyncMock, patch

from aiohttp import ClientSession
from anova_wifi import (
    AnovaApi,
    AnovaWebsocketHandler,
    InvalidLogin,
    NoDevicesFound,
    WebsocketFailure,
)
import pytest

from homeassistant.core import HomeAssistant

DUMMY_ID = "anova_id"


@dataclass
class MockedanovaWebsocketMessage:
    """Mock the websocket message for Anova."""

    input_data: dict[str, Any]
    data: str = ""

    def __post_init__(self) -> None:
        """Set up data after creation."""
        self.data = json.dumps(self.input_data)


class MockedAnovaWebsocketStream:
    """Mock the websocket stream for Anova."""

    def __init__(self, messages: list[MockedanovaWebsocketMessage]) -> None:
        """Initialize a Anova Websocket Stream that can be manipulated for tests."""
        self.messages = messages

    def __aiter__(self) -> MockedAnovaWebsocketStream:
        """Handle async iteration."""
        return self

    async def __anext__(self) -> MockedanovaWebsocketMessage:
        """Get the next message in the websocket stream."""
        if self.messages:
            return self.messages.pop(0)
        raise StopAsyncIteration

    def clear(self) -> None:
        """Clear the Websocket stream."""
        self.messages.clear()


class MockedAnovaWebsocketHandler(AnovaWebsocketHandler):
    """Mock the Anova websocket handler."""

    def __init__(
        self,
        firebase_jwt: str,
        jwt: str,
        session: ClientSession,
        connect_messages: list[MockedanovaWebsocketMessage],
        post_connect_messages: list[MockedanovaWebsocketMessage],
    ) -> None:
        """Initialize the websocket handler with whatever messages you want."""
        super().__init__(firebase_jwt, jwt, session)
        self.connect_messages = connect_messages
        self.post_connect_messages = post_connect_messages

    async def connect(self) -> None:
        """Create a future for the message listener."""
        self.ws = MockedAnovaWebsocketStream(self.connect_messages)
        await self.message_listener()
        self.ws = MockedAnovaWebsocketStream(self.post_connect_messages)
        # RUF006 ignored as it replicates the parent library
        # https://github.com/Lash-L/anova_wifi/issues/35
        asyncio.ensure_future(self.message_listener())  # noqa: RUF006


def anova_api_mock(
    connect_messages: list[MockedanovaWebsocketMessage] | None = None,
    post_connect_messages: list[MockedanovaWebsocketMessage] | None = None,
) -> AsyncMock:
    """Mock the api for Anova."""
    api_mock = AsyncMock()

    async def authenticate_side_effect() -> None:
        api_mock.jwt = "my_test_jwt"
        api_mock._firebase_jwt = "my_test_firebase_jwt"

    async def create_websocket_side_effect() -> None:
        api_mock.websocket_handler = MockedAnovaWebsocketHandler(
            firebase_jwt=api_mock._firebase_jwt,
            jwt=api_mock.jwt,
            session=AsyncMock(),
            connect_messages=connect_messages
            if connect_messages is not None
            else [
                MockedanovaWebsocketMessage(
                    {
                        "command": "EVENT_APC_WIFI_LIST",
                        "payload": [
                            {
                                "cookerId": DUMMY_ID,
                                "type": "a5",
                                "pairedAt": "2023-08-12T02:33:20.917716Z",
                                "name": "Anova Precision Cooker",
                            }
                        ],
                    }
                ),
            ],
            post_connect_messages=post_connect_messages
            if post_connect_messages is not None
            else [
                MockedanovaWebsocketMessage(
                    {
                        "command": "EVENT_APC_STATE",
                        "payload": {
                            "cookerId": DUMMY_ID,
                            "state": {
                                "boot-id": "8620610049456548422",
                                "job": {
                                    "cook-time-seconds": 0,
                                    "id": "8759286e3125b0c547",
                                    "mode": "IDLE",
                                    "ota-url": "",
                                    "target-temperature": 54.72,
                                    "temperature-unit": "F",
                                },
                                "job-status": {
                                    "cook-time-remaining": 0,
                                    "job-start-systick": 599679,
                                    "provisioning-pairing-code": 7514,
                                    "state": "",
                                    "state-change-systick": 599679,
                                },
                                "pin-info": {
                                    "device-safe": 0,
                                    "water-leak": 0,
                                    "water-level-critical": 0,
                                    "water-temp-too-high": 0,
                                },
                                "system-info": {
                                    "class": "A5",
                                    "firmware-version": "2.2.0",
                                    "type": "RA2L1-128",
                                },
                                "system-info-details": {
                                    "firmware-version-raw": "VM178_A_02.02.00_MKE15-128",
                                    "systick": 607026,
                                    "version-string": "VM171_A_02.02.00 RA2L1-128",
                                },
                                "temperature-info": {
                                    "heater-temperature": 22.37,
                                    "triac-temperature": 36.04,
                                    "water-temperature": 18.33,
                                },
                            },
                        },
                    }
                ),
            ],
        )
        await api_mock.websocket_handler.connect()
        if not api_mock.websocket_handler.devices:
            raise NoDevicesFound("No devices were found on the websocket.")

    api_mock.authenticate.side_effect = authenticate_side_effect
    api_mock.create_websocket.side_effect = create_websocket_side_effect
    return api_mock


@pytest.fixture
async def anova_api(
    hass: HomeAssistant,
) -> AnovaApi:
    """Mock the api for Anova."""
    api_mock = anova_api_mock()

    with (
        patch("homeassistant.components.anova.AnovaApi", return_value=api_mock),
        patch(
            "homeassistant.components.anova.config_flow.AnovaApi", return_value=api_mock
        ),
    ):
        api = AnovaApi(
            None,
            "sample@gmail.com",
            "sample",
        )
        yield api


@pytest.fixture
async def anova_api_no_devices(
    hass: HomeAssistant,
) -> AnovaApi:
    """Mock the api for Anova with no online devices."""
    api_mock = anova_api_mock(connect_messages=[], post_connect_messages=[])

    with (
        patch("homeassistant.components.anova.AnovaApi", return_value=api_mock),
        patch(
            "homeassistant.components.anova.config_flow.AnovaApi", return_value=api_mock
        ),
    ):
        api = AnovaApi(
            None,
            "sample@gmail.com",
            "sample",
        )
        yield api


@pytest.fixture
async def anova_api_wrong_login(
    hass: HomeAssistant,
) -> AnovaApi:
    """Mock the api for Anova with a wrong login."""
    api_mock = anova_api_mock()

    async def authenticate_side_effect():
        raise InvalidLogin

    api_mock.authenticate.side_effect = authenticate_side_effect

    with patch("homeassistant.components.anova.AnovaApi", return_value=api_mock):
        api = AnovaApi(
            None,
            "sample@gmail.com",
            "sample",
        )
        yield api


@pytest.fixture
async def anova_api_no_data(
    hass: HomeAssistant,
) -> AnovaApi:
    """Mock the api for Anova with a wrong login."""
    api_mock = anova_api_mock(post_connect_messages=[])

    with patch("homeassistant.components.anova.AnovaApi", return_value=api_mock):
        api = AnovaApi(
            None,
            "sample@gmail.com",
            "sample",
        )
        yield api


@pytest.fixture
async def anova_api_websocket_failure(
    hass: HomeAssistant,
) -> AnovaApi:
    """Mock the api for Anova with a websocket failure."""
    api_mock = anova_api_mock()

    async def create_websocket_side_effect():
        raise WebsocketFailure

    api_mock.create_websocket.side_effect = create_websocket_side_effect

    with patch("homeassistant.components.anova.AnovaApi", return_value=api_mock):
        api = AnovaApi(
            None,
            "sample@gmail.com",
            "sample",
        )
        yield api