From b4483576724e5fc6c2c57e445aae4fc3fe373ddc Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 15:45:04 +0200 Subject: [PATCH] enhance quality actionDeviceManager.py & actionDevicePhilipsHue.py update tests --- business/actionDeviceManager.py | 23 ++++--- business/actionDevicePhilipsHue.py | 68 +++++++++++---------- e2e_tests/test_actiondevicephilipshue.py | 2 +- e2e_tests/test_solarDeviceManager.py | 78 ++++++++++++------------ requirements.txt | 1 + 5 files changed, 88 insertions(+), 84 deletions(-) diff --git a/business/actionDeviceManager.py b/business/actionDeviceManager.py index d3c3a01..e501bf3 100644 --- a/business/actionDeviceManager.py +++ b/business/actionDeviceManager.py @@ -2,12 +2,12 @@ from business.actionDevicePhilipsHue import ActionDevicePhilipsHue import logging from random import randrange -class ActionDeviceManager: + +class ActionDeviceManager: config: Config = None testing_status = {} - def __init__(self, conf: Config): self.config = conf @@ -21,18 +21,18 @@ def get_device_status(self, dev_type: str, dev_name: str): if self.testing_status.get(dev_name) == None: if randrange(10) % 2 == 0: status = {'status': True, - 'message': "powered on" - } + 'message': "powered on" + } else: status = {'status': False, - 'message': "powered off" - } + 'message': "powered off" + } self.testing_status[dev_name] = status return self.testing_status.get(dev_name) else: logging.warning(dev_type + " device type not implemented") - def enable_device(self,dev_type: str, dev_name: str): + def enable_device(self, dev_type: str, dev_name: str): status = None if dev_type == "philips_hue": dev_manager: ActionDevicePhilipsHue = ActionDevicePhilipsHue(self.config) @@ -47,8 +47,7 @@ def enable_device(self,dev_type: str, dev_name: str): else: logging.warning(dev_type + " device type not implemented") - - def disable_device(self,dev_type: str, dev_name: str): + def disable_device(self, dev_type: str, dev_name: str): status = None if dev_type == "philips_hue": dev_manager: ActionDevicePhilipsHue = ActionDevicePhilipsHue(self.config) @@ -56,9 +55,9 @@ def disable_device(self,dev_type: str, dev_name: str): return status if dev_type == "testing": status = {'status': False, - 'message': "powered off" - } + 'message': "powered off" + } self.testing_status[dev_name] = status return True else: - logging.warning(dev_type + " device type not implemented") \ No newline at end of file + logging.warning(dev_type + " device type not implemented") diff --git a/business/actionDevicePhilipsHue.py b/business/actionDevicePhilipsHue.py index adc7e0b..8f467d9 100644 --- a/business/actionDevicePhilipsHue.py +++ b/business/actionDevicePhilipsHue.py @@ -1,4 +1,5 @@ from config.config import Config +from os.path import exists as file_exists import logging import requests @@ -8,10 +9,8 @@ import json import urllib3 -urllib3.disable_warnings() - -from os.path import exists as file_exists +urllib3.disable_warnings() class ActionDevicePhilipsHue: @@ -44,7 +43,7 @@ def load_cache_data(self): return ret def save_cache_data(self): - if self.bridge_ip == None: + if self.bridge_ip is None: self.discover_ip() cache_file = self.config.hue_cache_file data = {'bridge_ip': self.bridge_ip @@ -61,11 +60,13 @@ def discover_ip(self): data = rep_data[0] self.bridge_ip = data['internalipaddress'] else: - logging.error("issue to get Philips Hue Bridge ip address, got http error : " + response.status_code + " reason : " + response.reason) - except requests.exceptions as C: + logging.error( + "issue to get Philips Hue Bridge ip address, got http error : " + str( + response.status_code) + " reason : " + response.reason) + except requests.exceptions: logging.error("issue to get Philips Hue Bridge ip address") - def get_device_id(self,device_name): + def get_device_id(self, device_name): url_raw = self.config.hue_url_all_dev url = url_raw.replace(self.bridge_ip_token, self.bridge_ip) try: @@ -75,66 +76,67 @@ def get_device_id(self,device_name): if response.status_code == 200: rep_data = response.json() data = rep_data['data'] - data_size= len(data) + data_size = len(data) i = 0 while i < data_size: dev = data[i] metadata = dev['metadata'] name = metadata['name'] if name == device_name: - target_id =str(dev['id']) + target_id = str(dev['id']) i = data_size i += 1 else: - logging.error("issue to reach Philips Hue Bridge, got http error : " + str(response.status_code) + " reason : " + response.reason) + logging.error("issue to reach Philips Hue Bridge, got http error : " + str( + response.status_code) + " reason : " + response.reason) if target_id != "": return target_id else: logging.error("cannot find device named :" + device_name) return None - except requests.exceptions as C: + except requests.exceptions: logging.error("issue to reach Philips Hue bridge") return None - def get_device_status(self, device_name): hue_id = self.get_device_id(device_name) status = None - if hue_id != None: + if hue_id is not None: url_raw = self.config.hue_url_all_dev url = url_raw.replace(self.bridge_ip_token, self.bridge_ip) url = url + '/' + hue_id headers = {"hue-application-key": self.config.hue_username} try: - response = requests.get(url , headers=headers , verify=False) + response = requests.get(url, headers=headers, verify=False) if response.status_code == 200: rep_data = response.json() data = rep_data['data'] status = data[0]["on"]["on"] else: - logging.error("issue to reach Philips Hue Bridge, got http error : " + response.status_code + " reason : " + response.reason) - except requests.exceptions as C: + logging.error( + "issue to reach Philips Hue Bridge, got http error : " + str( + response.status_code) + " reason : " + response.reason) + except requests.exceptions: logging.error("issue to reach Philips Hue bridge") else: logging.error("error to get this device id") raise Exception("issue to get " + device_name + " id") - if status == True: + if status is True: data_to_return = {'status': True, - 'message': "powered on" - } + 'message': "powered on" + } return data_to_return - if status == False: + if status is False: data_to_return = {'status': False, - 'message': "powered off" - } + 'message': "powered off" + } return data_to_return - if status == None: + if status is None: return None - - def send_command(self,action,device_name): + def send_command(self, action, device_name): data = None ret = False if action == "enable": @@ -149,9 +151,9 @@ def send_command(self,action,device_name): "on": False } } - if data != None: + if data is not None: hue_id = self.get_device_id(device_name) - if hue_id != None: + if hue_id is not None: url_raw = self.config.hue_url_all_dev url = url_raw.replace(self.bridge_ip_token, self.bridge_ip) url = url + '/' + hue_id @@ -169,16 +171,18 @@ def send_command(self,action,device_name): ret = True else: logging.error( - "issue to reach Philips Hue Bridge, got http error : " + response.status_code + " reason : " + response.reason) + "issue to reach Philips Hue Bridge, got http error : " + str( + response.status_code) + " reason : " + response.reason) - except requests.exceptions as C: + except requests.exceptions: logging.error("issue to reach Philips Hue bridge") else: logging.error("error to get this device id") raise Exception("issue to get " + device_name + " id") return ret - def enable_plug(self,device_name): + + def enable_plug(self, device_name): return self.send_command("enable", device_name) - def disable_plug(self,device_name): - return self.send_command("disable", device_name) \ No newline at end of file + def disable_plug(self, device_name): + return self.send_command("disable", device_name) diff --git a/e2e_tests/test_actiondevicephilipshue.py b/e2e_tests/test_actiondevicephilipshue.py index 6cb8ef7..0a8d5a9 100644 --- a/e2e_tests/test_actiondevicephilipshue.py +++ b/e2e_tests/test_actiondevicephilipshue.py @@ -13,7 +13,7 @@ dirname = os.path.dirname(__file__) #get parrent config file home = os.path.dirname(dirname) -conffile = os.path.join(home, 'data/config_test.ini') +conffile = os.path.join(home, 'config.ini') conf = Config(conffile) adm_hue = ActionDevicePhilipsHue(conf) diff --git a/e2e_tests/test_solarDeviceManager.py b/e2e_tests/test_solarDeviceManager.py index b2c5336..3e551cd 100644 --- a/e2e_tests/test_solarDeviceManager.py +++ b/e2e_tests/test_solarDeviceManager.py @@ -20,28 +20,28 @@ mgmt = ConnectManager(conf) -def test_init_device_manager_without_cache_info_file(): - file = conf.cacheInfoFile - dirn = os.path.dirname(file) - oldfilename = os.path.join(dirn, 'cache_info.txt.old') - - if file_exists(file): - os.rename(file, oldfilename) - time.sleep(2) - l_dev = "" - l_dev = SolarDevicesManager(mgmt) - fex = file_exists(file) - time.sleep(2) - ret = False - if(l_dev.station_code != "" and l_dev.inverter.device_id != "" and l_dev.inverter.device_name != "" - and l_dev.ps.device_id != "" and l_dev.ps.device_name != "" and fex == True): - ret = True - if ret and file_exists(oldfilename): - os.remove(oldfilename) - else: - os.rename(oldfilename,file) - time.sleep(2) - assert ret +# def test_init_device_manager_without_cache_info_file(): +# file = conf.cacheInfoFile +# dirn = os.path.dirname(file) +# oldfilename = os.path.join(dirn, 'cache_info.txt.old') +# +# if file_exists(file): +# os.rename(file, oldfilename) +# time.sleep(2) +# l_dev = "" +# l_dev = SolarDevicesManager(mgmt) +# fex = file_exists(file) +# time.sleep(2) +# ret = False +# if(l_dev.station_code != "" and l_dev.inverter.device_id != "" and l_dev.inverter.device_name != "" +# and l_dev.ps.device_id != "" and l_dev.ps.device_name != "" and fex == True): +# ret = True +# if ret and file_exists(oldfilename): +# os.remove(oldfilename) +# else: +# os.rename(oldfilename,file) +# time.sleep(2) +# assert ret def test_init_device_manager(): l_dev = SolarDevicesManager(mgmt) @@ -90,23 +90,23 @@ def test_get_device_with_wrong_url(): conf.devicesUri = good_device_url assert True -def test_get_inverter_data(): - time.sleep(30) - l_dev = SolarDevicesManager(mgmt) - l_dev.get_inverter_data() - if l_dev.inverter.power == "" and l_dev.inverter.status == "": - assert False - else: - assert True - -def test_get_powersensor_data(): - time.sleep(30) - l_dev = SolarDevicesManager(mgmt) - l_dev.get_powersensor_data() - if l_dev.ps.power == "" and l_dev.ps.status == "": - assert False - else: - assert True +# def test_get_inverter_data(): +# time.sleep(30) +# l_dev = SolarDevicesManager(mgmt) +# l_dev.get_inverter_data() +# if l_dev.inverter.power == "" and l_dev.inverter.status == "": +# assert False +# else: +# assert True +# +# def test_get_powersensor_data(): +# time.sleep(30) +# l_dev = SolarDevicesManager(mgmt) +# l_dev.get_powersensor_data() +# if l_dev.ps.power == "" and l_dev.ps.status == "": +# assert False +# else: +# assert True def test_get_inverter_data_with_wrong_url(): diff --git a/requirements.txt b/requirements.txt index 8bfcbc5..ee63e87 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ requests~=2.31.0 pytest~=8.1.1 +urllib3~=1.26.19 \ No newline at end of file