1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
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()
|
