diff --git a/pyproject.toml b/pyproject.toml index 82f0fa0..7df3dde 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,14 +1,12 @@ [tool.poetry] name = "robotframework-heal" -version = "0.3.0dev3" +version = "0.3.0" description = "A Robot Framework Listener for library agnostic self-healing and smart recovery of tests" authors = ["Many Kasiriha "] maintainers = ["Many Kasiriha "] readme = "README.md" license = "Apache-2.0" -packages = [ - { include = "SelfHealing", from = "src" } -] +packages = [{ include = "SelfHealing", from = "src" }] exclude = ["src/SelfHealing/.env"] homepage = "https://github.com/manykarim/robotframework-heal" @@ -30,7 +28,7 @@ jinja2 = "*" litellm = "*" -[tool.poetry.group.dev.dependencies] +[tool.poetry.group.dev.dependencies] pytest = "*" invoke = "*" coverage = "*" diff --git a/src/SelfHealing/appium_healing.py b/src/SelfHealing/appium_healing.py index 531cdfa..7e173e9 100644 --- a/src/SelfHealing/appium_healing.py +++ b/src/SelfHealing/appium_healing.py @@ -1,17 +1,17 @@ -from robot.libraries.BuiltIn import BuiltIn -from robot.api import logger import json -from bs4 import BeautifulSoup -import re -from uuid import uuid4 -from .llm_client import LLM_TEXT_MODEL, completion -from .utils import extract_json_objects, filter_dict, compare_dict, xpath_to_browser, get_xpath_selector, get_simplified_dom_tree, generate_unique_css_selector, generate_unique_xpath_selector, is_leaf_or_lowest, has_parent_dialog_without_open, has_child_dialog_without_open, has_direct_text, is_headline, is_div_in_li, is_p, filter_locator_list_with_fuzz -from .locator_db import LocatorDetailsDB -from tinydb import Query -from cssify import cssify import pprint +import re from time import sleep + +from bs4 import BeautifulSoup from lxml import etree +from robot.api import logger +from robot.libraries.BuiltIn import BuiltIn +from litellm import completion + +from .llm_client import LLM_TEXT_MODEL +from .utils import extract_json_objects, generate_unique_xpath_selector + try: from appium.webdriver.common.appiumby import AppiumBy except ImportError: @@ -24,7 +24,7 @@ class AppiumHealer: _instance = None locators = {} dom_tree = None - + def __new__(cls, *args, **kwargs): if cls._instance is None: cls._instance = super(AppiumHealer, cls).__new__(cls) @@ -32,18 +32,19 @@ def __new__(cls, *args, **kwargs): def __init__(self, instance=None, **kwargs): if instance: - self.appium=instance._current_application() - self.use_locator_db = kwargs.get('use_locator_db', False) - self.use_llm_for_locator_proposals = kwargs.get('use_llm_for_locator_proposals', True) - self.parse_full_page = kwargs.get('parse_full_page', False) - + self.appium = instance._current_application() + self.use_locator_db = kwargs.get("use_locator_db", False) + self.use_llm_for_locator_proposals = kwargs.get( + "use_llm_for_locator_proposals", True + ) + self.parse_full_page = kwargs.get("parse_full_page", False) def set_appium_instance(self, instance): self.appium = instance._current_application() def is_locator_broken(self, message) -> bool: - return ('did not match any elements' in message) - + return "did not match any elements" in message + def is_element_not_ready(self, message) -> bool: pass @@ -59,30 +60,37 @@ def close_modal_dialog(self) -> bool: def is_permission_popup_open(self) -> bool: try: - if len(self.appium.find_elements(AppiumBy.XPATH, "//*[contains(@resource-id,'dialog_container')]")) == 1: + if ( + len( + self.appium.find_elements( + AppiumBy.XPATH, "//*[contains(@resource-id,'dialog_container')]" + ) + ) + == 1 + ): return True except: pass return False def confirm_permission_popup(self) -> bool: - - buttons = ['ALLOW', 'TURN', 'CONFIRM', 'OK', 'OKAY'] + buttons = ["ALLOW", "TURN", "CONFIRM", "OK", "OKAY"] for button in buttons: try: - elem = self.appium.find_element(AppiumBy.XPATH, f"//*[contains(@text, '{button}')] ") + elem = self.appium.find_element( + AppiumBy.XPATH, f"//*[contains(@text, '{button}')] " + ) elem.click() break except: pass - logger.error(f"Popup Dialog could not be closed") + logger.error("Popup Dialog could not be closed") - def get_fixed_locator(self, data, result) -> str: - output_dir = BuiltIn().get_variable_value('${OUTPUT_DIR}') - testsuite = BuiltIn().get_variable_value('${SUITE NAME}') - -# source = self.appium.page_source + output_dir = BuiltIn().get_variable_value("${OUTPUT_DIR}") + testsuite = BuiltIn().get_variable_value("${SUITE NAME}") + + # source = self.appium.page_source if self.parse_full_page: soup_list = self.get_full_dom() else: @@ -95,90 +103,121 @@ def get_fixed_locator(self, data, result) -> str: for soup in soup_list: if self.use_llm_for_locator_proposals: - fixed_locator_list += self.get_locator_proposals_from_llm(data, result, str(soup.hierarchy)) + fixed_locator_list += self.get_locator_proposals_from_llm( + data, result, str(soup.hierarchy) + ) else: - fixed_locator_list += self.get_locator_proposals_from_parser(data, result, str(soup.hierarchy)) - - + fixed_locator_list += self.get_locator_proposals_from_parser( + data, result, str(soup.hierarchy) + ) print(f"{fixed_locator_list}") - existing_locators = {d['fixed_locator'] for d in fixed_locators_with_info} - - fixed_locator_list = [item for item in fixed_locator_list if item not in existing_locators] + existing_locators = {d["fixed_locator"] for d in fixed_locators_with_info} + + fixed_locator_list = [ + item for item in fixed_locator_list if item not in existing_locators + ] for retry_selector in fixed_locator_list: try: - retry_locator_info = self.get_locator_info(str(soup.hierarchy), retry_selector) - fixed_locators_with_info.append({"index": int(index), "fixed_locator": retry_selector, "additional_info": retry_locator_info}) + retry_locator_info = self.get_locator_info( + str(soup.hierarchy), retry_selector + ) + fixed_locators_with_info.append( + { + "index": int(index), + "fixed_locator": retry_selector, + "additional_info": retry_locator_info, + } + ) except: pass index += 1 print(f"{fixed_locators_with_info}") - - logger.info(f"Pre-filtered locator candidates with info: {pprint.pformat(fixed_locators_with_info)}", also_console=False) + logger.info( + f"Pre-filtered locator candidates with info: {pprint.pformat(fixed_locators_with_info)}", + also_console=False, + ) messages = [] - messages.append({ - 'role': 'system', - 'content': ( - "You are a xpath locator self-healing tool for Robot Framework.\n" - "You will select exactly one fixed_locator and index from a list of Locators Proposals.\n" - "Always return the unchanged fixed_locator.\n" - 'Respond using the following json schema: {"index": "index of fixed_locator", "fixed_locator": "selected fixed_locator"}' - ) - - }) - - messages.append({ - 'role': 'user', - 'content': ( - f"Broken Locator : `{failed_locator}`\n" - f"Keyword : `{data.name}`\n" - - f"Analyse the `additional_info` for each `fixed_locator` to select the `fixed_locator` which most likely matches to failed_locator=`{failed_locator}`.\n" + messages.append( + { + "role": "system", + "content": ( + "You are a xpath locator self-healing tool for Robot Framework.\n" + "You will select exactly one fixed_locator and index from a list of Locators Proposals.\n" + "Always return the unchanged fixed_locator.\n" + 'Respond using the following json schema: {"index": "index of fixed_locator", "fixed_locator": "selected fixed_locator"}' + ), + } + ) - f"** Locators Proposals **\n" - f"```{json.dumps(fixed_locators_with_info)}```" - ) + messages.append( + { + "role": "user", + "content": ( + f"Broken Locator : `{failed_locator}`\n" + f"Keyword : `{data.name}`\n" + f"Analyse the `additional_info` for each `fixed_locator` to select the `fixed_locator` which most likely matches to failed_locator=`{failed_locator}`.\n" + f"** Locators Proposals **\n" + f"```{json.dumps(fixed_locators_with_info)}```" + ), + } + ) + try: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + temperature=0.1, + response_format={"type": "json_object"}, + ) + except Exception: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + response_format={"type": "json_object"}, + ) - }) + solution_text = response["choices"][0]["message"]["content"] - response = completion( - model = LLM_TEXT_MODEL, - messages = messages, - temperature = 0.1, - response_format = { "type": "json_object" } + logger.info( + f"2nd LLM response with sorted candidates: {solution_text}", + also_console=False, ) - solution_text = response['choices'][0]['message']['content'] - - logger.info(f"2nd LLM response with sorted candidates: {solution_text}", also_console=False) sorted_retry_locators = list(extract_json_objects(solution_text))[0] if isinstance(sorted_retry_locators, dict): - if 'index' in sorted_retry_locators: - retry_locator = next(item["fixed_locator"] for item in fixed_locators_with_info if int(item['index']) == int(sorted_retry_locators['index'])) + if "index" in sorted_retry_locators: + retry_locator = next( + item["fixed_locator"] + for item in fixed_locators_with_info + if int(item["index"]) == int(sorted_retry_locators["index"]) + ) return retry_locator - elif 'fixed_locators' in sorted_retry_locators: + elif "fixed_locators" in sorted_retry_locators: retry_locator = sorted_retry_locators["fixed_locators"] - elif 'fixed_locator' in sorted_retry_locators: + elif "fixed_locator" in sorted_retry_locators: retry_locator = sorted_retry_locators["fixed_locator"] - + if isinstance(retry_locator, list): retry_locator = retry_locator[0] if isinstance(retry_locator, str): - return retry_locator - + return retry_locator + return None - def rerun_keyword(self, data, fixed_locator = None) -> str: + def rerun_keyword(self, data, fixed_locator=None) -> str: if fixed_locator: data.args = list(data.args) data.args[0] = fixed_locator try: - logger.info(f"Re-trying Keyword '{data.name}' with arguments '{data.args}'.", also_console=True) + logger.info( + f"Re-trying Keyword '{data.name}' with arguments '{data.args}'.", + also_console=True, + ) BuiltIn().run_keyword(data.name, *data.args) BuiltIn().run_keyword("Capture Page Screenshot") return "PASS" @@ -186,18 +225,17 @@ def rerun_keyword(self, data, fixed_locator = None) -> str: logger.debug(f"Unexpected error: {e}") return "FAIL" - def is_scrollable(self): dom_list = [] driver = self.appium initial_dom = driver.page_source - + # Perform a swipe to check if more elements are available driver.swipe(500, 1500, 500, 500, 1000) sleep(1) new_dom = driver.page_source - + return new_dom != initial_dom def get_locator_info(self, source, locator): @@ -206,15 +244,26 @@ def get_locator_info(self, source, locator): element = tree.xpath(locator)[0] locator_info = {} locator_info["locator"] = locator - attribute_list = ["package", "content-desc", "index", "hint", "bounds", "text", "clickable", "checkable", "scrollable", "class"] - + attribute_list = [ + "package", + "content-desc", + "index", + "hint", + "bounds", + "text", + "clickable", + "checkable", + "scrollable", + "class", + ] + for attribute in attribute_list: try: value = element.get(attribute) except: value = None if value: - locator_info[attribute]=value + locator_info[attribute] = value testsuite = BuiltIn().get_variable_value("${SUITE NAME}") locator_info["testsuite"] = testsuite return locator_info @@ -223,7 +272,7 @@ def get_current_dom(self): dom_list = [] driver = self.appium initial_dom = driver.page_source - dom_list.append(BeautifulSoup(initial_dom, 'xml')) + dom_list.append(BeautifulSoup(initial_dom, "xml")) return dom_list def get_full_dom(self): @@ -232,34 +281,34 @@ def get_full_dom(self): driver = self.appium initial_dom = driver.page_source - dom_list.append(BeautifulSoup(initial_dom, 'xml')) + dom_list.append(BeautifulSoup(initial_dom, "xml")) deviceSize = driver.get_window_size() - screenWidth = deviceSize['width'] - screenHeight = deviceSize['height'] - startx = screenWidth/2 - endx = screenWidth/2 - starty = screenHeight*8/9 - endy = screenHeight/9 + screenWidth = deviceSize["width"] + screenHeight = deviceSize["height"] + startx = screenWidth / 2 + endx = screenWidth / 2 + starty = screenHeight * 8 / 9 + endy = screenHeight / 9 # Perform a swipe to check if more elements are available # driver.swipe(500, 1500, 500, 500, 1000) driver.swipe(startx, starty, endx, endy, 1000) new_dom = driver.page_source - + # If the DOM changes after the swipe, continue swiping while new_dom != initial_dom: - dom_list.append(BeautifulSoup(new_dom, 'xml')) + dom_list.append(BeautifulSoup(new_dom, "xml")) initial_dom = new_dom driver.swipe(startx, starty, endx, endy, 1000) new_dom = driver.page_source # Swipe to top again - startx2 = screenWidth/2 - endx2 = screenWidth/2 - starty2 = screenHeight*2/9 - endy2 = screenHeight*8/9 + startx2 = screenWidth / 2 + endx2 = screenWidth / 2 + starty2 = screenHeight * 2 / 9 + endy2 = screenHeight * 8 / 9 driver.swipe(startx2, starty2, endx2, endy2, 1000) new_dom = driver.page_source @@ -268,12 +317,11 @@ def get_full_dom(self): driver.swipe(startx2, starty2, endx2, endy2, 1000) new_dom = driver.page_source - #full_dom_tree = self.merge_dom_trees(dom_list) - + # full_dom_tree = self.merge_dom_trees(dom_list) + return dom_list def is_element_visible_with_swiping(self, locator): - driver = self.appium if len(self.appium.find_elements(AppiumBy.XPATH, locator)) == 1: return True @@ -281,12 +329,12 @@ def is_element_visible_with_swiping(self, locator): initial_dom = driver.page_source deviceSize = driver.get_window_size() - screenWidth = deviceSize['width'] - screenHeight = deviceSize['height'] - startx = screenWidth/2 - endx = screenWidth/2 - starty = screenHeight*8/9 - endy = screenHeight/9 + screenWidth = deviceSize["width"] + screenHeight = deviceSize["height"] + startx = screenWidth / 2 + endx = screenWidth / 2 + starty = screenHeight * 8 / 9 + endy = screenHeight / 9 driver.swipe(startx, starty, endx, endy, 1000) new_dom = driver.page_source @@ -298,11 +346,11 @@ def is_element_visible_with_swiping(self, locator): driver.swipe(startx, starty, endx, endy, 1000) sleep(1) new_dom = driver.page_source - - startx2 = screenWidth/2 - endx2 = screenWidth/2 - starty2 = screenHeight*2/9 - endy2 = screenHeight*8/9 + + startx2 = screenWidth / 2 + endx2 = screenWidth / 2 + starty2 = screenHeight * 2 / 9 + endy2 = screenHeight * 8 / 9 driver.swipe(startx2, starty2, endx2, endy2, 1000) new_dom = driver.page_source @@ -312,89 +360,86 @@ def is_element_visible_with_swiping(self, locator): new_dom = driver.page_source return False - def merge_dom_trees(self, dom_list): # Parse the first DOM tree - full_tree = BeautifulSoup(dom_list[0], 'xml') - + full_tree = BeautifulSoup(dom_list[0], "xml") + def add_elements(soup, elements): for element in elements: - if not any(existing_element == element for existing_element in soup.find_all(element.name)): + if not any( + existing_element == element + for existing_element in soup.find_all(element.name) + ): soup.append(element) - + for dom in dom_list[1:]: - new_tree = BeautifulSoup(dom, 'xml') + new_tree = BeautifulSoup(dom, "xml") add_elements(full_tree, new_tree.find_all()) - + return full_tree def get_locator_proposals_from_llm(self, data, result, source): - failed_locator = BuiltIn().replace_variables(str(result.args[0])) - schema = { - "fixed_locator": "The fixed xpath locator." - } + schema = {"fixed_locator": "The fixed xpath locator."} locator_has_been_fixed = False retry_count = 0 error_message = result.message prompt_content = { - 'error_message': error_message, - 'failed_locator': failed_locator, - 'keyword': data.name, - 'page_source': source - } + "error_message": error_message, + "failed_locator": failed_locator, + "keyword": data.name, + "page_source": source, + } messages = [] - + android_widgets = [ - 'android.widget.Button', - 'android.widget.EditText', - 'android.widget.TextView', - 'android.widget.ImageView', - 'android.widget.CheckBox', - 'android.widget.RadioButton', - 'android.widget.ToggleButton', - 'android.widget.ProgressBar', - 'android.widget.SeekBar', - 'android.widget.Spinner', + "android.widget.Button", + "android.widget.EditText", + "android.widget.TextView", + "android.widget.ImageView", + "android.widget.CheckBox", + "android.widget.RadioButton", + "android.widget.ToggleButton", + "android.widget.ProgressBar", + "android.widget.SeekBar", + "android.widget.Spinner", # 'android.widget.ListView', # 'android.widget.GridView', # 'android.widget.ScrollView', - 'android.widget.Switch', - 'android.widget.RatingBar' + "android.widget.Switch", + "android.widget.RatingBar", ] - SYS_PROMPT= ( - "You are a xpath locator self-healing tool." - "You will provide a fixed_locator for a failed_locator." - "The User prompt will contain data for 'error_message' , 'failed_locator', 'keyword' and 'page_source'." - "You will analyze the `page_source` and the `error_message` and find the eight best alternative fixed_locators for the failed_locator." - f"Only elements of class {android_widgets} are candidates" - "Keywords to fill or enter text are always related to `android.widget.EditText` elements." - "Keywords like `Click` are often related to 'android.widget.Button','android.widget.CheckBox', 'android.widget.ToggleButton', 'android.widget.RadioButton' or 'android.widget.TextView' elements." - 'Respond using the following json schema: {"fixed_locators": ["locator1", "locator2", "locator3", ... ]}.' - 'Example: {"fixed_locators": ["css=input[id=\'my_id\']", "//*[contains(@text,\'Login\')]", "//*[@resource-id=\'android:id/content\')", "//android.widget.EditText[@content-desc=\'password\')"}' - ) + SYS_PROMPT = ( + "You are a xpath locator self-healing tool." + "You will provide a fixed_locator for a failed_locator." + "The User prompt will contain data for 'error_message' , 'failed_locator', 'keyword' and 'page_source'." + "You will analyze the `page_source` and the `error_message` and find the eight best alternative fixed_locators for the failed_locator." + f"Only elements of class {android_widgets} are candidates" + "Keywords to fill or enter text are always related to `android.widget.EditText` elements." + "Keywords like `Click` are often related to 'android.widget.Button','android.widget.CheckBox', 'android.widget.ToggleButton', 'android.widget.RadioButton' or 'android.widget.TextView' elements." + 'Respond using the following json schema: {"fixed_locators": ["locator1", "locator2", "locator3", ... ]}.' + 'Example: {"fixed_locators": ["css=input[id=\'my_id\']", "//*[contains(@text,\'Login\')]", "//*[@resource-id=\'android:id/content\')", "//android.widget.EditText[@content-desc=\'password\')"}' + ) - messages.append({ - 'role': 'system', - 'content': SYS_PROMPT - }) + messages.append({"role": "system", "content": SYS_PROMPT}) - messages.append({ - 'role': 'user', - 'content': json.dumps(prompt_content), - }) + messages.append( + { + "role": "user", + "content": json.dumps(prompt_content), + } + ) - while not locator_has_been_fixed: retry_count += 1 if retry_count > 3: - break + break llm_max_retries = 3 llm_attempts = 0 @@ -402,20 +447,27 @@ def get_locator_proposals_from_llm(self, data, result, source): while llm_attempts < llm_max_retries and not llm_success: try: - response = completion( - model = LLM_TEXT_MODEL, - messages = messages, - temperature = 0.1, - response_format = { "type": "json_object" } - ) + try: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + temperature=0.1, + response_format={"type": "json_object"}, + ) + except Exception: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + response_format={"type": "json_object"}, + ) llm_success = True except Exception as e: llm_attempts += 1 print(f"Attempt {llm_attempts} failed: {e}") - solution_text = response['choices'][0]['message']['content'] - + solution_text = response["choices"][0]["message"]["content"] + # messages.append({ # "role": "assistant", # "content": solution_text @@ -425,7 +477,10 @@ def get_locator_proposals_from_llm(self, data, result, source): # f.write(solution_text) # f.close() - logger.info(f"1st LLM response: {pprint.pformat(solution_text)}\n", also_console=False) + logger.info( + f"1st LLM response: {pprint.pformat(solution_text)}\n", + also_console=False, + ) try: fixed_locator_dict = list(extract_json_objects(solution_text))[0] @@ -434,7 +489,7 @@ def get_locator_proposals_from_llm(self, data, result, source): # Attempt to fix common JSON formatting issues solution_text = solution_text.replace("}}", "}") solution_text = solution_text.replace("{{", "{") - pattern = r'```(.*?)```' + pattern = r"```(.*?)```" match = re.search(pattern, solution_text, re.DOTALL) if match: solution_text = match.group(1) @@ -448,37 +503,36 @@ def get_locator_proposals_from_llm(self, data, result, source): logger.debug(f"Unexpected error: {e}") locator_has_been_fixed = False continue - + try: - fixed_locator_list = fixed_locator_dict['fixed_locators'] + fixed_locator_list = fixed_locator_dict["fixed_locators"] return fixed_locator_list except: locator_has_been_fixed = False continue def get_locator_proposals_from_parser(self, data, result, source): - soup = BeautifulSoup(source, 'xml') + soup = BeautifulSoup(source, "xml") locators = [] android_widgets = [ - 'android.widget.Button', - 'android.widget.EditText', - 'android.widget.TextView', - 'android.widget.ImageView', - 'android.widget.CheckBox', - 'android.widget.RadioButton', - 'android.widget.ToggleButton', - 'android.widget.ProgressBar', - 'android.widget.SeekBar', - 'android.widget.Spinner', + "android.widget.Button", + "android.widget.EditText", + "android.widget.TextView", + "android.widget.ImageView", + "android.widget.CheckBox", + "android.widget.RadioButton", + "android.widget.ToggleButton", + "android.widget.ProgressBar", + "android.widget.SeekBar", + "android.widget.Spinner", # 'android.widget.ListView', # 'android.widget.GridView', # 'android.widget.ScrollView', - 'android.widget.Switch', - 'android.widget.RatingBar' + "android.widget.Switch", + "android.widget.RatingBar", ] elements = soup.find_all(android_widgets) - for elem in elements: if self.locators.get(str(elem)): locator = self.locators.get(str(elem)) @@ -488,7 +542,7 @@ def get_locator_proposals_from_parser(self, data, result, source): if locator: locators.append(locator) self.locators[f"{str(elem)}"] = locator - + return locators @@ -496,4 +550,4 @@ def get_locator(elem, soup): selector = generate_unique_xpath_selector(elem, soup) if selector: return selector - return None \ No newline at end of file + return None diff --git a/src/SelfHealing/browser_healing.py b/src/SelfHealing/browser_healing.py index f14fc92..d1f0b64 100644 --- a/src/SelfHealing/browser_healing.py +++ b/src/SelfHealing/browser_healing.py @@ -1,19 +1,38 @@ -from robot.libraries.BuiltIn import BuiltIn -from robot.api import logger import json -from bs4 import BeautifulSoup -import re -from .llm_client import LLM_TEXT_MODEL, LLM_LOCATOR_MODEL, litellm, completion -from .utils import extract_json_objects, filter_dict, compare_dict, xpath_to_browser, get_xpath_selector, get_simplified_dom_tree, generate_unique_css_selector, generate_unique_xpath_selector, is_leaf_or_lowest, has_parent_dialog_without_open, has_child_dialog_without_open, has_direct_text, is_headline, is_div_in_li, is_p, filter_locator_list_with_fuzz, filter_locator_list_with_fuzz_median -from .locator_db import LocatorDetailsDB -from tinydb import Query -from cssify import cssify import pprint +import re import time from concurrent.futures import ProcessPoolExecutor, as_completed +from bs4 import BeautifulSoup +from cssify import cssify +from robot.api import logger +from robot.libraries.BuiltIn import BuiltIn +from tinydb import Query +from litellm import completion + + # Try relative imports first (when used as a package) +from .llm_client import LLM_TEXT_MODEL +from .locator_db import LocatorDetailsDB +from .utils import ( + compare_dict, + extract_json_objects, + filter_locator_list_with_fuzz_median, + generate_unique_css_selector, + get_simplified_dom_tree, + has_child_dialog_without_open, + has_direct_text, + has_parent_dialog_without_open, + is_div_in_li, + is_headline, + is_leaf_or_lowest, + is_p, + xpath_to_browser, +) + PARALLEL = True + class BrowserHealer: _instance = None @@ -23,22 +42,29 @@ def __new__(cls, *args, **kwargs): return cls._instance def __init__(self, instance=None, **kwargs): - self.browser=instance - self.use_locator_db = kwargs.get('use_locator_db', False) - self.use_llm_for_locator_proposals = kwargs.get('use_llm_for_locator_proposals', True) - self.read_clickable_info = kwargs.get('read_clickable_info', True) - + self.browser = instance + self.use_locator_db = kwargs.get("use_locator_db", False) + self.use_llm_for_locator_proposals = kwargs.get( + "use_llm_for_locator_proposals", True + ) + self.read_clickable_info = kwargs.get("read_clickable_info", True) + def is_locator_broken(self, message) -> bool: - return ('waiting for' in message or 'Element is not an' in message) and (not 'waiting for element to be' in message) - + return ("waiting for" in message or "Element is not an" in message) and ( + "waiting for element to be" not in message + ) + def is_element_not_ready(self, message) -> bool: - return 'element is not visible' in message or 'waiting for element to be' in message + return ( + "element is not visible" in message + or "waiting for element to be" in message + ) def is_modal_dialog_open(self) -> bool: - soup = BeautifulSoup(self.browser.get_page_source(), 'html.parser') + soup = BeautifulSoup(self.browser.get_page_source(), "html.parser") # Find all elements with 'display: none' - dialogs = soup.find_all('dialog', {"open": True}) + dialogs = soup.find_all("dialog", {"open": True}) if len(dialogs) > 0: return True @@ -64,15 +90,15 @@ def is_page_ready(self) -> bool: return bool(is_loading) def wait_until_page_is_ready(self, timeout=20): - BuiltIn().wait_until_keyword_succeeds(timeout, "1s", "Browser.Wait For Load State", "load", "1s" ) + BuiltIn().wait_until_keyword_succeeds( + timeout, "1s", "Browser.Wait For Load State", "load", "1s" + ) def close_modal_dialog(self): - soup = BeautifulSoup(self.browser.get_page_source(), 'html.parser') - dialog = soup.find('dialog', {"open": True}) - - prompt_content = { - 'page_source': str(dialog) - } + soup = BeautifulSoup(self.browser.get_page_source(), "html.parser") + dialog = soup.find("dialog", {"open": True}) + + prompt_content = {"page_source": str(dialog)} schema = { "fixed_locator": "The fixed css or xpath locator. Starts with xpath= or css=" @@ -80,42 +106,46 @@ def close_modal_dialog(self): messages = [] - messages.append({ - 'role': 'system', - 'content': ( - "You are a xpath and css selector tool that shall close a dialog." - "You will analyze the `page_source` and find a short and unique xpath or css selector to close the dialog." - "Most likely a button or a link needs to be clicked" - "Respond only in valid json that looks like this: {'fixed_locator': }" - "When the 'fixed_locator' is an xpath, always add a xpath= prefix to the locator." - "When the 'fixed_locator' is an css selector, always add a css= prefix to the locator." - f"Use the following schema: ```json{json.dumps(schema)}```." - ) + messages.append( + { + "role": "system", + "content": ( + "You are a xpath and css selector tool that shall close a dialog." + "You will analyze the `page_source` and find a short and unique xpath or css selector to close the dialog." + "Most likely a button or a link needs to be clicked" + "Respond only in valid json that looks like this: {'fixed_locator': }" + "When the 'fixed_locator' is an xpath, always add a xpath= prefix to the locator." + "When the 'fixed_locator' is an css selector, always add a css= prefix to the locator." + f"Use the following schema: ```json{json.dumps(schema)}```." + ), + } + ) - }) - - messages.append({ - 'role': 'user', - 'content': ( - f"'page_source': ```{str(dialog)}```\n" - ) - }) + messages.append( + {"role": "user", "content": (f"'page_source': ```{str(dialog)}```\n")} + ) locator_has_been_fixed = False retry_count = 0 while not locator_has_been_fixed: retry_count += 1 if retry_count > 5: - break - - response = litellm.completion( - model = LLM_TEXT_MODEL, - messages = messages, - temperature = 0.1, - # top_k = 1, - response_format = { "type": "json_object" } - ) - solution_text = response['choices'][0]['message']['content'] + break + + try: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + temperature=0.1, + response_format={"type": "json_object"}, + ) + except Exception: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + response_format={"type": "json_object"}, + ) + solution_text = response["choices"][0]["message"]["content"] try: solution_dict = list(extract_json_objects(solution_text))[0] @@ -124,7 +154,7 @@ def close_modal_dialog(self): # Attempt to fix common JSON formatting issues solution_text = solution_text.replace("}}", "}") solution_text = solution_text.replace("{{", "{") - pattern = r'```(.*?)```' + pattern = r"```(.*?)```" match = re.search(pattern, solution_text, re.DOTALL) if match: solution_text = match.group(1) @@ -138,32 +168,35 @@ def close_modal_dialog(self): logger.debug(f"Unexpected error: {e}") locator_has_been_fixed = False continue - + try: - fixed_locator = str(solution_dict['fixed_locator']) + fixed_locator = str(solution_dict["fixed_locator"]) except: locator_has_been_fixed = False continue # Search for xpath= or css= in the solution_text - if fixed_locator.startswith('xpath'): + if fixed_locator.startswith("xpath"): # Remove xpath= from the string and store it as a variable - new_locator = re.sub('xpath.', '', fixed_locator) + new_locator = re.sub("xpath.", "", fixed_locator) retry_selector = "xpath=" + new_locator - elif fixed_locator.startswith('/'): + elif fixed_locator.startswith("/"): new_locator = fixed_locator retry_selector = "xpath=" + new_locator - elif fixed_locator.startswith('css'): + elif fixed_locator.startswith("css"): # Remove css= from the string and store it as a variable - new_locator = re.sub('css.', '', fixed_locator) - retry_selector = "css=" + new_locator + new_locator = re.sub("css.", "", fixed_locator) + retry_selector = "css=" + new_locator else: retry_selector = fixed_locator try: if self.browser.get_element_count(retry_selector) == 1: locator_has_been_fixed = True - logger.info(f"Locator to close dialog has been found: {retry_selector}", also_console=True) + logger.info( + f"Locator to close dialog has been found: {retry_selector}", + also_console=True, + ) self.browser.click(retry_selector) except: locator_has_been_fixed = False @@ -172,17 +205,14 @@ def close_modal_dialog(self): else: return None - def get_fixed_locator(self, data, result) -> str: try: old_log_level = BuiltIn().set_log_level("NONE") except Exception as e: logger.info(f"Error when setting log level: {e}") - output_dir = BuiltIn().get_variable_value('${OUTPUT_DIR}') - testsuite = BuiltIn().get_variable_value('${SUITE NAME}') - - + output_dir = BuiltIn().get_variable_value("${OUTPUT_DIR}") + testsuite = BuiltIn().get_variable_value("${SUITE NAME}") script = """() => { @@ -238,58 +268,68 @@ def get_fixed_locator(self, data, result) -> str: } """ - shadowdom_script ="""{ + shadowdom_script = """{ let html = document.documentElement.outerHTML for (e of Array.from(document.documentElement.querySelectorAll('*')).filter(el => el.shadowRoot)) {e.shadowRoot.innerHTML} }""" - + shadowdom_exist_script = """ () => { return Array.from(document.querySelectorAll('*')).some(el => el.shadowRoot); } """ try: - shadowdom_exists = self.browser.evaluate_javascript(None, shadowdom_exist_script) + shadowdom_exists = self.browser.evaluate_javascript( + None, shadowdom_exist_script + ) if shadowdom_exists: soup = BeautifulSoup( - self.browser.evaluate_javascript(None, - script), - 'html.parser' - ) + self.browser.evaluate_javascript(None, script), "html.parser" + ) else: - soup = BeautifulSoup(self.browser.get_page_source(), 'html.parser') + soup = BeautifulSoup(self.browser.get_page_source(), "html.parser") except: - soup = BeautifulSoup(self.browser.get_page_source(), 'html.parser') + soup = BeautifulSoup(self.browser.get_page_source(), "html.parser") source = get_simplified_dom_tree(str(soup.body)) failed_locator = BuiltIn().replace_variables(str(result.args[0])) - + if self.use_llm_for_locator_proposals: - fixed_locator_list = self.get_locator_proposals_from_llm(data, result, source) + fixed_locator_list = self.get_locator_proposals_from_llm( + data, result, source + ) else: - fixed_locator_list = self.get_locator_proposals_from_parser(data, result, source) - + fixed_locator_list = self.get_locator_proposals_from_parser( + data, result, source + ) + fixed_locators_with_info = [] fixed_locators_with_similarity = [] - if self.use_locator_db: locator_db = LocatorDetailsDB().db for fixed_locator in fixed_locator_list: try: retry_selector = get_locator_with_prefix(fixed_locator.strip()) - - if self.browser.get_element_count(retry_selector) == 1: + if self.browser.get_element_count(retry_selector) == 1: try: retry_locator_info = self.get_locator_info(retry_selector) - original_locator_info = locator_db.search(Query().locator == failed_locator & Query().testsuite == testsuite)[-1] - added, removed, modified, same, similarity = compare_dict(original_locator_info, retry_locator_info) + original_locator_info = locator_db.search( + Query().locator + == failed_locator & Query().testsuite + == testsuite + )[-1] + added, removed, modified, same, similarity = compare_dict( + original_locator_info, retry_locator_info + ) except: similarity = 0.01 - fixed_locators_with_similarity.append((retry_selector, similarity)) + fixed_locators_with_similarity.append( + (retry_selector, similarity) + ) except: continue @@ -298,40 +338,63 @@ def get_fixed_locator(self, data, result) -> str: except Exception as e: logger.info(f"Error when setting log level: {e}") - fixed_locators_with_similarity.sort(key=lambda x: x[1], reverse=True) for fixed_locator in fixed_locators_with_similarity: retry_selector = fixed_locator[0] locator_has_been_fixed = True return retry_selector - - - else: - - - index = 0 - while index < len(fixed_locator_list): try: - retry_selector = get_locator_with_prefix(fixed_locator_list[index].strip()) + retry_selector = get_locator_with_prefix( + fixed_locator_list[index].strip() + ) if self.browser.get_element_count(retry_selector) == 0: retry_selector = retry_selector.replace("button", "*") if self.browser.get_element_count(retry_selector) == 1: states = self.browser.get_element_states(retry_selector) - if 'visible' in states: - retry_locator_info = self.get_locator_info(retry_selector, read_clickable_info=self.read_clickable_info) - retry_locator_info = {key: value for key, value in retry_locator_info.items() if key not in ["testsuite"]} - fixed_locators_with_info.append({"index": int(index), "fixed_locator": retry_selector, "additional_info": retry_locator_info}) + if "visible" in states: + retry_locator_info = self.get_locator_info( + retry_selector, + read_clickable_info=self.read_clickable_info, + ) + retry_locator_info = { + key: value + for key, value in retry_locator_info.items() + if key not in ["testsuite"] + } + fixed_locators_with_info.append( + { + "index": int(index), + "fixed_locator": retry_selector, + "additional_info": retry_locator_info, + } + ) elif self.browser.get_element_count(retry_selector) > 1: - if self.browser.get_element_count(f"{retry_selector}:visible") == 1: + if ( + self.browser.get_element_count(f"{retry_selector}:visible") + == 1 + ): retry_selector = f"{retry_selector}:visible >> nth=0" - retry_locator_info = self.get_locator_info(retry_selector, read_clickable_info=self.read_clickable_info) - retry_locator_info = {key: value for key, value in retry_locator_info.items() if key not in ["testsuite"]} - fixed_locators_with_info.append({"index": int(index), "fixed_locator": retry_selector, "additional_info": retry_locator_info}) + retry_locator_info = self.get_locator_info( + retry_selector, + read_clickable_info=self.read_clickable_info, + ) + retry_locator_info = { + key: value + for key, value in retry_locator_info.items() + if key not in ["testsuite"] + } + fixed_locators_with_info.append( + { + "index": int(index), + "fixed_locator": retry_selector, + "additional_info": retry_locator_info, + } + ) # for element in self.browser.get_elements(retry_selector): # fixed_locator_list.append(str(element)) except: @@ -339,130 +402,171 @@ def get_fixed_locator(self, data, result) -> str: index += 1 if len(fixed_locators_with_info) > 50: - fixed_locators_with_info = filter_locator_list_with_fuzz_median(fixed_locators_with_info, failed_locator) + fixed_locators_with_info = filter_locator_list_with_fuzz_median( + fixed_locators_with_info, failed_locator + ) try: BuiltIn().set_log_level(old_log_level) except Exception as e: logger.info(f"Error when setting log level: {e}") - - logger.info(f"Pre-filtered locator candidates with info: {pprint.pformat(fixed_locators_with_info)}", also_console=False) + logger.info( + f"Pre-filtered locator candidates with info: {pprint.pformat(fixed_locators_with_info)}", + also_console=False, + ) messages = [] - messages.append({ - 'role': 'system', - 'content': ( - "You are a xpath and css selector self-healing tool for Robot Framework." - "You will select exactly one fixed_locator from a list of Locator Proposals." - 'Respond only using the following json schema: {"index": "index of locator", "fixed_locator": "locator1"}' - "NO COMMENTS. NO DESCRIPTIONS. NO ADDITIONAL INFORMATION." - ) - - }) - + messages.append( + { + "role": "system", + "content": ( + "You are a xpath and css selector self-healing tool for Robot Framework." + "You will select exactly one fixed_locator from a list of Locator Proposals." + 'Respond only using the following json schema: {"index": "index of locator", "fixed_locator": "locator1"}' + "NO COMMENTS. NO DESCRIPTIONS. NO ADDITIONAL INFORMATION." + ), + } + ) match data.name: - case "Fill Text" | "Type Text" | "Press Keys" | "Fill Secret" | "Type Secret" | "Clear Text": - messages.append({ - 'role': 'user', - 'content': ( + case ( + "Fill Text" + | "Type Text" + | "Press Keys" + | "Fill Secret" + | "Type Secret" + | "Clear Text" + ): + messages.append( + { + "role": "user", + "content": ( f"fixed_locators with `input` or `textarea` elements and that are similar to failed_locator=`{failed_locator}` have a priority." "Always return the unchanged fixed_locator." - ) - }) - - case "Click" | "Click With Options": - messages.append({ - 'role': 'user', - 'content': ( + ), + } + ) + + case "Click" | "Click With Options": + messages.append( + { + "role": "user", + "content": ( f"fixed_locators with tagName `button`,`checkbox`, `a`, `li` or `input` elements and that are similar to failed_locator=`{failed_locator}` have a priority." "Check for `clickable: True` property." "Always return the unchanged fixed_locator." - ) - }) - if self.read_clickable_info: - fixed_locators_with_info = [x for x in fixed_locators_with_info if x['additional_info']['clickable']==True] - - case "Select Options By" | "Deselect Options": - messages.append({ - 'role': 'user', - 'content': ( - f"fixed_locators with `select` elements have a priority" + ), + } + ) + if self.read_clickable_info: + fixed_locators_with_info = [ + x + for x in fixed_locators_with_info + if x["additional_info"]["clickable"] == True + ] + + case "Select Options By" | "Deselect Options": + messages.append( + { + "role": "user", + "content": ( + "fixed_locators with `select` elements have a priority" "Always return the unchanged fixed_locator." - ) - }) - case "Check Checkbox" | "Uncheck Checkbox": - messages.append({ - 'role': 'user', - 'content': ( + ), + } + ) + case "Check Checkbox" | "Uncheck Checkbox": + messages.append( + { + "role": "user", + "content": ( f"fixed_locators with `checkbox`, `button` and `input` elements and that are similar to failed_locator=`{failed_locator}` have a priority." "Always return the unchanged fixed_locator." - ) - }) + ), + } + ) - case "Get Text": - messages.append({ - 'role': 'user', - 'content': ( + case "Get Text": + messages.append( + { + "role": "user", + "content": ( f"fixed_locators with `label` or `span` elements and that are similar to failed_locator=`{failed_locator}` have a priority" "Always return the unchanged fixed_locator." - ) - }) + ), + } + ) + messages.append( + { + "role": "user", + "content": ( + f"Broken Locator : `{failed_locator}`\n" + f"Keyword : `{data.name}`\n" + f"** Locators Proposals **\n" + f"```{json.dumps(fixed_locators_with_info)}```\n" + f"Analyse the `additional_info` for each `fixed_locator` to select the `fixed_locator` which most likely matches to failed_locator=`{failed_locator}`.\n" + f"Priority for matching: innerText > previousSibling.innerText > nextSibling.innerText > parentElement.innerText\n" + f"Also consider fixed_locators, where the items from additional_info only match partially. " + ), + } + ) + try: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + temperature=0.1, + response_format={"type": "json_object"}, + ) + except Exception: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + response_format={"type": "json_object"}, + ) + solution_text = response["choices"][0]["message"]["content"] - messages.append({ - 'role': 'user', - 'content': ( - f"Broken Locator : `{failed_locator}`\n" - f"Keyword : `{data.name}`\n" - f"** Locators Proposals **\n" - f"```{json.dumps(fixed_locators_with_info)}```\n" - f"Analyse the `additional_info` for each `fixed_locator` to select the `fixed_locator` which most likely matches to failed_locator=`{failed_locator}`.\n" - f"Priority for matching: innerText > previousSibling.innerText > nextSibling.innerText > parentElement.innerText\n" - f"Also consider fixed_locators, where the items from additional_info only match partially. " - ) - }) - - response = litellm.completion( - model = LLM_TEXT_MODEL, - messages = messages, - temperature = 0.1, - # top_k = 10, - response_format = { "type": "json_object" } + logger.info( + f"2nd LLM response with sorted candidates: {solution_text}", + also_console=False, ) - solution_text = response['choices'][0]['message']['content'] - - logger.info(f"2nd LLM response with sorted candidates: {solution_text}", also_console=False) sorted_retry_locators = list(extract_json_objects(solution_text))[0] if isinstance(sorted_retry_locators, dict): - if 'index' in sorted_retry_locators: + if "index" in sorted_retry_locators: try: - retry_locator = next(item["fixed_locator"] for item in fixed_locators_with_info if int(item['index']) == int(sorted_retry_locators['index'])) + retry_locator = next( + item["fixed_locator"] + for item in fixed_locators_with_info + if int(item["index"]) == int(sorted_retry_locators["index"]) + ) except: retry_locator = sorted_retry_locators["fixed_locator"] return retry_locator - elif 'fixed_locators' in sorted_retry_locators: + elif "fixed_locators" in sorted_retry_locators: retry_locator = sorted_retry_locators["fixed_locators"] - elif 'fixed_locator' in sorted_retry_locators: + elif "fixed_locator" in sorted_retry_locators: retry_locator = sorted_retry_locators["fixed_locator"] - + if isinstance(retry_locator, list): retry_locator = retry_locator[0] if isinstance(retry_locator, str): - return retry_locator - + return retry_locator + return None - def rerun_keyword(self, data, fixed_locator = None) -> str: + def rerun_keyword(self, data, fixed_locator=None) -> str: if fixed_locator: data.args = list(data.args) data.args[0] = fixed_locator try: - logger.info(f"Re-trying Keyword '{data.name}' with arguments '{data.args}'.", also_console=True) + logger.info( + f"Re-trying Keyword '{data.name}' with arguments '{data.args}'.", + also_console=True, + ) return_value = BuiltIn().run_keyword(data.name, *data.args) BuiltIn().run_keyword("Take Screenshot") return return_value @@ -474,19 +578,38 @@ def get_element_count_for_locator(self, locator) -> int: return self.browser.get_element_count(locator) def has_locator(self, result) -> bool: - if ("PageContent" in result.tags): + if "PageContent" in result.tags: return True - else: + else: return False - def get_locator_info(self, locator, read_clickable_info = False): + def get_locator_info(self, locator, read_clickable_info=False): locator_info = {} locator_info["locator"] = locator - property_list = ["tagName", "childElementCount", "parentElement.tagName", "previousSibling.tagName", "nextSibling.tagName"] + property_list = [ + "tagName", + "childElementCount", + "parentElement.tagName", + "previousSibling.tagName", + "nextSibling.tagName", + ] inner_text_property = "innerText" - additional_text_property_list = [ "parentElement.innerText", "previousSibling.innerText", "nextSibling.innerText"] + additional_text_property_list = [ + "parentElement.innerText", + "previousSibling.innerText", + "nextSibling.innerText", + ] # property_list = ["tagName", "innerText", "childElementCount", "parentElement.tagName", "parentElement.innerText"] - allowed_attributes = ['id', 'class', 'value', 'name', 'type', 'placeholder', 'role', 'innerText'] + allowed_attributes = [ + "id", + "class", + "value", + "name", + "type", + "placeholder", + "role", + "innerText", + ] attribute_list = self.browser.get_attribute_names(locator) for attribute in allowed_attributes: if attribute in attribute_list: @@ -496,29 +619,32 @@ def get_locator_info(self, locator, read_clickable_info = False): for property in property_list: value = self.get_property_value(locator, property) if value: - locator_info[property]=value + locator_info[property] = value inner_text = self.get_property_value(locator, inner_text_property) if inner_text: - locator_info["innerText"]=inner_text + locator_info["innerText"] = inner_text else: for property in additional_text_property_list: value = self.get_property_value(locator, property) if value: - locator_info[property]=value + locator_info[property] = value testsuite = BuiltIn().get_variable_value("${SUITE NAME}") locator_info["testsuite"] = testsuite if read_clickable_info: - - clickable_tags = ['BUTTON', 'A', 'INPUT', 'SELECT'] + clickable_tags = ["BUTTON", "A", "INPUT", "SELECT"] if locator_info["tagName"] in clickable_tags: - locator_info["clickable"]= True + locator_info["clickable"] = True else: - if self.does_cursor_pointer_style_exist(locator) or self.does_checked_property_exist(locator) or self.does_clickable_control_property_exist(locator) or self.does_value_property_exist(locator): - locator_info["clickable"]= True + if ( + self.does_cursor_pointer_style_exist(locator) + or self.does_checked_property_exist(locator) + or self.does_clickable_control_property_exist(locator) + or self.does_value_property_exist(locator) + ): + locator_info["clickable"] = True else: - locator_info["clickable"]= False - + locator_info["clickable"] = False return locator_info @@ -528,27 +654,48 @@ def does_cursor_pointer_style_exist(self, locator): return cursor_style == "pointer" except: return False - + def does_value_property_exist(self, locator): try: - return self.browser.evaluate_javascript(locator, f"(elem) => elem.{'value'}") == 'on' or self.browser.evaluate_javascript(locator, f"(elem) => elem.{'value'}") == 'off' + return ( + self.browser.evaluate_javascript(locator, f"(elem) => elem.{'value'}") + == "on" + or self.browser.evaluate_javascript( + locator, f"(elem) => elem.{'value'}" + ) + == "off" + ) except: return False def does_checked_property_exist(self, locator): try: - return self.browser.evaluate_javascript(locator, f"(elem) => elem.{'checked'}") != '' + return ( + self.browser.evaluate_javascript(locator, f"(elem) => elem.{'checked'}") + != "" + ) except: return False def does_clickable_control_property_exist(self, locator): try: - tag = self.browser.evaluate_javascript(locator, f"(elem) => elem.{'control.tagName'}") - if tag == 'BUTTON' or tag == 'A': + tag = self.browser.evaluate_javascript( + locator, f"(elem) => elem.{'control.tagName'}" + ) + if tag == "BUTTON" or tag == "A": return True - elif tag == 'INPUT': - type = self.browser.evaluate_javascript(locator, f"(elem) => elem.{'control.type'}") - if type == "button" or type == "radio" or type == "checkbox" or type == "search" or type == "reset" or type == "submit": + elif tag == "INPUT": + type = self.browser.evaluate_javascript( + locator, f"(elem) => elem.{'control.type'}" + ) + if ( + type == "button" + or type == "radio" + or type == "checkbox" + or type == "search" + or type == "reset" + or type == "submit" + ): return True return False except: @@ -556,18 +703,19 @@ def does_clickable_control_property_exist(self, locator): def get_property_value(self, locator, property): try: - return self.browser.evaluate_javascript(locator, f"(elem) => elem.{property}") + return self.browser.evaluate_javascript( + locator, f"(elem) => elem.{property}" + ) except: return None - + def get_attribute_value(self, locator, attribute): try: return self.browser.get_attribute(locator, attribute) except: return None - - def get_locator_proposals_from_llm(self, data, result, source): + def get_locator_proposals_from_llm(self, data, result, source): failed_locator = BuiltIn().replace_variables(str(result.args[0])) schema = { @@ -578,11 +726,9 @@ def get_locator_proposals_from_llm(self, data, result, source): retry_count = 0 error_message = result.message - - messages = [] - - SYS_PROMPT= ( + + SYS_PROMPT = ( "You are a xpath and css selector self-healing tool.\n" "You will provide a fixed_locator for a failed_locator.\n" "The User prompt will contain data for 'error_message' , 'failed_locator', 'keyword' and 'page_source'.\n" @@ -597,28 +743,25 @@ def get_locator_proposals_from_llm(self, data, result, source): 'Example: {"fixed_locators": ["css=input[id=\'my_id\']", "xpath=//*[contains(text(),\'Login\')]", "xpath=//label[contains(text(),\'Speeding\')]/..//input", "xpath=//*[contains(@class, \'submitBtn\')]", "css=button.class1.class2"]}\n' ) + messages.append({"role": "system", "content": SYS_PROMPT}) + + messages.append( + { + "role": "user", + "content": ( + f"error_message: `{error_message}`\n" + f"failed_locator: `{failed_locator}`\n" + f"keyword : `{data.name}`\n" + f"arguments `{data.args}`\n" + f"page_source: ```{source}```" + ), + } + ) - messages.append({ - 'role': 'system', - 'content': SYS_PROMPT - }) - - messages.append({ - 'role': 'user', - 'content': ( - f"error_message: `{error_message}`\n" - f"failed_locator: `{failed_locator}`\n" - f"keyword : `{data.name}`\n" - f"arguments `{data.args}`\n" - f"page_source: ```{source}```" - ) - }) - - while not locator_has_been_fixed: retry_count += 1 if retry_count > 3: - break + break llm_max_retries = 3 llm_attempts = 0 @@ -626,21 +769,26 @@ def get_locator_proposals_from_llm(self, data, result, source): while llm_attempts < llm_max_retries and not llm_success: try: - response = litellm.completion( - model = LLM_TEXT_MODEL, - messages = messages, - temperature = 0.1, - # top_k = 10, - response_format = { "type": "json_object" } + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + temperature=0.1, + response_format={"type": "json_object"}, ) - + except Exception: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + response_format={"type": "json_object"}, + ) + llm_success = True except Exception as e: llm_attempts += 1 print(f"Attempt {llm_attempts} failed: {e}") - solution_text = response['choices'][0]['message']['content'] - + solution_text = response["choices"][0]["message"]["content"] + # messages.append({ # "role": "assistant", # "content": solution_text @@ -650,7 +798,10 @@ def get_locator_proposals_from_llm(self, data, result, source): # f.write(solution_text) # f.close() - logger.info(f"1st LLM response: {pprint.pformat(solution_text)}\n", also_console=False) + logger.info( + f"1st LLM response: {pprint.pformat(solution_text)}\n", + also_console=False, + ) try: fixed_locator_dict = list(extract_json_objects(solution_text))[0] @@ -659,7 +810,7 @@ def get_locator_proposals_from_llm(self, data, result, source): # Attempt to fix common JSON formatting issues solution_text = solution_text.replace("}}", "}") solution_text = solution_text.replace("{{", "{") - pattern = r'```(.*?)```' + pattern = r"```(.*?)```" match = re.search(pattern, solution_text, re.DOTALL) if match: solution_text = match.group(1) @@ -673,45 +824,62 @@ def get_locator_proposals_from_llm(self, data, result, source): logger.debug(f"Unexpected error: {e}") locator_has_been_fixed = False continue - + try: - fixed_locator_list = fixed_locator_dict['fixed_locators'] + fixed_locator_list = fixed_locator_dict["fixed_locators"] return fixed_locator_list except: locator_has_been_fixed = False continue - def get_locator_proposals_from_parser(self, data, result, source): - soup = BeautifulSoup(source, 'html.parser') + soup = BeautifulSoup(source, "html.parser") locators = [] match data.name: - case "Fill Text" | "Type Text" | "Press Keys" | "Fill Secret" | "Type Secret" | "Clear Text": - element_types = ['textarea', 'input' ] - elements = soup.find_all(element_types) - case "Click" | "Click With Options": - element_types = ['a', 'button', 'checkbox', 'link', 'input', 'label', 'li', has_direct_text] - elements = soup.find_all(element_types) - case "Select Options By" | "Deselect Options": - element_types = ['select'] - elements = soup.find_all(element_types) - case "Check Checkbox" | "Uncheck Checkbox": - element_types = ['input', 'button', 'checkbox'] - elements = soup.find_all(element_types) - case "Get Text": - element_types = ['label', 'div', 'span', has_direct_text] - elements = soup.find_all(element_types) - + case ( + "Fill Text" + | "Type Text" + | "Press Keys" + | "Fill Secret" + | "Type Secret" + | "Clear Text" + ): + element_types = ["textarea", "input"] + elements = soup.find_all(element_types) + case "Click" | "Click With Options": + element_types = [ + "a", + "button", + "checkbox", + "link", + "input", + "label", + "li", + has_direct_text, + ] + elements = soup.find_all(element_types) + case "Select Options By" | "Deselect Options": + element_types = ["select"] + elements = soup.find_all(element_types) + case "Check Checkbox" | "Uncheck Checkbox": + element_types = ["input", "button", "checkbox"] + elements = soup.find_all(element_types) + case "Get Text": + element_types = ["label", "div", "span", has_direct_text] + elements = soup.find_all(element_types) if PARALLEL: - # *** Parallel Processing *** + # *** Parallel Processing *** with ProcessPoolExecutor() as executor: # Submit tasks to the executor - futures = [executor.submit(get_locators_for_element_type, item, str(soup)) for item in element_types] - + futures = [ + executor.submit(get_locators_for_element_type, item, str(soup)) + for item in element_types + ] + # Collect the results as they complete for future in as_completed(futures): if future.result(): @@ -719,13 +887,23 @@ def get_locator_proposals_from_parser(self, data, result, source): else: # Filter elements to include only leaves or lowest elements of their type - filtered_elements = [elem for elem in elements if ((is_leaf_or_lowest(elem) or has_direct_text(elem)) and (not has_parent_dialog_without_open(elem)) and (not has_child_dialog_without_open(elem)) and (not is_headline(elem)) and (not is_div_in_li(elem)) and (not is_p(elem)))] + filtered_elements = [ + elem + for elem in elements + if ( + (is_leaf_or_lowest(elem) or has_direct_text(elem)) + and (not has_parent_dialog_without_open(elem)) + and (not has_child_dialog_without_open(elem)) + and (not is_headline(elem)) + and (not is_div_in_li(elem)) + and (not is_p(elem)) + ) + ] - # Generate and display unique selectors for elem in filtered_elements: with open("locator_stats.csv", "a") as f: - f.write(str(elem).replace('\n', '') + ";") + f.write(str(elem).replace("\n", "") + ";") start = time.time() locator = get_locator(elem, soup) if locator: @@ -739,72 +917,81 @@ def get_locator_proposals_from_parser(self, data, result, source): with open("locator_stats.csv", "a") as f: f.write(f"No Locator found;{total}\n") - - - - return locators def get_locator_with_prefix(locator): - # Search for xpath= or css= in the solution_text + # Search for xpath= or css= in the solution_text # output_dir = BuiltIn().get_variable_value('${OUTPUT_DIR}') # with open(f"{output_dir}/locator_proposals.csv", 'a') as f: # f.write(f"{locator}") if locator: - if locator.startswith('xpath'): + if locator.startswith("xpath"): try: locator_with_prefix = xpath_to_browser(locator) - except: - if 'text()' not in locator: - new_locator = re.sub('xpath.', '', locator) + except: + if "text()" not in locator: + new_locator = re.sub("xpath.", "", locator) try: new_locator = cssify(new_locator) locator_with_prefix = "css=" + new_locator except: locator_with_prefix = "xpath=" + new_locator else: - new_locator = re.sub('xpath.', '', locator) + new_locator = re.sub("xpath.", "", locator) locator_with_prefix = "xpath=" + new_locator - elif locator.startswith('/'): + elif locator.startswith("/"): try: locator_with_prefix = xpath_to_browser(locator) - except: - if 'text()' not in locator: - new_locator = re.sub('xpath.', '', locator) + except: + if "text()" not in locator: + new_locator = re.sub("xpath.", "", locator) try: new_locator = cssify(new_locator) locator_with_prefix = "css=" + new_locator except: locator_with_prefix = "xpath=" + new_locator else: - new_locator = re.sub('xpath.', '', locator) + new_locator = re.sub("xpath.", "", locator) locator_with_prefix = "xpath=" + new_locator - elif locator.startswith('css'): + elif locator.startswith("css"): # Remove css= from the string and store it as a variable - new_locator = re.sub('css.', '', locator) - locator_with_prefix = "css=" + new_locator + new_locator = re.sub("css.", "", locator) + locator_with_prefix = "css=" + new_locator else: locator_with_prefix = locator locator_with_prefix = locator_with_prefix.replace(":contains", ":has-text") - locator_with_prefix = locator_with_prefix.replace(":-soup-contains-own", ":text") - locator_with_prefix = locator_with_prefix.replace(":-soup-contains", ":has-text") - + locator_with_prefix = locator_with_prefix.replace( + ":-soup-contains-own", ":text" + ) + locator_with_prefix = locator_with_prefix.replace( + ":-soup-contains", ":has-text" + ) + # with open(f"{output_dir}/locator_proposals.csv", 'a') as f: # f.write(f"{locator_with_prefix}") - + return locator_with_prefix else: return None + def get_locators_for_element_type(element_type, source): - soup = BeautifulSoup( - source, - 'html.parser' - ) + soup = BeautifulSoup(source, "html.parser") elements = soup.find_all(element_type) - filtered_elements = [elem for elem in elements if ((is_leaf_or_lowest(elem) or has_direct_text(elem)) and (not has_parent_dialog_without_open(elem)) and (not has_child_dialog_without_open(elem)) and (not is_headline(elem)) and (not is_div_in_li(elem)) and (not is_p(elem)))] + filtered_elements = [ + elem + for elem in elements + if ( + (is_leaf_or_lowest(elem) or has_direct_text(elem)) + and (not has_parent_dialog_without_open(elem)) + and (not has_child_dialog_without_open(elem)) + and (not is_headline(elem)) + and (not is_div_in_li(elem)) + and (not is_p(elem)) + ) + ] locators = [] # Generate and display unique selectors @@ -814,6 +1001,7 @@ def get_locators_for_element_type(element_type, source): locators.append(locator) return locators + def get_locator(elem, soup): selector = generate_unique_css_selector(elem, soup) if selector: @@ -823,4 +1011,3 @@ def get_locator(elem, soup): # if selector: # return "xpath=" + selector return None - diff --git a/src/SelfHealing/llm_client.py b/src/SelfHealing/llm_client.py index cd74e53..8fec55b 100644 --- a/src/SelfHealing/llm_client.py +++ b/src/SelfHealing/llm_client.py @@ -1,19 +1,41 @@ -import httpx +import logging import os -from dotenv import load_dotenv -import base64 +import warnings + import litellm -from litellm import completion +from dotenv import find_dotenv, load_dotenv + +# Configure logging +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__) +# Suppress LiteLLM INFO messages +logging.getLogger("LiteLLM").setLevel(logging.WARNING) +# Suppress httpx INFO messages +logging.getLogger("httpx").setLevel(logging.WARNING) + +litellm.suppress_debug_info = True + +# Suppress specific Pydantic UserWarning from LiteLLM, which is noisy as of v1.72.6 +# See: https://github.com/BerriAI/litellm/issues/11759 +warnings.filterwarnings( + "ignore", + message="Pydantic serializer warnings.*", + category=UserWarning, +) -load_dotenv() +env_file = find_dotenv() +if env_file: + load_dotenv(env_file, override=True) -LLM_API_KEY = os.environ.get('LLM_API_KEY', None) -LLM_API_BASE = os.environ.get('LLM_API_BASE', None) -LLM_TEXT_MODEL = os.environ.get('LLM_TEXT_MODEL', "ollama_chat/llama3.1") -LLM_LOCATOR_MODEL = os.environ.get('LLM_LOCATOR_MODEL', "ollama_chat/llama3.1") -LLM_VISION_MODEL = os.environ.get('LLM_VISION_MODEL', "ollama_chat/llama3.2-vision") +LLM_API_KEY = os.environ.get("LLM_API_KEY", None) +LLM_API_BASE = os.environ.get("LLM_API_BASE", None) +LLM_TEXT_MODEL = os.environ.get("LLM_TEXT_MODEL", "ollama_chat/llama3.1") +LLM_LOCATOR_MODEL = os.environ.get("LLM_LOCATOR_MODEL", "ollama_chat/llama3.1") +LLM_VISION_MODEL = os.environ.get("LLM_VISION_MODEL", "ollama_chat/llama3.2-vision") if LLM_API_KEY: litellm.api_key = LLM_API_KEY if LLM_API_BASE: - litellm.api_base = LLM_API_BASE \ No newline at end of file + litellm.api_base = LLM_API_BASE diff --git a/src/SelfHealing/visual_healing.py b/src/SelfHealing/visual_healing.py index e866241..57e5780 100644 --- a/src/SelfHealing/visual_healing.py +++ b/src/SelfHealing/visual_healing.py @@ -1,17 +1,22 @@ -import cv2 -from .llm_client import LLM_TEXT_MODEL, LLM_VISION_MODEL, completion -from io import BytesIO import base64 import json +from io import BytesIO + +import cv2 from robot.api import logger from robot.libraries.BuiltIn import BuiltIn +from litellm import completion + +from .llm_client import LLM_TEXT_MODEL, LLM_VISION_MODEL from .utils import extract_json_objects + try: import pyautogui except: _has_pyautogui = False else: - _has_pyautogui= True + _has_pyautogui = True + class VisualHealer: def __init__(self, instance=None, **kwargs): @@ -21,97 +26,86 @@ def get_screenshot_as_base64(self, image_path=None): if image_path: image = cv2.imread(image_path) # Convert to base64 - im_data = cv2.imencode('.png', image)[1] - base64_image = base64.b64encode(im_data).decode('utf-8') + im_data = cv2.imencode(".png", image)[1] + base64_image = base64.b64encode(im_data).decode("utf-8") else: photo = pyautogui.screenshot() output = BytesIO() - photo.save(output, format='PNG') + photo.save(output, format="PNG") im_data = output.getvalue() base64_image = base64.b64encode(im_data).decode() return base64_image def get_image_description(self, image_as_base64): - response = completion( - model = LLM_VISION_MODEL, + model=LLM_VISION_MODEL, messages=[ { "role": "user", "content": [ - { - "type": "text", - "text": "Whats in this image?" - }, - { - "type": "image_url", - "image_url": { - "url": image_as_base64 - } - } - ] + {"type": "text", "text": "Whats in this image?"}, + {"type": "image_url", "image_url": {"url": image_as_base64}}, + ], } ], - temperature=0.1 + temperature=0.1, ) - text = response['choices'][0]['message']['content'] + text = response["choices"][0]["message"]["content"] return text - - def get_error_explanation(self, data, result, image_as_base64): + def get_error_explanation(self, data, result, image_as_base64): schema = { - - "keyword": { - "type": "string", - "description": "The Keyword Name" - }, - "args": { + "keyword": {"type": "string", "description": "The Keyword Name"}, + "args": { "type": "array", "description": "The Keyword Arguments", - "items": { - "type": "string" - }, - } - } + "items": {"type": "string"}, + }, + } messages = [] - prompt_content=( - "You are a tool to verify automated test results from Robot Framework\n" - "You will receive the error_message, keyword, arguments and a screenshot\n" - "You will analyze the screenshot, the error_message and the identified texts and the check if the error_message makes sense.\n" - "The error message will use the format: Text '' should be ''\n" - - f"error_message: {result.message}\n" - f"keyword: {data.name}\n" - f"arguments: {BuiltIn().replace_variables(str(data.args))}\n" - - "You will return all identified texts in the screenshot.\n" - "You will analyze the screenshot, the error_message and the identified texts and the check if the error_message makes sense.\n" - "If the arguments are incorrect, they need to be adjusted\n" - "Consider the , when adjusting the keyword arguments." - f"If adjustment is needed you will add a headline ***Adjustment*** and respond with the updated keyword and arguments using the following schema: {json.dumps(schema)}\n" - "Example: {'keyword': 'Do Something', 'args': ['hello', 'world'] }\n" - "Your answer will be short and clear\n" - ) - - messages.append({ - 'role': 'user', - 'content': prompt_content, - }) - + prompt_content = ( + "You are a tool to verify automated test results from Robot Framework\n" + "You will receive the error_message, keyword, arguments and a screenshot\n" + "You will analyze the screenshot, the error_message and the identified texts and the check if the error_message makes sense.\n" + "The error message will use the format: Text '' should be ''\n" + f"error_message: {result.message}\n" + f"keyword: {data.name}\n" + f"arguments: {BuiltIn().replace_variables(str(data.args))}\n" + "You will return all identified texts in the screenshot.\n" + "You will analyze the screenshot, the error_message and the identified texts and the check if the error_message makes sense.\n" + "If the arguments are incorrect, they need to be adjusted\n" + "Consider the , when adjusting the keyword arguments." + f"If adjustment is needed you will add a headline ***Adjustment*** and respond with the updated keyword and arguments using the following schema: {json.dumps(schema)}\n" + "Example: {'keyword': 'Do Something', 'args': ['hello', 'world'] }\n" + "Your answer will be short and clear\n" + ) - response = completion( - model = LLM_TEXT_MODEL, - messages = messages, - images=[f"data:image/png;base64,{image_as_base64}"], - temperature = 0.5, + messages.append( + { + "role": "user", + "content": prompt_content, + } ) - analysis = response['choices'][0]['message']['content'] + + try: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + temperature=0.1, + response_format={"type": "json_object"}, + ) + except Exception: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + response_format={"type": "json_object"}, + ) + analysis = response["choices"][0]["message"]["content"] logger.info(f"Visual LLM explanation:\n{analysis}\n", also_console=True) - messages = [] SYS_PROMPT = ( @@ -121,74 +115,68 @@ def get_error_explanation(self, data, result, image_as_base64): f"If adjustment is needed you will add a headline ***Adjustment*** and respond with the updated keyword and arguments using the following schema: {json.dumps(schema)}\n" "Example: {'keyword': 'Do Something', 'args': ['hello', 'world'] }\n" "Your answer will be short and clear\n" - ) - messages.append( - { - "role": "system", - "content": SYS_PROMPT - - } ) + messages.append({"role": "system", "content": SYS_PROMPT}) prompt_content = { - 'error_message': result.message, - 'keyword': data.name, - 'arguments': BuiltIn().replace_variables(str(data.args)), - 'analysis': analysis - } - - messages.append({ - 'role': 'user', - 'content': json.dumps(prompt_content), - }) - - messages.append({ - "role": "assistant", - "content": f"Analysis: {analysis}" - }) + "error_message": result.message, + "keyword": data.name, + "arguments": BuiltIn().replace_variables(str(data.args)), + "analysis": analysis, + } - messages.append({ - "role": "user", - "content": "Check if the analysis is consistent and if the recommended adjustment is correct\n" - "Especially check for small differences between the error_message and the updated args/arguments, e.g. spaces, symbols or special characters\n" - f"Add a headline ***Adjustment*** and respond with the updated keyword and arguments using the following schema: {json.dumps(schema)}\n" - }) + messages.append( + { + "role": "user", + "content": json.dumps(prompt_content), + } + ) + messages.append({"role": "assistant", "content": f"Analysis: {analysis}"}) - response = completion( - model = LLM_TEXT_MODEL, - messages = messages, - temperature = 0.01, + messages.append( + { + "role": "user", + "content": "Check if the analysis is consistent and if the recommended adjustment is correct\n" + "Especially check for small differences between the error_message and the updated args/arguments, e.g. spaces, symbols or special characters\n" + f"Add a headline ***Adjustment*** and respond with the updated keyword and arguments using the following schema: {json.dumps(schema)}\n", + } ) - text = response['choices'][0]['message']['content'] + + try: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + temperature=0.01, + response_format={"type": "json_object"}, + ) + except Exception: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + response_format={"type": "json_object"}, + ) + text = response["choices"][0]["message"]["content"] logger.info(f"LLM verification:\n{text}\n", also_console=True) return text - + def verify_error_analysis(self, explanation, data, result): - prompt_content = { - 'error_message': result.message, - 'keyword': data.name, - 'args': BuiltIn().replace_variables(str(data.args)), - 'analysis': explanation - } - - schema = { + "error_message": result.message, + "keyword": data.name, + "args": BuiltIn().replace_variables(str(data.args)), + "analysis": explanation, + } - "keyword": { - "type": "string", - "description": "The Keyword Name" - }, - "args": { + schema = { + "keyword": {"type": "string", "description": "The Keyword Name"}, + "args": { "type": "array", "description": "The Keyword Arguments", - "items": { - "type": "string" - }, - } - } - + "items": {"type": "string"}, + }, + } SYS_PROMPT = ( "You are a tool to verify the analysis of automated test results\n" @@ -197,53 +185,55 @@ def verify_error_analysis(self, explanation, data, result): "Especially check for small differences between the error_message and the updated args/arguments, e.g. spaces, symbols or special characters\n" f"Add a headline ***Adjustment*** and respond using the following json schema: {'keyword': 'Do Something', 'args': ['hello', 'world']}" "Your answer will be short and clear\n" - ) + ) messages = [] - messages.append( - { - "role": "system", - "content": SYS_PROMPT + messages.append({"role": "system", "content": SYS_PROMPT}) - } + messages.append( + { + "role": "user", + "content": json.dumps(prompt_content), + } ) - - messages.append({ - 'role': 'user', - 'content': json.dumps(prompt_content), - }) - response = completion( - model = LLM_TEXT_MODEL, - messages = messages, - temperature = 0.01, - ) - text = response['choices'][0]['message']['content'] + try: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + temperature=0.01, + response_format={"type": "json_object"}, + ) + except Exception: + response = completion( + model=LLM_TEXT_MODEL, + messages=messages, + response_format={"type": "json_object"}, + ) + text = response["choices"][0]["message"]["content"] logger.info(f"LLM verification:\n{text}\n", also_console=True) return text def is_application_still_loading(self, data, result, image_as_base64): - SYS_PROMPT = ( "Your are a test tool that analyses screeshots and checks if the application is currently loading or if the application is ready\n" "You will receive the error_message, keyword, arguments and a screenshot\n" "Analyse the screenshit\n" "If the application is still loading, respond with 'True'. If the application is ready, respond with 'False'\n" - 'Respond only in valid json that looks like this: ```{"result": "True/False", "reason": "An explanation of your decision" "screenshot_description": "A short description of the screenshot"}```\n' + 'Respond only in valid json that looks like this: ```{"result": "True/False", "reason": "An explanation of your decision" "screenshot_description": "A short description of the screenshot"}```\n' # 'Example: {"result": true, "reason": "A spinner is shown, which indicates that the screen is still loading"}\n' # 'Example: {"result": true, "reason": "Labels for buttons are missing, which indicates that the screen is still loading"}\n' # 'Example: {"result": false, "reason": "All elements are loaded and visible"}\n' - ) + ) messages = [] - + prompt_content = { - 'error_message': result.message, - 'keyword': data.name, - 'arguments': BuiltIn().replace_variables(str(data.args)) - } + "error_message": result.message, + "keyword": data.name, + "arguments": BuiltIn().replace_variables(str(data.args)), + } - # messages.append( # { # "role": "system", @@ -252,105 +242,73 @@ def is_application_still_loading(self, data, result, image_as_base64): # } # ) - messages.append( - { - 'role': 'user', - 'content': ( + { + "role": "user", + "content": ( "You are a test tool that analyses screenshots and check if the application is currently loading or if the application is ready\n" - f"error_message: {result.message}\n" f"keyword: {data.name}\n" f"arguments: {BuiltIn().replace_variables(str(data.args))}\n" - "If the application is still loading, respond with 'True'. If the application is ready, respond with 'False'\n" "If the issue is due to an error in the keyword or argument, respond with 'False'\n" - 'Respond only in valid json that looks like this: ```{"result": "True/False", "reason": "An explanation of your decision", "image_description": "A short description of the screenshot"}```\n' - - ) - }) + 'Respond only in valid json that looks like this: ```{"result": "True/False", "reason": "An explanation of your decision", "image_description": "A short description of the screenshot"}```\n' + ), + } + ) - messages.append( - { - 'role': 'user', - 'content': '', - 'images': [image_as_base64] - }) - - messages.append( - { - "role": "assistant", - "content": "```json" + messages.append({"role": "user", "content": "", "images": [image_as_base64]}) - } - ) + messages.append({"role": "assistant", "content": "```json"}) - prompt_content=( - "You are a test tool that analyses screenshots and check if the application is currently loading or if the application is ready\n" - - f"error_message: {result.message}\n" - f"keyword: {data.name}\n" - f"arguments: {BuiltIn().replace_variables(str(data.args))}\n" - f"screenshot: [img-0]\n" - - "If the application is still loading, respond with 'True'. If the application is ready, respond with 'False'\n" - "If the issue is due to an error in the keyword or argument, respond with 'False'\n" - 'Respond only in valid json that looks like this: ```{"result": "True/False", "reason": "An explanation of your decision", "image_description": "A short description of the screenshot"}```\n' - - ) + prompt_content = ( + "You are a test tool that analyses screenshots and check if the application is currently loading or if the application is ready\n" + f"error_message: {result.message}\n" + f"keyword: {data.name}\n" + f"arguments: {BuiltIn().replace_variables(str(data.args))}\n" + f"screenshot: [img-0]\n" + "If the application is still loading, respond with 'True'. If the application is ready, respond with 'False'\n" + "If the issue is due to an error in the keyword or argument, respond with 'False'\n" + 'Respond only in valid json that looks like this: ```{"result": "True/False", "reason": "An explanation of your decision", "image_description": "A short description of the screenshot"}```\n' + ) response = completion( - model = LLM_VISION_MODEL, + model=LLM_VISION_MODEL, messages=[ { "role": "user", "content": [ - { - "type": "text", - "text": prompt_content - }, - { - "type": "image_url", - "image_url": { - "url": image_as_base64 - } - } - ] + {"type": "text", "text": prompt_content}, + {"type": "image_url", "image_url": {"url": image_as_base64}}, + ], } ], - temperature=0.1 + temperature=0.1, ) - analysis = response['choices'][0]['message']['content'] + analysis = response["choices"][0]["message"]["content"] logger.info(f"Visual LLM explanation:\n{analysis}\n", also_console=True) return list(extract_json_objects(analysis))[0] - - def is_modal_dialog_open(self,image_as_base64): + def is_modal_dialog_open(self, image_as_base64): messages = [] - prompt_content=( - 'Check if a popup dialog or a notification dialog is visible in the screenshot\n' - 'Respond with with {"result": true} (if a dialog is open) or {"result": false} (if no dialog is open)\n' - 'Only respond with valid json\n' - ) - - messages.append({ - "role": "user", - "content": prompt_content - }) - + prompt_content = ( + "Check if a popup dialog or a notification dialog is visible in the screenshot\n" + 'Respond with with {"result": true} (if a dialog is open) or {"result": false} (if no dialog is open)\n' + "Only respond with valid json\n" + ) + messages.append({"role": "user", "content": prompt_content}) response = completion( - model = LLM_VISION_MODEL, - messages = messages, + model=LLM_VISION_MODEL, + messages=messages, images=[f"data:image/png;base64,{image_as_base64}"], - temperature = 0.5, - ) - analysis = response['choices'][0]['message']['content'] + temperature=0.5, + ) + analysis = response["choices"][0]["message"]["content"] logger.info(f"Visual LLM explanation:\n{analysis}\n", also_console=True) - return list(extract_json_objects(analysis))[0]['result'] - \ No newline at end of file + return list(extract_json_objects(analysis))[0]["result"]