1
|
|
|
import pytest |
2
|
|
|
import os, struct, copy |
3
|
|
|
import logging |
4
|
|
|
from asyncio import coroutine |
5
|
|
|
from curio import kernel, sleep |
6
|
|
|
|
7
|
|
|
from mock import Mock |
8
|
|
|
from mock import patch, call |
9
|
|
|
from mock import MagicMock |
10
|
|
|
from mock import PropertyMock |
11
|
|
|
|
12
|
|
|
from hypothesis import given, example, settings |
13
|
|
|
from hypothesis import strategies as st |
14
|
|
|
|
15
|
|
|
from bricknil.sensor.light import * |
16
|
|
|
from bricknil.sensor.motor import * |
17
|
|
|
from bricknil.const import Color |
18
|
|
|
|
19
|
|
|
class AsyncMock(MagicMock): |
20
|
|
|
async def __call__(self, *args, **kwargs): |
21
|
|
|
return super(AsyncMock, self).__call__(*args, **kwargs) |
22
|
|
|
|
23
|
|
|
class DirectWrite: |
24
|
|
|
def get_bytes(self, port, mode, value): |
25
|
|
|
return [0x00, 0x81, port, 0x01, 0x51, mode, value ] |
26
|
|
|
|
27
|
|
|
def get_bytes_for_set_pos(self, port, pos, speed, max_power): |
28
|
|
|
abs_pos = list(struct.pack('i', pos)) |
29
|
|
|
return [0x00, 0x81, port, 0x01, 0x0d] + abs_pos + [speed, max_power, 126, 3] |
30
|
|
|
|
31
|
|
|
def get_bytes_for_rotate(self, port, angle, speed, max_power): |
32
|
|
|
angle = list(struct.pack('i',angle)) |
33
|
|
|
return [0x00, 0x81, port, 0x01, 0x0b] + angle + [speed, max_power, 126, 3] |
34
|
|
|
|
35
|
|
|
class TestLED: |
36
|
|
|
|
37
|
|
|
def setup(self): |
38
|
|
|
self.l = LED(name='led') |
39
|
|
|
self.l.send_message = Mock(side_effect=coroutine(lambda x,y: "the awaitable should return this")) |
40
|
|
|
self.write = DirectWrite() |
41
|
|
|
|
42
|
|
|
@pytest.mark.curio |
43
|
|
|
async def test_set_color(self): |
44
|
|
|
port = 10 |
45
|
|
|
self.l.port = port |
46
|
|
|
await self.l.set_color(Color.blue) |
47
|
|
|
self.l.send_message.ask_called_once() |
48
|
|
|
args, kwargs = self.l.send_message.call_args |
49
|
|
|
assert args[1] == self.write.get_bytes(port, 0, Color.blue.value) |
50
|
|
|
|
51
|
|
|
class TestLight: |
52
|
|
|
|
53
|
|
|
def setup(self): |
54
|
|
|
self.l = Light(name='light') |
55
|
|
|
self.l.send_message = Mock(side_effect=coroutine(lambda x,y: "the awaitable should return this")) |
56
|
|
|
self.write = DirectWrite() |
57
|
|
|
|
58
|
|
|
@given( brightness = st.integers(0,100), |
59
|
|
|
port = st.integers(0,255) |
60
|
|
|
) |
61
|
|
|
def test_set_brightness(self, port, brightness): |
62
|
|
|
self.l.port = port |
63
|
|
|
|
64
|
|
|
async def child(): |
65
|
|
|
await self.l.set_brightness(brightness) |
66
|
|
|
kernel.run(child) |
67
|
|
|
|
68
|
|
|
self.l.send_message.ask_called_once() |
69
|
|
|
args, kwargs = self.l.send_message.call_args |
70
|
|
|
assert args[1] == self.write.get_bytes(port, 0, brightness) |
71
|
|
|
|
72
|
|
|
class TestMotor: |
73
|
|
|
|
74
|
|
|
def setup(self): |
75
|
|
|
#self.m = TrainMotor(name='motor') |
76
|
|
|
#self.m.send_message = Mock(side_effect=coroutine(lambda x,y: "the awaitable should return this")) |
77
|
|
|
self.write = DirectWrite() |
78
|
|
|
|
79
|
|
|
def _create_motor(self, cls): |
80
|
|
|
self.m = cls(name='motor') |
81
|
|
|
self.m.send_message = Mock(side_effect=coroutine(lambda x,y: "the awaitable should return this")) |
82
|
|
|
|
83
|
|
|
@given( speed = st.integers(-100,100), |
84
|
|
|
port = st.integers(0,255), |
85
|
|
|
cls = st.sampled_from([TrainMotor, DuploTrainMotor, WedoMotor, |
86
|
|
|
ExternalMotor, InternalMotor]) |
87
|
|
|
) |
88
|
|
|
def test_set_speed(self, cls, port, speed): |
89
|
|
|
self._create_motor(cls) |
90
|
|
|
self.m.port = port |
91
|
|
|
|
92
|
|
|
async def child(): |
93
|
|
|
await self.m.set_speed(speed) |
94
|
|
|
kernel.run(child) |
95
|
|
|
|
96
|
|
|
self.m.send_message.ask_called_once() |
97
|
|
|
args, kwargs = self.m.send_message.call_args |
98
|
|
|
assert args[1] == self.write.get_bytes(port, 0, self.m._convert_speed_to_val(speed)) |
99
|
|
|
|
100
|
|
View Code Duplication |
@given( speed = st.integers(-100,100), |
|
|
|
|
101
|
|
|
port = st.integers(0,255), |
102
|
|
|
cls = st.sampled_from([TrainMotor, DuploTrainMotor, WedoMotor, |
103
|
|
|
ExternalMotor, InternalMotor]) |
104
|
|
|
) |
105
|
|
|
def test_ramp_speed(self, cls, port, speed): |
106
|
|
|
self._create_motor(cls) |
107
|
|
|
self.m.port = port |
108
|
|
|
|
109
|
|
|
async def child(): |
110
|
|
|
await self.m.ramp_speed(speed, 200) |
111
|
|
|
await self.m.ramp_in_progress_task.join() |
112
|
|
|
async def main(): |
113
|
|
|
t = await spawn(child()) |
|
|
|
|
114
|
|
|
await t.join() |
115
|
|
|
assert self.m.speed == speed |
116
|
|
|
kernel.run(main) |
117
|
|
|
|
118
|
|
View Code Duplication |
@given( speed = st.sampled_from([-50,0,100]), |
|
|
|
|
119
|
|
|
port = st.integers(0,255), |
120
|
|
|
cls = st.sampled_from([TrainMotor, DuploTrainMotor, WedoMotor, |
121
|
|
|
ExternalMotor, InternalMotor]) |
122
|
|
|
) |
123
|
|
|
def test_ramp_cancel_speed(self, cls, port, speed): |
124
|
|
|
self._create_motor(cls) |
125
|
|
|
self.m.port = port |
126
|
|
|
|
127
|
|
|
async def child(): |
128
|
|
|
await self.m.ramp_speed(speed, 2000) |
129
|
|
|
await sleep(0.1) |
130
|
|
|
await self.m.set_speed(speed+10) |
131
|
|
|
|
132
|
|
|
async def main(): |
133
|
|
|
t = await spawn(child()) |
|
|
|
|
134
|
|
|
await t.join() |
135
|
|
|
assert self.m.speed == speed+10 |
136
|
|
|
kernel.run(main) |
137
|
|
|
|
138
|
|
View Code Duplication |
@given( pos = st.integers(-2147483648, 2147483647), |
|
|
|
|
139
|
|
|
port = st.integers(0,255), |
140
|
|
|
cls = st.sampled_from([ExternalMotor, InternalMotor]) |
141
|
|
|
) |
142
|
|
|
def test_set_pos(self, cls, port, pos): |
143
|
|
|
self._create_motor(cls) |
144
|
|
|
self.m.port = port |
145
|
|
|
speed = 50 |
146
|
|
|
max_power = 50 |
147
|
|
|
|
148
|
|
|
async def child(): |
149
|
|
|
await self.m.set_pos(pos, speed, max_power) |
150
|
|
|
|
151
|
|
|
async def main(): |
152
|
|
|
t = await spawn(child()) |
|
|
|
|
153
|
|
|
await t.join() |
154
|
|
|
kernel.run(main) |
155
|
|
|
|
156
|
|
|
args, kwargs = self.m.send_message.call_args |
157
|
|
|
assert args[1] == self.write.get_bytes_for_set_pos(port, pos, self.m._convert_speed_to_val(speed), max_power) |
158
|
|
|
|
159
|
|
View Code Duplication |
@given( angle = st.integers(0, 2147483647), |
|
|
|
|
160
|
|
|
speed = st.integers(-100,100), |
161
|
|
|
port = st.integers(0,255), |
162
|
|
|
cls = st.sampled_from([ExternalMotor, InternalMotor]) |
163
|
|
|
) |
164
|
|
|
def test_rotate(self, cls, port, angle, speed): |
165
|
|
|
self._create_motor(cls) |
166
|
|
|
self.m.port = port |
167
|
|
|
max_power = 50 |
168
|
|
|
|
169
|
|
|
async def child(): |
170
|
|
|
await self.m.rotate(angle, speed, max_power) |
171
|
|
|
|
172
|
|
|
async def main(): |
173
|
|
|
t = await spawn(child()) |
|
|
|
|
174
|
|
|
await t.join() |
175
|
|
|
kernel.run(main) |
176
|
|
|
|
177
|
|
|
args, kwargs = self.m.send_message.call_args |
178
|
|
|
assert args[1] == self.write.get_bytes_for_rotate(port, angle, self.m._convert_speed_to_val(speed), max_power) |