Python Selenium 自动化测试框架：基于 BaseSpider 和 SeleniumUtil 的网页操作

"class SeleniumUtil():\n\tdef init(self, driver):\n\t\tself.driver = driver\n\t\t# self.display = settings['CHROME_DISPLAY']\n\n\tdef set_driver(self, driver):\n\t\tself.driver = driver\n\n\n\n\tdef safe_get_elements_(self, by, xpath, err):\n\t\tels = self.driver.find_elements(by=by, value=xpath)\n\t\tif not els:\n\t\t\traise SpiderException(err, xpath) # 返回错误信息 和 xpath\n\t\treturn els[0]\n\n\tdef get_page(self, url, sleep=1):\n\t\tself.driver.get(url)\n\t\ttime.sleep(sleep)\n\t\t# 打开网页\n\n\tdef input_with_xpath(self, by, xpath, keys, err):\n\t\telement = self.safe_get_elements_(by, xpath, err)\n\t\treturn element.send_keys(keys)\n\n\tdef click_with_xpath(self, by, xpath, err_msg, front_sleep=3):\n\t\ttime.sleep(front_sleep)\n\t\telement = self.safe_get_elements_(by, xpath, err_msg)\n\t\tif element.is_enabled():\n\t\t\treturn element.click()\n\t\telse:\n\t\t\traise SpiderException(err_msg, xpath)\n\n\tdef check_loging_with(self ,by, xpath):\n\t\t"""登录验证"""\n\t\t# try:\n\n\t\terror_element = self.driver.find_elements(by, xpath)\n\t\tfor err in error_element:\n\t\t\terror_message = err.text # 获取登录失败返回的信息\n\t\t\tif error_message:\n\t\t\t\traise SpiderException(error_message, xpath=xpath)\n\t\t# 页面跳转失败\n\t\tcurrent_url = self.driver.current_url\n\t\tlogin_url = LOGIN_URL\n\t\tif current_url == login_url:\n\t\t\traise SpiderException(LogErrorMsg.LOGIN_PAGE_JUMP_FAILED, xpath=xpath)\n\n\t\t# 页面跳转成功未刷新\n\t\tlogout = self.driver.find_elements(by, BUTTON_LOGOUT_ELEMENT)\n\t\tif not logout:\n\t\t\traise SpiderException(LogErrorMsg.LOGIN_PAGE_NOT_REFRESHED, xpath=xpath)\n\n\n\nclass BaseSpider(ABC):\n\n\tdef init(self):\n\t\tself.driver = None\n\t\tself.browser_ = None\n\t\tself.records = list()\n\t\tself.now_time = datetime.datetime.now()\n\n\tdef set_driver(self, driver):\n\t\tself.driver = driver\n\t\tself.browser_ = SeleniumUtil(self.driver)\n\n\tdef add_record(self, urn, class_name, operate, exception, ele_message):\n\t\t"""错误记录\n\t\turn:urn class_name:类名 operate:操作 exception:错误信息 ele_message:元素信息\n\t\t"""\n\t\tself.records.extend([self.now_time, urn, class_name, operate, exception, ele_message])\n\t\tQueryResult.insert_data(INSERT_OPERATION_LOG, self.records)\n\t\tself.records.clear()\n\n\tdef get_search_url(self, url):\n\t\tself.browser_.get_page(url)\n\n\tdef check_login(self):\n\t\t"""登录验证逻辑实现"""\n\t\tself.browser_.get_page(LOGIN_URL) # 登录\n\t\ttry:\n\t\t\tself.browser_.input_with_xpath(By.NAME, TEXTBOX_LOGIN_ELEMENT, settings['dfm_username'],\n\t\t\t\t\t\t\t\t\t\t\tErrorMsg.TEXT_BOX_NOT_DISPLAYED)\n\t\t\tself.browser_.input_with_xpath(By.NAME, TEXTBOX_PWD_ELEMENT, \n\t\t\t\t\t\t\t\t\t\t\tgenerate_utils.transform_str(settings['dfm_pwd']),\n\t\t\t\t\t\t\t\t\t\t\tErrorMsg.TEXT_BOX_NOT_DISPLAYED)\n\t\t\tself.browser_.click_with_xpath(By.NAME, BUTTON_LOGIN_ELEMENT, \n\t\t\t\t\t\t\t\t\t\t\tLogErrorMsg.LOGIN_CLICK_BUTTON_NOT_DISPLAYED)\n\t\t\tself.browser_.check_loging_with(By.XPATH, CUSTOM_LOGIN_ELEMENT)\n\t\t\treturn True\n\n\t\texcept SpiderException as e:\n\t\t\tlog_error(e)\n\t\t\tself.add_record(False, self.class.name, Operate.LOGIN.value, e.message, e.xpath)\n\t\t\treturn False\n\n\nclass EmailSpider(BaseSpider):\n\n\tdef int(self):\n\t\tBaseSpider.init(self) # 调用父类构造方法\n\n\tdef get_advanced_and_ingestion(self):\n\t\t"""advanced 和ingestion 按钮"""\n\t\tself.get_search_url(SEARCH_PAGE) # search页面\n\t\tself.browser_.click_with_xpath(By.LINK_TEXT, ADVANCED_SEARCH_ELEMENT, \n\t\t\t\t\t\t\t\t\t\t\tEmailErrorMsg.ADVANCED_SEARCH_NOR_DISPLAYED) # Advanced Search 按钮\n\t\tself.browser_.click_with_xpath(By.LINK_TEXT, EMAIL_INGESTION_DATA_ELEMENT, \n\t\t\t\t\t\t\t\t\t\t\tEmailErrorMsg.INGESTION_DATA_NOR_DISPLAYED) # Email Ingestion data 按钮\n\n\nclass DFMProcessor(BaseProcessor):\n\tdef init(self):\n\t\tBaseProcessor.init(self)\n\t\tself.name = 'DFMProcessor'\n\t\tself.lock = ''\n\t\tself.display = settings['CHROME_DISPLAY']\n\t\tself.baseSpider = BaseSpider()\n\n\tdef process(self):\n\t\tpass\n\n\tdef get_download_path(self):\n\t\treturn ''\n\n\tdef start(self):\n\t\ttry:\n\t\t\tself.lock.acquire()\n\t\t\tlogger.info(r" >>> 开始执行 {}".format(self.name))\n\t\t\tdriver = get_chrome_driver(self.get_download_path(), display=self.display)\n\t\t\tself.baseSpider.set_driver(driver) # 设置driver给BaseSpider\n\t\t\tstatus = self.baseSpider.check_login()\n\t\t\tif status:\n\t\t\t\tself.process()\n\t\t\tlogger.info(r" >>> 完成执行 {}".format(self.name))\n\t\texcept Exception as e:\n\t\t\tlog_error(e)\n\t\tfinally:\n\t\t\tself.baseSpider.driver.quit()\n\t\t\tself.lock.release()\n\n\n@singleton\nclass EmailIngestionProcessor(DFMProcessor):\n\tdef init(self):\n\t\tDFMProcessor.init(self)\n\t\tself.lock = threading.Lock()\n\t\tself.name = "EmailIngestionProcessor"\n\t\tself.emailSpider = EmailSpider()\n\n\tdef get_download_path(self):\n\t\tcur_date = get_strtime('%Y-%m-%d')\n\t\tdownload_path = os.path.join(settings['email_ingestion_path'], cur_date)\n\t\tif not os.path.exists(download_path):\n\t\t\tos.makedirs(download_path)\n\t\treturn download_path\n\n\tdef process(self):\n\t\tself.emailSpider.get_advanced_and_ingestion()\n\n\n\n#\n# lock = threading.Lock()\n#\n#\ndef run():\n\t# while True:\n\tEmailIngestionProcessor().start()\n\n\n#\nif name == 'main':\n\trun()\n