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 01/15] 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 From 8311fdc791866d891b04bbbe21acdc2c4f14e0d7 Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 16:11:46 +0200 Subject: [PATCH 02/15] Update python-app.yml --- .github/workflows/python-app.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index 0984e61..ab7d3a1 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -7,7 +7,7 @@ on: push: branches: [ "staging" ] pull_request: - branches: [ "main", "staging" ] + branches: [ "main" ] permissions: contents: read From d8d405eb46171987f12cd66dbc24f36afa3d8985 Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 16:16:11 +0200 Subject: [PATCH 03/15] enhance quality actionDevicePhilipsHue.py --- business/actionDeviceManager.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/business/actionDeviceManager.py b/business/actionDeviceManager.py index e501bf3..18d75cd 100644 --- a/business/actionDeviceManager.py +++ b/business/actionDeviceManager.py @@ -18,7 +18,7 @@ def get_device_status(self, dev_type: str, dev_name: str): status = dev_manager.get_device_status(dev_name) return status if dev_type == "testing": - if self.testing_status.get(dev_name) == None: + if self.testing_status.get(dev_name) is None: if randrange(10) % 2 == 0: status = {'status': True, 'message': "powered on" @@ -31,6 +31,7 @@ def get_device_status(self, dev_type: str, dev_name: str): return self.testing_status.get(dev_name) else: logging.warning(dev_type + " device type not implemented") + return status def enable_device(self, dev_type: str, dev_name: str): status = None @@ -46,6 +47,7 @@ def enable_device(self, dev_type: str, dev_name: str): return True else: logging.warning(dev_type + " device type not implemented") + return status def disable_device(self, dev_type: str, dev_name: str): status = None @@ -61,3 +63,4 @@ def disable_device(self, dev_type: str, dev_name: str): return True else: logging.warning(dev_type + " device type not implemented") + return status From c23f7637df3a315d5346acbd50c100b2d7a3202c Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 16:19:59 +0200 Subject: [PATCH 04/15] Dev merge into staging (#7) * enhance quality actionDeviceManager.py & actionDevicePhilipsHue.py update tests * Update python-app.yml * enhance quality actionDevicePhilipsHue.py --- .github/workflows/python-app.yml | 2 +- business/actionDeviceManager.py | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index 99ddc47..82378fb 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -7,7 +7,7 @@ on: push: branches: [ "staging" ] pull_request: - branches: [ "main", "staging" ] + branches: [ "main" ] permissions: diff --git a/business/actionDeviceManager.py b/business/actionDeviceManager.py index e501bf3..f6f7a8b 100644 --- a/business/actionDeviceManager.py +++ b/business/actionDeviceManager.py @@ -18,7 +18,7 @@ def get_device_status(self, dev_type: str, dev_name: str): status = dev_manager.get_device_status(dev_name) return status if dev_type == "testing": - if self.testing_status.get(dev_name) == None: + if self.testing_status.get(dev_name) is None: if randrange(10) % 2 == 0: status = {'status': True, 'message': "powered on" @@ -31,6 +31,7 @@ def get_device_status(self, dev_type: str, dev_name: str): return self.testing_status.get(dev_name) else: logging.warning(dev_type + " device type not implemented") + return status def enable_device(self, dev_type: str, dev_name: str): status = None @@ -46,6 +47,7 @@ def enable_device(self, dev_type: str, dev_name: str): return True else: logging.warning(dev_type + " device type not implemented") + return status def disable_device(self, dev_type: str, dev_name: str): status = None @@ -61,3 +63,5 @@ def disable_device(self, dev_type: str, dev_name: str): return True else: logging.warning(dev_type + " device type not implemented") + return status + From f300caa35a4df5e2db273a1725a15447744d82d8 Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 16:22:32 +0200 Subject: [PATCH 05/15] enhance quality actionDeviceManager.py update github action file --- .github/workflows/python-app.yml | 4 ++-- business/actionDeviceManager.py | 1 - 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index 82378fb..f5bdb65 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -5,9 +5,9 @@ name: Python application on: push: - branches: [ "staging" ] + branches: [ "dev", "staging", "main" ] pull_request: - branches: [ "main" ] + branches: [ "main", "staging" ] permissions: diff --git a/business/actionDeviceManager.py b/business/actionDeviceManager.py index f6f7a8b..18d75cd 100644 --- a/business/actionDeviceManager.py +++ b/business/actionDeviceManager.py @@ -64,4 +64,3 @@ def disable_device(self, dev_type: str, dev_name: str): else: logging.warning(dev_type + " device type not implemented") return status - From 035bf2c1d68b40e4af82797d894b4add5e7b459c Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 16:29:56 +0200 Subject: [PATCH 06/15] enhance global code quality --- business/alertManager.py | 7 +--- business/connectManager.py | 12 +++--- business/inverterRules.py | 76 +++++++++++++++++------------------- business/powersensorRules.py | 2 +- business/rulesManager.py | 57 +++++++++++++++------------ 5 files changed, 75 insertions(+), 79 deletions(-) diff --git a/business/alertManager.py b/business/alertManager.py index e0e26ed..d448e98 100644 --- a/business/alertManager.py +++ b/business/alertManager.py @@ -10,12 +10,11 @@ class AlertManager: def __init__(self, conf: Config): self.config = conf - def sendAlert(self, message): + def send_alert(self, message): url = self.config.alert_configured_url headers = { "Title": self.config.alert_title, - #"Priority": "urgent", "Tags": "warning" } @@ -27,7 +26,3 @@ def sendAlert(self, message): else: logging.error("cannot send alert") return False - - - - diff --git a/business/connectManager.py b/business/connectManager.py index a9ee9e6..2409b33 100644 --- a/business/connectManager.py +++ b/business/connectManager.py @@ -12,6 +12,11 @@ from config.config import Config +def decode_pwd(encoded): + decoded_bytes = base64.b64decode(encoded) + decoded_string = decoded_bytes.decode("utf-8").rstrip() + return decoded_string + class ConnectManager: session_cookie = {} @@ -29,14 +34,9 @@ def init_session(self): if not self.load_session(): self.login() - def decode_pwd(self, encoded): - decoded_bytes = base64.b64decode(encoded) - decoded_string = decoded_bytes.decode("utf-8").rstrip() - return decoded_string - def login(self): uri = self.config.loginUri - data = {"userName": self.config.userName, "systemCode": self.decode_pwd(self.config.systemCode)} + data = {"userName": self.config.userName, "systemCode": decode_pwd(self.config.systemCode)} json_data = json.dumps(data) try: response = requests.post(uri, data=json_data) diff --git a/business/inverterRules.py b/business/inverterRules.py index 908feee..a4b5b02 100644 --- a/business/inverterRules.py +++ b/business/inverterRules.py @@ -1,46 +1,42 @@ - class InverterRules: - ok_status = 512 ko_status = 768 - status_list = { - "0":{"message": "Standby: initializing", "type": "warning"}, - "1":{"message":"Standby: insulation resistance detecting", "type": "warning"}, - "2":{"message":"Standby: irradiation detecting", "type": "warning"}, - "3":{"message":"Standby: grid detecting", "type": "warning"}, - "256":{"message":"Start", "type": "ok"}, - "512":{"message":"Grid-connected", "type": "ok"}, - "513":{"message":"Grid-connected: power limited", "type": "ok"}, - "514":{"message":"Grid-connected: self-derating", "type": "ok"}, - "515":{"message":"Off-grid operation","type": "warning"}, - "768":{"message":"Shutdown: on fault", "type": "error"}, - "769":{"message":"Shutdown: on command", "type": "error"}, - "770":{"message":"Shutdown: OVGR", "type": "error"}, - "771":{"message":"Shutdown: communication interrupted", "type": "error"}, - "772":{"message":"Shutdown: power limited", "type": "error"}, - "773":{"message":"Shutdown: manual startup required", "type": "error"}, - "774":{"message":"Shutdown: DC switch disconnected", "type": "error"}, - "775":{"message":"Shutdown: rapid shutdown","type": "error"}, - "776":{"message":"Shutdown: input underpower","type": "error"}, - "777":{"message":"Shutdown: NS protection","type": "error"}, - "778":{"message":"Shutdown: commanded rapid shutdown","type": "error"}, - "1025":{"message":"Grid scheduling: cosφ-P curve", "type": "warning"}, - "1026":{"message":"Grid scheduling: Q-U curve","type": "warning"}, - "1027":{"message":"Power grid scheduling: PF-U characteristic curve","type": "warning"}, - "1028":{"message":"Grid scheduling: dry contact","type": "warning"}, - "1029":{"message":"Power grid scheduling: Q-P characteristic curve","type": "warning"}, - "1280":{"message":"Ready for terminal test","type": "warning"}, - "1281":{"message":"Terminal testing...","type": "warning"}, - "1536":{"message":"Inspection in progress","type": "warning"}, - "1792":{"message":"AFCI self-check","type": "warning"}, - "2048":{"message":"I-V scanning","type": "warning"}, - "2304":{"message":"DC input detection","type": "warning"}, - "2560":{"message":"Off-grid charging","type": "warning"}, - "40960":{"message":"Standby: no irradiation","type": "warning"}, - "40961":{"message":"Standby: no DC input","type": "warning"}, - "45056":{"message":"Communication interrupted (written by SmartLogger)","type": "error"}, - "49152":{"message":"Loading... (written by SmartLogger)","type": "warning"}, + "0": {"message": "Standby: initializing", "type": "warning"}, + "1": {"message": "Standby: insulation resistance detecting", "type": "warning"}, + "2": {"message": "Standby: irradiation detecting", "type": "warning"}, + "3": {"message": "Standby: grid detecting", "type": "warning"}, + "256": {"message": "Start", "type": "ok"}, + "512": {"message": "Grid-connected", "type": "ok"}, + "513": {"message": "Grid-connected: power limited", "type": "ok"}, + "514": {"message": "Grid-connected: self-derating", "type": "ok"}, + "515": {"message": "Off-grid operation", "type": "warning"}, + "768": {"message": "Shutdown: on fault", "type": "error"}, + "769": {"message": "Shutdown: on command", "type": "error"}, + "770": {"message": "Shutdown: OVGR", "type": "error"}, + "771": {"message": "Shutdown: communication interrupted", "type": "error"}, + "772": {"message": "Shutdown: power limited", "type": "error"}, + "773": {"message": "Shutdown: manual startup required", "type": "error"}, + "774": {"message": "Shutdown: DC switch disconnected", "type": "error"}, + "775": {"message": "Shutdown: rapid shutdown", "type": "error"}, + "776": {"message": "Shutdown: input underpower", "type": "error"}, + "777": {"message": "Shutdown: NS protection", "type": "error"}, + "778": {"message": "Shutdown: commanded rapid shutdown", "type": "error"}, + "1025": {"message": "Grid scheduling: cosφ-P curve", "type": "warning"}, + "1026": {"message": "Grid scheduling: Q-U curve", "type": "warning"}, + "1027": {"message": "Power grid scheduling: PF-U characteristic curve", "type": "warning"}, + "1028": {"message": "Grid scheduling: dry contact", "type": "warning"}, + "1029": {"message": "Power grid scheduling: Q-P characteristic curve", "type": "warning"}, + "1280": {"message": "Ready for terminal test", "type": "warning"}, + "1281": {"message": "Terminal testing...", "type": "warning"}, + "1536": {"message": "Inspection in progress", "type": "warning"}, + "1792": {"message": "AFCI self-check", "type": "warning"}, + "2048": {"message": "I-V scanning", "type": "warning"}, + "2304": {"message": "DC input detection", "type": "warning"}, + "2560": {"message": "Off-grid charging", "type": "warning"}, + "40960": {"message": "Standby: no irradiation", "type": "warning"}, + "40961": {"message": "Standby: no DC input", "type": "warning"}, + "45056": {"message": "Communication interrupted (written by SmartLogger)", "type": "error"}, + "49152": {"message": "Loading... (written by SmartLogger)", "type": "warning"}, } - diff --git a/business/powersensorRules.py b/business/powersensorRules.py index 9099d9b..0b9b9e2 100644 --- a/business/powersensorRules.py +++ b/business/powersensorRules.py @@ -3,4 +3,4 @@ class PowersensorRules: ok_status = 1 # 1 mean --> OK - # other mean --KO \ No newline at end of file + # other mean --KO diff --git a/business/rulesManager.py b/business/rulesManager.py index de2f470..71a94ae 100644 --- a/business/rulesManager.py +++ b/business/rulesManager.py @@ -12,6 +12,7 @@ from exception.HuaweiApiFrequencyException import * from exception.HuaweiApiException import * + class RulesManager: dev_mgmt: SolarDevicesManager = None last_update: time = None @@ -25,10 +26,11 @@ def __init__(self, conf: Config): self.dev_mgmt = SolarDevicesManager(ConnectManager(conf)) self.alerter = AlertManager(conf) self.action_devices_mgmt = ActionDeviceManager(conf) + def refresh_data(self): age = (time.time() - 600) my_timing = 1 - if self.last_update != None: + if self.last_update is not None: my_timing = self.last_update if age > my_timing: try: @@ -51,13 +53,13 @@ def control_status(self): log_warning = "" self.refresh_data() status = False - if self.dev_mgmt.inverter.status != None: + if self.dev_mgmt.inverter.status is not None: # we got a status from Huawei ... - str_statusCode = str(self.dev_mgmt.inverter.status) - current_status = self.inverter_rules.status_list.get(str_statusCode) + str_status_code = str(self.dev_mgmt.inverter.status) + current_status = self.inverter_rules.status_list.get(str_status_code) inverter_message = "" - if current_status != None: - #and ... we know the status + if current_status is not None: + # and ... we know the status status_inverter = True str_status = current_status["message"] message_type = current_status["type"] @@ -66,31 +68,33 @@ def control_status(self): # error case: log and alert if message_type == 'error': inverter_message = ("Inverter status KO, code : " + str( - self.dev_mgmt.inverter.status) + " --> " + str_status + "\n" - "Need manual action on AC/DC module as soon as possible\n") + self.dev_mgmt.inverter.status) + " --> " + str_status + "\n" + "Need manual action on AC/DC " + "module as soon as possible\n") - #warning case: log but not alert + # warning case: log but not alert if message_type == 'warning': - logging.warning ("Inverter status different from OK ("+ str(self.inverter_rules.ok_status) +")") - logging.warning("Code received is : " + str_statusCode + " --> " + str_status) - #good case here : + logging.warning( + "Inverter status different from OK (" + str(self.inverter_rules.ok_status) + ")") + logging.warning("Code received is : " + str_status_code + " --> " + str_status) + # good case here : else: logging.info("Inverter OK") - #unknown status here : + # unknown status here : else: - inverter_message = ("Inverter unknowed status code : "+ str_statusCode +" \n") + inverter_message = ("Inverter unknowed status code : " + str_status_code + " \n") status_inverter = False # cannot get status here : else: - inverter_message = ("Issue : impossible to get inverter status \n") + inverter_message = "Issue : impossible to get inverter status \n" status_inverter = False - #stack messages + # stack messages if inverter_message != "": messages = messages + inverter_message + "\n" else: messages = "" - if self.dev_mgmt.ps.status != None: + if self.dev_mgmt.ps.status is not None: status_ps = True if self.dev_mgmt.ps.status != self.ps_rules.ok_status: # need to alert @@ -105,7 +109,7 @@ def control_status(self): if messages != "": logging.error(messages) - status = self.alerter.sendAlert(messages) + status = self.alerter.send_alert(messages) else: status = status_ps and status_inverter @@ -123,9 +127,9 @@ def control_rules(self): dev_type = rule['action_device_type'] status = self.action_devices_mgmt.get_device_status(dev_type, dev_name) bool_status = None - if status != None: + if status is not None: bool_status = status["status"] - if estimated_remaining_power != None: + if estimated_remaining_power is not None: if estimated_remaining_power >= 0: # if there is power if estimated_remaining_power >= rule['rule_remaining_power_value']: @@ -135,7 +139,8 @@ def control_rules(self): if bool_status == False and rule['action_threshold_crossed'] == "enable": if self.action_devices_mgmt.enable_device(dev_type, dev_name): logging.info(dev_name + " successfully power ON !") - estimated_remaining_power = estimated_remaining_power - rule['rule_remaining_power_value'] + estimated_remaining_power = estimated_remaining_power - rule[ + 'rule_remaining_power_value'] else: logging.error("Error when try to power ON : " + dev_name) @@ -147,7 +152,8 @@ def control_rules(self): if bool_status == True and rule['action_threshold_crossed'] == "disable": if self.action_devices_mgmt.disable_device(dev_type, dev_name): logging.info(dev_name + " successfully power OFF") - estimated_remaining_power = estimated_remaining_power - rule['rule_remaining_power_value'] + estimated_remaining_power = estimated_remaining_power - rule[ + 'rule_remaining_power_value'] else: logging.error("Error when try to power OFF: " + dev_name) @@ -165,10 +171,10 @@ def control_rules(self): logging.info("device " + dev_name + " already ON") # case device powered OFF, keep it as-is else: - logging.info("not enough power to start : "+ dev_name) + logging.info("not enough power to start : " + dev_name) else: - #case we cannot get value from Huawei cloud + # case we cannot get value from Huawei cloud logging.info("Not enough power produced : Disable") disable_flag = True else: @@ -176,8 +182,7 @@ def control_rules(self): logging.info("Issue to get data from Huawei : Disable") disable_flag = True if disable_flag: - if self.action_devices_mgmt.disable_device(dev_type,dev_name): + if self.action_devices_mgmt.disable_device(dev_type, dev_name): logging.info(dev_name + " successfully power OFF") else: logging.error("Error when trying to power off : " + dev_name) - From 271408f189d8c5380be2fae5b8b56e4023de8586 Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 16:30:38 +0200 Subject: [PATCH 07/15] enhance global code quality --- business/powerUsageRules.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/business/powerUsageRules.py b/business/powerUsageRules.py index d3b5602..10602d9 100644 --- a/business/powerUsageRules.py +++ b/business/powerUsageRules.py @@ -1,6 +1,4 @@ - -class PowerUsageRules(): - +class PowerUsageRules: rules = [{'rule_id': 1, 'rule_name': "Pool Heating Threshold", 'rule_remaining_power_value': 1000, @@ -9,4 +7,4 @@ class PowerUsageRules(): 'action_device_name': 'PAC piscine', 'action_device_type': 'philips_hue' } - ] + ] From 85371c3f01395b4074cbc724e9a081b638dc89c6 Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 16:33:10 +0200 Subject: [PATCH 08/15] enhance ruleManager.py code quality --- business/rulesManager.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/business/rulesManager.py b/business/rulesManager.py index 71a94ae..2fbd521 100644 --- a/business/rulesManager.py +++ b/business/rulesManager.py @@ -50,7 +50,6 @@ def refresh_data(self): def control_status(self): messages = "" - log_warning = "" self.refresh_data() status = False if self.dev_mgmt.inverter.status is not None: @@ -136,7 +135,7 @@ def control_rules(self): # if there is enough power for this device # case device status OFF and need to be power ON: - if bool_status == False and rule['action_threshold_crossed'] == "enable": + if bool_status is False and rule['action_threshold_crossed'] == "enable": if self.action_devices_mgmt.enable_device(dev_type, dev_name): logging.info(dev_name + " successfully power ON !") estimated_remaining_power = estimated_remaining_power - rule[ @@ -145,11 +144,11 @@ def control_rules(self): logging.error("Error when try to power ON : " + dev_name) # case device status ON and need to be power ON: - if bool_status == True and rule['action_threshold_crossed'] == "enable": + if bool_status is True and rule['action_threshold_crossed'] == "enable": logging.info("device " + dev_name + " already ON") # case device status ON and need to be power OFF (why disable when generate power? idk): - if bool_status == True and rule['action_threshold_crossed'] == "disable": + if bool_status is True and rule['action_threshold_crossed'] == "disable": if self.action_devices_mgmt.disable_device(dev_type, dev_name): logging.info(dev_name + " successfully power OFF") estimated_remaining_power = estimated_remaining_power - rule[ @@ -158,16 +157,16 @@ def control_rules(self): logging.error("Error when try to power OFF: " + dev_name) # case device status OFF and need to be power OFF (why disable when generate power ? idk): - if bool_status == False and rule['action_threshold_crossed'] == "disable": + if bool_status is False and rule['action_threshold_crossed'] == "disable": logging.info("device " + dev_name + " already OFF") else: # if there is NOT enough power for this device : - if bool_status == False: + if bool_status is False: logging.info("device " + dev_name + " already OFF") # case device already ON, keep it as-is - if bool_status == True: + if bool_status is True: logging.info("device " + dev_name + " already ON") # case device powered OFF, keep it as-is else: From 2b7748a85ab25a9970063d219e83c424f4ba57ae Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 16:37:36 +0200 Subject: [PATCH 09/15] enhance ruleManager.py powersensorDevice.py and inverterDevice.py --- business/inverterDevice.py | 4 ---- business/powersensorDevice.py | 2 +- business/rulesManager.py | 8 ++++---- 3 files changed, 5 insertions(+), 9 deletions(-) diff --git a/business/inverterDevice.py b/business/inverterDevice.py index 9fed7d1..e1a6890 100644 --- a/business/inverterDevice.py +++ b/business/inverterDevice.py @@ -11,7 +11,3 @@ class InverterDevice: # 768 mean --KO power = "" temperature = "" - - - - diff --git a/business/powersensorDevice.py b/business/powersensorDevice.py index 7f83643..b317d35 100644 --- a/business/powersensorDevice.py +++ b/business/powersensorDevice.py @@ -11,4 +11,4 @@ class PowersensorDevice: power: float = None # 100 - # -3159.0 \ No newline at end of file + # -3159.0 diff --git a/business/rulesManager.py b/business/rulesManager.py index 2fbd521..3bdbe30 100644 --- a/business/rulesManager.py +++ b/business/rulesManager.py @@ -9,8 +9,8 @@ from business.powerUsageRules import PowerUsageRules from business.alertManager import AlertManager from business.actionDeviceManager import ActionDeviceManager -from exception.HuaweiApiFrequencyException import * -from exception.HuaweiApiException import * +from exception.HuaweiApiFrequencyException import HuaweiApiFrequencyException +from exception.HuaweiApiException import HuaweiApiException class RulesManager: @@ -51,7 +51,7 @@ def refresh_data(self): def control_status(self): messages = "" self.refresh_data() - status = False + if self.dev_mgmt.inverter.status is not None: # we got a status from Huawei ... str_status_code = str(self.dev_mgmt.inverter.status) @@ -101,7 +101,7 @@ def control_status(self): messages = messages + ps_message else: logging.info("Powersensor OK") - status = True + else: status_ps = False messages = messages + "Issue : impossible to get powersensor status \n" From 7e864036a190c39df8f208df6e994fa937d41730 Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 17:01:47 +0200 Subject: [PATCH 10/15] enhance solarDevicesManager.py and config.py code --- business/solarDevicesManager.py | 50 ++++++++++++++++----------------- config/config.py | 48 ++++++++++++++----------------- 2 files changed, 46 insertions(+), 52 deletions(-) diff --git a/business/solarDevicesManager.py b/business/solarDevicesManager.py index 68b99c4..ef522c1 100644 --- a/business/solarDevicesManager.py +++ b/business/solarDevicesManager.py @@ -12,8 +12,8 @@ from exception.HuaweiApiFrequencyException import HuaweiApiFrequencyException from exception.HuaweiApiException import HuaweiApiException -class SolarDevicesManager: +class SolarDevicesManager: session_mgmt = "" station_code = "" inverter = InverterDevice() @@ -32,12 +32,12 @@ def load_station_and_devices_info(self): cache_info_file = self.session_mgmt.config.cacheInfoFile ret = False if file_exists(cache_info_file): - with open(cache_info_file, 'rb') as f1: - #in case of cacheInfoFile is older than 24h --> need to delete it and create another - c_timestamp = os.path.getctime(cache_info_file) - duration = self.session_mgmt.config.cacheInfoFileDuration - age = (time.time() - c_timestamp) - if age < duration: + with open(cache_info_file, 'rb') as f1: + # in case of cacheInfoFile is older than 24h --> need to delete it and create another + c_timestamp = os.path.getctime(cache_info_file) + duration = self.session_mgmt.config.cacheInfoFileDuration + age = (time.time() - c_timestamp) + if age < duration: data = pickle.load(f1) self.station_code = data['station_code'] self.inverter.device_id = data['inverter_device_id'] @@ -64,8 +64,9 @@ def save_station_and_devices_info(self): def get_station_code(self): # max 24 calls per day uri = self.session_mgmt.config.stationUri - headers = {"XSRF-TOKEN": self.session_mgmt.session_cookie["XSRF-TOKEN"], "Content-Type": self.session_mgmt.contentType["Content-Type"]} - data = {"pageNo":1, "pageSize":100} + headers = {"XSRF-TOKEN": self.session_mgmt.session_cookie["XSRF-TOKEN"], + "Content-Type": self.session_mgmt.contentType["Content-Type"]} + data = {"pageNo": 1, "pageSize": 100} json_data = json.dumps(data) try: response = requests.post(uri, headers=headers, data=json_data) @@ -81,15 +82,15 @@ def get_station_code(self): logging.error("Error Connecting:", errc) raise SystemExit(errc) - def get_devices(self): - #max 24 calls per day + # max 24 calls per day uri = self.session_mgmt.config.devicesUri - headers = {"XSRF-TOKEN": self.session_mgmt.session_cookie["XSRF-TOKEN"], "Content-Type": self.session_mgmt.contentType["Content-Type"]} + headers = {"XSRF-TOKEN": self.session_mgmt.session_cookie["XSRF-TOKEN"], + "Content-Type": self.session_mgmt.contentType["Content-Type"]} data = {"stationCodes": self.station_code} json_data = json.dumps(data) try: - response = requests.post(uri, headers=headers , data=json_data) + response = requests.post(uri, headers=headers, data=json_data) rep_data = response.json() if rep_data['success']: data = rep_data["data"] @@ -116,12 +117,13 @@ def get_devices(self): def get_inverter_data(self): uri = self.session_mgmt.config.deviceKpiUri - headers = {"XSRF-TOKEN": self.session_mgmt.session_cookie["XSRF-TOKEN"], "Content-Type": self.session_mgmt.contentType["Content-Type"]} + headers = {"XSRF-TOKEN": self.session_mgmt.session_cookie["XSRF-TOKEN"], + "Content-Type": self.session_mgmt.contentType["Content-Type"]} data = {"devIds": self.inverter.device_id, "devTypeId": self.inverter.device_type_code} json_data = json.dumps(data) + response = requests.post(uri, headers=headers, data=json_data) + rep_data = response.json() try: - response = requests.post(uri, headers=headers , data=json_data) - rep_data = response.json() if rep_data['success']: data = rep_data["data"][0] kpi = data["dataItemMap"] @@ -131,12 +133,12 @@ def get_inverter_data(self): elif not rep_data["success"] and rep_data["failCode"] == 407: err = rep_data["data"] msg = "Query Frequency Too High" - logging.error(msg,err) + logging.error(msg, err) raise HuaweiApiFrequencyException(msg, err) else: err = rep_data["data"] msg = rep_data["message"] - logging.error(msg,err) + logging.error(msg, err) raise HuaweiApiException("issue with Huawei API", rep_data) except requests.exceptions.HTTPError as err: logging.error("Error HTTP:", err) @@ -150,20 +152,20 @@ def get_inverter_data(self): if rep_data["failCode"] == 305: err = rep_data["failCode"] msg = "User must Relogin" - logging.error(msg,err) + logging.error(msg, err) logging.info("deleting session...") - if (self.session_mgmt.delete_session()): + if self.session_mgmt.delete_session(): logging.info("Session Deleted !") raise HuaweiApiException(msg, rep_data) - def get_powersensor_data(self): uri = self.session_mgmt.config.deviceKpiUri - headers = {"XSRF-TOKEN": self.session_mgmt.session_cookie["XSRF-TOKEN"], "Content-Type": self.session_mgmt.contentType["Content-Type"]} + headers = {"XSRF-TOKEN": self.session_mgmt.session_cookie["XSRF-TOKEN"], + "Content-Type": self.session_mgmt.contentType["Content-Type"]} data = {"devIds": self.ps.device_id, "devTypeId": self.ps.device_type_code} json_data = json.dumps(data) try: - response = requests.post(uri, headers=headers , data=json_data) + response = requests.post(uri, headers=headers, data=json_data) rep_data = response.json() if rep_data['success']: data = rep_data["data"][0] @@ -183,5 +185,3 @@ def get_powersensor_data(self): except requests.exceptions.ConnectionError as errc: logging.error("Error Connecting:", errc) raise SystemExit(errc) - - diff --git a/config/config.py b/config/config.py index 4c21bfd..99052df 100644 --- a/config/config.py +++ b/config/config.py @@ -18,7 +18,6 @@ class Config: devicesUri = '' deviceKpiUri = '' - hue_discovery_url = '' hue_url_all_dev = '' hue_username = '' @@ -29,8 +28,7 @@ class Config: alert_title = "" alert_configured_url = "" - - def __init__(self, file: str) -> object: + def __init__(self, file: str): """ :rtype: Config object with all vars @@ -38,18 +36,16 @@ def __init__(self, file: str) -> object: config = configparser.ConfigParser() config.read(file) - configDir =os.path.dirname(__file__) - pvHome = os.path.dirname(configDir) + config_dir = os.path.dirname(__file__) + pv_home = os.path.dirname(config_dir) self.baseUrl = config.get('General', 'domain') self.userName = config.get('General', 'user') self.systemCode = config.get('General', 'password') - sessFile = config.get('Session', 'connectionFile') - self.sessionFile = os.path.join(pvHome, sessFile) - self.checkDirs(self.sessionFile) - - + sess_file = config.get('Session', 'connectionFile') + self.sessionFile = os.path.join(pv_home, sess_file) + self.check_dirs(self.sessionFile) self.sessionDuration = int(config.get('Session', 'duration')) @@ -57,32 +53,30 @@ def __init__(self, file: str) -> object: self.logoutUri = urljoin(self.baseUrl, config.get('Logout', 'urn')) self.stationUri = urljoin(self.baseUrl, config.get('Station', 'urn')) - stationInfoFile = config.get('Station', 'cacheInfoFile') - self.cacheInfoFile = os.path.join(pvHome,stationInfoFile) - self.checkDirs(self.cacheInfoFile) - + station_info_file = config.get('Station', 'cacheInfoFile') + self.cacheInfoFile = os.path.join(pv_home, station_info_file) + self.check_dirs(self.cacheInfoFile) self.cacheInfoFileDuration = int(config.get('Station', 'duration')) self.devicesUri = urljoin(self.baseUrl, config.get('Devices', 'urn')) self.deviceKpiUri = urljoin(self.baseUrl, config.get('DevicesKpi', 'urn')) - self.alert_configured_url = urljoin(config.get("alert","url"),config.get("alert","topic")) - self.alert_title = config.get("alert","title") - - self.hue_discovery_url = config.get("action_philips_hue","discover_url") - self.hue_url_all_dev = config.get("action_philips_hue","bridge_all_light_devices") - self.hue_username = config.get("action_philips_hue","hue_username") - self.hue_client_key = config.get("action_philips_hue","hue_client_key") + self.alert_configured_url = urljoin(config.get("alert", "url"), config.get("alert", "topic")) + self.alert_title = config.get("alert", "title") - hue_cache = config.get("action_philips_hue","cacheFile") - self.hue_cache_file = os.path.join(pvHome,hue_cache) - self.checkDirs(self.hue_cache_file) + self.hue_discovery_url = config.get("action_philips_hue", "discover_url") + self.hue_url_all_dev = config.get("action_philips_hue", "bridge_all_light_devices") + self.hue_username = config.get("action_philips_hue", "hue_username") + self.hue_client_key = config.get("action_philips_hue", "hue_client_key") - self.hue_cache_duration = int(config.get("action_philips_hue","cache_duration")) + hue_cache = config.get("action_philips_hue", "cacheFile") + self.hue_cache_file = os.path.join(pv_home, hue_cache) + self.check_dirs(self.hue_cache_file) + self.hue_cache_duration = int(config.get("action_philips_hue", "cache_duration")) - def checkDirs(self,path): + @staticmethod + def check_dirs(path): parent = os.path.dirname(path) if not (os.path.exists(parent)): os.makedirs(parent) - From be979066ff1ffbce73dc3f838cbe27fbcf17b74e Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 17:10:01 +0200 Subject: [PATCH 11/15] enhance test_actiondevicephilipshue.py code --- e2e_tests/test_actiondevicephilipshue.py | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/e2e_tests/test_actiondevicephilipshue.py b/e2e_tests/test_actiondevicephilipshue.py index 0a8d5a9..5f3dc79 100644 --- a/e2e_tests/test_actiondevicephilipshue.py +++ b/e2e_tests/test_actiondevicephilipshue.py @@ -1,45 +1,46 @@ import ipaddress import os from os.path import exists as file_exists +from config.config import Config +from business.actionDevicePhilipsHue import ActionDevicePhilipsHue import sys sys.path.insert(0, '../pvAlert') -from config.config import Config -from business.actionDevicePhilipsHue import ActionDevicePhilipsHue # should have a configured philips hue tor testing dirname = os.path.dirname(__file__) -#get parrent config file +# get parrent config file home = os.path.dirname(dirname) conffile = os.path.join(home, 'config.ini') conf = Config(conffile) adm_hue = ActionDevicePhilipsHue(conf) + def test_save_cache(): - if adm_hue.bridge_ip != None: + if adm_hue.bridge_ip is not None: adm_hue.save_cache_data() file = conf.hue_cache_file assert file_exists(file) else: assert False + def test_load_cache(): - adm_hue.bridge_ip = None + adm_hue.bridge_ip = '' adm_hue.load_cache_data() # test if it is an v4 ip address assert ipaddress.ip_address(adm_hue.bridge_ip) + def test_bridge_ip_discovery(): - adm_hue.bridge_ip = None + adm_hue.bridge_ip = '' adm_hue.discover_ip() # test if it is an v4 ip address assert ipaddress.ip_address(adm_hue.bridge_ip) -def test_get_device_id_KO(): +def test_get_device_id_ko(): hue_id = adm_hue.get_device_id("azerzrtzretzertezrt") - assert hue_id == None - - + assert hue_id is None From 3c95bcbb80547e2a94140793827ca55c4d1f082f Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 17:43:18 +0200 Subject: [PATCH 12/15] enhance test_actiondevicephilipshue.py code --- business/actionDevicePhilipsHue.py | 2 +- e2e_tests/test_actiondevicephilipshue.py | 16 +- e2e_tests/test_locator.py | 2 + e2e_tests/test_rulesManager.py | 616 +++++++++++------------ e2e_tests/test_solarDeviceManager.py | 76 +-- 5 files changed, 327 insertions(+), 385 deletions(-) create mode 100644 e2e_tests/test_locator.py diff --git a/business/actionDevicePhilipsHue.py b/business/actionDevicePhilipsHue.py index 8f467d9..f18980b 100644 --- a/business/actionDevicePhilipsHue.py +++ b/business/actionDevicePhilipsHue.py @@ -15,7 +15,7 @@ class ActionDevicePhilipsHue: config: Config = None - bridge_ip: str = None + bridge_ip: str = '' bridge_ip_token: str = "" def __init__(self, conf: Config): diff --git a/e2e_tests/test_actiondevicephilipshue.py b/e2e_tests/test_actiondevicephilipshue.py index 5f3dc79..744c269 100644 --- a/e2e_tests/test_actiondevicephilipshue.py +++ b/e2e_tests/test_actiondevicephilipshue.py @@ -1,20 +1,20 @@ import ipaddress import os +import test_locator + from os.path import exists as file_exists from config.config import Config from business.actionDevicePhilipsHue import ActionDevicePhilipsHue -import sys -sys.path.insert(0, '../pvAlert') - - +# dumb test for quality +test_locator = test_locator # should have a configured philips hue tor testing -dirname = os.path.dirname(__file__) +current_dir = os.path.dirname(__file__) # get parrent config file -home = os.path.dirname(dirname) -conffile = os.path.join(home, 'config.ini') -conf = Config(conffile) +home = os.path.dirname(current_dir) +conf_file = os.path.join(home, 'config.ini') +conf = Config(conf_file) adm_hue = ActionDevicePhilipsHue(conf) diff --git a/e2e_tests/test_locator.py b/e2e_tests/test_locator.py new file mode 100644 index 0000000..e2e9e97 --- /dev/null +++ b/e2e_tests/test_locator.py @@ -0,0 +1,2 @@ +import sys +sys.path.insert(0, '../pvAlert') diff --git a/e2e_tests/test_rulesManager.py b/e2e_tests/test_rulesManager.py index 79457a7..f8ba052 100644 --- a/e2e_tests/test_rulesManager.py +++ b/e2e_tests/test_rulesManager.py @@ -1,20 +1,18 @@ import time import os - -import sys -sys.path.insert(0, '../pvAlert') +import test_locator from config.config import Config from business.connectManager import ConnectManager from business.rulesManager import RulesManager - - +# dumb test +test_locator = test_locator # need a configured config.ini file !!! dirname = os.path.dirname(__file__) -#get parrent config file +# get parrent config file home = os.path.dirname(dirname) conffile = os.path.join(home, 'config.ini') conf = Config(conffile) @@ -22,124 +20,129 @@ mgmt = ConnectManager(conf) my_rm = RulesManager(conf) + def test_refresh_data_timeout(): my_rm.last_update = (time.time() - 605) my_rm.refresh_data() - if my_rm.last_update != None: + if my_rm.last_update is not None: assert True else: assert False + def test_refresh_data_intime(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.refresh_data() - if my_rm.last_update == myTime: + if my_rm.last_update == my_time: assert True else: assert False def test_control_status_ok_case(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.inverter.status = 512 my_rm.dev_mgmt.ps.status = 1 assert my_rm.control_status() + def test_control_status_both_none_case(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.inverter.status = None my_rm.dev_mgmt.ps.status = None assert my_rm.control_status() def test_control_status_one_ok_and_none_case(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.inverter.status = 512 my_rm.dev_mgmt.ps.status = None assert my_rm.control_status() + def test_control_status_one_ko_and_none_case_other(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.inverter.status = None my_rm.dev_mgmt.ps.status = 0 assert my_rm.control_status() def test_control_status_both_ko(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.inverter.status = 768 my_rm.dev_mgmt.ps.status = 0 assert my_rm.control_status() def test_control_status_one_ko_and_inverter_warning(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.inverter.status = 771 my_rm.dev_mgmt.ps.status = 0 assert my_rm.control_status() + def test_control_status_one_ok_and_inverter_warning(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.inverter.status = 771 my_rm.dev_mgmt.ps.status = 1 assert my_rm.control_status() def test_control_status_one_ok_and_inverter_unknowned_code(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.inverter.status = 9999 my_rm.dev_mgmt.ps.status = 1 assert my_rm.control_status() -def test_control_rules_case_remaining_ok_for_one_device_when_devs_OFF(): - myTime = (time.time() - 60) - my_rm.last_update = myTime +def test_control_rules_case_remaining_ok_for_one_device_when_devs_off(): + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.ps.power = 1500 - my_test_rules = [{'rule_id': 1, - 'rule_name': "Pool Heating Threshold", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1000, - 'rule_priority': 1, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev1', - 'action_device_type': 'testing' - }, - {'rule_id': 2, - 'rule_name': "Water eater", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1500, - 'rule_priority': 3, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev2', - 'action_device_type': 'testing' - }, - {'rule_id': 3, - 'rule_name': "Water eater 2", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1500, - 'rule_priority': 2, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev3', - 'action_device_type': 'testing' - }, - ] + my_test_rules = [{'rule_id': 1, + 'rule_name': "Pool Heating Threshold", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1000, + 'rule_priority': 1, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev1', + 'action_device_type': 'testing' + }, + {'rule_id': 2, + 'rule_name': "Water eater", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1500, + 'rule_priority': 3, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev2', + 'action_device_type': 'testing' + }, + {'rule_id': 3, + 'rule_name': "Water eater 2", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1500, + 'rule_priority': 2, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev3', + 'action_device_type': 'testing' + }, + ] my_rm.power_rules.rules = my_test_rules status_off = {'status': False, - 'message': "powered off" - } + 'message': "powered off" + } - #status_on = {'status': True, + # status_on = {'status': True, # 'message': "powered on" # } @@ -147,59 +150,58 @@ def test_control_rules_case_remaining_ok_for_one_device_when_devs_OFF(): my_rm.action_devices_mgmt.testing_status["dev2"] = status_off my_rm.action_devices_mgmt.testing_status["dev3"] = status_off - my_rm.control_rules() - dev1 = my_rm.action_devices_mgmt.get_device_status("testing","dev1") + dev1 = my_rm.action_devices_mgmt.get_device_status("testing", "dev1") dev1_status = dev1["status"] - dev2 = my_rm.action_devices_mgmt.get_device_status("testing","dev2") + dev2 = my_rm.action_devices_mgmt.get_device_status("testing", "dev2") dev2_status = dev2["status"] - dev3 = my_rm.action_devices_mgmt.get_device_status("testing","dev3") + dev3 = my_rm.action_devices_mgmt.get_device_status("testing", "dev3") dev3_status = dev3["status"] - if dev1_status == True and dev2_status == False and dev3_status ==False: + if dev1_status is True and dev2_status is False and dev3_status is False: assert True else: assert False -def test_control_rules_case_remaining_ok_for_x_devices_when_devs_OFF(): - myTime = (time.time() - 60) - my_rm.last_update = myTime +def test_control_rules_case_remaining_ok_for_x_devices_when_devs_off(): + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.ps.power = 1500 - my_test_rules = [{'rule_id': 1, - 'rule_name': "Pool Heating Threshold", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1000, - 'rule_priority': 1, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev1', - 'action_device_type': 'testing' - }, - {'rule_id': 2, - 'rule_name': "Water eater", - 'rule_type': "threshold", - 'rule_remaining_power_value': 500, - 'rule_priority': 3, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev2', - 'action_device_type': 'testing' - }, - {'rule_id': 3, - 'rule_name': "Water eater 2", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1500, - 'rule_priority': 2, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev3', - 'action_device_type': 'testing' - }, - ] + my_test_rules = [{'rule_id': 1, + 'rule_name': "Pool Heating Threshold", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1000, + 'rule_priority': 1, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev1', + 'action_device_type': 'testing' + }, + {'rule_id': 2, + 'rule_name': "Water eater", + 'rule_type': "threshold", + 'rule_remaining_power_value': 500, + 'rule_priority': 3, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev2', + 'action_device_type': 'testing' + }, + {'rule_id': 3, + 'rule_name': "Water eater 2", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1500, + 'rule_priority': 2, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev3', + 'action_device_type': 'testing' + }, + ] my_rm.power_rules.rules = my_test_rules status_off = {'status': False, - 'message': "powered off" - } + 'message': "powered off" + } - #status_on = {'status': True, + # status_on = {'status': True, # 'message': "powered on" # } @@ -207,315 +209,305 @@ def test_control_rules_case_remaining_ok_for_x_devices_when_devs_OFF(): my_rm.action_devices_mgmt.testing_status["dev2"] = status_off my_rm.action_devices_mgmt.testing_status["dev3"] = status_off - my_rm.control_rules() - dev1 = my_rm.action_devices_mgmt.get_device_status("testing","dev1") + dev1 = my_rm.action_devices_mgmt.get_device_status("testing", "dev1") dev1_status = dev1["status"] - dev2 = my_rm.action_devices_mgmt.get_device_status("testing","dev2") + dev2 = my_rm.action_devices_mgmt.get_device_status("testing", "dev2") dev2_status = dev2["status"] - dev3 = my_rm.action_devices_mgmt.get_device_status("testing","dev3") + dev3 = my_rm.action_devices_mgmt.get_device_status("testing", "dev3") dev3_status = dev3["status"] - if dev1_status == True and dev2_status == True and dev3_status == False: + if dev1_status is True and dev2_status is True and dev3_status is False: assert True else: assert False -def test_control_rules_case_remaining_ok_with_several_dev_ON(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + +def test_control_rules_case_remaining_ok_with_several_dev_on(): + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.ps.power = 1500 - my_test_rules = [{'rule_id': 1, - 'rule_name': "Pool Heating Threshold", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1000, - 'rule_priority': 1, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev1', - 'action_device_type': 'testing' - }, - {'rule_id': 2, - 'rule_name': "Water eater", - 'rule_type': "threshold", - 'rule_remaining_power_value': 500, - 'rule_priority': 3, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev2', - 'action_device_type': 'testing' - }, - {'rule_id': 3, - 'rule_name': "Water eater 2", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1500, - 'rule_priority': 2, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev3', - 'action_device_type': 'testing' - }, - ] + my_test_rules = [{'rule_id': 1, + 'rule_name': "Pool Heating Threshold", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1000, + 'rule_priority': 1, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev1', + 'action_device_type': 'testing' + }, + {'rule_id': 2, + 'rule_name': "Water eater", + 'rule_type': "threshold", + 'rule_remaining_power_value': 500, + 'rule_priority': 3, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev2', + 'action_device_type': 'testing' + }, + {'rule_id': 3, + 'rule_name': "Water eater 2", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1500, + 'rule_priority': 2, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev3', + 'action_device_type': 'testing' + }, + ] my_rm.power_rules.rules = my_test_rules status_off = {'status': False, - 'message': "powered off" - } + 'message': "powered off" + } status_on = {'status': True, - 'message': "powered on" - } + 'message': "powered on" + } - #prio 1: + # prio 1: my_rm.action_devices_mgmt.testing_status["dev1"] = status_on - #prio 3: + # prio 3: my_rm.action_devices_mgmt.testing_status["dev2"] = status_on - #prio 2: + # prio 2: my_rm.action_devices_mgmt.testing_status["dev3"] = status_off my_rm.control_rules() - dev1 = my_rm.action_devices_mgmt.get_device_status("testing","dev1") + dev1 = my_rm.action_devices_mgmt.get_device_status("testing", "dev1") dev1_status = dev1["status"] - dev2 = my_rm.action_devices_mgmt.get_device_status("testing","dev2") + dev2 = my_rm.action_devices_mgmt.get_device_status("testing", "dev2") dev2_status = dev2["status"] - dev3 = my_rm.action_devices_mgmt.get_device_status("testing","dev3") + dev3 = my_rm.action_devices_mgmt.get_device_status("testing", "dev3") dev3_status = dev3["status"] - if dev1_status == True and dev2_status == True and dev3_status == True: + if dev1_status is True and dev2_status is True and dev3_status is True: assert True else: assert False def test_control_rules_case_remaining_ok_when_on(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.ps.power = 1500 - my_test_rules = [{'rule_id': 1, - 'rule_name': "Pool Heating Threshold", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1000, - 'rule_priority': 1, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev1', - 'action_device_type': 'testing' - }, - {'rule_id': 2, - 'rule_name': "Water eater", - 'rule_type': "threshold", - 'rule_remaining_power_value': 500, - 'rule_priority': 3, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev2', - 'action_device_type': 'testing' - }, - {'rule_id': 3, - 'rule_name': "Water eater 2", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1500, - 'rule_priority': 2, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev3', - 'action_device_type': 'testing' - }, - ] + my_test_rules = [{'rule_id': 1, + 'rule_name': "Pool Heating Threshold", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1000, + 'rule_priority': 1, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev1', + 'action_device_type': 'testing' + }, + {'rule_id': 2, + 'rule_name': "Water eater", + 'rule_type': "threshold", + 'rule_remaining_power_value': 500, + 'rule_priority': 3, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev2', + 'action_device_type': 'testing' + }, + {'rule_id': 3, + 'rule_name': "Water eater 2", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1500, + 'rule_priority': 2, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev3', + 'action_device_type': 'testing' + }, + ] my_rm.power_rules.rules = my_test_rules status_off = {'status': False, - 'message': "powered off" - } + 'message': "powered off" + } status_on = {'status': True, - 'message': "powered on" - } + 'message': "powered on" + } - #prio 1: + # prio 1: my_rm.action_devices_mgmt.testing_status["dev1"] = status_on - #prio 3: + # prio 3: my_rm.action_devices_mgmt.testing_status["dev2"] = status_off - #prio 2: + # prio 2: my_rm.action_devices_mgmt.testing_status["dev3"] = status_off - my_rm.control_rules() - dev1 = my_rm.action_devices_mgmt.get_device_status("testing","dev1") + dev1 = my_rm.action_devices_mgmt.get_device_status("testing", "dev1") dev1_status = dev1["status"] - dev2 = my_rm.action_devices_mgmt.get_device_status("testing","dev2") + dev2 = my_rm.action_devices_mgmt.get_device_status("testing", "dev2") dev2_status = dev2["status"] - dev3 = my_rm.action_devices_mgmt.get_device_status("testing","dev3") + dev3 = my_rm.action_devices_mgmt.get_device_status("testing", "dev3") dev3_status = dev3["status"] - if dev1_status == True and dev2_status == False and dev3_status == True: + if dev1_status is True and dev2_status is False and dev3_status is True: assert True else: assert False + def test_control_rules_case_remaining_is_none(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.ps.power = None - my_test_rules = [{'rule_id': 1, - 'rule_name': "Pool Heating Threshold", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1000, - 'rule_priority': 1, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev1', - 'action_device_type': 'testing' - }, - {'rule_id': 2, - 'rule_name': "Water eater", - 'rule_type': "threshold", - 'rule_remaining_power_value': 500, - 'rule_priority': 3, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev2', - 'action_device_type': 'testing' - }, - {'rule_id': 3, - 'rule_name': "Water eater 2", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1500, - 'rule_priority': 2, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev3', - 'action_device_type': 'testing' - }, - ] + my_test_rules = [{'rule_id': 1, + 'rule_name': "Pool Heating Threshold", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1000, + 'rule_priority': 1, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev1', + 'action_device_type': 'testing' + }, + {'rule_id': 2, + 'rule_name': "Water eater", + 'rule_type': "threshold", + 'rule_remaining_power_value': 500, + 'rule_priority': 3, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev2', + 'action_device_type': 'testing' + }, + {'rule_id': 3, + 'rule_name': "Water eater 2", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1500, + 'rule_priority': 2, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev3', + 'action_device_type': 'testing' + }, + ] my_rm.power_rules.rules = my_test_rules - - my_rm.control_rules() - dev1 = my_rm.action_devices_mgmt.get_device_status("testing","dev1") + dev1 = my_rm.action_devices_mgmt.get_device_status("testing", "dev1") dev1_status = dev1["status"] - dev2 = my_rm.action_devices_mgmt.get_device_status("testing","dev2") + dev2 = my_rm.action_devices_mgmt.get_device_status("testing", "dev2") dev2_status = dev2["status"] - dev3 = my_rm.action_devices_mgmt.get_device_status("testing","dev3") + dev3 = my_rm.action_devices_mgmt.get_device_status("testing", "dev3") dev3_status = dev3["status"] - if dev1_status == False and dev2_status == False and dev3_status == False: + if dev1_status is False and dev2_status is False and dev3_status is False: assert True else: assert False def test_control_rules_case_remaining_is_negative(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.ps.power = -186.0 - my_test_rules = [{'rule_id': 1, - 'rule_name': "Pool Heating Threshold", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1000, - 'rule_priority': 1, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev1', - 'action_device_type': 'testing' - }, - {'rule_id': 2, - 'rule_name': "Water eater", - 'rule_type': "threshold", - 'rule_remaining_power_value': 500, - 'rule_priority': 3, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev2', - 'action_device_type': 'testing' - }, - {'rule_id': 3, - 'rule_name': "Water eater 2", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1500, - 'rule_priority': 2, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev3', - 'action_device_type': 'testing' - }, - ] + my_test_rules = [{'rule_id': 1, + 'rule_name': "Pool Heating Threshold", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1000, + 'rule_priority': 1, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev1', + 'action_device_type': 'testing' + }, + {'rule_id': 2, + 'rule_name': "Water eater", + 'rule_type': "threshold", + 'rule_remaining_power_value': 500, + 'rule_priority': 3, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev2', + 'action_device_type': 'testing' + }, + {'rule_id': 3, + 'rule_name': "Water eater 2", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1500, + 'rule_priority': 2, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev3', + 'action_device_type': 'testing' + }, + ] my_rm.power_rules.rules = my_test_rules - - status_off = {'status': False, - 'message': "powered off" - } - status_on = {'status': True, - 'message': "powered on" - } + 'message': "powered on" + } - #prio 1: + # prio 1: my_rm.action_devices_mgmt.testing_status["dev1"] = status_on - #prio 3: + # prio 3: my_rm.action_devices_mgmt.testing_status["dev2"] = status_on - #prio 2: + # prio 2: my_rm.action_devices_mgmt.testing_status["dev3"] = status_on - my_rm.control_rules() - dev1 = my_rm.action_devices_mgmt.get_device_status("testing","dev1") + dev1 = my_rm.action_devices_mgmt.get_device_status("testing", "dev1") dev1_status = dev1["status"] - dev2 = my_rm.action_devices_mgmt.get_device_status("testing","dev2") + dev2 = my_rm.action_devices_mgmt.get_device_status("testing", "dev2") dev2_status = dev2["status"] - dev3 = my_rm.action_devices_mgmt.get_device_status("testing","dev3") + dev3 = my_rm.action_devices_mgmt.get_device_status("testing", "dev3") dev3_status = dev3["status"] - if dev1_status == False and dev2_status == False and dev3_status == False: + if dev1_status is False and dev2_status is False and dev3_status is False: assert True else: assert False - def test_control_rules_case_remaining_low_when_on(): - myTime = (time.time() - 60) - my_rm.last_update = myTime + my_time = (time.time() - 60) + my_rm.last_update = my_time my_rm.dev_mgmt.ps.power = 100 - my_test_rules = [{'rule_id': 1, - 'rule_name': "Pool Heating Threshold", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1000, - 'rule_priority': 1, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev1', - 'action_device_type': 'testing' - }, - {'rule_id': 2, - 'rule_name': "Water eater", - 'rule_type': "threshold", - 'rule_remaining_power_value': 500, - 'rule_priority': 3, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev2', - 'action_device_type': 'testing' - }, - {'rule_id': 3, - 'rule_name': "Water eater 2", - 'rule_type': "threshold", - 'rule_remaining_power_value': 1500, - 'rule_priority': 2, - 'action_threshold_crossed': 'enable', - 'action_device_name': 'dev3', - 'action_device_type': 'testing' - }, - ] + my_test_rules = [{'rule_id': 1, + 'rule_name': "Pool Heating Threshold", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1000, + 'rule_priority': 1, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev1', + 'action_device_type': 'testing' + }, + {'rule_id': 2, + 'rule_name': "Water eater", + 'rule_type': "threshold", + 'rule_remaining_power_value': 500, + 'rule_priority': 3, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev2', + 'action_device_type': 'testing' + }, + {'rule_id': 3, + 'rule_name': "Water eater 2", + 'rule_type': "threshold", + 'rule_remaining_power_value': 1500, + 'rule_priority': 2, + 'action_threshold_crossed': 'enable', + 'action_device_name': 'dev3', + 'action_device_type': 'testing' + }, + ] my_rm.power_rules.rules = my_test_rules status_off = {'status': False, - 'message': "powered off" - } + 'message': "powered off" + } status_on = {'status': True, - 'message': "powered on" - } + 'message': "powered on" + } - #prio 1: + # prio 1: my_rm.action_devices_mgmt.testing_status["dev1"] = status_on - #prio 3: + # prio 3: my_rm.action_devices_mgmt.testing_status["dev2"] = status_off - #prio 2: + # prio 2: my_rm.action_devices_mgmt.testing_status["dev3"] = status_off - my_rm.control_rules() - dev1 = my_rm.action_devices_mgmt.get_device_status("testing","dev1") + dev1 = my_rm.action_devices_mgmt.get_device_status("testing", "dev1") dev1_status = dev1["status"] - dev2 = my_rm.action_devices_mgmt.get_device_status("testing","dev2") + dev2 = my_rm.action_devices_mgmt.get_device_status("testing", "dev2") dev2_status = dev2["status"] - dev3 = my_rm.action_devices_mgmt.get_device_status("testing","dev3") + dev3 = my_rm.action_devices_mgmt.get_device_status("testing", "dev3") dev3_status = dev3["status"] - if dev1_status == True and dev2_status == False and dev3_status == False: + if dev1_status is True and dev2_status is False and dev3_status is False: assert True else: assert False diff --git a/e2e_tests/test_solarDeviceManager.py b/e2e_tests/test_solarDeviceManager.py index 3e551cd..1353e4a 100644 --- a/e2e_tests/test_solarDeviceManager.py +++ b/e2e_tests/test_solarDeviceManager.py @@ -1,47 +1,22 @@ import os -import time -from os.path import exists as file_exists -import sys -sys.path.insert(0, '../pvAlert') - -from config.config import Config +import test_locator from business.connectManager import ConnectManager from business.solarDevicesManager import SolarDevicesManager +from config.config import Config +from exception.HuaweiApiException import HuaweiApiException - +test_locator = test_locator # need a configured config.ini file !!! dirname = os.path.dirname(__file__) -#get parrent config file +# get parrent config file home = os.path.dirname(dirname) conffile = os.path.join(home, 'config.ini') conf = Config(conffile) 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(): l_dev = SolarDevicesManager(mgmt) @@ -65,15 +40,14 @@ def test_init_device_manager_outdated_cache_info_file(): else: assert False + def test_get_station_with_wrong_url(): good_station_url = conf.stationUri fake_station_url = "https://dummy.url/thirdData/stations" conf.stationUri = fake_station_url try: SolarDevicesManager(mgmt) - except Exception as err: - template = "An exception of type {0} occurred. Arguments:\n{1!r}" - message = template.format(type(err).__name__, err.args) + except HuaweiApiException: conf.stationUri = good_station_url assert True @@ -83,31 +57,11 @@ def test_get_device_with_wrong_url(): fake_device_url = "https://dummy.url/thirdData/getDevList" conf.devicesUri = fake_device_url try: - dev = SolarDevicesManager(mgmt) - except Exception as err: - template = "An exception of type {0} occurred. Arguments:\n{1!r}" - message = template.format(type(err).__name__, err.args) + SolarDevicesManager(mgmt) + except HuaweiApiException: 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_with_wrong_url(): good_url = conf.devicesUri @@ -116,9 +70,7 @@ def test_get_inverter_data_with_wrong_url(): dev = SolarDevicesManager(mgmt) try: dev.get_inverter_data() - except Exception as err: - template = "An exception of type {0} occurred. Arguments:\n{1!r}" - message = template.format(type(err).__name__, err.args) + except BaseException: conf.deviceKpiUri = good_url assert True @@ -130,12 +82,11 @@ def test_get_powersensor_data_with_wrong_url(): dev = SolarDevicesManager(mgmt) try: dev.get_powersensor_data() - except Exception as err: - template = "An exception of type {0} occurred. Arguments:\n{1!r}" - message = template.format(type(err).__name__, err.args) + except BaseException: conf.deviceKpiUri = good_url assert True + def test_get_inverter_data_frequency_exception(): dev = SolarDevicesManager(mgmt) result = False @@ -152,6 +103,3 @@ def test_get_inverter_data_frequency_exception(): i = 5 i += 1 assert result - - - From 51a254508240d5819093105971dafd3057136288 Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 17:45:55 +0200 Subject: [PATCH 13/15] update e2e tests --- e2e_tests/test_actiondevicephilipshue.py | 2 +- e2e_tests/test_locator.py | 2 ++ e2e_tests/test_rulesManager.py | 2 +- e2e_tests/test_solarDeviceManager.py | 2 +- 4 files changed, 5 insertions(+), 3 deletions(-) diff --git a/e2e_tests/test_actiondevicephilipshue.py b/e2e_tests/test_actiondevicephilipshue.py index 744c269..27c4564 100644 --- a/e2e_tests/test_actiondevicephilipshue.py +++ b/e2e_tests/test_actiondevicephilipshue.py @@ -6,7 +6,7 @@ from config.config import Config from business.actionDevicePhilipsHue import ActionDevicePhilipsHue -# dumb test for quality + test_locator = test_locator # should have a configured philips hue tor testing diff --git a/e2e_tests/test_locator.py b/e2e_tests/test_locator.py index e2e9e97..cec1cb1 100644 --- a/e2e_tests/test_locator.py +++ b/e2e_tests/test_locator.py @@ -1,2 +1,4 @@ import sys sys.path.insert(0, '../pvAlert') + +# for pytest \ No newline at end of file diff --git a/e2e_tests/test_rulesManager.py b/e2e_tests/test_rulesManager.py index f8ba052..022a574 100644 --- a/e2e_tests/test_rulesManager.py +++ b/e2e_tests/test_rulesManager.py @@ -6,7 +6,7 @@ from business.connectManager import ConnectManager from business.rulesManager import RulesManager -# dumb test + test_locator = test_locator # need a configured config.ini file !!! diff --git a/e2e_tests/test_solarDeviceManager.py b/e2e_tests/test_solarDeviceManager.py index 1353e4a..8f63f5a 100644 --- a/e2e_tests/test_solarDeviceManager.py +++ b/e2e_tests/test_solarDeviceManager.py @@ -7,7 +7,7 @@ from exception.HuaweiApiException import HuaweiApiException test_locator = test_locator -# need a configured config.ini file !!! +# need a configured config.ini file !! dirname = os.path.dirname(__file__) # get parrent config file From 3cc9b62eecd512d733982b6f9ba041bcb5c9d75e Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 17:50:01 +0200 Subject: [PATCH 14/15] update tests quality --- e2e_tests/test_locator.py | 2 +- tests/test_config.py | 39 +++++++++++++++++++++++++----------- tests/test_connectManager.py | 30 ++++++++++----------------- tests/test_locator.py | 4 ++++ 4 files changed, 43 insertions(+), 32 deletions(-) create mode 100644 tests/test_locator.py diff --git a/e2e_tests/test_locator.py b/e2e_tests/test_locator.py index cec1cb1..b6cae56 100644 --- a/e2e_tests/test_locator.py +++ b/e2e_tests/test_locator.py @@ -1,4 +1,4 @@ import sys sys.path.insert(0, '../pvAlert') -# for pytest \ No newline at end of file +# for pytest diff --git a/tests/test_config.py b/tests/test_config.py index a4ed788..eaccd2b 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,76 +1,91 @@ import os - -import sys -sys.path.insert(0, '../pvAlert') - - +import test_locator from config.config import Config +test_locator = test_locator -dirname = os.path.dirname(__file__) -#get parrent config file -home = os.path.dirname(dirname) +current_dirname = os.path.dirname(__file__) +# get parrent config file +home = os.path.dirname(current_dirname) conffile = os.path.join(home, 'data/config_test.ini') conf = Config(conffile) - -def test_config_userName(): +def test_config_user_name(): assert conf.userName == "Eugen_API" + def test_config_password(): assert conf.systemCode == "SHVhd2VpQDIwMTk=" + def test_config_domain(): assert conf.baseUrl == "https://eu5.fusionsolar.huawei.com/thirdData/" + def test_config_session(): assert conf.sessionFile != "" + def test_config_session_duration(): assert conf.sessionDuration == 1620 + def test_config_login(): assert conf.loginUri == "https://eu5.fusionsolar.huawei.com/thirdData/login" + def test_config_logout(): assert conf.logoutUri == "https://eu5.fusionsolar.huawei.com/thirdData/logout" + def test_config_stations(): assert conf.stationUri == "https://eu5.fusionsolar.huawei.com/thirdData/stations" + def test_config_station_cache_info_file(): assert conf.cacheInfoFile != "" + def test_config_station_duration(): assert conf.cacheInfoFileDuration == 86400 + def test_config_devices(): assert conf.devicesUri == "https://eu5.fusionsolar.huawei.com/thirdData/getDevList" -def test_config_devicesKpi(): + +def test_config_devices_kpi(): assert conf.deviceKpiUri == "https://eu5.fusionsolar.huawei.com/thirdData/getDevRealKpi" + def test_config_hue_discovery(): assert conf.hue_discovery_url == "https://discovery.meethue.com" + def test_config_hue_all_dev(): assert conf.hue_url_all_dev == "https:///clip/v2/resource/light" + def test_config_hue_username(): assert conf.hue_username == "youpi" + def test_config_hue_client_key(): assert conf.hue_client_key == "matin" + def test_config_hue_cache_file(): assert conf.hue_cache_file != "" + def test_config_hue_cache_file_duration(): assert conf.hue_cache_duration == 86400 + def test_config_alert_url_configured(): assert conf.alert_configured_url == "https://ntfy.sh/a-Testing-topic" + def test_config_alert_title(): - assert conf.alert_title == "a title" \ No newline at end of file + assert conf.alert_title == "a title" diff --git a/tests/test_connectManager.py b/tests/test_connectManager.py index 1a09977..5b7ef6f 100644 --- a/tests/test_connectManager.py +++ b/tests/test_connectManager.py @@ -1,26 +1,21 @@ import time import pytest import shutil - +import test_locator import os from os.path import exists as file_exists from shutil import copyfile - -import sys -sys.path.insert(0, '../pvAlert') - - from config.config import Config from business.connectManager import ConnectManager +test_locator = test_locator -dirname = os.path.dirname(__file__) -#get parrent config file -home = os.path.dirname(dirname) +current_dir = os.path.dirname(__file__) +# get parrent config file +home = os.path.dirname(current_dir) conffile = os.path.join(home, 'data/config_test.ini') conf = Config(conffile) - mgmt = ConnectManager(conf) current_session_file = conf.sessionFile + ".bak" @@ -76,6 +71,7 @@ def test_delete_session(manage_session_safe_delete_session_file): mgmt.delete_session() assert mgmt.session_cookie == {} + def test_load_session_without_file(manage_session_safe_delete_session_file): ret = mgmt.load_session() if not ret: @@ -85,12 +81,12 @@ def test_load_session_without_file(manage_session_safe_delete_session_file): def test_load_session_with_outdated_file(backup_restore_session_file): - mgmt = ConnectManager(conf) + conn_mg = ConnectManager(conf) old = conf.sessionDuration conf.sessionDuration = 1 # await for file system to update time.sleep(2) - ret = mgmt.load_session() + ret = conn_mg.load_session() conf.sessionDuration = old if not ret: assert True @@ -105,12 +101,11 @@ def test_logout_wrong_url(): conf.logoutUri = fake_logout_url try: mgmt.logout(mgmt.session_cookie) - except Exception as err: - template = "An exception of type {0} occurred. Arguments:\n{1!r}" - message = template.format(type(err).__name__, err.args) + except Exception: conf.logoutUri = good_url assert True + def test_login_with_wrong_credentials(): mgmt.session_cookie = "" mgmt.login() @@ -127,9 +122,6 @@ def test_login_with_wrong_url(): conf.loginUri = fake_login_url try: mgmt.login() - except Exception as err: - template = "An exception of type {0} occurred. Arguments:\n{1!r}" - message = template.format(type(err).__name__, err.args) + except Exception: conf.loginUri = good_url assert True - diff --git a/tests/test_locator.py b/tests/test_locator.py new file mode 100644 index 0000000..b6cae56 --- /dev/null +++ b/tests/test_locator.py @@ -0,0 +1,4 @@ +import sys +sys.path.insert(0, '../pvAlert') + +# for pytest From 56cf4f3f5b2304841bd8e7810e2599f2f4f7a6d4 Mon Sep 17 00:00:00 2001 From: bastoscorp <70770115+bastoscorp@users.noreply.github.com> Date: Wed, 17 Jul 2024 17:52:23 +0200 Subject: [PATCH 15/15] update main and exception quality --- exception/HuaweiApiException.py | 2 +- main.py | 23 +++++++++-------------- 2 files changed, 10 insertions(+), 15 deletions(-) diff --git a/exception/HuaweiApiException.py b/exception/HuaweiApiException.py index 47792fc..21b7549 100644 --- a/exception/HuaweiApiException.py +++ b/exception/HuaweiApiException.py @@ -6,4 +6,4 @@ def __init__(self, message, errors): # Now for your custom code... errStr = "Error : " + errors["data"] + " failCode : " + errors["failCode"] - self.errors = errStr \ No newline at end of file + self.errors = errStr diff --git a/main.py b/main.py index 6d04141..c71b4dc 100644 --- a/main.py +++ b/main.py @@ -6,7 +6,7 @@ logging.root.setLevel(logging.INFO) -#formatter = logging.Formatter('%(asctime)s - %(module)s %(filename)s %(funcName)s:%(lineno)s - %(name)s -%(message)s') +# formatter = logging.Formatter('%(asctime)s - %(module)s %(filename)s %(funcName)s:%(lineno)s - %(name)s -%(message)s') formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(module)s - %(funcName)s - %(message)s') @@ -16,33 +16,28 @@ console_handler.addFilter(logging.Filter()) logging.root.addHandler(console_handler) -dirname = os.path.dirname(__file__) -dirlogs = os.path.join(dirname,"logs") -if os.path.exists(dirlogs) == False: +current_dir = os.path.dirname(__file__) +dirlogs = os.path.join(current_dir, "logs") +if not os.path.exists(dirlogs): os.mkdir(dirlogs) filename = os.path.join(dirlogs, 'pvAlert.log') - - - - myTimeHandler = TimedRotatingFileHandler(filename, when='midnight', backupCount=60, encoding="utf-8") myTimeHandler.suffix = "%Y-%m-%d.log" -#for testing only -#myTimeHandler = TimedRotatingFileHandler(filename, when='s', interval=5 , backupCount=60, encoding="utf-8") -#myTimeHandler.suffix = "%Y-%m-%d--%H-%M.log" +# for testing only +# myTimeHandler = TimedRotatingFileHandler(filename, when='s', interval=5 , backupCount=60, encoding="utf-8") +# myTimeHandler.suffix = "%Y-%m-%d--%H-%M.log" -myTimeHandler.filter(logging.Filter()) +# myTimeHandler.filter(logging.Filter()) myTimeHandler.setFormatter(formatter) myTimeHandler.setLevel(logging.INFO) logging.root.addHandler(myTimeHandler) - logging.info("Start pvAlert Checking ....") -conf_filename = os.path.join(dirname, "config.ini") +conf_filename = os.path.join(current_dir, "config.ini") config = Config(conf_filename) conso_manager = RulesManager(config)