File: test_barrier.py

package info (click to toggle)
python-nice-go 1.0.1-2
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 608 kB
  • sloc: python: 1,981; makefile: 3
file content (156 lines) | stat: -rw-r--r-- 4,091 bytes parent folder | download | duplicates (2)
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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
from unittest.mock import AsyncMock, MagicMock

import pytest

from nice_go import Barrier
from nice_go._barrier import BarrierState


async def test_open() -> None:
    barrier = Barrier(
        "barrier_id",
        "barrier_type",
        "control_level",
        [],
        BarrierState(
            deviceId="device_id",
            reported={"key": "value"},
            timestamp="timestamp",
            version="version",
            connectionState=None,
        ),
        MagicMock(open_barrier=AsyncMock(return_value=True)),
    )
    assert await barrier.open() is True


async def test_close() -> None:
    barrier = Barrier(
        "barrier_id",
        "barrier_type",
        "control_level",
        [],
        BarrierState(
            deviceId="device_id",
            reported={"key": "value"},
            timestamp="timestamp",
            version="version",
            connectionState=None,
        ),
        MagicMock(close_barrier=AsyncMock(return_value=True)),
    )
    assert await barrier.close() is True


async def test_light_on() -> None:
    barrier = Barrier(
        "barrier_id",
        "barrier_type",
        "control_level",
        [],
        BarrierState(
            deviceId="device_id",
            reported={"key": "value"},
            timestamp="timestamp",
            version="version",
            connectionState=None,
        ),
        MagicMock(light_on=AsyncMock(return_value=True)),
    )
    assert await barrier.light_on() is True


async def test_light_off() -> None:
    barrier = Barrier(
        "barrier_id",
        "barrier_type",
        "control_level",
        [],
        BarrierState(
            deviceId="device_id",
            reported={"key": "value"},
            timestamp="timestamp",
            version="version",
            connectionState=None,
        ),
        MagicMock(light_off=AsyncMock(return_value=True)),
    )
    assert await barrier.light_off() is True


async def test_get_attr() -> None:
    barrier = Barrier(
        "barrier_id",
        "barrier_type",
        "control_level",
        [{"key": "key", "value": "value"}],
        BarrierState(
            deviceId="device_id",
            reported={"key": "value"},
            timestamp="timestamp",
            version="version",
            connectionState=None,
        ),
        MagicMock(),
    )
    assert await barrier.get_attr("key") == "value"


async def test_get_attr_not_found() -> None:
    barrier = Barrier(
        "barrier_id",
        "barrier_type",
        "control_level",
        [{"key": "key", "value": "value"}],
        BarrierState(
            deviceId="device_id",
            reported={"key": "value"},
            timestamp="timestamp",
            version="version",
            connectionState=None,
        ),
        MagicMock(),
    )
    with pytest.raises(KeyError) as exc_info:
        await barrier.get_attr("not_found")
    assert str(exc_info.value) == "'Attribute with key not_found not found.'"


async def test_vacation_mode_on() -> None:
    mock_api = MagicMock(vacation_mode_on=AsyncMock())
    barrier = Barrier(
        "barrier_id",
        "barrier_type",
        "control_level",
        [],
        BarrierState(
            deviceId="device_id",
            reported={"key": "value"},
            timestamp="timestamp",
            version="version",
            connectionState=None,
        ),
        mock_api,
    )
    await barrier.vacation_mode_on()
    mock_api.vacation_mode_on.assert_called_once_with("barrier_id")


async def test_vacation_mode_off() -> None:
    mock_api = MagicMock(vacation_mode_off=AsyncMock())
    barrier = Barrier(
        "barrier_id",
        "barrier_type",
        "control_level",
        [],
        BarrierState(
            deviceId="device_id",
            reported={"key": "value"},
            timestamp="timestamp",
            version="version",
            connectionState=None,
        ),
        mock_api,
    )
    await barrier.vacation_mode_off()
    mock_api.vacation_mode_off.assert_called_once_with("barrier_id")