test_hub

Complexity

Total Complexity

43

Size/Duplication

Total Lines	234
Duplicated Lines	0 %

Importance

Changes

0

27 Methods

Rating	Name	Duplication	Size	Complexity
A	MockBLE.clear_cached_data()	0	2	1
A	MockDevice.__init__()	0	4	1
A	MockBLE.find_devices()	0	3	1
A	TestSensors.setup()	0	16	1
A	TestSensors._wait_send_message()	0	7	3
A	MockBLE.initialize()	0	2	1
A	MockBLE.run_mainloop_with()	0	3	1
A	TestSensors._with_header()	0	4	1
A	MockDevice.discover()	0	4	1
A	MockDevice.disconnect()	0	2	1
A	MockUart.__init__()	0	2	1
A	TestSensors._get_hub_class()	0	11	1
A	MockUart.find_characteristic()	0	3	1
A	TestSensors.test_run_hub()	0	20	3
A	MockAdapter.stop_scan()	0	2	1
A	MockBLE.get_default_adapter()	0	3	1
A	MockDevice.find_service()	0	3	1
A	MockAdapter.power_on()	0	2	1
A	MockBLE.__init__()	0	2	1
A	MockUart.start_notify()	0	3	1
A	MockAdapter.__init__()	0	2	1
A	TestSensors._draw_capabilities()	0	14	2
A	MockAdapter.start_scan()	0	2	1
A	MockDevice.connect()	0	2	1
A	TestSensors.test_attach_sensor()	0	14	1
A	MockUart.write_value()	0	2	1
D	TestSensors._emit_control()	0	41	12

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

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

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

from bricknil.message_dispatch import MessageDispatch
from bricknil.messages import UnknownMessageError, HubPropertiesMessage
from bricknil.sensor import *
from bricknil.const import DEVICES
from bricknil import attach, start
from bricknil.hub import PoweredUpHub, Hub, BoostHub, DuploTrainHub
import bricknil
import bricknil.const


class TestSensors:

    def setup(self):
        # Create the main dispatch
        self.hub = MagicMock()
        self.m = MessageDispatch(self.hub)
        self.sensor_list = [ CurrentSensor,
                             DuploSpeedSensor,
                             VisionSensor,
                             InternalTiltSensor,
                             ExternalMotionSensor,
                             ExternalTiltSensor,
                             RemoteButtons,
                             Button,
                             DuploVisionSensor,
                             VoltageSensor,
        ]
        self.hub_list = [ PoweredUpHub, BoostHub, DuploTrainHub]
    
    def _with_header(self, msg:bytearray):
        l = len(msg)+2
        assert l<127
        return bytearray([l, 0]+list(msg))

    def _draw_capabilities(self, data, sensor):
        if len(sensor.allowed_combo) > 0:
            # test capabilities 1 by 1, 
            # or some combination of those in the allowed_combo list
            capabilities = data.draw(
                    st.one_of(
                        st.lists(st.sampled_from(sensor.capability), min_size=1, max_size=1),
                        st.lists(st.sampled_from(sensor.allowed_combo), min_size=1, unique=True)
                    )
                )
        else:
            # if no combos allowed, then just test 1 by 1
            capabilities = data.draw(st.lists(st.sampled_from(sensor.capability), min_size=1, max_size=1))
        return capabilities


    def _get_hub_class(self, hub_type, sensor, sensor_name, capabilities):
        stop_evt = Event()
        @attach(sensor, name=sensor_name, capabilities=capabilities)
        class TestHub(hub_type):
            async def sensor_change(self):
                pass
            async def run(self):
                pass
                await stop_evt.wait()

        return TestHub, stop_evt

    #@patch('bricknil.hub.PoweredUpHub', autospec=True, create=True)
    @given(data = st.data())
    def test_attach_sensor(self, data):
        
        sensor_name = 'sensor'
        sensor = data.draw(st.sampled_from(self.sensor_list))
        capabilities = self._draw_capabilities(data, sensor)

        hub_type = data.draw(st.sampled_from(self.hub_list))
        TestHub, stop_evt = self._get_hub_class(hub_type, sensor, sensor_name, capabilities)
        hub = TestHub('testhub')
        # Check to make sure we have the peripheral attached
        # and the sensor inserted as an attribute
        assert sensor_name in hub.peripherals
        assert hasattr(hub, sensor_name)

    @given(data = st.data())
    def test_run_hub(self, data):

        Hub.hubs = []
        sensor_name = 'sensor'
        sensor = data.draw(st.sampled_from(self.sensor_list))
        capabilities = self._draw_capabilities(data, sensor)

        hub_type = data.draw(st.sampled_from(self.hub_list))
        TestHub, stop_evt = self._get_hub_class(hub_type, sensor, sensor_name, capabilities)
        hub = TestHub('test_hub')

        # Start the hub
        #kernel.run(self._emit_control(TestHub))
        with patch('Adafruit_BluefruitLE.get_provider') as ble,\
             patch('bricknil.ble_queue.USE_BLEAK', False) as use_bleak:
            ble.return_value = MockBLE(hub)
            sensor_obj = getattr(hub, sensor_name)
            sensor_obj.send_message = Mock(side_effect=coroutine(lambda x,y: "the awaitable should return this"))
            kernel.run(self._emit_control, data, hub, stop_evt, ble(), sensor_obj)
            #start(system)

    async def _wait_send_message(self, mock_call, msg):
        print("in mock")
        while not mock_call.call_args:
            await sleep(0.01)
        while not msg in mock_call.call_args[0][0]:
            print(mock_call.call_args)
            await sleep(0.01)

    async def _emit_control(self, data, hub, hub_stop_evt, ble, sensor):
        async def dummy():
            pass
        system = await spawn(bricknil.bricknil._run_all(ble, dummy))
        while not hub.peripheral_queue:
            await sleep(0.1)
        #await sleep(3)
        port = data.draw(st.integers(0,254))
        await hub.peripheral_queue.put( ('attach', (port, sensor.sensor_name)) )

        # Now, make sure the sensor sent an activate updates message
        if sensor.sensor_name == "Button":
            await self._wait_send_message(sensor.send_message, 'Activate button')
        else:
            await self._wait_send_message(sensor.send_message, 'Activate SENSOR')
        # Need to generate a value on the port
        # if False:
        msg = []
        if len(sensor.capabilities) == 1:
            # Handle single capability
            for cap in sensor.capabilities:
                n_datasets, byte_count = sensor.datasets[cap][0:2]
                for i in range(n_datasets):
                    for b in range(byte_count):
                        msg.append(data.draw(st.integers(0,255)))
            msg = bytearray(msg)
            await hub.peripheral_queue.put( ('value_change', (port, msg)))
        elif len(sensor.capabilities) > 1:
            modes = 1
            msg.append(modes)
            for cap_i, cap in enumerate(sensor.capabilities):
                if modes & (1<<cap_i): 
                    n_datasets, byte_count = sensor.datasets[cap][0:2]
                    for i in range(n_datasets):
                        for b in range(byte_count):
                            msg.append(data.draw(st.integers(0,255)))
            msg = bytearray(msg)
            await hub.peripheral_queue.put( ('value_change', (port, msg)))
        
        await hub_stop_evt.set()
        await system.join()

class MockBLE:
    def __init__(self, hub):
        self.hub = hub

    def initialize(self):
        print("initialized")
    
    def clear_cached_data(self):
        pass

    def get_default_adapter(self):
        self.mock_adapter = MockAdapter()
        return self.mock_adapter

    def find_devices(self, service_uuids):
        self.device = MockDevice(hub_name = self.hub.ble_name, hub_id = self.hub.manufacturer_id)
        return [self.device]

    def run_mainloop_with(self, func):
        print("run mainloop")
        func()

class MockAdapter:
    def __init__(self):
        self.name = 'Mock adapter'
    def power_on(self):
        pass

    def start_scan(self):
        print("start scan called")

    def stop_scan(self):
        print("stop scan called")

class MockDevice:
    def __init__(self, hub_name, hub_id):
        self.advertised = [-1, -1, -1, -1, hub_id]
        self.id = 'XX:XX:XX:XX:XX:XX'
        self.name = hub_name

    def connect(self):
        print("device connect called")

    def discover(self, uart_uuid, char_uuid):
        print(f'discover called on uart {uart_uuid}, char {char_uuid}')
        self.uart_uuid = uart_uuid
        self.char = char_uuid
    
    def find_service(self, uart_uuid):
        self.uart = MockUart()
        return self.uart

    def disconnect(self):
        print('device disconnect called')


class MockUart:
    def __init__(self):
        pass
    def find_characteristic(self, char_uuid):
        self.char_uuid = char_uuid
        return self

    def start_notify(self, callback):
        # Spawn a task to do the attachments, etc
        self.notify = callback

    def write_value(self, values):
        print(f'received values: {values}')