Fixed following issues for feature test:

1. Modified the get_chromedriver utility for supporting python version below 3.5.
2. Handled some exceptions getting intermittently on some databases.
3. Generalized some functions with additional parameters.
4. Disabled the auto-expansion of the children nodes for maintaining the synchronization.
This commit is contained in:
Shubham Agarwal 2019-12-03 19:35:48 +05:30 committed by Akshay Joshi
parent f7cc79fab1
commit 5093e6db5e
8 changed files with 93 additions and 60 deletions

View File

@ -17,7 +17,12 @@ import os
import platform
import subprocess
import sys
import urllib.request
try:
from urllib.request import urlopen, urlretrieve
from urllib.error import URLError
except Exception:
from urllib import urlopen, urlretrieve
URLError = Exception
import zipfile
@ -85,13 +90,13 @@ def get_chrome_version(args):
# On Linux/Mac we run the Chrome executable with the --version flag,
# then parse the output.
try:
result = subprocess.run([args.chrome, '--version'],
stdout=subprocess.PIPE)
except FileNotFoundError as e:
result = subprocess.Popen([args.chrome, '--version'],
stdout=subprocess.PIPE)
except FileNotFoundError:
print('The specified Chrome executable could not be found.')
sys.exit(1)
version_str = result.stdout.decode("utf-8").strip()
version_str = result.stdout.read().decode("utf-8")
# Check for 'Chrom' not 'Chrome' in case the user is using Chromium.
if "Chrom" not in version_str:
print('The specified Chrome executable output an unexpected '
@ -120,8 +125,8 @@ def get_chromedriver_version(chrome_version):
.format(chrome_version)
try:
fp = urllib.request.urlopen(url)
except urllib.error.URLError as e:
fp = urlopen(url)
except URLError as e:
print('The chromedriver catalog URL could not be accessed: {}'
.format(e))
sys.exit(1)
@ -173,8 +178,8 @@ print('Downloading chromedriver v{} for Chrome v{} on {}...'
.format(chromedriver_version, chrome_version, system))
try:
file, headers = urllib.request.urlretrieve(url)
except urllib.error.URLError as e:
file, headers = urlretrieve(url)
except URLError as e:
print('The chromedriver download URL could not be accessed: {}'
.format(e))
sys.exit(1)

View File

@ -164,11 +164,7 @@ class PGDataypeFeatureTest(BaseFeatureTest):
self._create_enum_type()
for batch in config_data:
query = self.construct_select_query(batch)
self.page.fill_codemirror_area_with(query)
execute_query = self.page.find_by_css_selector(
QueryToolLocators.btn_execute_query_css)
execute_query.click()
self.page.execute_query(query)
wait = WebDriverWait(self.page.driver, 5)
# wait for the visibility of the grid to appear

View File

@ -118,8 +118,8 @@ class PGUtilitiesBackupFeatureTest(BaseFeatureTest):
(By.XPATH,
NavMenuLocators.process_watcher_alertfier))
self.page.wait_for_element_to_disappear(
lambda driver: driver.find_element_by_css_selector(".loading-logs")
)
lambda driver: driver.find_element_by_css_selector(
".loading-logs"), 10)
if status != "Successfully completed.":
self.assertEquals(status, "Successfully completed.")
@ -188,8 +188,8 @@ class PGUtilitiesBackupFeatureTest(BaseFeatureTest):
(By.XPATH,
NavMenuLocators.process_watcher_alertfier))
self.page.wait_for_element_to_disappear(
lambda driver: driver.find_element_by_css_selector(".loading-logs")
)
lambda driver: driver.find_element_by_css_selector(
".loading-logs"), 10)
if status != "Successfully completed.":
self.assertEquals(status, "Successfully completed.")
@ -215,6 +215,7 @@ class PGUtilitiesBackupFeatureTest(BaseFeatureTest):
os.remove(backup_file)
def after(self):
test_gui_helper.close_process_watcher(self)
test_gui_helper.close_bgprocess_popup(self)
self.page.remove_server(self.server)
connection = test_utils.get_db_connection(

View File

@ -37,22 +37,22 @@ class QueryToolAutoCompleteFeatureTest(BaseFeatureTest):
self.page.add_server(self.server)
self.first_schema_name = "test_schema" + \
str(random.randint(1000, 3000))
str(random.randint(1000, 2000))
test_utils.create_schema(self.server, self.test_db,
self.first_schema_name)
self.second_schema_name = "comp_schema" + \
str(random.randint(1000, 3000))
str(random.randint(2000, 3000))
test_utils.create_schema(self.server, self.test_db,
self.second_schema_name)
self.first_table_name = "auto_comp_" + \
str(random.randint(1000, 3000))
str(random.randint(1000, 2000))
test_utils.create_table(self.server, self.test_db,
self.first_table_name)
self.second_table_name = "auto_comp_" + \
str(random.randint(1000, 3000))
str(random.randint(2000, 3000))
test_utils.create_table(self.server, self.test_db,
self.second_table_name)

View File

@ -218,12 +218,6 @@ SELECT generate_series(1, {}) as id1, 'dummy' as id2""".format(
QueryToolLocators.query_output_cells))
)
# click on first data column to select all column.
column_1 = \
self.page.find_by_css_selector(
QueryToolLocators.output_column_header_css.format('id1'))
column_1.click()
canvas = self.wait.until(EC.presence_of_element_located(
(By.CSS_SELECTOR, QueryToolLocators.query_output_canvas_css)))
@ -235,6 +229,11 @@ SELECT generate_series(1, {}) as id1, 'dummy' as id2""".format(
scroll = 10
status = False
while scroll:
# click on first data column to select all column.
column_1 = \
self.page.find_by_css_selector(
QueryToolLocators.output_column_header_css.format('id1'))
column_1.click()
canvas_ele = self.page.find_by_css_selector('.grid-canvas')
scrolling_height = canvas_ele.size['height']
self.driver.execute_script(

View File

@ -173,7 +173,7 @@ class PgadminPage:
self.fill_codemirror_area_with(query)
self.click_execute_query_button()
def click_execute_query_button(self):
def click_execute_query_button(self, timeout=20):
retry = 5
execute_button = self.find_by_css_selector(
QueryToolLocators.btn_execute_query_css)
@ -189,7 +189,7 @@ class PgadminPage:
break
else:
retry -= 1
self.wait_for_query_tool_loading_indicator_to_disappear()
self.wait_for_query_tool_loading_indicator_to_disappear(timeout)
def check_execute_option(self, option):
""""This function will check auto commit or auto roll back based on
@ -689,15 +689,17 @@ class PgadminPage:
try:
webdriver.ActionChains(self.driver).double_click(
server_element).perform()
if self.wait_for_element_to_appear(self.driver,
ConnectToServerDiv.ok_button):
self.fill_input_by_xpath(
ConnectToServerDiv.password_field, password)
if self.check_if_element_exist_by_xpath(
ConnectToServerDiv.ok_button):
field = self.find_by_xpath(
ConnectToServerDiv.password_field)
self.fill_input(field, password)
self.find_by_xpath(ConnectToServerDiv.ok_button).click()
self.wait_until_element_not_visible(
ConnectToServerDiv.ok_button)
if self.wait_for_element_to_be_visible(
self.driver, ConnectToServerDiv.error_message, 2):
self.wait_for_element_to_disappear(
lambda driver: driver.find_element_by_xpath(
ConnectToServerDiv.ok_button))
if self.check_if_element_exist_by_xpath(
ConnectToServerDiv.error_message, 2):
print(
"While entering password in click_and_connect_server "
"function, error is occurred : " + str(
@ -983,7 +985,7 @@ class PgadminPage:
self._wait_for("element to exist", element_if_it_exists)
return find_method_with_args(self.driver)
def wait_for_element_to_disappear(self, find_method_with_args):
def wait_for_element_to_disappear(self, find_method_with_args, timeout=5):
def element_if_it_disappears(driver):
try:
element = find_method_with_args(driver)
@ -994,7 +996,8 @@ class PgadminPage:
except (NoSuchElementException, StaleElementReferenceException):
return True
return self._wait_for("element to disappear", element_if_it_disappears)
return self._wait_for("element to disappear",
element_if_it_disappears, timeout)
def wait_for_reloading_indicator_to_disappear(self):
def reloading_indicator_has_disappeared(driver):
@ -1017,7 +1020,7 @@ class PgadminPage:
self._wait_for("spinner to disappear", spinner_has_disappeared, 20)
def wait_for_query_tool_loading_indicator_to_disappear(self):
def wait_for_query_tool_loading_indicator_to_disappear(self, timeout=20):
def spinner_has_disappeared(driver):
try:
spinner = driver.find_element_by_css_selector(
@ -1029,7 +1032,8 @@ class PgadminPage:
time.sleep(0.5)
return True
self._wait_for("spinner to disappear", spinner_has_disappeared, 20)
self._wait_for(
"spinner to disappear", spinner_has_disappeared, timeout)
def wait_for_query_tool_loading_indicator_to_appear(self):
status = self.check_if_element_exist_by_xpath(
@ -1151,7 +1155,7 @@ class PgadminPage:
try:
element = self.driver.find_element(*click_locator)
element.click()
WebDriverWait(self.driver, 2).until(
WebDriverWait(self.driver, 10).until(
EC.visibility_of_element_located(verify_locator))
click_status = True
except Exception:

View File

@ -47,11 +47,14 @@ def close_bgprocess_popup(tester):
def close_process_watcher(tester):
attempt = 10
while attempt > 0:
close_btn = tester.page.find_by_xpath(
NavMenuLocators.process_watcher_close_button_xpath)
close_btn.click()
if not tester.page.check_if_element_exist_by_xpath(
NavMenuLocators.process_watcher_close_button_xpath, 1):
break
else:
try:
if not tester.page.check_if_element_exist_by_xpath(
NavMenuLocators.process_watcher_close_button_xpath, 1):
break
else:
close_btn = tester.page.find_by_xpath(
NavMenuLocators.process_watcher_close_button_xpath)
close_btn.click()
attempt -= 1
except Exception:
attempt -= 1

View File

@ -108,6 +108,7 @@ def clear_node_info_dict():
def create_database(server, db_name, encoding=None):
"""This function used to create database and returns the database id"""
db_id = ''
try:
connection = get_db_connection(
server['db'],
@ -135,13 +136,13 @@ def create_database(server, db_name, encoding=None):
pg_cursor.execute("SELECT db.oid from pg_database db WHERE"
" db.datname='%s'" % db_name)
oid = pg_cursor.fetchone()
db_id = ''
if oid:
db_id = oid[0]
connection.close()
return db_id
except Exception:
traceback.print_exc(file=sys.stderr)
return db_id
def create_table(server, db_name, table_name, extra_columns=[]):
@ -777,6 +778,27 @@ def configure_preferences(default_binary_path=None):
' WHERE PID = ?', (-1, pref_tree_state_save_interval.pid)
)
# Disable auto expand sole children tree state for tests
pref_auto_expand_sol_children = \
browser_pref.preference('auto_expand_sole_children')
user_pref = cur.execute(
'SELECT pid, uid FROM user_preferences '
'where pid=?', (pref_auto_expand_sol_children.pid,)
)
if len(user_pref.fetchall()) == 0:
cur.execute(
'INSERT INTO user_preferences(pid, uid, value)'
' VALUES (?,?,?)', (pref_auto_expand_sol_children.pid, 1, 'False')
)
else:
cur.execute(
'UPDATE user_preferences'
' SET VALUE = ?'
' WHERE PID = ?', ('False', pref_auto_expand_sol_children.pid)
)
# Disable reload warning on browser
pref_confirm_on_refresh_close = \
browser_pref.preference('confirm_on_refresh_close')
@ -1103,18 +1125,21 @@ def get_watcher_dialogue_status(self):
"""This will get watcher dialogue status"""
import time
attempts = 120
status = None
while attempts > 0:
status = self.page.find_by_css_selector(
".pg-bg-status-text").text
try:
status = self.page.find_by_css_selector(
".pg-bg-status-text").text
if 'Failed' in status:
break
if status == 'Started' or status == 'Running...':
if 'Failed' in status:
break
if status == 'Started' or status == 'Running...':
attempts -= 1
time.sleep(.5)
else:
break
except Exception:
attempts -= 1
time.sleep(.5)
else:
break
return status