Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use set_cb method in fMBT

Best Python code snippet using fMBT_python

test_model.py

Source:test_model.py

1from collections import OrderedDict2from copy import deepcopy3from datetime import datetime4from enum import Enum5from unittest.mock import Mock6import pytest7from pyzkaccess.device_data.model import Field, Model, models_registry8class EnumStub(Enum):9    val1 = 12310    val2 = 45611class ModelStub(Model):12    table_name = 'table1'13    incremented_field = Field(14        'IncField', int, lambda x: int(x) + 1, lambda x: x - 1, lambda x: x > 015    )16    append_foo_field = Field(17        'FooField', str, lambda x: x + 'Foo', lambda x: x[:-3], lambda x: len(x) > 018    )19class TestField:20    def test_init__should_set_properties(self):21        field_datatype, get_cb, set_cb, validation_cb = Mock(), Mock(), Mock(), Mock()22        obj = Field('my_name', field_datatype, get_cb, set_cb, validation_cb)23        assert obj._raw_name == 'my_name'24        assert obj._field_datatype is field_datatype25        assert obj._get_cb is get_cb26        assert obj._set_cb is set_cb27        assert obj._validation_cb is validation_cb28    def test_init__should_set_default_properties(self):29        obj = Field('my_name')30        assert obj._raw_name == 'my_name'31        assert obj._field_datatype == str32        assert obj._get_cb is None33        assert obj._set_cb is None34        assert obj._validation_cb is None35    def test_raw_name_prop__should_return_raw_name(self):36        obj = Field('my_name')37        assert obj.raw_name == 'my_name'38    def test_field_datatype_prop__should_return_field_datatype(self):39        obj = Field('my_name', int)40        assert obj.field_datatype == int41    def test_to_raw_value__on_all_defaults__should_return_the_same_string(self):42        obj = Field('my_name')43        res = obj.to_raw_value('value1')44        assert res == 'value1'45    @pytest.mark.parametrize('datatype,value', (46        (str, '123'),47        (int, 123),48        (datetime, datetime(2020, 12, 12, 12, 12, 12)),49        (tuple, (1, 2, 3))50    ))51    def test_to_raw_value__if_type_set__should_return_string_representation(self, datatype, value):52        obj = Field('my_name', datatype)53        res = obj.to_raw_value(value)54        assert type(res) == str and res == str(value)55    def test_to_raw_value__if_type_is_enum__should_return_its_value_string_representation(self):56        obj = Field('my_name', EnumStub)57        res = obj.to_raw_value(EnumStub.val2)58        assert res == '456'59    @pytest.mark.parametrize('datatype,value,set_cb,expect', (60        (str, '123', int, '123'),  # str=>int=>str61        (int, 123, lambda x: x + 1, '124'),  # int=>[increment]=>str62        (datetime, datetime(2020, 12, 12, 12, 12, 12), lambda x: x.day, '12'),  # dtime=>[.day]=>str63        (tuple, ('1', '2', '3'), lambda x: ''.join(x), '123'),  # tuple=>[join to string]=>str64        (Enum, EnumStub.val2, lambda x: x + 1, '457')  # Enum=>[.value]=>[increment]=>str65    ))66    def test_to_raw_value__if_set_cb_set__should_use_its_value_and_cast_to_string(67        self, datatype, value, set_cb, expect68    ):69        get_cb = Mock()70        obj = Field('my_name', datatype, get_cb, set_cb)71        res = obj.to_raw_value(value)72        assert res == expect and type(res) == type(expect)73        get_cb.assert_not_called()74    @pytest.mark.parametrize('datatype,value,set_cb,expect', (75        (str, '123', int, '123'),  # str=>int=>str76        (int, 123, lambda x: x + 1, '124'),  # int=>[increment]=>str77        (datetime, datetime(2020, 12, 12, 12, 12, 12), lambda x: x.day, '12'),  # dtime=>[.day]=>str78        (tuple, ('1', '2', '3'), lambda x: ''.join(x), '123'),  # tuple=>[join to string]=>str79        (Enum, EnumStub.val2, lambda x: x + 1, '457')  # Enum=>[.value]=>[increment]=>str80    ))81    def test_to_raw_value__if_set_cb_and_validation_cb_passed__should_return_string_of_get_cb_value(82        self, datatype, value, set_cb, expect83    ):84        get_cb = Mock()85        validation_cb = Mock(return_value=True)86        obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)87        res = obj.to_raw_value(value)88        assert res == expect and type(res) == type(expect)89        get_cb.assert_not_called()90        validation_cb.assert_called_once_with(value)91    @pytest.mark.parametrize('datatype,value', (92        (str, '123'),93        (int, 123),94        (datetime, datetime(2020, 12, 12, 12, 12, 12)),95        (tuple, (1, 2, 3)),96        (Enum, EnumStub.val2)97    ))98    def test_to_raw_value__if_set_cb_and_validation_cb_failed__should_raise_error(99        self, datatype, value100    ):101        get_cb = Mock()102        set_cb = Mock(return_value=555)103        validation_cb = Mock(return_value=False)104        obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)105        with pytest.raises(ValueError):106            obj.to_raw_value(value)107    @pytest.mark.parametrize('datatype,value', (108        (str, 123),109        (str, None),110        (int, '123'),111        (datetime, 5),112        (tuple, datetime.now()),113        (Enum, 456)114    ))115    def test_to_raw_value__if_value_has_another_type__should_raise_error(self, datatype, value):116        get_cb = Mock()117        set_cb = Mock(return_value=555)118        validation_cb = Mock(return_value=False)119        obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)120        with pytest.raises(TypeError):121            obj.to_raw_value(value)122    def test_to_field_value__on_all_defaults__should_return_the_same_string(self):123        obj = Field('my_name')124        res = obj.to_field_value('value1')125        assert res == 'value1'126    @pytest.mark.parametrize('datatype,value,expect', (127        (str, '123', '123'),128        (int, '123', 123),129        (EnumStub, 456, EnumStub.val2)130    ))131    def test_to_field_value__if_type_set__should_return_value_of_this_type(132            self, datatype, value, expect133    ):134        obj = Field('my_name', datatype)135        res = obj.to_field_value(value)136        assert res == expect and type(res) == type(expect)137    @pytest.mark.parametrize('datatype,value,get_cb,expect', (138        (139            datetime,140            '2020-12-13 14:15:16',141            lambda x: datetime.strptime(x, '%Y-%m-%d %H:%M:%S'),142            datetime(2020, 12, 13, 14, 15, 16)143        ),  # str=>datetime144        (tuple, '123', lambda x: tuple(x), ('1', '2', '3')),  # str=>tuple145        (EnumStub, '456', lambda x: EnumStub(int(x)), EnumStub.val2),  # str=>Enum146        (bool, '1', lambda x: bool(int(x)), True),  # str=>bool147        (bool, '0', lambda x: bool(int(x)), False)  # str=>bool148    ))149    def test_to_field_value__if_get_cb_returns_value_of_datatype__should_convert_value(150        self, datatype, value, get_cb, expect151    ):152        set_cb = Mock()153        validation_cb = Mock()154        obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)155        res = obj.to_field_value(value)156        assert res == expect and type(res) == type(expect)157        set_cb.assert_not_called()158        validation_cb.assert_not_called()159    @pytest.mark.parametrize('datatype,value', (160        (datetime, '2020-12-13 14:15:16'),161        (tuple, '123'),162        (EnumStub, '456'),163        (bool, '1'),164        (bool, '0')165    ))166    def test_to_field_value__if_get_cb_returns_none__should_return_none(self, datatype, value):167        set_cb = Mock()168        validation_cb = Mock()169        obj = Field('my_name', datatype, lambda _: None, set_cb, validation_cb)170        res = obj.to_field_value(value)171        assert res is None172        set_cb.assert_not_called()173        validation_cb.assert_not_called()174    @pytest.mark.parametrize('datatype,value,get_cb,expect', (175        (176            str,177            '2020-12-13 14:15:16',178            lambda x: datetime.strptime(x, '%Y-%m-%d %H:%M:%S'),179            '2020-12-13 14:15:16'180        ),  # str=>datetime=>str181        (tuple, '123', lambda x: list(x), ('1', '2', '3')),  # str=>list=>tuple182        (EnumStub, '456', int, EnumStub.val2),  # str=>int=>Enum183        (bool, '1', int, True),  # str=>int=>bool184        (bool, '0', int, False)  # str=>int=>bool185    ))186    def test_to_field_value__if_get_cb_returns_value_not_of_datatype__should_also_cast_to_datatype(187        self, datatype, value, get_cb, expect188    ):189        set_cb = Mock()190        validation_cb = Mock()191        obj = Field('my_name', datatype, get_cb, set_cb, validation_cb)192        res = obj.to_field_value(value)193        assert res == expect and type(res) == type(expect)194        set_cb.assert_not_called()195        validation_cb.assert_not_called()196    def test_hash__should_be_hashable(self):197        obj = Field('my_name')198        assert hash(obj) == hash('my_name')199    def test_get_descriptor__should_correct_field_values(self):200        obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'}, False)201        assert obj._dirty is False202        assert obj.append_foo_field == 'MagicFoo'203        assert obj.incremented_field == 124204        assert obj._dirty is False205    def test_get_descriptor__if_no_such_raw_field__should_return_none(self):206        obj = ModelStub().with_raw_data({'FooField': 'Magic'})207        assert obj.incremented_field is None208    def test_get_descriptor__if_class_instance__should_return_itself(self):209        assert isinstance(ModelStub.incremented_field, Field)210    def test_set_descriptor__should_set_raw_data(self):211        obj = ModelStub()212        obj.append_foo_field = "WowFoo"213        obj.incremented_field = 123214        assert obj._raw_data == {'IncField': '122', 'FooField': 'Wow'}215    def test_set_descriptor__should_set_dirty_flag(self):216        obj = ModelStub().with_raw_data({}, False)217        assert obj._dirty is False218        obj.incremented_field = 123219        assert obj._dirty is True220    def test_set_descriptor__if_none_is_given__should_delete_field_from_raw_data(self):221        obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'})222        obj.incremented_field = None223        assert obj._raw_data == {'FooField': 'Magic'}224    def test_set_descriptor__should_consider_field_validation(self):225        obj = ModelStub()226        with pytest.raises(ValueError):227            obj.incremented_field = -1228    def test_set_descriptor__if_class_instance__should_do_nothing(self):229        class ModelStub2(Model):230            table_name = 'test'231            field1 = Field('field')232        ModelStub2.field1 = 'testvalue'233        assert object.__getattribute__(ModelStub2, 'field1') == 'testvalue'234    def test_del_descriptor__should_delete_field_from_raw_data(self):235        obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'})236        del obj.incremented_field237        assert obj._raw_data == {'FooField': 'Magic'}238    def test_del_descriptor__should_set_dirty_flag(self):239        obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'}, False)240        assert obj._dirty is False241        del obj.incremented_field242        assert obj._dirty is True243    def test_del_descriptor__if_class_instance__should_do_nothing(self):244        class ModelStub2(Model):245            table_name = 'test'246            field1 = Field('field')247        del ModelStub2.field1248        with pytest.raises(AttributeError):249            object.__getattribute__(ModelStub2, 'field1')250class TestModelMeta:251    @pytest.fixture252    def test_registry(self):253        orig_model_registry = deepcopy(models_registry)254        models_registry.clear()255        yield models_registry256        models_registry.clear()257        models_registry.update(orig_model_registry)258    def test_metaclass__should_add_class_to_registry(self, test_registry):259        class MyModel(Model):260            table_name = 'test'261            field1 = Field('FieldOne')262        assert test_registry == {'MyModel': MyModel}263    def test_metaclass__should_fill_fields_mapping_for_each_class(self):264        class MyModel(Model):265            table_name = 'test'266            field1 = Field('FieldOne')267            field2 = Field('FieldTwo')268        class MyModel2(Model):269            table_name = 'test'270            field3 = Field('FieldThree')271            field4 = Field('FieldFour')272        assert MyModel._fields_mapping == {'field1': 'FieldOne', 'field2': 'FieldTwo'}273        assert MyModel2._fields_mapping == {'field3': 'FieldThree', 'field4': 'FieldFour'}274    def test_metaclass__should_set_field_objects_doc_attribute(self):275        class MyModel(Model):276            table_name = 'test'277            field1 = Field('FieldOne')278            field2 = Field('FieldTwo', int)279        assert MyModel.field1.__doc__ == 'MyModel.field1'280        assert MyModel.field2.__doc__ == 'MyModel.field2'281    def test_metaclass__should_set_field_object_class_var_annotation(self):282        class MyModel(Model):283            table_name = 'test'284            field1 = Field('FieldOne')285            field2 = Field('FieldTwo', int)286        assert {'field1': str, 'field2': int}.items() <= MyModel.__annotations__.items()287class TestModel:288    def test_init__should_set_default_attributes(self):289        obj = ModelStub()290        assert obj._sdk is None291        assert obj._dirty is True292        assert obj._raw_data == {}293    def test_init__if_fields_has_passed__should_initialize_raw_data_only_with_given_fields(self):294        obj = ModelStub(incremented_field=123)295        assert obj._raw_data == {'IncField': '122'}296    def test_init__if_nones_has_passed_in_fields__should_ignore_them(self):297        obj = ModelStub(incremented_field=123, append_foo_field=None)298        assert obj._raw_data == {'IncField': '122'}299    def test_init__if_unknown_fields_has_passed__should_raise_error(self):300        with pytest.raises(TypeError):301            ModelStub(incremented_field=123, unknown_field=3)302    def test_init__should_consider_fields_validation(self):303        with pytest.raises(ValueError):304            ModelStub(incremented_field=-1)305    def test_dict__should_return_fields_value(self):306        obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'})307        assert obj.dict == {'incremented_field': 124, 'append_foo_field': 'MagicFoo'}308    def test_dict__if_no_certain_field_in_raw_data__should_return_nones(self):309        obj = ModelStub().with_raw_data({'IncField': '123'})310        assert obj.dict == {'incremented_field': 124, 'append_foo_field': None}311    def test_raw_data__should_return_raw_data_appended_with_empty_string_for_absend_keys(self):312        obj = ModelStub().with_raw_data({'IncField': '123'})313        assert obj.raw_data == {'IncField': '123', 'FooField': ''}314    def test_fields_mapping__should_return_fields_mapping(self):315        assert ModelStub.fields_mapping() == {316            'incremented_field': 'IncField', 'append_foo_field': 'FooField'317        }318        assert ModelStub().fields_mapping() == {319            'incremented_field': 'IncField', 'append_foo_field': 'FooField'320        }321    def test_delete__should_delete_current_record_and_set_dirty_flag(322            self, generator_sends_collector323    ):324        items = []325        sdk = Mock()326        sdk.delete_device_data.side_effect = generator_sends_collector(items)327        obj = ModelStub().with_sdk(sdk).with_raw_data(328            {'IncField': '123', 'FooField': 'Magic'}, False329        )330        assert obj._dirty is False331        obj.delete()332        sdk.delete_device_data.assert_called_once_with('table1')333        assert items == [{'IncField': '123', 'FooField': 'Magic'}, None]334        assert obj._dirty is True335    def test_delete__if_manually_created_record__should_raise_error(self):336        obj = ModelStub(incremented_field=123, append_foo_field='MagicFoo')337        with pytest.raises(TypeError):338            obj.delete()339    def test_save__should_upsert_current_record_and_reset_dirty_flag(340            self, generator_sends_collector341    ):342        items = []343        sdk = Mock()344        sdk.set_device_data.side_effect = generator_sends_collector(items)345        obj = ModelStub().with_sdk(sdk).with_raw_data({'IncField': '123', 'FooField': 'Magic'})346        assert obj._dirty is True347        obj.save()348        sdk.set_device_data.assert_called_once_with('table1')349        assert items == [{'IncField': '123', 'FooField': 'Magic'}, None]350        assert obj._dirty is False351    def test_save__if_manually_created_record__should_raise_error(self):352        obj = ModelStub(incremented_field=123, append_foo_field='MagicFoo')353        with pytest.raises(TypeError):354            obj.save()355    @pytest.mark.parametrize('dirty_flag', (True, False))356    def test_with_raw_data__should_set_raw_data_and_dirty_flag(self, dirty_flag):357        obj = ModelStub().with_raw_data({'IncField': '123', 'FooField': 'Magic'}, dirty_flag)358        assert obj._raw_data == {'IncField': '123', 'FooField': 'Magic'}359        assert obj._dirty == dirty_flag360    def test_with_raw_data__should_return_self(self):361        obj = ModelStub()362        assert obj.with_raw_data({}) is obj363    def test_with_sdk__should_set_sdk(self):364        sdk = Mock()365        obj = ModelStub().with_raw_data({'A': 'val1'}, False).with_sdk(sdk)366        assert obj._sdk is sdk367        assert obj._raw_data == {'A': 'val1'}368        assert obj._dirty is False369    def test_with_sdk__should_return_self(self):370        obj = ModelStub()371        assert obj.with_sdk(Mock()) is obj372    def test_with_zk__should_set_sdk(self):373        zk = Mock()374        obj = ModelStub().with_raw_data({'A': 'val1'}, False).with_zk(zk)375        assert obj._sdk is zk.sdk376        assert obj._raw_data == {'A': 'val1'}377        assert obj._dirty is False378    def test_with_zk__should_return_self(self):379        obj = ModelStub()380        assert obj.with_zk(Mock()) is obj381    def test_eq_ne__if_raw_data_and_table_are_equal__should_return_true(self):382        obj1 = ModelStub(incremented_field=123, append_foo_field='MagicFoo')383        obj2 = ModelStub(incremented_field=123, append_foo_field='MagicFoo')384        assert obj1 == obj2385        assert not(obj1 != obj2)386    @pytest.mark.parametrize('table_name,kwargs', (387        ('table1', {'incremented_field': 1, 'append_foo_field': 'Magic'}),388        ('table2', {'incremented_field': 122, 'append_foo_field': 'Magic'}),389        ('table2', {'incremented_field': 1, 'append_foo_field': 'Magic'}),390    ))391    def test_eq_ne__if_raw_data_or_table_are_not_equal__should_return_false(392        self, table_name, kwargs393    ):394        obj1 = ModelStub(incremented_field=123, append_foo_field='MagicFoo')395        obj2 = ModelStub(**kwargs)396        obj2.table_name = table_name397        assert obj1 != obj2398        assert not(obj1 == obj2)399    def test_repr__should_return_data_table_name_and_fields_and_their_raw_values(self):400        raw_data = OrderedDict((('IncField', '123'), ('FooField', 'Magic')))401        obj = ModelStub().with_raw_data(raw_data, False)402        assert repr(obj) == 'ModelStub(append_foo_field=Magic, incremented_field=123)'403    def test_repr__if_dirty_flag_is_set__should_reflect_this_fact_in_string(self):404        raw_data = OrderedDict((('IncField', '123'), ('FooField', 'Magic')))405        obj = ModelStub().with_raw_data(raw_data, True)...

ccwi9p9215_gpio.py

Source:ccwi9p9215_gpio.py

1############################################################################2#                                                                          #3# Copyright (c)2008, 2009, Digi International (Digi). All Rights Reserved. #4#                                                                          #5# Permission to use, copy, modify, and distribute this software and its    #6# documentation, without fee and without a signed licensing agreement, is  #7# hereby granted, provided that the software is used on Digi products only #8# and that the software contain this copyright notice,  and the following  #9# two paragraphs appear in all copies, modifications, and distributions as #10# well. Contact Product Management, Digi International, Inc., 11001 Bren   #11# Road East, Minnetonka, MN, +1 952-912-3444, for commercial licensing     #12# opportunities for non-Digi products.                                     #13#                                                                          #14# DIGI SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT LIMITED   #15# TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A          #16# PARTICULAR PURPOSE. THE SOFTWARE AND ACCOMPANYING DOCUMENTATION, IF ANY, #17# PROVIDED HEREUNDER IS PROVIDED "AS IS" AND WITHOUT WARRANTY OF ANY KIND. #18# DIGI HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,         #19# ENHANCEMENTS, OR MODIFICATIONS.                                          #20#                                                                          #21# IN NO EVENT SHALL DIGI BE LIABLE TO ANY PARTY FOR DIRECT, INDIRECT,      #22# SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING LOST PROFITS,   #23# ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF   #24# DIGI HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGES.                #25#                                                                          #26############################################################################27# Imports28import digihw29from devices.device_base import DeviceBase30from settings.settings_base import SettingsBase, Setting31from channels.channel_source_device_property import *32from samples.sample import Sample33from common.shutdown import SHUTDOWN_WAIT34import threading35import time36#Main class37class ModuleGPIOs(DeviceBase, threading.Thread):38    """39    This class extends one of our base classes and is intended as an40    example of a concrete, example implementation, but it is not itself41    meant to be included as part of our developer API. Please consult the42    base class documentation for the API and the source code for this file43    for an example implementation.44    """45    46    #Class vars47    gpios = {}48    gpios_ind = {}49    setting_gpio=False50    for i in xrange (0,32):51        gpios["GPIO_"+str(i)]=0        52    for i in xrange (0,32):53        gpios_ind["GPIO_"+str(i)]=i54    55    def __init__(self, name, core_services):56        self.__name = name57        self.__core = core_services58        self.setting_gpio = False59        60        from core.tracing import get_tracer61        self.__tracer = get_tracer(name)62        63        ## Settings Table Definition:64        settings_list = [           65            Setting(66                name='input_gpios', type=list, required=True),67            Setting(68                name='output_gpios', type=list, required=True),69            Setting(                70                name='update_rate', type=float, required=False, 71                default_value=0.1,verify_function=lambda x: x > 0.0)72        ]73        74        ## Declare the GPIOs channels75        property_list = [        76            ChannelSourceDeviceProperty(name="GPIO_0",77                  type=bool,initial=Sample(0, False),                  78                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,79                  options=DPROP_OPT_AUTOTIMESTAMP,80                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_0",81                                                      sample=sample)),82            ChannelSourceDeviceProperty(name="GPIO_1",83                  type=bool,initial=Sample(0, False),                  84                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,85                  options=DPROP_OPT_AUTOTIMESTAMP,86                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_1",87                                                      sample=sample)),88            ChannelSourceDeviceProperty(name="GPIO_2",89                  type=bool,initial=Sample(0, False),                  90                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,91                  options=DPROP_OPT_AUTOTIMESTAMP,92                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_2",93                                                      sample=sample)),94            ChannelSourceDeviceProperty(name="GPIO_3",95                  type=bool,initial=Sample(0, False),                  96                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,97                  options=DPROP_OPT_AUTOTIMESTAMP,98                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_3",99                                                      sample=sample)),100            ChannelSourceDeviceProperty(name="GPIO_4",101                  type=bool,initial=Sample(0, False),                  102                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,103                  options=DPROP_OPT_AUTOTIMESTAMP,104                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_4",105                                                      sample=sample)),106            ChannelSourceDeviceProperty(name="GPIO_5",107                  type=bool,initial=Sample(0, False),                  108                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,109                  options=DPROP_OPT_AUTOTIMESTAMP,110                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_5",111                                                      sample=sample)),112            ChannelSourceDeviceProperty(name="GPIO_6",113                  type=bool,initial=Sample(0, False),                  114                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,115                  options=DPROP_OPT_AUTOTIMESTAMP,116                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_6",117                                                      sample=sample)),118            ChannelSourceDeviceProperty(name="GPIO_7",119                  type=bool,initial=Sample(0, False),                  120                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,121                  options=DPROP_OPT_AUTOTIMESTAMP,122                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_7",123                                                      sample=sample)),124            ChannelSourceDeviceProperty(name="GPIO_8",125                  type=bool,initial=Sample(0, False),                  126                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,127                  options=DPROP_OPT_AUTOTIMESTAMP,128                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_8",129                                                      sample=sample)),130            ChannelSourceDeviceProperty(name="GPIO_9",131                  type=bool,initial=Sample(0, False),                  132                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,133                  options=DPROP_OPT_AUTOTIMESTAMP,134                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_9",135                                                      sample=sample)),136            ChannelSourceDeviceProperty(name="GPIO_10",137                  type=bool,initial=Sample(0, False),                  138                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,139                  options=DPROP_OPT_AUTOTIMESTAMP,140                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_10",141                                                      sample=sample)),142            ChannelSourceDeviceProperty(name="GPIO_11",143                  type=bool,initial=Sample(0, False),                  144                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,145                  options=DPROP_OPT_AUTOTIMESTAMP,146                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_11",147                                                      sample=sample)),148            ChannelSourceDeviceProperty(name="GPIO_12",149                  type=bool,initial=Sample(0, False),                  150                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,151                  options=DPROP_OPT_AUTOTIMESTAMP,152                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_12",153                                                      sample=sample)),154            ChannelSourceDeviceProperty(name="GPIO_13",155                  type=bool,initial=Sample(0, False),                  156                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,157                  options=DPROP_OPT_AUTOTIMESTAMP,158                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_13",159                                                      sample=sample)),160            ChannelSourceDeviceProperty(name="GPIO_14",161                  type=bool,initial=Sample(0, False),                  162                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,163                  options=DPROP_OPT_AUTOTIMESTAMP,164                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_14",165                                                      sample=sample)),166            ChannelSourceDeviceProperty(name="GPIO_15",167                  type=bool,initial=Sample(0, False),                  168                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,169                  options=DPROP_OPT_AUTOTIMESTAMP,170                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_15",171                                                      sample=sample)),172            ChannelSourceDeviceProperty(name="GPIO_16",173                  type=bool,initial=Sample(0, False),                  174                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,175                  options=DPROP_OPT_AUTOTIMESTAMP,176                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_16",177                                                      sample=sample)),178            ChannelSourceDeviceProperty(name="GPIO_17",179                  type=bool,initial=Sample(0, False),                  180                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,181                  options=DPROP_OPT_AUTOTIMESTAMP,182                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_17",183                                                      sample=sample)),184            ChannelSourceDeviceProperty(name="GPIO_18",185                  type=bool,initial=Sample(0, False),                  186                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,187                  options=DPROP_OPT_AUTOTIMESTAMP,188                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_18",189                                                      sample=sample)),190            ChannelSourceDeviceProperty(name="GPIO_19",191                  type=bool,initial=Sample(0, False),                  192                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,193                  options=DPROP_OPT_AUTOTIMESTAMP,194                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_19",195                                                      sample=sample)),196            ChannelSourceDeviceProperty(name="GPIO_20",197                  type=bool,initial=Sample(0, False),                  198                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,199                  options=DPROP_OPT_AUTOTIMESTAMP,200                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_20",201                                                      sample=sample)),202            ChannelSourceDeviceProperty(name="GPIO_21",203                  type=bool,initial=Sample(0, False),                  204                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,205                  options=DPROP_OPT_AUTOTIMESTAMP,206                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_21",207                                                      sample=sample)),208            ChannelSourceDeviceProperty(name="GPIO_22",209                  type=bool,initial=Sample(0, False),                  210                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,211                  options=DPROP_OPT_AUTOTIMESTAMP,212                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_22",213                                                      sample=sample)),214            ChannelSourceDeviceProperty(name="GPIO_23",215                  type=bool,initial=Sample(0, False),                  216                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,217                  options=DPROP_OPT_AUTOTIMESTAMP,218                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_23",219                                                      sample=sample)),220            ChannelSourceDeviceProperty(name="GPIO_24",221                  type=bool,initial=Sample(0, False),                  222                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,223                  options=DPROP_OPT_AUTOTIMESTAMP,224                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_24",225                                                      sample=sample)),226            ChannelSourceDeviceProperty(name="GPIO_25",227                  type=bool,initial=Sample(0, False),                  228                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,229                  options=DPROP_OPT_AUTOTIMESTAMP,230                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_25",231                                                      sample=sample)),232            ChannelSourceDeviceProperty(name="GPIO_26",233                  type=bool,initial=Sample(0, False),                  234                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,235                  options=DPROP_OPT_AUTOTIMESTAMP,236                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_26",237                                                      sample=sample)),238            ChannelSourceDeviceProperty(name="GPIO_27",239                  type=bool,initial=Sample(0, False),                  240                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,241                  options=DPROP_OPT_AUTOTIMESTAMP,242                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_27",243                                                      sample=sample)),244            ChannelSourceDeviceProperty(name="GPIO_28",245                  type=bool,initial=Sample(0, False),                  246                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,247                  options=DPROP_OPT_AUTOTIMESTAMP,248                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_28",249                                                      sample=sample)),250            ChannelSourceDeviceProperty(name="GPIO_29",251                  type=bool,initial=Sample(0, False),                  252                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,253                  options=DPROP_OPT_AUTOTIMESTAMP,254                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_29",255                                                      sample=sample)),256            ChannelSourceDeviceProperty(name="GPIO_30",257                  type=bool,initial=Sample(0, False),                  258                  perms_mask=DPROP_PERM_GET|DPROP_PERM_SET,259                  options=DPROP_OPT_AUTOTIMESTAMP,260                  set_cb=lambda sample: self.set_gpio(gpio="GPIO_30",261                                                      sample=sample))262        ]263                                                                        264        ## Initialize the DeviceBase interface:265        DeviceBase.__init__(self, self.__name, self.__core,266                                settings_list, property_list)267        ## Thread initialization:268        self.__stopevent = threading.Event()269        threading.Thread.__init__(self, name=name)270        threading.Thread.setDaemon(self, True)271    ## Functions which must be implemented to conform to the DeviceBase272    ## interface:273    def apply_settings(self):274        SettingsBase.merge_settings(self)275        accepted, rejected, not_found = SettingsBase.verify_settings(self)276        if len(rejected) or len(not_found):277            self.__tracer.error("Settings rejected/not found: %s %s", 278                                rejected, not_found)279        if (('update_rate' in accepted) and 280            (accepted['update_rate'] > SHUTDOWN_WAIT)):281            self.__tracer.warning("Long update_rate setting may " + 282                                  "interfere with shutdown of Dia")283            284        SettingsBase.commit_settings(self, accepted)285        return (accepted, rejected, not_found)286    def start(self):287        threading.Thread.start(self)288        return True289    def stop(self):290        self.__stopevent.set()291        return True        292    # Threading related functions:293    def run(self):294        295        #Get the device properties296        time_sleep = SettingsBase.get_setting(self,"update_rate")297        self.input_gpios = SettingsBase.get_setting(self,"input_gpios")298        self.output_gpios = SettingsBase.get_setting(self,"output_gpios")299        300        #Call the GPIOs initializer method301        self.initialize_gpios()302        303        #Start the refresh thread304        while 1:305            if self.__stopevent.isSet():306                self.__stopevent.clear()307                break308            try:309                while self.setting_gpio:310                    pass311                self.get_GPIOs()312            except Exception, e:313                self.__tracer.error("Unable to update values: %s", str(e))314            315            #Sleep the time configured in settings (in seconds)316            time.sleep(time_sleep)317    def set_gpio(self,gpio,sample):318        319        self.setting_gpio = True320        #Update the channel and the module GPIO321        if sample.value==True:            322            self.property_set(gpio,Sample(time.time(), True))323            digihw.gpio_set_value(self.gpios_ind[gpio], 1)324        else:325            self.property_set(gpio,Sample(time.time(), False))326            digihw.gpio_set_value(self.gpios_ind[gpio], 0)327        self.setting_gpio = False328    def get_GPIOs(self):329        330        for gpio in self.input_gpios:331            val = digihw.gpio_get_value(gpio)332            #If the GPIO value has changed, update its channel333            if self.gpios["GPIO_"+str(gpio)]!=val:334                self.gpios["GPIO_"+str(gpio)]=val            335                if val==0 :336                    self.property_set("GPIO_"+str(gpio),337                        Sample(time.time(), False))338                else:339                    self.property_set("GPIO_"+str(gpio),340                        Sample(time.time(), True))341    342    def initialize_gpios(self):343        344        for gpio in self.output_gpios:345            digihw.gpio_set_value(gpio,0)346        for gpio in self.input_gpios:347            digihw.gpio_set_input(gpio)348            ...

task.py

Source:task.py

...3class Task(object):4    lock = _thread.allocate_lock()5    def __init__(self, event=(lambda args: print('task running')), args=None):6        self.alive = False7        self.set_cb(event, args)8    def set_cb(self, event, args):9        self.event = event10        self.args = args11    def run(self):12        if Task.lock.acquire():13            while self.alive:14                self.event(self.args)15            Task.lock.release()16        _thread.exit()17    def stop(self):18        if self.alive is True:19            self.alive = False20            if Task.lock.acquire():21                # print("stop")22                Task.lock.release()23    def start(self, size=2048):24        self.stop()25        self.alive = True26        if Task.lock.acquire():27            # print("start")28            import gc29            gc.collect()30            _thread.stack_size(size)31            _thread.start_new_thread(self.run, ())32            _thread.stack_size()33            Task.lock.release()34if __name__ == "__main__":35    import time36    def unit_test(args):37        print('unit_test', args)38        sleep_ms(500)39    tmp = Task(unit_test, {'name': 'unit_test_0'})40    tmp.start()41    time.sleep(1)42    tmp.set_cb(unit_test, {'name': 'unit_test_1'})43    tmp.start()44    time.sleep(2)45    tmp.set_cb(unit_test, {'name': 'unit_test_2'})46    tmp.start()47    # time.sleep(3)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.