How to use connect_wrapper method in toolium

Best Python code snippet using toolium_python

test_Smile.py

Source:test_Smile.py

...151        except (Smile.DeviceTimeoutError, Smile.InvalidXMLError) as e:152            await self.disconnect(server, client)153            raise e154    # Wrap connect for invalid connections155    async def connect_wrapper(self, put_timeout=False):156        """Wrap connect to try negative testing before positive testing."""157        if put_timeout:158            _LOGGER.warning("Connecting to device exceeding timeout in handling:")159            return await self.connect(put_timeout=True)160        try:161            _LOGGER.warning("Connecting to device exceeding timeout in response:")162            await self.connect(timeout=True)163            _LOGGER.error(" - timeout not handled")164            raise self.ConnectError165        except (Smile.DeviceTimeoutError, Smile.ResponseError):166            _LOGGER.info(" + succesfully passed timeout handling.")167        try:168            _LOGGER.warning("Connecting to device with missing data:")169            await self.connect(broken=True)170            _LOGGER.error(" - broken information not handled")171            raise self.ConnectError172        except Smile.InvalidXMLError:173            _LOGGER.info(" + succesfully passed XML issue handling.")174        _LOGGER.info("Connecting to functioning device:")175        return await self.connect()176    # Generic disconnect177    @pytest.mark.asyncio178    async def disconnect(self, server, client):179        """Disconnect from webserver."""180        await client.session.close()181        await server.close()182    @staticmethod183    def show_setup(location_list, device_list):184        """Show informative outline of the setup."""185        _LOGGER.info("This environment looks like:")186        for loc_id, loc_info in location_list.items():187            _LOGGER.info(188                "  --> Location: %s", "{} ({})".format(loc_info["name"], loc_id)189            )190            device_count = 0191            for dev_id, dev_info in device_list.items():192                if dev_info["location"] == loc_id:193                    device_count += 1194                    _LOGGER.info(195                        "      + Device: %s",196                        "{} ({} - {})".format(197                            dev_info["name"], dev_info["class"], dev_id198                        ),199                    )200            if device_count == 0:201                _LOGGER.info("      ! no devices found in this location")202    @pytest.mark.asyncio203    async def device_test(self, smile=Smile, testdata=None):204        """Perform basic device tests."""205        _LOGGER.info("Asserting testdata:")206        device_list = smile.get_all_devices()207        self._write_json("get_all_devices", device_list)208        location_list, dummy = smile.scan_thermostats()209        _LOGGER.info("Gateway id = %s", smile.gateway_id)210        _LOGGER.info("Hostname = %s", smile.smile_hostname)211        self.show_setup(location_list, device_list)212        pp4 = PrettyPrinter(indent=4)213        pp8 = PrettyPrinter(indent=8)214        _LOGGER.debug("Device list:\n%s", pp4.pformat(device_list))215        for dev_id, details in device_list.items():216            data = smile.get_device_data(dev_id)217            self._write_json("get_device_data/" + dev_id, data)218            _LOGGER.debug(219                "%s",220                "Device {} id:{}\nDetails: {}\nData: {}".format(221                    details["name"], dev_id, pp4.pformat(details), pp8.pformat(data)222                ),223            )224        for testdevice, measurements in testdata.items():225            assert testdevice in device_list226            # if testdevice not in device_list:227            #    _LOGGER.info("Device {} to test against {} not found in device_list for {}".format(testdevice,measurements,self.smile_setup))228            # else:229            #    _LOGGER.info("Device {} to test found in {}".format(testdevice,device_list))230            for dev_id, details in device_list.items():231                if testdevice == dev_id:232                    data = smile.get_device_data(dev_id)233                    _LOGGER.info(234                        "%s",235                        "- Testing data for device {} ({})".format(236                            details["name"], dev_id237                        ),238                    )239                    _LOGGER.info("  + Device data: %s", data)240                    for measure_key, measure_assert in measurements.items():241                        _LOGGER.info(242                            "%s",243                            "  + Testing {} (should be {})".format(244                                measure_key, measure_assert245                            ),246                        )247                        if isinstance(data[measure_key], float):248                            if float(data[measure_key]) < 10:249                                measure = float(250                                    "{:.2f}".format(round(float(data[measure_key]), 2))251                                )252                            else:253                                measure = float(254                                    "{:.1f}".format(round(float(data[measure_key]), 1))255                                )256                            assert measure == measure_assert257                        else:258                            assert data[measure_key] == measure_assert259    @pytest.mark.asyncio260    async def tinker_relay(self, smile, dev_ids=None, unhappy=False):261        """Switch a relay on and off to test functionality."""262        _LOGGER.info("Asserting modifying settings for relay devices:")263        for dev_id in dev_ids:264            _LOGGER.info("- Devices (%s):", dev_id)265            for new_state in [False, True, False]:266                _LOGGER.info("- Switching %s", new_state)267                try:268                    relay_change = await smile.set_relay_state(dev_id, None, new_state)269                    assert relay_change270                    _LOGGER.info("  + worked as intended")271                except (Smile.ErrorSendingCommandError, Smile.ResponseError):272                    if unhappy:273                        _LOGGER.info("  + failed as expected")274                    else:275                        _LOGGER.info("  - failed unexpectedly")276                        raise self.UnexpectedError277    @pytest.mark.asyncio278    async def tinker_thermostat(self, smile, loc_id, good_schemas=None, unhappy=False):279        """Toggle various climate settings to test functionality."""280        if good_schemas is None:281            good_schemas = ["Weekschema"]282        _LOGGER.info("Asserting modifying settings in location (%s):", loc_id)283        for new_temp in [20.0, 22.9]:284            _LOGGER.info("- Adjusting temperature to %s", new_temp)285            try:286                temp_change = await smile.set_temperature(loc_id, new_temp)287                assert temp_change288                _LOGGER.info("  + worked as intended")289            except (Smile.ErrorSendingCommandError, Smile.ResponseError):290                if unhappy:291                    _LOGGER.info("  + failed as expected")292                else:293                    _LOGGER.info("  - failed unexpectedly")294                    raise self.UnexpectedError295        for new_preset in ["asleep", "home", "!bogus"]:296            assert_state = True297            warning = ""298            if new_preset[0] == "!":299                assert_state = False300                warning = " Negative test"301                new_preset = new_preset[1:]302            _LOGGER.info("%s", "- Adjusting preset to {}{}".format(new_preset, warning))303            try:304                preset_change = await smile.set_preset(loc_id, new_preset)305                assert preset_change == assert_state306                _LOGGER.info("  + worked as intended")307            except (Smile.ErrorSendingCommandError, Smile.ResponseError):308                if unhappy:309                    _LOGGER.info("  + failed as expected")310                else:311                    _LOGGER.info("  - failed unexpectedly")312                    raise self.UnexpectedError313        if good_schemas is not []:314            good_schemas.append("!VeryBogusSchemaNameThatNobodyEverUsesOrShouldUse")315            for new_schema in good_schemas:316                assert_state = True317                warning = ""318                if new_schema[0] == "!":319                    assert_state = False320                    warning = " Negative test"321                    new_schema = new_schema[1:]322                _LOGGER.info(323                    "- Adjusting schedule to %s", "{}{}".format(new_schema, warning)324                )325                try:326                    schema_change = await smile.set_schedule_state(327                        loc_id, new_schema, "auto"328                    )329                    assert schema_change == assert_state330                    _LOGGER.info("  + failed as intended")331                except (Smile.ErrorSendingCommandError, Smile.ResponseError):332                    if unhappy:333                        _LOGGER.info("  + failed as expected before intended failure")334                    else:335                        _LOGGER.info("  - suceeded unexpectedly for some reason")336                        raise self.UnexpectedError337        else:338            _LOGGER.info("- Skipping schema adjustments")339    @pytest.mark.asyncio340    async def test_connect_legacy_anna(self):341        """Test a legacy Anna device."""342        # testdata is a dictionary with key ctrl_id_dev_id => keys:values343        # testdata={344        #             'ctrl_id': { 'outdoor+temp': 20.0, }345        #             'ctrl_id:dev_id': { 'type': 'thermostat', 'battery': None, }346        #         }347        testdata = {348            # Anna349            "0d266432d64443e283b5d708ae98b455": {350                "setpoint": 20.5,351                "temperature": 20.4,352                "illuminance": 151,353            },354            # Central355            "04e4cbfe7f4340f090f85ec3b9e6a950": {356                "water_temperature": 23.6,357                "water_pressure": 1.2,358                "modulation_level": 0.0,359                "heating_state": True,360            },361        }362        self.smile_setup = "legacy_anna"363        server, smile, client = await self.connect_wrapper()364        assert smile.smile_hostname is None365        _LOGGER.info("Basics:")366        _LOGGER.info(" # Assert type = thermostat")367        assert smile.smile_type == "thermostat"368        _LOGGER.info(" # Assert version")369        assert smile.smile_version[0] == "1.8.0"370        _LOGGER.info(" # Assert legacy")371        assert smile._smile_legacy  # pylint: disable=protected-access372        _LOGGER.info(" # Assert master thermostat")373        assert smile.single_master_thermostat()374        assert not smile.notifications375        await self.device_test(smile, testdata)376        await self.tinker_thermostat(377            smile,378            "c34c6864216446528e95d88985e714cc",379            good_schemas=[380                "Thermostat schedule",381            ],382        )383        await smile.close_connection()384        await self.disconnect(server, client)385        server, smile, client = await self.connect_wrapper(put_timeout=True)386        await self.tinker_thermostat(387            smile,388            "c34c6864216446528e95d88985e714cc",389            good_schemas=[390                "Thermostat schedule",391            ],392            unhappy=True,393        )394        await smile.close_connection()395        await self.disconnect(server, client)396    @pytest.mark.asyncio397    async def test_connect_legacy_anna_2(self):398        """Test a legacy Anna device."""399        # testdata is a dictionary with key ctrl_id_dev_id => keys:values400        # testdata={401        #             'ctrl_id': { 'outdoor+temp': 20.0, }402        #             'ctrl_id:dev_id': { 'type': 'thermostat', 'battery': None, }403        #         }404        testdata = {405            # Anna406            "9e7377867dc24e51b8098a5ba02bd89d": {407                "setpoint": 15.0,408                "temperature": 21.4,409                "illuminance": 19.5,410            },411            # Central412            "ea5d8a7177e541b0a4b52da815166de4": {413                "water_pressure": 1.7,414            },415        }416        self.smile_setup = "legacy_anna_2"417        server, smile, client = await self.connect_wrapper()418        assert smile.smile_hostname is None419        _LOGGER.info("Basics:")420        _LOGGER.info(" # Assert type = thermostat")421        assert smile.smile_type == "thermostat"422        _LOGGER.info(" # Assert version")423        assert smile.smile_version[0] == "1.8.0"424        _LOGGER.info(" # Assert legacy")425        assert smile._smile_legacy  # pylint: disable=protected-access426        _LOGGER.info(" # Assert master thermostat")427        assert smile.single_master_thermostat()428        assert not smile.notifications429        await self.device_test(smile, testdata)430        await self.tinker_thermostat(431            smile,432            "c34c6864216446528e95d88985e714cc",433            good_schemas=[434                "Thermostat schedule",435            ],436        )437        await smile.close_connection()438        await self.disconnect(server, client)439        server, smile, client = await self.connect_wrapper(put_timeout=True)440        await self.tinker_thermostat(441            smile,442            "c34c6864216446528e95d88985e714cc",443            good_schemas=[444                "Thermostat schedule",445            ],446            unhappy=True,447        )448        await smile.close_connection()449        await self.disconnect(server, client)450    @pytest.mark.asyncio451    async def test_connect_smile_p1_v2(self):452        """Test a legacy P1 device."""453        # testdata dictionary with key ctrl_id_dev_id => keys:values454        testdata = {455            # Gateway / P1 itself456            "938696c4bcdb4b8a9a595cb38ed43913": {457                "electricity_consumed_peak_point": 458.0,458                "net_electricity_point": 458.0,459                "gas_consumed_cumulative": 584.4,460                "electricity_produced_peak_cumulative": 1296136.0,461                "electricity_produced_off_peak_cumulative": 482598.0,462            }463        }464        self.smile_setup = "smile_p1_v2"465        server, smile, client = await self.connect_wrapper()466        assert smile.smile_hostname == "smile000000"467        _LOGGER.info("Basics:")468        _LOGGER.info(" # Assert type = power")469        assert smile.smile_type == "power"470        _LOGGER.info(" # Assert version")471        assert smile.smile_version[0] == "2.5.9"472        _LOGGER.info(" # Assert legacy")473        assert smile._smile_legacy  # pylint: disable=protected-access474        _LOGGER.info(" # Assert no master thermostat")475        assert smile.single_master_thermostat() is None  # it's not a thermostat :)476        assert not smile.notifications477        await self.device_test(smile, testdata)478        await smile.close_connection()479        await self.disconnect(server, client)480        server, smile, client = await self.connect_wrapper(put_timeout=True)481    @pytest.mark.asyncio482    async def test_connect_smile_p1_v2_2(self):483        """Test another legacy P1 device."""484        # testdata dictionary with key ctrl_id_dev_id => keys:values485        testdata = {486            # Gateway / P1 itself487            "199aa40f126840f392983d171374ab0b": {488                "electricity_consumed_peak_point": 368.0,489                "net_electricity_point": 368.0,490                "gas_consumed_cumulative": 2638.0,491                "electricity_produced_peak_cumulative": 0.0,492            }493        }494        self.smile_setup = "smile_p1_v2_2"495        server, smile, client = await self.connect_wrapper()496        assert smile.smile_hostname == "smile000000"497        _LOGGER.info("Basics:")498        _LOGGER.info(" # Assert type = power")499        assert smile.smile_type == "power"500        _LOGGER.info(" # Assert version")501        assert smile.smile_version[0] == "2.5.9"502        _LOGGER.info(" # Assert legacy")503        assert smile._smile_legacy  # pylint: disable=protected-access504        _LOGGER.info(" # Assert no master thermostat")505        assert smile.single_master_thermostat() is None  # it's not a thermostat :)506        assert not smile.notifications507        await self.device_test(smile, testdata)508        await smile.close_connection()509        await self.disconnect(server, client)510    @pytest.mark.asyncio511    async def test_connect_anna_v4(self):512        """Test an Anna firmware 4 setup without a boiler."""513        # testdata is a dictionary with key ctrl_id_dev_id => keys:values514        # testdata={515        #             'ctrl_id': { 'outdoor+temp': 20.0, }516        #             'ctrl_id:dev_id': { 'type': 'thermostat', 'battery': None, }517        #         }518        testdata = {519            # Anna520            "01b85360fdd243d0aaad4d6ac2a5ba7e": {521                "selected_schedule": None,522                "illuminance": 60.0,523                "active_preset": "home",524            },525            # Central526            "cd0e6156b1f04d5f952349ffbe397481": {527                "heating_state": True,528                "water_pressure": 2.1,529                "water_temperature": 52.0,530            },531            "0466eae8520144c78afb29628384edeb": {532                "outdoor_temperature": 7.44,533            },534        }535        self.smile_setup = "anna_v4"536        server, smile, client = await self.connect_wrapper()537        assert smile.smile_hostname == "smile000000"538        _LOGGER.info("Basics:")539        _LOGGER.info(" # Assert type = thermostat")540        assert smile.smile_type == "thermostat"541        _LOGGER.info(" # Assert version")542        assert smile.smile_version[0] == "4.0.15"543        _LOGGER.info(" # Assert no legacy")544        assert not smile._smile_legacy  # pylint: disable=protected-access545        _LOGGER.info(" # Assert master thermostat")546        assert smile.single_master_thermostat()547        assert not smile.notifications548        await self.device_test(smile, testdata)549        await self.tinker_thermostat(550            smile,551            "eb5309212bf5407bb143e5bfa3b18aee",552            good_schemas=["Standaard", "Thuiswerken"],553        )554        await smile.close_connection()555        await self.disconnect(server, client)556        server, smile, client = await self.connect_wrapper(put_timeout=True)557        await self.tinker_thermostat(558            smile,559            "eb5309212bf5407bb143e5bfa3b18aee",560            good_schemas=["Standaard", "Thuiswerken"],561            unhappy=True,562        )563        await smile.close_connection()564        await self.disconnect(server, client)565    @pytest.mark.asyncio566    async def test_connect_anna_without_boiler_fw3(self):567        """Test an Anna firmware 3 without a boiler."""568        # testdata is a dictionary with key ctrl_id_dev_id => keys:values569        # testdata={570        #             'ctrl_id': { 'outdoor+temp': 20.0, }571        #             'ctrl_id:dev_id': { 'type': 'thermostat', 'battery': None, }572        #         }573        testdata = {574            # Anna575            "7ffbb3ab4b6c4ab2915d7510f7bf8fe9": {576                "selected_schedule": "Normal",577                "illuminance": 35.0,578                "active_preset": "away",579            },580            "a270735e4ccd45239424badc0578a2b1": {581                "outdoor_temperature": 10.8,582            },583            # Central584            "c46b4794d28149699eacf053deedd003": {585                "heating_state": False,586            },587        }588        self.smile_setup = "anna_without_boiler_fw3"589        server, smile, client = await self.connect_wrapper()590        assert smile.smile_hostname == "smile000000"591        _LOGGER.info("Basics:")592        _LOGGER.info(" # Assert type = thermostat")593        assert smile.smile_type == "thermostat"594        _LOGGER.info(" # Assert version")595        assert smile.smile_version[0] == "3.1.11"596        _LOGGER.info(" # Assert no legacy")597        assert not smile._smile_legacy  # pylint: disable=protected-access598        _LOGGER.info(" # Assert master thermostat")599        assert smile.single_master_thermostat()600        assert not smile.notifications601        await self.device_test(smile, testdata)602        await self.tinker_thermostat(603            smile, "c34c6864216446528e95d88985e714cc", good_schemas=["Test", "Normal"]604        )605        await smile.close_connection()606        await self.disconnect(server, client)607        server, smile, client = await self.connect_wrapper(put_timeout=True)608        await self.tinker_thermostat(609            smile,610            "c34c6864216446528e95d88985e714cc",611            good_schemas=["Test", "Normal"],612            unhappy=True,613        )614        await smile.close_connection()615        await self.disconnect(server, client)616    @pytest.mark.asyncio617    async def test_connect_anna_without_boiler_fw4(self):618        """Test an Anna firmware 4 without a boiler."""619        # testdata is a dictionary with key ctrl_id_dev_id => keys:values620        # testdata={621        #             'ctrl_id': { 'outdoor+temp': 20.0, }622        #             'ctrl_id:dev_id': { 'type': 'thermostat', 'battery': None, }623        #         }624        testdata = {625            # Anna626            "7ffbb3ab4b6c4ab2915d7510f7bf8fe9": {627                "selected_schedule": "Normal",628                "illuminance": 44.8,629                "active_preset": "home",630            },631            "a270735e4ccd45239424badc0578a2b1": {632                "outdoor_temperature": 16.6,633            },634            # Central635            "c46b4794d28149699eacf053deedd003": {636                "heating_state": True,637            },638        }639        self.smile_setup = "anna_without_boiler_fw4"640        server, smile, client = await self.connect_wrapper()641        assert smile.smile_hostname == "smile000000"642        _LOGGER.info("Basics:")643        _LOGGER.info(" # Assert type = thermostat")644        assert smile.smile_type == "thermostat"645        _LOGGER.info(" # Assert version")646        assert smile.smile_version[0] == "4.0.15"647        _LOGGER.info(" # Assert no legacy")648        assert not smile._smile_legacy  # pylint: disable=protected-access649        _LOGGER.info(" # Assert master thermostat")650        assert smile.single_master_thermostat()651        assert not smile.notifications652        await self.device_test(smile, testdata)653        await self.tinker_thermostat(654            smile, "c34c6864216446528e95d88985e714cc", good_schemas=["Test", "Normal"]655        )656        await smile.close_connection()657        await self.disconnect(server, client)658        server, smile, client = await self.connect_wrapper(put_timeout=True)659        await self.tinker_thermostat(660            smile,661            "c34c6864216446528e95d88985e714cc",662            good_schemas=["Test", "Normal"],663            unhappy=True,664        )665        await smile.close_connection()666        await self.disconnect(server, client)667    """668    # TODO: This device setup needs work - doesn't seem to work straightforard669    # currently breaks on setting thermostat setpoint670    # Actual test for directory 'Adam'671    # living room floor radiator valve and separate zone thermostat672    # an three rooms with conventional radiators673    @pytest.mark.asyncio674    async def test_connect_adam(self):675        testdata = {676            "95395fb15c814a1f8bba88363e4a5833": { "temperature": 19.8, 'active_preset': 'home',},677            "450d49ef2e8942f78c1242cdd8dfecd0": { "temperature": 20.18, 'battery':  0.77, 'selected_schedule': 'Kira' },678            "bc9e18756ad04c3f9f35298cbe537c8e": { "temperature": 20.63, 'thermostat': 20.0 },679        }680        self.smile_setup = 'adam_living_floor_plus_3_rooms'681        server, smile, client = await self.connect_wrapper()682        assert smile.smile_type == "thermostat"683        assert smile.smile_version[0] == "2.3.35"684        assert not smile._smile_legacy685        await self.device_test(smile, testdata)686        await self.tinker_thermostat(687            smile, "95395fb15c814a1f8bba88363e4a5833", good_schemas=["Living room"]688        )689        await smile.close_connection()690        await self.disconnect(server, client)691        server, smile, client = await self.connect_wrapper(put_timeout=True)692        await self.tinker_thermostat(693            smile, "95395fb15c814a1f8bba88363e4a5833", good_schemas=["Living room"],694            unhappy=True,695        )696        await smile.close_connection()697        await self.disconnect(server, client)698    """699    @pytest.mark.asyncio700    async def test_connect_adam_plus_anna(self):701        """Test outdated information for Adam with Anna setup."""702        # testdata dictionary with key ctrl_id_dev_id => keys:values703        testdata = {704            # Anna705            "ee62cad889f94e8ca3d09021f03a660b": {706                "selected_schedule": "Weekschema",707                "last_used": "Weekschema",708                "active_preset": "home",709                "setpoint": 20.5,  # HA setpoint_temp710                "temperature": 20.5,  # HA current_temp711            },712            # Central713            "2743216f626f43948deec1f7ab3b3d70": {714                "heating_state": False,715            },716            "b128b4bbbd1f47e9bf4d756e8fb5ee94": {717                "outdoor_temperature": 11.9,718            },719            # Plug MediaCenter720            "aa6b0002df0a46e1b1eb94beb61eddfe": {721                "electricity_consumed": 10.3,722                "relay": True,723            },724        }725        self.smile_setup = "adam_plus_anna"726        server, smile, client = await self.connect_wrapper()727        assert smile.smile_hostname == "smile000000"728        _LOGGER.info("Basics:")729        _LOGGER.info(" # Assert type = thermostat")730        assert smile.smile_type == "thermostat"731        _LOGGER.info(" # Assert version")732        assert smile.smile_version[0] == "3.0.15"733        _LOGGER.info(" # Assert legacy")734        assert not smile._smile_legacy  # pylint: disable=protected-access735        _LOGGER.info(" # Assert master thermostat")736        assert smile.single_master_thermostat()737        assert not smile.notifications738        await self.device_test(smile, testdata)739        await self.tinker_thermostat(740            smile, "009490cc2f674ce6b576863fbb64f867", good_schemas=["Weekschema"]741        )742        await self.tinker_relay(smile, ["aa6b0002df0a46e1b1eb94beb61eddfe"])743        await smile.close_connection()744        await self.disconnect(server, client)745        server, smile, client = await self.connect_wrapper(put_timeout=True)746        await self.tinker_thermostat(747            smile,748            "009490cc2f674ce6b576863fbb64f867",749            good_schemas=["Weekschema"],750            unhappy=True,751        )752        await self.tinker_relay(753            smile, ["aa6b0002df0a46e1b1eb94beb61eddfe"], unhappy=True754        )755        await smile.close_connection()756        await self.disconnect(server, client)757    @pytest.mark.asyncio758    async def test_connect_adam_zone_per_device(self):759        """Test a broad setup of Adam with a zone per device setup."""760        # testdata dictionary with key ctrl_id_dev_id => keys:values761        testdata = {762            # Lisa WK763            "b59bcebaf94b499ea7d46e4a66fb62d8": {764                "setpoint": 21.5,765                "temperature": 21.1,766                "battery": 0.34,767            },768            # Floor WK769            "b310b72a0e354bfab43089919b9a88bf": {770                "setpoint": 21.5,771                "temperature": 26.2,772                "valve_position": 1.0,773            },774            # CV pomp775            "78d1126fc4c743db81b61c20e88342a7": {776                "electricity_consumed": 35.8,777                "relay": True,778            },779            # Lisa Bios780            "df4a4a8169904cdb9c03d61a21f42140": {781                "setpoint": 13.0,782                "temperature": 16.5,783                "battery": 0.67,784            },785            # Adam786            "90986d591dcd426cae3ec3e8111ff730": {"intended_boiler_temperature": 70.0},787            "fe799307f1624099878210aa0b9f1475": {788                "outdoor_temperature": 7.69,789            },790            # Modem791            "675416a629f343c495449970e2ca37b5": {792                "electricity_consumed": 12.2,793                "relay": True,794            },795        }796        self.smile_setup = "adam_zone_per_device"797        server, smile, client = await self.connect_wrapper()798        assert smile.smile_hostname == "smile000000"799        _LOGGER.info("Basics:")800        _LOGGER.info(" # Assert type = thermostat")801        assert smile.smile_type == "thermostat"802        _LOGGER.info(" # Assert version")803        assert smile.smile_version[0] == "3.0.15"804        _LOGGER.info(" # Assert legacy")805        assert not smile._smile_legacy  # pylint: disable=protected-access806        _LOGGER.info(" # Assert master thermostat")807        assert not smile.single_master_thermostat()808        assert "af82e4ccf9c548528166d38e560662a4" in smile.notifications809        await self.device_test(smile, testdata)810        await self.tinker_thermostat(811            smile, "c50f167537524366a5af7aa3942feb1e", good_schemas=["GF7  Woonkamer"]812        )813        await self.tinker_thermostat(814            smile, "82fa13f017d240daa0d0ea1775420f24", good_schemas=["CV Jessie"]815        )816        await self.tinker_relay(smile, ["675416a629f343c495449970e2ca37b5"])817        await smile.close_connection()818        await self.disconnect(server, client)819        server, smile, client = await self.connect_wrapper(put_timeout=True)820        await self.tinker_thermostat(821            smile,822            "c50f167537524366a5af7aa3942feb1e",823            good_schemas=["GF7  Woonkamer"],824            unhappy=True,825        )826        await self.tinker_thermostat(827            smile,828            "82fa13f017d240daa0d0ea1775420f24",829            good_schemas=["CV Jessie"],830            unhappy=True,831        )832        await smile.close_connection()833        await self.disconnect(server, client)834    @pytest.mark.asyncio835    async def test_connect_adam_multiple_devices_per_zone(self):836        """Test a broad setup of Adam with multiple devices per zone setup."""837        # testdata dictionary with key ctrl_id_dev_id => keys:values838        testdata = {839            # Lisa WK840            "b59bcebaf94b499ea7d46e4a66fb62d8": {841                "setpoint": 21.5,842                "temperature": 20.9,843                "battery": 0.34,844            },845            # Floor WK846            "b310b72a0e354bfab43089919b9a88bf": {847                "setpoint": 21.5,848                "temperature": 26.0,849                "valve_position": 1.0,850            },851            # CV pomp852            "78d1126fc4c743db81b61c20e88342a7": {853                "electricity_consumed": 35.6,854                "relay": True,855            },856            # Lisa Bios857            "df4a4a8169904cdb9c03d61a21f42140": {858                "setpoint": 13.0,859                "temperature": 16.5,860                "battery": 0.67,861            },862            # Adam863            "90986d591dcd426cae3ec3e8111ff730": {"intended_boiler_temperature": 70.0},864            "fe799307f1624099878210aa0b9f1475": {865                "outdoor_temperature": 7.81,866            },867            # Modem868            "675416a629f343c495449970e2ca37b5": {869                "electricity_consumed": 12.2,870                "relay": True,871            },872        }873        self.smile_setup = "adam_multiple_devices_per_zone"874        server, smile, client = await self.connect_wrapper()875        assert smile.smile_hostname == "smile000000"876        _LOGGER.info("Basics:")877        _LOGGER.info(" # Assert type = thermostat")878        assert smile.smile_type == "thermostat"879        _LOGGER.info(" # Assert version")880        assert smile.smile_version[0] == "3.0.15"881        _LOGGER.info(" # Assert legacy")882        assert not smile._smile_legacy  # pylint: disable=protected-access883        _LOGGER.info(" # Assert master thermostat")884        assert not smile.single_master_thermostat()885        assert "af82e4ccf9c548528166d38e560662a4" in smile.notifications886        await self.device_test(smile, testdata)887        await self.tinker_thermostat(888            smile, "c50f167537524366a5af7aa3942feb1e", good_schemas=["GF7  Woonkamer"]889        )890        await self.tinker_thermostat(891            smile, "82fa13f017d240daa0d0ea1775420f24", good_schemas=["CV Jessie"]892        )893        await self.tinker_relay(smile, ["675416a629f343c495449970e2ca37b5"])894        await smile.close_connection()895        await self.disconnect(server, client)896        server, smile, client = await self.connect_wrapper(put_timeout=True)897        await self.tinker_thermostat(898            smile,899            "c50f167537524366a5af7aa3942feb1e",900            good_schemas=["GF7  Woonkamer"],901            unhappy=True,902        )903        await self.tinker_thermostat(904            smile,905            "82fa13f017d240daa0d0ea1775420f24",906            good_schemas=["CV Jessie"],907            unhappy=True,908        )909        await smile.close_connection()910        await self.disconnect(server, client)911    @pytest.mark.asyncio912    async def test_connect_p1v3(self):913        """Test a P1 firmware 3 with only electricity setup."""914        # testdata dictionary with key ctrl_id_dev_id => keys:values915        testdata = {916            # Gateway / P1 itself917            "ba4de7613517478da82dd9b6abea36af": {918                "electricity_consumed_peak_point": 650.0,919                "electricity_produced_peak_cumulative": 0.0,920                "electricity_consumed_off_peak_cumulative": 10263159.0,921            }922        }923        self.smile_setup = "p1v3"924        server, smile, client = await self.connect_wrapper()925        assert smile.smile_hostname == "smile000000"926        _LOGGER.info("Basics:")927        _LOGGER.info(" # Assert type = power")928        assert smile.smile_type == "power"929        _LOGGER.info(" # Assert version")930        assert smile.smile_version[0] == "3.3.6"931        _LOGGER.info(" # Assert no master thermostat")932        assert not smile._smile_legacy  # pylint: disable=protected-access933        assert smile.single_master_thermostat() is None  # it's not a thermostat :)934        assert not smile.notifications935        await self.device_test(smile, testdata)936        await smile.close_connection()937        await self.disconnect(server, client)938    @pytest.mark.asyncio939    async def test_connect_p1v3solarfake(self):940        """Test a P1 firmware 3 with manually added solar setup."""941        # testdata dictionary with key ctrl_id_dev_id => keys:values942        testdata = {943            # Gateway / P1 itself944            "ba4de7613517478da82dd9b6abea36af": {945                "electricity_consumed_peak_point": 644.0,946                "electricity_produced_peak_cumulative": 20000.0,947                "electricity_consumed_off_peak_cumulative": 10263159.0,948                "net_electricity_point": 244,949            }950        }951        self.smile_setup = "p1v3solarfake"952        server, smile, client = await self.connect_wrapper()953        assert smile.smile_hostname == "smile000000"954        _LOGGER.info("Basics:")955        _LOGGER.info(" # Assert type = power")956        assert smile.smile_type == "power"957        _LOGGER.info(" # Assert version")958        assert smile.smile_version[0] == "3.3.6"959        _LOGGER.info(" # Assert no legacy")960        assert not smile._smile_legacy  # pylint: disable=protected-access961        _LOGGER.info(" # Assert nomaster thermostat")962        assert smile.single_master_thermostat() is None  # it's not a thermostat :)963        assert not smile.notifications964        await self.device_test(smile, testdata)965        await smile.close_connection()966        await self.disconnect(server, client)967    @pytest.mark.asyncio968    async def test_connect_p1v3_full_option(self):969        """Test a P1 firmware 3 full option (gas and solar) setup."""970        # testdata dictionary with key ctrl_id_dev_id => keys:values971        testdata = {972            # Gateway / P1 itself973            "e950c7d5e1ee407a858e2a8b5016c8b3": {974                "electricity_consumed_peak_point": 0.0,975                "electricity_produced_peak_cumulative": 396559.0,976                "electricity_consumed_off_peak_cumulative": 551090.0,977                "electricity_produced_peak_point": 2761.0,978                "net_electricity_point": -2761.0,979                "gas_consumed_cumulative": 584.9,980            }981        }982        self.smile_setup = "p1v3_full_option"983        server, smile, client = await self.connect_wrapper()984        assert smile.smile_hostname == "smile000000"985        _LOGGER.info("Basics:")986        _LOGGER.info(" # Assert type = power")987        assert smile.smile_type == "power"988        _LOGGER.info(" # Assert version")989        assert smile.smile_version[0] == "3.3.9"990        _LOGGER.info(" # Assert legacy")991        assert not smile._smile_legacy  # pylint: disable=protected-access992        _LOGGER.info(" # Assert no master thermostat")993        assert smile.single_master_thermostat() is None  # it's not a thermostat :)994        assert not smile.notifications995        await self.device_test(smile, testdata)996        await smile.close_connection()997        await self.disconnect(server, client)998    @pytest.mark.asyncio999    async def test_connect_anna_heatpump(self):1000        """Test a Anna with Elga setup in idle mode."""1001        # testdata dictionary with key ctrl_id_dev_id => keys:values1002        testdata = {1003            # Anna1004            "3cb70739631c4d17a86b8b12e8a5161b": {1005                "selected_schedule": "standaard",1006                "illuminance": 86.0,1007                "active_preset": "home",1008            },1009            # Central1010            "1cbf783bb11e4a7c8a6843dee3a86927": {1011                "dhw_state": False,1012                "water_temperature": 29.1,1013                "water_pressure": 1.57,1014            },1015            "015ae9ea3f964e668e490fa39da3870b": {1016                "outdoor_temperature": 20.2,1017            },1018        }1019        self.smile_setup = "anna_heatpump"1020        server, smile, client = await self.connect_wrapper()1021        assert smile.smile_hostname == "smile000000"1022        _LOGGER.info("Basics:")1023        _LOGGER.info(" # Assert type = thermostat")1024        assert smile.smile_type == "thermostat"1025        _LOGGER.info(" # Assert version")1026        assert smile.smile_version[0] == "4.0.15"1027        _LOGGER.info(" # Assert no legacy")1028        assert not smile._smile_legacy  # pylint: disable=protected-access1029        _LOGGER.info(" # Assert master thermostat")1030        assert smile.single_master_thermostat()1031        assert not smile.notifications1032        await self.device_test(smile, testdata)1033        await smile.close_connection()1034        await self.disconnect(server, client)1035    @pytest.mark.asyncio1036    async def test_connect_anna_heatpump_cooling(self):1037        """Test a Anna with Elga setup in cooling mode."""1038        # testdata dictionary with key ctrl_id_dev_id => keys:values1039        testdata = {1040            # Anna1041            "3cb70739631c4d17a86b8b12e8a5161b": {1042                "selected_schedule": None,1043                "illuminance": 24.5,1044                "active_preset": "home",1045            },1046            # Central1047            "1cbf783bb11e4a7c8a6843dee3a86927": {1048                "dhw_state": False,1049                "water_temperature": 24.7,1050                "water_pressure": 1.61,1051            },1052            "015ae9ea3f964e668e490fa39da3870b": {1053                "outdoor_temperature": 22.0,1054            },1055        }1056        self.smile_setup = "anna_heatpump_cooling"1057        server, smile, client = await self.connect_wrapper()1058        assert smile.smile_hostname == "smile000000"1059        _LOGGER.info("Basics:")1060        _LOGGER.info(" # Assert type = thermostat")1061        assert smile.smile_type == "thermostat"1062        _LOGGER.info(" # Assert version")1063        assert smile.smile_version[0] == "4.0.15"1064        _LOGGER.info(" # Assert no legacy")1065        assert not smile._smile_legacy  # pylint: disable=protected-access1066        _LOGGER.info(" # Assert master thermostat")1067        assert smile.single_master_thermostat()1068        assert not smile.notifications1069        await self.device_test(smile, testdata)1070        await smile.close_connection()1071        await self.disconnect(server, client)1072    @pytest.mark.asyncio1073    async def test_connect_adam_plus_anna_copy_with_error_domain_added(self):1074        """Test erronous domain_objects file from user."""1075        # testdata dictionary with key ctrl_id_dev_id => keys:values1076        self.smile_setup = "adam_plus_anna_copy_with_error_domain_added"1077        server, smile, client = await self.connect_wrapper()1078        assert smile.smile_hostname == "smile000000"1079        _LOGGER.info("Basics:")1080        _LOGGER.info(" # Assert type = thermostat")1081        assert smile.smile_type == "thermostat"1082        _LOGGER.info(" # Assert version")1083        assert smile.smile_version[0] == "3.0.23"1084        _LOGGER.info(" # Assert legacy")1085        assert not smile._smile_legacy  # pylint: disable=protected-access1086        _LOGGER.info(" # Assert master thermostat")1087        assert smile.single_master_thermostat()1088        assert "3d28a20e17cb47dca210a132463721d5" in smile.notifications1089        await smile.close_connection()1090        await self.disconnect(server, client)1091    @pytest.mark.asyncio1092    async def test_connect_stretch_v31(self):1093        """Test erronous domain_objects file from user."""1094        # testdata dictionary with key ctrl_id_dev_id => keys:values1095        testdata = {1096            # Koelkast1097            "e1c884e7dede431dadee09506ec4f859": {1098                "electricity_consumed": 53.2,1099                "relay": True,1100            },1101            # Droger1102            "cfe95cf3de1948c0b8955125bf754614": {1103                "electricity_consumed_interval": 1.06,1104            },1105        }1106        self.smile_setup = "stretch_v31"1107        server, smile, client = await self.connect_wrapper()1108        assert smile.smile_hostname == "stretch000000"1109        _LOGGER.info("Basics:")1110        _LOGGER.info(" # Assert type = thermostat")1111        assert smile.smile_type == "stretch"1112        _LOGGER.info(" # Assert version")1113        assert smile.smile_version[0] == "3.1.11"1114        _LOGGER.info(" # Assert legacy")1115        assert smile._smile_legacy  # pylint: disable=protected-access1116        await self.device_test(smile, testdata)1117        await smile.close_connection()1118        await self.disconnect(server, client)1119    @pytest.mark.asyncio1120    async def test_fail_legacy_system(self):1121        """Test erronous legacy stretch system."""1122        self.smile_setup = "faulty_stretch"1123        try:1124            server, smile, client = await self.connect_wrapper()1125            assert False1126        except Smile.ConnectionFailedError:1127            assert True1128    class PlugwiseTestError(Exception):1129        """Plugwise test exceptions class."""1130    class ConnectError(PlugwiseTestError):1131        """Raised when connectivity test fails."""1132    class UnexpectedError(PlugwiseTestError):...

backend.py

Source:backend.py

...21    cur = conn.cursor()22    cur.execute("CREATE TABLE IF NOT EXISTS rides (id INTEGER PRIMARY KEY, date DATE, Karol TEXT, Pioter TEXT)")23    conn.commit()24    conn.close()25def connect_wrapper(func):26    def func_with_wrapper(*args, **kwargs):27        conn = sqlite3.connect('rides')28        cur = conn.cursor()29        result = func(cur, *args, **kwargs)30        conn.commit()31        conn.close()32        return result33    return func_with_wrapper34@connect_wrapper35def check_calendar_day(cur, day):36    cur.execute("SELECT * FROM rides WHERE date=?", (day,))37    results = cur.fetchone()38    if results:39        car_karol = results[2]...

MySQLdb.py

Source:MySQLdb.py

1# pylint: disable=C01032"""3MySQLdb patcher module4"""5from __future__ import absolute_import6import wrapt7from .db_wrapper import connect_wrapper8def patch():9    """10    patch module.11    :return: None12    """13    wrapt.wrap_function_wrapper(14        'MySQLdb',15        'connect',16        connect_wrapper17    )18    wrapt.wrap_function_wrapper(19        'MySQLdb',20        'Connection',21        connect_wrapper22    )23    wrapt.wrap_function_wrapper(24        'MySQLdb',25        'Connect',26        connect_wrapper...

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.