From 53227825caec2e621aa043707993462dd0623a9f Mon Sep 17 00:00:00 2001 From: naisanzaa Date: Sat, 27 Jul 2024 02:01:02 -0700 Subject: [PATCH] selenium: overhaul logging much better now --- .../integrations/seleniumWrapper/browser.py | 592 +++++++++++------- .../integrations/seleniumWrapper/config.py | 14 +- .../seleniumWrapper/webdriver_chrome.py | 69 +- 3 files changed, 405 insertions(+), 270 deletions(-) diff --git a/automon/integrations/seleniumWrapper/browser.py b/automon/integrations/seleniumWrapper/browser.py index 4bd4feeb..a3393965 100644 --- a/automon/integrations/seleniumWrapper/browser.py +++ b/automon/integrations/seleniumWrapper/browser.py @@ -62,14 +62,16 @@ def config(self): return self._config async def cookie_file_to_dict(self, file: str = 'cookies.json') -> list: - logger.debug(dict( - cookie_file_to_dict=file - )) + """load cookie file as dict""" + logger.debug(f'cookie_file_to_dict :: {file=}') with open(file, 'r') as file: + logger.debug(f'cookie_file_to_dict :: read :: {file=}') + logger.info(f'cookie_file_to_dict :: read :: done') return json.loads(file.read()) @property def _current_url(self): + """get browser current url""" try: return self.webdriver.current_url except Exception as error: @@ -77,11 +79,15 @@ def _current_url(self): @property def webdriver(self): + """get webdriver""" return self.config.webdriver async def get_log(self, log_type: str) -> list: """Get logs for log type""" - return self.webdriver.get_log(log_type) + logger.debug(f'get_log :: {log_type=}') + get_log = self.webdriver.get_log(log_type) + logger.info(f'get_log :: done') + return get_log async def get_logs(self) -> dict: """Get all logs @@ -89,41 +95,64 @@ async def get_logs(self) -> dict: you can only run this once afterwards the logs are cleared from the webdriver """ + logger.debug(f'get_logs') + logger.debug(f'get_logs :: {len(self.webdriver.log_types)} types found') + for log_type in self.webdriver.log_types: - self.logs.update( - { - log_type: self.webdriver.get_log(log_type) - } - ) + logger.debug(f'get_logs :: {log_type}') + self.logs.update({ + log_type: self.webdriver.get_log(log_type) + }) + logger.debug(f'get_logs :: {log_type} :: {len(self.logs[log_type])} logs') + logger.info(f'get_logs :: {log_type} :: done') + + logger.debug(f'get_logs :: {len(self.logs)} logs') + logger.info(f'get_logs :: done') return self.logs async def get_log_browser(self) -> list: """Get browser logs""" + logger.debug(f'get_log_browser') logs = await self.get_log('browser') + logger.debug(f'get_log_browser :: {len(logs)} logs') + logger.info(f'get_log_browser :: done') return logs async def get_log_driver(self) -> list: """Get driver logs""" + logger.debug(f'get_log_driver') logs = await self.get_log('driver') + logger.debug(f'get_log_driver :: {len(logs)} logs') + logger.info(f'get_log_driver :: done') return logs async def get_log_performance(self) -> list: """Get performance logs""" + logger.debug(f'get_log_performance') logs = await self.get_log('performance') + logger.debug(f'get_log_performance :: {len(logs)} logs') + logger.info(f'get_log_performance :: done') return logs async def check_page_load_finished(self) -> bool: """Checks for `frameStoppedLoading` string in performance logs""" + logger.debug(f'check_page_load_finished :: checking') + logs = await self.get_log_performance() + logger.debug(f'check_page_load_finished :: checking :: {len(logs)} logs found') check = [] for log_dict in logs: + # logger.debug(f'check_page_load_finished :: checking :: {log_dict}') if 'frameStoppedLoading' in log_dict.get('message'): + logger.debug(f'check_page_load_finished :: checking :: frameStoppedLoading :: found :: {log_dict}') check.append(log_dict) if check: + logger.info(f'check_page_load_finished :: checking :: done') return True + logger.error(f'check_page_load_finished :: checking :: not finished loading') return False @property @@ -132,8 +161,11 @@ def keys(self): return selenium.webdriver.common.keys.Keys async def refresh(self) -> None: - logger.debug(f'refreshing page: {self.current_url=}') - return self.webdriver.refresh() + """refresh the page""" + logger.debug(f'refresh :: {self.current_url=}') + refresh = self.webdriver.refresh() + logger.info(f'refresh :: done') + return refresh @property def url(self): @@ -160,36 +192,51 @@ def window_size(self): def _screenshot_name(self, prefix=None): """Generate a unique filename""" + logger.debug(f'_screenshot_name :: {prefix=}') title = self.webdriver.title url = self.current_url + hostname = urlparse(url).hostname + timestamp = Dates.filename_timestamp() hostname_ = Sanitation.ascii_numeric_only(hostname) title_ = Sanitation.ascii_numeric_only(title) - timestamp = Dates.filename_timestamp() + + logger.debug(f'_screenshot_name :: {url=}') + logger.debug(f'_screenshot_name :: {timestamp=}') + logger.debug(f'_screenshot_name :: {title_=}') + logger.debug(f'_screenshot_name :: {hostname_=}') if prefix: prefix = Sanitation.safe_string(prefix) - return f'{prefix}_{timestamp}_{hostname_}_{title_}.png' + _screenshot_name = f'{prefix}_{timestamp}_{hostname_}_{title_}.png' + logger.info(f'_screenshot_name :: {_screenshot_name=}') + return _screenshot_name - return f'{timestamp}_{hostname_}_{title_}.png' + _screenshot_name = f'{timestamp}_{hostname_}_{title_}.png' + logger.debug(f'_screenshot_name :: {_screenshot_name=}') + + logger.info(f'_screenshot_name :: done') + return _screenshot_name async def action_click( self, element: selenium.webdriver.remote.webelement.WebElement, **kwargs): """perform mouse click""" + logger.debug(f'action_click :: {element=} :: {kwargs=}') try: - logger.debug(dict( - tag_name=element.tag_name, - text=element.text, - accessible_name=element.accessible_name, - aria_role=element.aria_role)) + logger.debug(f'action_click :: tag_name :: {element.tag_name}') + logger.debug(f'action_click :: text :: {element.text}') + logger.debug(f'action_click :: accessible_name :: {element.accessible_name}') + logger.debug(f'action_click :: aria_role :: {element.aria_role}') + element.click(**kwargs) + logger.info(f'action_click :: done') return True except Exception as error: - raise Exception(error) + raise Exception(f'action_click :: failed :: {error=}') async def action_type( self, @@ -198,16 +245,18 @@ async def action_type( """perform keyboard command""" if secret: - logger.debug(dict(send_keys=f'*' * len(f'{key}'))) + logger.debug(f'action_type :: {f"*" * len(f"{key}")}') else: - logger.debug(dict(send_keys=key)) + logger.debug(f'action_type :: {key}') try: - return selenium.webdriver.common.action_chains.ActionChains( + action_type = selenium.webdriver.common.action_chains.ActionChains( self.webdriver).send_keys(key).perform() + logger.info(f'action_type :: done') + return action_type except Exception as error: - raise Exception(error) + raise Exception(f'action_type :: failed :: {error=}') async def action_type_up( self, @@ -216,16 +265,18 @@ async def action_type_up( """release key""" if secret: - logger.debug(dict(send_keys=f'*' * len(f'{key}'))) + logger.debug(f'action_type_up :: {"*" * len(f"{key}")}') else: - logger.debug(dict(send_keys=key)) + logger.debug(f'action_type_up :: {key}') try: - return selenium.webdriver.common.action_chains.ActionChains( + action_type_up = selenium.webdriver.common.action_chains.ActionChains( self.webdriver).key_up(key).perform() + logger.info(f'action_type_up :: done') + return action_type_up except Exception as error: - raise Exception(error) + raise Exception(f'action_type_up :: failed :: {error=}') async def action_type_down( self, @@ -234,18 +285,22 @@ async def action_type_down( """hold key down""" if secret: - logger.debug(dict(send_keys=f'*' * len(f'{key}'))) + logger.debug(f'action_type_down :: {"*" * len(f"{key}")}') else: - logger.debug(dict(send_keys=key)) + logger.debug(f'action_type_down :: {key}') try: - return selenium.webdriver.common.action_chains.ActionChains( + action_type_down = selenium.webdriver.common.action_chains.ActionChains( self.webdriver).key_down(key).perform() + logger.info(f'action_type_down :: done') + return action_type_down except Exception as error: - raise Exception(error) + raise Exception(f'action_type_down :: failed :: {error=}') async def add_cookie(self, cookie_dict: dict) -> bool: + logger.debug(f'add_cookie :: {cookie_dict=}') + result = self.webdriver.add_cookie(cookie_dict=cookie_dict) if result is None: @@ -256,130 +311,163 @@ async def add_cookie(self, cookie_dict: dict) -> bool: expiry=cookie_dict.get('expiry'), name=cookie_dict.get('name'), )) + logger.debug(f'add_cookie :: domain :: {cookie_dict.get('domain')}') + logger.debug(f'add_cookie :: path :: {cookie_dict.get('path')}') + logger.debug(f'add_cookie :: secure :: {cookie_dict.get('secure')}') + logger.debug(f'add_cookie :: expiry :: {cookie_dict.get('expiry')}') + logger.debug(f'add_cookie :: name :: {cookie_dict.get('name')}') + + logger.info(f'add_cookie :: done') return True - logger.error(dict( - add_cookie=cookie_dict - )) - return False + raise Exception(f'add_cookie :: failed :: {cookie_dict=}') async def add_cookie_from_file(self, file: str) -> bool: """add cookies from file""" + logger.debug(f'add_cookie_from_file :: {file=}') + if os.path.exists(file): await self.add_cookies_from_list( await self.cookie_file_to_dict(file=file) ) + logger.info(f'add_cookie_from_file :: done') return True - logger.error(f'{file}') - return False + raise Exception(f'add_cookie_from_file :: failed :: {file=}') async def add_cookies_from_list(self, cookies_list: list) -> bool: + """add cookies from a list of cookies""" + logger.debug(f'add_cookies_from_list :: start') + logger.debug(f'add_cookies_from_list :: {len(cookies_list)} cookie(s) found') + for cookie in cookies_list: + logger.debug(f'add_cookies_from_list :: {cookie=}') await self.add_cookie(cookie_dict=cookie) - logger.debug(dict( - add_cookies_from_list=len(cookies_list) - )) + logger.debug(f'add_cookies_from_list :: {len(cookies_list)} cookie(s) added') + logger.info(f'add_cookies_from_list :: done') return True async def add_cookie_from_current_url(self): - logger.debug(dict( - add_cookie_from_current_url=self.url, - )) + """add cookies from the current url""" + logger.debug(f'add_cookie_from_current_url :: {self.url=}') + logger.info(f'add_cookie_from_current_url :: done') return self.add_cookie_from_url(self.url) async def add_cookie_from_url(self, url: str) -> bool: """add cookies from matching hostname""" + logger.debug(f'add_cookie_from_url :: {url=}') + cookie_file = await self._url_filename(url=url) + logger.debug(f'add_cookie_from_url :: {cookie_file=}') if os.path.exists(cookie_file): - logger.debug(dict( - add_cookie_from_url=url, - cookie_file=cookie_file, - )) - return await self.add_cookie_from_file(file=cookie_file) + logger.debug(f'add_cookie_from_url :: {cookie_file=} file found') + add_cookie_from_url = await self.add_cookie_from_file(file=cookie_file) + logger.info(f'add_cookie_from_url :: done') + return add_cookie_from_url - logger.error(dict( - add_cookie_from_url=url, - cookie_file=cookie_file, - )) + raise Exception(f'add_cookie_from_url :: failed :: {cookie_file=}') async def add_cookie_from_base64(self, base64_str: str) -> bool: + """add cookie from base64 string""" + logger.debug(f'add_cookie_from_base64 :: base64 :: {len(base64_str) / 1024} KB') + if base64_str: add_cookie_from_base64 = json.loads(base64.b64decode(base64_str)) + logger.debug(f'add_cookie_from_base64 :: str :: {len(add_cookie_from_base64) / 1024} KB') await self.add_cookies_from_list(add_cookie_from_base64) - logger.debug(dict( - add_cookie_from_base64=add_cookie_from_base64, - base64_str=base64_str, - )) + + logger.info(f'add_cookie_from_base64 :: done') return True - logger.error(dict( - base64_str=base64_str - )) - return False + raise Exception(f'add_cookie_from_base64 :: failed :: {len(base64_str) / 1024} KB') async def autosave_cookies(self) -> bool: + """auto save cookies for current url""" + logger.debug(f'autosave_cookies') + if self.current_url: + logger.debug(f'autosave_cookies :: {self.current_url=}') + if not self.autosaved: + logger.debug(f'autosave_cookies :: {self.autosaved=}') await self.add_cookie_from_current_url() await self.refresh() self.autosaved = True - return await self.save_cookies_for_current_url() + logger.debug(f'autosave_cookies :: {self.autosaved=}') + + autosave_cookies = await self.save_cookies_for_current_url() + logger.info(f'autosave_cookies :: done') + return autosave_cookies + + logger.debug(f'autosave_cookies :: failed :: no current url :: {self.current_url=}') async def delete_all_cookies(self) -> None: - result = self.webdriver.delete_all_cookies() - logger.debug(dict( - delete_all_cookies='done' - )) - return result + """delete all cookies""" + logger.debug(f'delete_all_cookies') + delete_all_cookies = self.webdriver.delete_all_cookies() + logger.info(f'delete_all_cookies :: done') + return delete_all_cookies async def _url_filename(self, url: str): + """turn url into a filename""" + logger.debug(f'_url_filename :: {url=}') + parsed = await self.urlparse(url) hostname = parsed.hostname cookie_file = f'cookies-{hostname}.json' - logger.debug(dict( - cookie_file=cookie_file - )) + + logger.debug(f'_url_filename :: {hostname=}') + logger.debug(f'_url_filename :: {cookie_file=}') + + logger.info(f'_url_filename :: done') return cookie_file async def get_cookie(self, name: str) -> dict: + """get cookie by name""" + logger.debug(f'get_cookie :: {name=}') get_cookie = self.webdriver.get_cookie(name=name) - logger.debug(dict( - name=name, - get_cookie=get_cookie, - )) + logger.debug(f'get_cookie :: {get_cookie=}') + logger.info(f'get_cookie :: done') return get_cookie async def get_cookies(self) -> [dict]: + """get all cookies""" + logger.debug(f'get_cookies :: ') get_cookies = self.webdriver.get_cookies() - logger.debug(dict( - get_cookies=len(get_cookies) - )) + logger.debug(f'get_cookies :: {len(get_cookies)} total cookie(s)') + logger.info(f'get_cookies :: done') return get_cookies async def get_cookies_base64(self) -> str: + """get cookies as base64""" + logger.debug(f'get_cookies_base64 ::') + get_cookies_base64 = await self.get_cookies() get_cookies_base64 = base64.b64encode( json.dumps(get_cookies_base64).encode() ).decode() + logger.debug(f'get_cookies_base64 :: {len(get_cookies_base64) / 1024} KB') - logger.debug(dict( - get_cookies_base64=get_cookies_base64 - )) + logger.info(f'get_cookies_base64 :: done') return get_cookies_base64 async def get_cookies_json(self) -> json.dumps: + """get cookies as json""" + logger.debug(f'get_cookies_json ::') + get_cookies_json = await self.get_cookies() get_cookies_json = json.dumps(get_cookies_json) + logger.debug(f'get_cookies_json :: {len(get_cookies_json) / 1024} KB') - logger.debug(dict( - get_cookies_json=f'{len(get_cookies_json)} B', - )) + logger.info(f'get_cookies_json :: done') return get_cookies_json async def get_cookies_summary(self) -> dict: + """get cookies summary""" + logger.debug(f'get_cookies_summary ::') + result = await self.get_cookies() summary = {} if result: @@ -389,21 +477,31 @@ async def get_cookies_summary(self) -> dict: name = cookie.get('name') expiry = cookie.get('expiry') + logger.debug(f'get_cookies_summary :: {cookie}') + logger.debug(f'get_cookies_summary :: domain :: {domain}') + logger.debug(f'get_cookies_summary :: name :: {name}') + logger.debug(f'get_cookies_summary :: expiry :: {expiry}') + if domain in summary.keys(): summary[domain].append(cookie) else: summary[domain] = [cookie] - logger.debug(summary) + logger.debug(f'get_cookies_summary :: summary :: {summary}') + logger.info(f'get_cookies_summary ::') return summary async def close(self): """close browser""" - logger.info(f'closing webdriver') + logger.debug(f'webdriver :: close') self.webdriver.close() + logger.info(f'webdriver :: close :: done') @staticmethod async def error_parsing(error) -> tuple: + """parse webdriver error""" + logger.debug(f'error_parsing :: {error=}') + try: error_parsed = f'{error}'.splitlines() error_parsed = [f'{x}'.strip() for x in error_parsed] @@ -412,15 +510,16 @@ async def error_parsing(error) -> tuple: stacktrace = error_parsed[2:] stacktrace = ' '.join(stacktrace) - return message, session, stacktrace + logger.debug(f'error_parsing :: {error_parsed}') + logger.debug(f'error_parsing :: {message}') + logger.debug(f'error_parsing :: {stacktrace}') - except Exception as e: - logger.error(dict( - exception=e, - error=error, - )) + logger.info(f'error_parsing :: done') + return message, session, stacktrace - return error, None, None + except Exception as exception: + logger.error(f'error_parsing :: failed :: {exception=}') + return error, None, None async def find_anything( self, @@ -436,14 +535,8 @@ async def find_anything( find all tags find all matches within meta data """ - logger.debug(dict( - find_anything=self.current_url, - value=value, - by=by, - case_sensitive=case_sensitive, - exact_match=exact_match, - kwargs=kwargs, - )) + logger.debug( + f'find_anything :: {match=} :: {value=} :: {by=} : {case_sensitive=} :: {exact_match=} :: {return_first=} :: {kwargs=}') by_types = [ self.by.TAG_NAME, @@ -494,16 +587,16 @@ async def find_anything( FOUND = element if FOUND and FOUND not in MATCHED: - logger.debug(dict( - MATCH=MATCH, - AGAINST=AGAINST, - attribute=dir_, - element=element, - )) + logger.debug(f'find_anything :: {MATCH=} :: {AGAINST=} :: attribute={dir_} :: {element=}') + logger.info(f'find_anything :: {MATCH=} :: found') MATCHED.append(FOUND) if return_first: + logger.info(f'find_anything :: done') return MATCHED + + logger.debug(f'find_anything :: {len(MATCH)} result(s) found') + logger.info(f'find_anything :: done') return MATCHED async def find_element( @@ -512,12 +605,13 @@ async def find_element( by: selenium.webdriver.common.by.By, **kwargs) -> selenium.webdriver.Chrome.find_element: """find element""" - logger.debug(dict( - find_element=self.current_url, - value=value, - by=by, - )) - return self.webdriver.find_element(value=value, by=by, **kwargs) + logger.debug(f'find_element :: {self.current_url} :: {value=} :: {by=} :: {kwargs=}') + + find_element = self.webdriver.find_element(value=value, by=by, **kwargs) + logger.debug(f'find_element :: {find_element=}') + + logger.info(f'find_element :: done') + return find_element async def find_elements( self, @@ -525,12 +619,13 @@ async def find_elements( by: selenium.webdriver.common.by.By, **kwargs) -> list: """find elements""" - logger.debug(dict( - find_elements=self.current_url, - value=value, - by=by, - )) - return self.webdriver.find_elements(value=value, by=by, **kwargs) + logger.debug(f'find_elements :: {self.current_url} :: {value=} :: {by=} :: {kwargs=}') + + find_elements = self.webdriver.find_elements(value=value, by=by, **kwargs) + logger.debug(f'find_elements :: {len(find_elements)} element(s) found') + + logger.info(f'find_elements :: done') + return find_elements async def find_xpath( self, @@ -538,71 +633,90 @@ async def find_xpath( by: selenium.webdriver.common.by.By = selenium.webdriver.common.by.By.XPATH, **kwargs) -> selenium.webdriver.Chrome.find_element: """find xpath""" - logger.debug(dict( - find_xpath=self.current_url, - value=value, - by=by, - )) - return await self.find_element(value=value, by=by, **kwargs) + logger.debug(f'find_xpath :: {self.current_url} :: {value=} :: {by=} :: {kwargs=}') + + find_xpath = await self.find_element(value=value, by=by, **kwargs) + logger.debug(f'find_xpath :: {find_xpath=}') + + logger.info(f'find_xpath :: done') + return find_xpath async def get(self, url: str, **kwargs) -> bool: """get url""" + logger.debug(f'browser :: get :: {url=} :: {kwargs=}') + if not self.webdriver: - logger.error(f'missing webdriver') - raise Exception(f'missing webdriver') + raise Exception(f'browser :: get :: failed :: missing webdriver') try: if self.webdriver.get(url, **kwargs) is None: - logger.debug(dict( - get=url, - current_url=self.current_url, - kwargs=kwargs - )) + logger.debug(f'browser :: get :: {url=} :: {self.current_url=} :: {kwargs=}') if self.config.cookies_autosave: await self.autosave_cookies() + logger.info(f'browser :: get :: done') return True - except Exception as error: - logger.error(dict( - error=error, - )) + except Exception as exception: + logger.error(f'browser :: get :: failed :: {exception=}') + + logger.error(f'browser :: get :: failed :: {url=}') return False async def get_page(self, *args, **kwargs) -> bool: """alias to get""" + logger.debug(f'get_page :: {args=} :: {kwargs=}') return await self.get(*args, **kwargs) async def get_page_source(self) -> str: """get page source""" - return self.webdriver.page_source + logger.debug(f'get_page_source :: ') + get_page_source = self.webdriver.page_source + logger.debug(f'get_page_source :: {len(get_page_source) / 1024} KB') + logger.info(f'get_page_source :: done') + return get_page_source async def get_page_source_beautifulsoup( self, - markdup: str = None, + markup: str = None, features: str = 'lxml') -> BeautifulSoup: """read page source with beautifulsoup""" - if not markdup: - markdup = await self.get_page_source() - return BeautifulSoup( - markup=markdup, + logger.debug(f'get_page_source_beautifulsoup :: {features=} :: {len(markup) / 1024} KB') + + if not markup: + markup = await self.get_page_source() + + get_page_source_beautifulsoup = BeautifulSoup( + markup=markup, features=features) + logger.debug(f'get_page_source_beautifulsoup :: {len(get_page_source_beautifulsoup)} size') + + logger.info(f'get_page_source_beautifulsoup :: done') + return get_page_source_beautifulsoup async def get_random_user_agent( self, filter: list or str = None, case_sensitive: bool = False) -> str: - return SeleniumUserAgentBuilder().get_random_agent( + """get a random user agent string""" + logger.debug(f'get_random_user_agent :: {filter=} :: {case_sensitive=}') + + get_random_user_agent = SeleniumUserAgentBuilder().get_random_agent( filter=filter, case_sensitive=case_sensitive) + logger.debug(f'get_random_user_agent :: {get_random_user_agent}') + + logger.info(f'get_random_user_agent :: done') + return get_random_user_agent async def get_screenshot_as_base64(self, **kwargs): """screenshot as base64""" - screenshot = self.webdriver.get_screenshot_as_base64(**kwargs) - logger.debug(f'get_screenshot_as_base64 ({round(len(screenshot) / 1024)} KB)') - - return screenshot + logger.debug(f'get_screenshot_as_base64 :: ') + get_screenshot_as_base64 = self.webdriver.get_screenshot_as_base64(**kwargs) + logger.debug(f'get_screenshot_as_base64 :: {round(len(get_screenshot_as_base64) / 1024)} KB') + logger.info(f'get_screenshot_as_base64 :: done') + return get_screenshot_as_base64 async def get_screenshot_as_file( self, @@ -610,97 +724,118 @@ async def get_screenshot_as_file( prefix: str = None, folder: str = None, **kwargs) -> bool: - return await self.save_screenshot( + """alias to save_screenshot""" + logger.debug(f'get_screenshot_as_file :: {filename=} :: {prefix=} :: {folder=} :: {kwargs=}') + + get_screenshot_as_file = await self.save_screenshot( self, filename=filename, prefix=prefix, folder=folder, **kwargs) + logger.info(f'get_screenshot_as_file :: done') + return get_screenshot_as_file + async def get_screenshot_as_png(self, **kwargs): """screenshot as png""" - screenshot = self.webdriver.get_screenshot_as_png(**kwargs) - logger.debug(f'{round(len(screenshot) / 1024)} KB') - - return screenshot + logger.debug(f'get_screenshot_as_png ::') + get_screenshot_as_png = self.webdriver.get_screenshot_as_png(**kwargs) + logger.debug(f'get_screenshot_as_png :: {round(len(get_screenshot_as_png) / 1024)} KB') + logger.info(f'get_screenshot_as_png :: done') + return get_screenshot_as_png async def is_running(self) -> bool: - """browser is running""" + """webdriver is running""" + logger.debug(f'webdriver :: ') + if self.webdriver: - logger.info(f'webdriver is running') + logger.info(f'webdriver :: is running') return True - logger.error(f'webdriver is not running') + logger.error(f'webdriver :: is not running') return False async def load_cookies_for_current_url(self) -> bool: + """load cookies from file for current url""" + logger.debug(f'load_cookies_for_current_url :: ') + filename = await self._url_filename(url=self.url) logger.debug(dict( load_cookies_for_current_url=filename, url=self.url, )) - return await self.add_cookie_from_file(file=filename) + + load_cookies_for_current_url = await self.add_cookie_from_file(file=filename) + logger.info(f'load_cookies_for_current_url :: done') + return load_cookies_for_current_url @property def page_source(self): return self.webdriver.page_source async def urlparse(self, url: str): + """parse url""" + logger.debug(f'urlparse :: {url=}') parsed = urlparse(url=url) - logger.debug(dict( - urlparse=parsed - )) + logger.info(f'urlparse :: {parsed=}') + logger.info(f'urlparse :: done') return parsed async def quit(self) -> bool: - """gracefully quit browser""" - try: - self.webdriver.close() - self.webdriver.quit() - self.webdriver.stop_client() - except Exception as error: - message, session, stacktrace = await self.error_parsing(error) - logger.error(str(dict( - message=message, - session=session, - stacktrace=stacktrace, - ))) - return False + """gracefully quit webdriver""" + logger.debug(f'webdriver :: quit') + + if self.webdriver: + try: + await self.close() + self.webdriver.quit() + self.webdriver.stop_client() + except Exception as error: + message, session, stacktrace = await self.error_parsing(error) + logger.error(f'webdriver :: quit :: failed :: {message=} :: {session=} :: {stacktrace=}') + return False + + logger.info(f'webdriver :: quit :: done') return True async def run(self) -> bool: - """run browser""" + """run webdriver""" + logger.debug(f'webdriver :: run') + try: - return await self.config.run() - except Exception as error: - logger.error(dict( - error=error - )) + run = await self.config.run() + logger.info(f'webdriver :: run :: done') + return run + except Exception as exception: + logger.error(f'webdriver :: run :: failed :: {exception=}') return False async def save_cookies_for_current_url(self) -> bool: + """save cookies for current url""" + logger.debug(f'save_cookies_for_current_url :: ') + filename = await self._url_filename(url=self.url) - logger.debug(dict( - save_cookies_for_current_url=self.url, - filename=filename, - )) - return await self.save_cookies_to_file(file=filename) + save_cookies_for_current_url = await self.save_cookies_to_file(file=filename) + logger.debug(f'save_cookies_for_current_url :: {self.current_url} :: {filename}') + + logger.info(f'save_cookies_for_current_url :: done') + return save_cookies_for_current_url async def save_cookies_to_file(self, file: str) -> bool: + """save cookies to file""" + logger.debug(f'save_cookies_to_file :: {file}') + with open(file, 'w') as cookies: cookies.write( await self.get_cookies_json() ) if os.path.exists(file): - logger.debug(dict( - save_cookies_to_file=os.path.abspath(file), - bytes=os.stat(file).st_size - )) + logger.debug(f'save_cookies_to_file :: {os.path.abspath(file)} :: {os.stat(file).st_size} B') + logger.info(f'save_cookies_to_file :: done') return True - logger.error(dict( - file=file - )) + logger.error(f'save_cookies_to_file :: failed :: {file=}') return False async def save_screenshot( @@ -710,14 +845,18 @@ async def save_screenshot( folder: str = None, **kwargs) -> bool: """save screenshot to file""" + logger.debug(f'save_screenshot :: {filename=} :: {prefix=} :: {folder=} :: {kwargs=}') if not filename: filename = self._screenshot_name(prefix) + logger.debug(f'save_screenshot :: {filename=}') if not folder: path = os.path.abspath(tempfile.gettempdir()) + logger.debug(f'save_screenshot :: {path=}') else: path = os.path.abspath(folder) + logger.debug(f'save_screenshot :: {path=}') if not os.path.exists(path): os.makedirs(path) @@ -725,13 +864,16 @@ async def save_screenshot( save = os.path.join(path, filename) if self.webdriver.save_screenshot(save, **kwargs): - logger.info(f'Saving screenshot to: {save} ({round(os.stat(save).st_size / 1024)} KB)') + logger.debug(f'save_screenshot :: {save} :: {round(os.stat(save).st_size / 1024)} KB') + logger.info(f'save_screenshot :: done') return True + logger.error(f'save_screenshot :: failed') return False async def set_window_size(self, width=1920, height=1080, device_type=None) -> bool: """set browser resolution""" + logger.debug(f'set_window_size :: {width=} :: {height=} :: {device_type=}') try: self.config.webdriver_wrapper.set_window_size( @@ -740,20 +882,17 @@ async def set_window_size(self, width=1920, height=1080, device_type=None) -> bo device_type=device_type) except Exception as error: message, session, stacktrace = await self.error_parsing(error) - logger.error(str(dict( - message=message, - session=session, - stacktrace=stacktrace, - ))) + logger.error(f'set_window_size :: failed :: {message=} :: {session=} :: {stacktrace=}') return False + + logger.info(f'set_window_size :: done') return True async def set_window_position(self, x: int = 0, y: int = 0): """set browser position""" + logger.debug(f'set_window_position :: {x=} :: {y=}') set_window_position = self.webdriver.set_window_position(x, y) - logger.debug(dict( - set_window_position=set_window_position - )) + logger.info(f'set_window_position :: done') return set_window_position async def start(self): @@ -771,6 +910,9 @@ async def wait_for_anything( return_first: bool = False, **kwargs) -> list: """wait for anything""" + logger.debug( + f'wait_for_anything :: {match=} :: {value=} :: {by=} :: {case_sensitive=} :: {exact_match=} :: {timeout=} :: {return_first=} :: {kwargs=}') + timeout_start = time.time() timeout_elapsed = round(abs(timeout_start - time.time()), 1) @@ -792,16 +934,19 @@ async def wait_for_anything( exact_match=exact_match, return_first=return_first, **kwargs) + logger.debug(f'wait_for_anything :: {len(find)} element(s) found') if find: + logger.info(f'wait_for_anything :: done') return find except Exception as error: - logger.error(error) + logger.error(f'wait_for_anything :: failed :: {error=}') timeout_elapsed = round(abs(timeout_start - time.time()), 1) + logger.debug(f'wait_for_anything :: {timeout_elapsed} seconds elapsed') - raise ElementNotFoundException(value) + raise ElementNotFoundException(f'wait_for_anything :: failed :: {value=}') async def wait_for_element( self, @@ -810,33 +955,34 @@ async def wait_for_element( timeout: int = 30, **kwargs) -> selenium.webdriver.Chrome.find_element: """wait for an element""" + logger.debug(f'wait_for_element :: {value=} :: {by=} :: {timeout=} :: {kwargs=}') + timeout_start = time.time() timeout_elapsed = round(abs(timeout_start - time.time()), 1) while timeout_elapsed < timeout: - logger.debug(str(dict( - timeout=f'{timeout_elapsed}/{timeout}', - by=by, - current_url=self.current_url, - value=value, - ))) + logger.debug( + f'wait_for_element :: {f"{timeout_elapsed}/{timeout}"} :: {by=} :: {self.current_url} :: {value=}') try: find = await self.find_element( value=value, by=by, **kwargs) + logger.debug(f'wait_for_element :: element found') if find: + logger.info(f'wait_for_element :: done') return find except Exception as error: - logger.error(error) + logger.error(f'wait_for_element :: failed :: {error=}') timeout_elapsed = round(abs(timeout_start - time.time()), 1) + logger.debug(f'wait_for_element :: {timeout_elapsed} seconds elapsed') - raise ElementNotFoundException(value) + raise ElementNotFoundException(f'wait_for_element :: failed :: {value=}') async def wait_for_elements( self, @@ -845,33 +991,34 @@ async def wait_for_elements( timeout: int = 30, **kwargs) -> list: """wait for all matching elements""" + logger.debug(f'wait_for_elements :: {value=} :: {by=} :: {timeout=} :: {kwargs=}') + timeout_start = time.time() timeout_elapsed = round(abs(timeout_start - time.time()), 1) while timeout_elapsed < timeout: - logger.debug(str(dict( - timeout=f'{timeout_elapsed}/{timeout}', - by=by, - current_url=self.current_url, - value=value, - ))) + logger.debug( + f'wait_for_element :: {f"{timeout_elapsed}/{timeout}"} :: {by=} :: {self.current_url} :: {value=}') try: find = await self.find_elements( value=value, by=by, **kwargs) + logger.debug(f'wait_for_elements :: {len(find)} element(s) found') if find: + logger.info(f'wait_for_elements :: done') return find except Exception as error: - logger.error(error) + logger.error(f'wait_for_elements :: failed :: {error=}') timeout_elapsed = round(abs(timeout_start - time.time()), 1) + logger.debug(f'wait_for_elements :: {timeout_elapsed} seconds elapsed') - raise ElementNotFoundException(value) + raise ElementNotFoundException(f'wait_for_elements :: failed :: {value=}') async def wait_for_id( self, @@ -879,20 +1026,29 @@ async def wait_for_id( timeout: int = 30, **kwargs) -> selenium.webdriver.Chrome.find_element: """wait for an element id""" - return await self.wait_for_element( + logger.debug(f'wait_for_id :: {value=} :: {timeout=} :: {kwargs=}') + wait_for_id = await self.wait_for_element( value=value, by=self.by.ID, timeout=timeout, **kwargs) + logger.info(f'wait_for_id :: done') + return wait_for_id + async def wait_for_xpath( self, value: str, timeout: int = 30, **kwargs) -> selenium.webdriver.Chrome.find_element: """wait for a xpath""" - return await self.wait_for_element( + logger.debug(f'wait_for_xpath :: {value=} :: {timeout=} :: {kwargs=}') + + wait_for_xpath = await self.wait_for_element( value=value, by=self.by.XPATH, timeout=timeout, **kwargs) + + logger.info(f'wait_for_xpath :: done') + return wait_for_xpath diff --git a/automon/integrations/seleniumWrapper/config.py b/automon/integrations/seleniumWrapper/config.py index 5d06d8f7..e75d13b5 100644 --- a/automon/integrations/seleniumWrapper/config.py +++ b/automon/integrations/seleniumWrapper/config.py @@ -25,29 +25,27 @@ def webdriver(self): @property def window_size(self): - """get window size - - """ + """get window size""" if self.webdriver_wrapper: return self.webdriver_wrapper.window_size @property def cookies_base64(self): - logger.debug(f'{len(self._cookies_base64) if self._cookies_base64 else None}') + logger.debug(f'cookies_base64 :: {len(self._cookies_base64) / 1024 if self._cookies_base64 else None} KB') return self._cookies_base64 @property def cookies_file(self): - logger.info(f'{self._cookies_file}') + logger.debug(f'cookies_file :: {self._cookies_file}') return self._cookies_file async def run(self): """run webdriver""" + logger.debug(f'webdriver :: config :: run') run = await self.webdriver_wrapper.run() self._webdriver = self.webdriver_wrapper.webdriver - logger.info(str(dict( - webdriver=self.webdriver - ))) + logger.debug(f'webdriver :: config :: run :: {self.webdriver=}') + logger.info(f'webdriver :: config :: run :: done') return run async def start(self): diff --git a/automon/integrations/seleniumWrapper/webdriver_chrome.py b/automon/integrations/seleniumWrapper/webdriver_chrome.py index 79eaec79..b6cb0fd4 100644 --- a/automon/integrations/seleniumWrapper/webdriver_chrome.py +++ b/automon/integrations/seleniumWrapper/webdriver_chrome.py @@ -74,25 +74,19 @@ def window_size(self): return self._window_size def disable_certificate_verification(self): + logger.debug(f'webdriver :: chrome :: add_argument :: --ignore-certificate-errors') logger.warning('Certificates are not verified') self.chrome_options.add_argument('--ignore-certificate-errors') - logger.debug(str(dict( - add_argument='--ignore-certificate-errors' - ))) return self def disable_extensions(self): + logger.debug(f'webdriver :: chrome :: add_argument :: --disable-extensions') self.chrome_options.add_argument("--disable-extensions") - logger.debug(str(dict( - add_argument=f'--disable-extensions' - ))) return self def disable_infobars(self): + logger.debug(f'webdriver :: chrome :: add_argument :: --disable-infobars') self.chrome_options.add_argument("--disable-infobars") - logger.debug(str(dict( - add_argument=f'--disable-infobars' - ))) return self def disable_notifications(self): @@ -109,20 +103,20 @@ def disable_notifications(self): return self def disable_sandbox(self): + logger.debug(f'webdriver :: chrome :: add_argument :: --no-sandbox') self.chrome_options.add_argument('--no-sandbox') - logger.debug(str(dict( - add_argument=f'--no-sandbox' - ))) return self def disable_shm(self): logger.warning('Disabled shm will use disk I/O, and will be slow') + logger.debug(f'webdriver :: chrome :: add_argument :: --disable-dev-shm-usage') self.chrome_options.add_argument('--disable-dev-shm-usage') - logger.debug(str(dict( - add_argument=f'--disable-dev-shm-usage' - ))) return self + def download_chromedriver(self): + versions = 'https://googlechromelabs.github.io/chrome-for-testing/latest-versions-per-milestone-with-downloads.json' + raise + def enable_bigshm(self): logger.warning('Big shm not yet implemented') return self @@ -133,24 +127,18 @@ def enable_defaults(self): return self def enable_fullscreen(self): + logger.debug(f'webdriver :: chrome :: add_argument :: --start-fullscreen') self.chrome_options.add_argument("--start-fullscreen") - logger.debug(str(dict( - add_argument=f'--start-fullscreen' - ))) return self def enable_headless(self): + logger.debug(f'webdriver :: chrome :: add_argument :: headless') self.chrome_options.add_argument('headless') - logger.debug(str(dict( - add_argument='headless' - ))) return self def enable_logging(self): + logger.debug(f'webdriver :: chrome :: set_capability :: "goog:loggingPrefs", {{"performance": "ALL"}}') self.chrome_options.set_capability('goog:loggingPrefs', {'performance': 'ALL'}) - logger.debug(dict( - set_capability=('goog:loggingPrefs', {'performance': 'ALL'}) - )) return self def enable_notifications(self): @@ -166,10 +154,8 @@ def enable_notifications(self): return self def enable_maximized(self): + logger.debug(f'webdriver :: chrome :: add_argument :: --start-maximized') self.chrome_options.add_argument('--start-maximized') - logger.debug(str(dict( - add_argument='--start-maximized' - ))) return self def enable_translate(self, native_language: str = 'en'): @@ -317,25 +303,24 @@ async def run(self) -> bool: self._ChromeService = selenium.webdriver.ChromeService( executable_path=self.chromedriver_path ) - logger.debug(str(dict( - ChromeService=self.ChromeService - ))) + + logger.debug(f'webdriver :: chrome :: run :: {self.ChromeService=}') self._webdriver = selenium.webdriver.Chrome( service=self.ChromeService, options=self.chrome_options ) - logger.info(f'{self}') + logger.debug(f'webdriver :: chrome :: run :: {self=}') + logger.info(f'webdriver :: chrome :: run :: done') return True self._webdriver = selenium.webdriver.Chrome(options=self.chrome_options) logger.info(f'{self}') return True - except Exception as error: - logger.error(f'{error}') - raise Exception(error) + except Exception as exception: + raise Exception(f'webdriver :: chrome :: run :: failed :: {exception}') async def set_chromedriver(self, chromedriver_path: str): logger.debug(f'{chromedriver_path}') @@ -344,10 +329,8 @@ async def set_chromedriver(self, chromedriver_path: str): return self def set_locale(self, locale: str = 'en'): + logger.debug(f'webdriver :: chrome :: add_argument :: "--lang={locale}"') self.chrome_options.add_argument(f"--lang={locale}") - logger.debug(str(dict( - add_argument=f"--lang={locale}" - ))) return self def set_locale_experimental(self, locale: str = 'en-US'): @@ -365,10 +348,8 @@ def set_locale_experimental(self, locale: str = 'en-US'): return self def set_user_agent(self, user_agent: str): + logger.debug(f'webdriver :: chrome :: add_argument :: f"user-agent={user_agent}"') self.chrome_options.add_argument(f"user-agent={user_agent}") - logger.debug(str(dict( - add_argument=f"user-agent={user_agent}" - ))) return self def set_window_size(self, *args, **kwargs): @@ -401,13 +382,13 @@ def update_paths(self, path: str): else: os.environ['PATH'] = f"{os.getenv('PATH')}:{path}" - logger.debug(str(dict( - SELENIUM_CHROMEDRIVER_PATH=path, - PATH=os.environ['PATH'] - ))) + # logger.debug(f'update_paths :: {path=} :: {os.environ['PATH']}') + logger.debug(f'update_paths :: {path}') + logger.info(f'update_paths :: done') return True + logger.error(f'update_paths :: failed :: {path=}') return False async def quit(self):