test_peripheral.DirectWrite.get_bytes_for_set_pos() - Code Metrics - Inspection of "added unit tests" - virantha/bricknil - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( b99246...4b835f )

by Virantha

created 2019-04-24 03:04 UTC

DirectWrite.get_bytes_for_set_pos() A

↳ Parent: test_peripheral

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	5
dl	0
loc	3
rs	10
c	0
b	0
f	0

import pytest
import os, struct, copy
import logging
from asyncio import coroutine
from curio import kernel, sleep

from mock import Mock
from mock import patch, call
from mock import MagicMock
from mock import PropertyMock

from hypothesis import given, example, settings
from hypothesis import strategies as st

from bricknil.sensor.light import *
from bricknil.sensor.motor import *
from bricknil.const import Color

class AsyncMock(MagicMock):
    async def __call__(self, *args, **kwargs):
        return super(AsyncMock, self).__call__(*args, **kwargs)

class DirectWrite:
    def get_bytes(self, port, mode, value):
        return [0x00, 0x81, port, 0x01, 0x51, mode, value ]

    def get_bytes_for_set_pos(self, port, pos, speed, max_power):
        abs_pos = list(struct.pack('i', pos))
        return [0x00, 0x81, port, 0x01, 0x0d] + abs_pos + [speed, max_power, 126, 3]
        
    def get_bytes_for_rotate(self, port, angle, speed, max_power):
        angle = list(struct.pack('i',angle))
        return [0x00, 0x81, port, 0x01, 0x0b] +  angle + [speed, max_power, 126, 3]

class TestLED:

    def setup(self):
        self.l = LED(name='led')
        self.l.send_message = Mock(side_effect=coroutine(lambda x,y: "the awaitable should return this"))
        self.write = DirectWrite()

    @pytest.mark.curio
    async def test_set_color(self):
        port = 10
        self.l.port = port
        await self.l.set_color(Color.blue)
        self.l.send_message.ask_called_once()
        args, kwargs = self.l.send_message.call_args
        assert args[1] == self.write.get_bytes(port, 0, Color.blue.value)

class TestLight:

    def setup(self):
        self.l = Light(name='light')
        self.l.send_message = Mock(side_effect=coroutine(lambda x,y: "the awaitable should return this"))
        self.write = DirectWrite()

    @given( brightness = st.integers(0,100),
            port = st.integers(0,255)
    )
    def test_set_brightness(self, port, brightness):
        self.l.port = port

        async def child():
            await self.l.set_brightness(brightness)
        kernel.run(child)

        self.l.send_message.ask_called_once()
        args, kwargs = self.l.send_message.call_args
        assert args[1] == self.write.get_bytes(port, 0, brightness)

class TestMotor:

    def setup(self):
        #self.m = TrainMotor(name='motor')
        #self.m.send_message = Mock(side_effect=coroutine(lambda x,y: "the awaitable should return this"))
        self.write = DirectWrite()

    def _create_motor(self, cls):
        self.m = cls(name='motor')
        self.m.send_message = Mock(side_effect=coroutine(lambda x,y: "the awaitable should return this"))

    @given( speed = st.integers(-100,100),
            port = st.integers(0,255),
            cls = st.sampled_from([TrainMotor, DuploTrainMotor, WedoMotor, 
                        ExternalMotor, InternalMotor])
    )
    def test_set_speed(self, cls, port, speed):
        self._create_motor(cls)
        self.m.port = port

        async def child():
            await self.m.set_speed(speed)
        kernel.run(child)

        self.m.send_message.ask_called_once()
        args, kwargs = self.m.send_message.call_args
        assert args[1] == self.write.get_bytes(port, 0, self.m._convert_speed_to_val(speed))

    @given( speed = st.integers(-100,100),

            port = st.integers(0,255),
            cls = st.sampled_from([TrainMotor, DuploTrainMotor, WedoMotor, 
                        ExternalMotor, InternalMotor])
    )
    def test_ramp_speed(self, cls, port, speed):
        self._create_motor(cls)
        self.m.port = port

        async def child():
            await self.m.ramp_speed(speed, 200)
            await self.m.ramp_in_progress_task.join()
        async def main():
            t = await spawn(child())

            await t.join()
            assert self.m.speed == speed
        kernel.run(main)

    @given( speed = st.sampled_from([-50,0,100]),

            port = st.integers(0,255),
            cls = st.sampled_from([TrainMotor, DuploTrainMotor, WedoMotor, 
                        ExternalMotor, InternalMotor])
    )
    def test_ramp_cancel_speed(self, cls, port, speed):
        self._create_motor(cls)
        self.m.port = port

        async def child():
            await self.m.ramp_speed(speed, 2000)
            await sleep(0.1)
            await self.m.set_speed(speed+10)

        async def main():
            t = await spawn(child())

            await t.join()
            assert self.m.speed == speed+10
        kernel.run(main)

    @given( pos = st.integers(-2147483648, 2147483647),

            port = st.integers(0,255),
            cls = st.sampled_from([ExternalMotor, InternalMotor])
    )
    def test_set_pos(self, cls, port, pos):
        self._create_motor(cls)
        self.m.port = port
        speed = 50
        max_power = 50

        async def child():
            await self.m.set_pos(pos, speed, max_power)

        async def main():
            t = await spawn(child())

            await t.join()
        kernel.run(main)

        args, kwargs = self.m.send_message.call_args
        assert args[1] == self.write.get_bytes_for_set_pos(port, pos, self.m._convert_speed_to_val(speed), max_power)

    @given( angle = st.integers(0, 2147483647),

            speed = st.integers(-100,100),
            port = st.integers(0,255),
            cls = st.sampled_from([ExternalMotor, InternalMotor])
    )
    def test_rotate(self, cls, port, angle, speed):
        self._create_motor(cls)
        self.m.port = port
        max_power = 50

        async def child():
            await self.m.rotate(angle, speed, max_power)

        async def main():
            t = await spawn(child())

            await t.join()
        kernel.run(main)

        args, kwargs = self.m.send_message.call_args
        assert args[1] == self.write.get_bytes_for_rotate(port, angle, self.m._convert_speed_to_val(speed), max_power)

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),
		0 ignored issues – show Duplication introduced 2019-04-24 03:06 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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())
		0 ignored issues – show Comprehensibility Best Practice introduced 2019-04-24 03:06 UTC by Report Bug Copy Issue Report The variable `spawn` does not seem to be defined. Loading history...
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]),
		0 ignored issues – show Duplication introduced 2019-04-24 03:06 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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())
		0 ignored issues – show Comprehensibility Best Practice introduced 2019-04-24 03:06 UTC by Report Bug Copy Issue Report The variable `spawn` does not seem to be defined. Loading history...
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),
		0 ignored issues – show Duplication introduced 2019-04-24 03:06 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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())
		0 ignored issues – show Comprehensibility Best Practice introduced 2019-04-24 03:06 UTC by Report Bug Copy Issue Report The variable `spawn` does not seem to be defined. Loading history...
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),
		0 ignored issues – show Duplication introduced 2019-04-24 03:06 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
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())
		0 ignored issues – show Comprehensibility Best Practice introduced 2019-04-24 03:06 UTC by Report Bug Copy Issue Report The variable `spawn` does not seem to be defined. Loading history...
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)

virantha / bricknil

GitHub Access Token became invalid

Push — master ( b99246...4b835f )

DirectWrite.get_bytes_for_set_pos() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like