Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Test base and button resource controller #419

Open
wants to merge 11 commits into
base: main
Choose a base branch
from
5 changes: 5 additions & 0 deletions aiohue/util.py
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,11 @@ def parse_utc_timestamp(datetimestr: str):
return datetime.fromisoformat(datetimestr.replace("Z", "+00:00"))


def format_utc_timestamp(time: datetime):
"""Format datetime to string."""
return time.strftime("%Y-%m-%dT%H:%M:%S.%fZ")


def _parse_value(name: str, value: Any, value_type: Any, default: Any = MISSING) -> Any:
"""Try to parse a value from raw (json) data and type annotations."""
# ruff: noqa: PLR0911, PLR0912
Expand Down
3 changes: 3 additions & 0 deletions aiohue/v2/controllers/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
import asyncio
from asyncio.coroutines import iscoroutinefunction
from collections.abc import Callable, Iterator

# pylint: disable=no-name-in-module
from typing import (
TYPE_CHECKING,
Any,
Expand Down Expand Up @@ -228,6 +230,7 @@ async def _handle_event(
# do not forward update event at reconnects if no keys were updated
if len(updated_keys) == 0 and is_reconnect:
return

# Do not forward update events for button resource if
# the button feature is missing in event data in an attempt to prevent
# ghost events at bridge reboots/firmware updates.
Expand Down
112 changes: 112 additions & 0 deletions tests/v2/test_base_resource_controller.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
"""Test base controller functions."""

from dataclasses import dataclass
from unittest.mock import Mock
from uuid import uuid4

from aiohue import HueBridgeV2
from aiohue.v2 import EventType
from aiohue.v2.controllers.base import BaseResourcesController
from aiohue.v2.models.resource import ResourceTypes


@dataclass
class MockData:
"""Mock data resource type."""

id: str
type: ResourceTypes = ResourceTypes.UNKNOWN

id_v1: str | None = None


class MockController(BaseResourcesController[type[MockData]]):
"""Controller for mock data resource."""

item_type = ResourceTypes.UNKNOWN
item_cls = MockData
allow_parser_error = False


async def test_handle_event():
"""Test handling of events."""
bridge = HueBridgeV2("127.0.0.1", "fake")
controller = MockController(bridge)
callback = Mock(return_value=None)
controller.subscribe(callback)

resource_id = str(uuid4())
other_id = str(uuid4())

evt_data = {
"id": resource_id,
"id_v1": "mock/1",
}

# Create a new resource
# pylint: disable=protected-access
await controller._handle_event(EventType.RESOURCE_ADDED, evt_data)

cur_data = MockData(
id=resource_id,
type=ResourceTypes.UNKNOWN,
id_v1="mock/1",
)
callback.assert_called_once_with(EventType.RESOURCE_ADDED, cur_data)
callback.reset_mock()

evt_data = {
"id": resource_id,
"id_v1": "mock/2",
}

# Update of a single property of an existing resource
# pylint: disable=protected-access
await controller._handle_event(EventType.RESOURCE_UPDATED, evt_data)

cur_data = MockData(
id=resource_id,
type=ResourceTypes.UNKNOWN,
id_v1="mock/2",
)
callback.assert_called_once_with(EventType.RESOURCE_UPDATED, cur_data)
callback.reset_mock()

evt_data = {
"id": other_id,
"id_v1": "mock/1",
}

# Update of a single property of a non-existing resource
# pylint: disable=protected-access
await controller._handle_event(EventType.RESOURCE_UPDATED, evt_data)

callback.assert_not_called()
callback.reset_mock()

evt_data = {
"id": resource_id,
}

# Remove of existing resource
# pylint: disable=protected-access
await controller._handle_event(EventType.RESOURCE_DELETED, evt_data)

cur_data = MockData(
id=resource_id,
type=ResourceTypes.UNKNOWN,
id_v1="mock/2",
)
callback.assert_called_once_with(EventType.RESOURCE_DELETED, cur_data)
callback.reset_mock()

evt_data = {
"id": resource_id,
"id_v1": "mock/2",
}

# Update of an already removed resource
# pylint: disable=protected-access
await controller._handle_event(EventType.RESOURCE_UPDATED, evt_data)

callback.assert_not_called()
Loading
Loading