caspia.toolbox.testing package

Submodules

caspia.toolbox.testing.service module

class caspia.toolbox.testing.service.ServiceReactiveMock(*args, **kwargs)

Bases: caspia.meadow.client.gateway.ServiceGatewayMixin, caspia.meadow.services.base.ServiceBase

Initialize new service.

Parameters

• name – Name of the service (or just its path)

• include – What optional characteristics to include. Tuple of their names.

property all_observables

attach(connection)

property attached

auto_discovery = True

characteristic_cached_read(characteristic)

characteristic_cached_write(characteristic, value_and_meta)

async characteristic_read(characteristic, **kwargs)

Read value of the characteristic.

Parameters

• characteristic – The characteristic to which the value should be written.

• value – The value to be written.

• kwargs – Dictionary with extra key (dictionary with metadata) and optionally a timeout (float specifing explicit timeout for the operation).

Returns

Tuple (value, extra).

characteristic_subscribe(characteristic, on_value, on_error=None)

async characteristic_write(characteristic, value, **kwargs)

Write a value to the characteristic.

Parameters

• characteristic – The characteristic to which the value should be written.

• value – The value to be written.

• kwargs – Dictionary with extra key (dictionary with metadata) and optionally a timeout (float specifing explicit timeout for the operation).

characteristics = {'$metadata': <Characteristic <class 'caspia.meadow.services.base.ServiceMeta'>:$metadata>}

detach()

static get_subclass(stype, mixin=None)

load_definition(spec)

Update characteristics of this instance based on serialized specs.

Parameters

spec – Dict most likely going to be coming from meadow - from some gateway informing us about its services and their characteristics.

metadata

async notify(characteristic, value, if_changed=False, extra=None)

on(name, *args, **kwargs)

optional = {}

async published()

serialize()

Return serialized version (most likely dict) if possible or None.

set(characteristic, value)

type = 'servicereactivemock'

async update(characteristic, value)

class caspia.toolbox.testing.service.keydefaultdict

Bases: collections.defaultdict

clear() → None. Remove all items from D.

copy() → a shallow copy of D.

default_factory

Factory for default value called by __missing__().

fromkeys()

Create a new dictionary with keys from iterable and values set to value.

get()

Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.

If key is not found, d is returned if given, otherwise KeyError is raised

popitem() → (k, v), remove and return some (key, value) pair as a

2-tuple; but raise KeyError if D is empty.

setdefault()

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None. Update D from dict/iterable E and F.

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values() → an object providing a view on D's values

Module contents