Source code for caspia.toolbox.testing.service

from collections import defaultdict

from caspia import reactive as r
from caspia.meadow.client.gateway import ServiceGatewayMixin
from caspia.meadow.services import ServiceBase


class keydefaultdict(defaultdict):
    def __missing__(self, key):
        # pylint: disable=not-callable
        self[key] = self.default_factory(key)
        return self[key]


class ServiceReactiveMock(ServiceGatewayMixin, ServiceBase):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.values = keydefaultdict(self._create_value)

    def _create_value(self, characteristic, value=None):
        value = r.Value(value)

        async def on_update(value, **kwargs):
            # pylint: disable=no-member
            await self.notify(characteristic, value)

        value.subscribe(on_update)
        return value

    def set(self, characteristic, value):
        if isinstance(characteristic, str):
            characteristic = self[characteristic]
        self.values[characteristic] = self._create_value(characteristic, value)

    async def update(self, characteristic, value):
        if isinstance(characteristic, str):
            characteristic = self[characteristic]
        await self.values[characteristic].update(value)

    async def characteristic_read(self, characteristic, **kwargs):
        return await self.values[characteristic].observe()

    async def characteristic_write(self, characteristic, value, **kwargs):
        await self.values[characteristic].update(value)