Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run toolium automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
