diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_add.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_add.py
index 3cd6b3b86..5b24e6e8f 100644
--- a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_add.py
+++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_add.py
@@ -43,31 +43,30 @@ class IndexConstraintAddTestCase(BaseTestGenerator):
dict(url='/browser/unique_constraint/obj/', data=unique_key_data))
]
- @classmethod
- def setUpClass(cls):
- cls.db_name = parent_node_dict["database"][-1]["db_name"]
+ def setUp(self):
+ self.db_name = parent_node_dict["database"][-1]["db_name"]
schema_info = parent_node_dict["schema"][-1]
- cls.server_id = schema_info["server_id"]
- cls.db_id = schema_info["db_id"]
- db_con = database_utils.connect_database(cls, utils.SERVER_GROUP,
- cls.server_id, cls.db_id)
+ self.server_id = schema_info["server_id"]
+ self.db_id = schema_info["db_id"]
+ db_con = database_utils.connect_database(self, utils.SERVER_GROUP,
+ self.server_id, self.db_id)
if not db_con['data']["connected"]:
raise Exception("Could not connect to database to add a "
"index constraint(primary key or unique key).")
- cls.schema_id = schema_info["schema_id"]
- cls.schema_name = schema_info["schema_name"]
- schema_response = schema_utils.verify_schemas(cls.server,
- cls.db_name,
- cls.schema_name)
+ self.schema_id = schema_info["schema_id"]
+ self.schema_name = schema_info["schema_name"]
+ schema_response = schema_utils.verify_schemas(self.server,
+ self.db_name,
+ self.schema_name)
if not schema_response:
raise Exception("Could not find the schema to add a index "
"constraint(primary key or unique key).")
- cls.table_name = "table_indexconstraint_%s" % \
- (str(uuid.uuid4())[1:8])
- cls.table_id = tables_utils.create_table(cls.server,
- cls.db_name,
- cls.schema_name,
- cls.table_name)
+ self.table_name = "table_indexconstraint_%s" % \
+ (str(uuid.uuid4())[1:8])
+ self.table_id = tables_utils.create_table(self.server,
+ self.db_name,
+ self.schema_name,
+ self.table_name)
def runTest(self):
"""This function will add index constraint(primary key or unique key)
@@ -80,7 +79,6 @@ class IndexConstraintAddTestCase(BaseTestGenerator):
content_type='html/json')
self.assertEquals(response.status_code, 200)
- @classmethod
- def tearDownClass(cls):
+ def tearDown(self):
# Disconnect the database
- database_utils.disconnect_database(cls, cls.server_id, cls.db_id)
+ database_utils.disconnect_database(self, self.server_id, self.db_id)
diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_delete.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_delete.py
index 7eea5e9bb..f35a00b4a 100644
--- a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_delete.py
+++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_delete.py
@@ -38,31 +38,30 @@ class IndexConstraintDeleteTestCase(BaseTestGenerator):
type="UNIQUE"))
]
- @classmethod
- def setUpClass(cls):
- cls.db_name = parent_node_dict["database"][-1]["db_name"]
+ def setUp(self):
+ self.db_name = parent_node_dict["database"][-1]["db_name"]
schema_info = parent_node_dict["schema"][-1]
- cls.server_id = schema_info["server_id"]
- cls.db_id = schema_info["db_id"]
- db_con = database_utils.connect_database(cls, utils.SERVER_GROUP,
- cls.server_id, cls.db_id)
+ self.server_id = schema_info["server_id"]
+ self.db_id = schema_info["db_id"]
+ db_con = database_utils.connect_database(self, utils.SERVER_GROUP,
+ self.server_id, self.db_id)
if not db_con['data']["connected"]:
raise Exception("Could not connect to database to add a "
"index constraint(primary key or unique key).")
- cls.schema_id = schema_info["schema_id"]
- cls.schema_name = schema_info["schema_name"]
- schema_response = schema_utils.verify_schemas(cls.server,
- cls.db_name,
- cls.schema_name)
+ self.schema_id = schema_info["schema_id"]
+ self.schema_name = schema_info["schema_name"]
+ schema_response = schema_utils.verify_schemas(self.server,
+ self.db_name,
+ self.schema_name)
if not schema_response:
raise Exception("Could not find the schema to add a index "
"constraint(primary key or unique key).")
- cls.table_name = "table_indexconstraint_%s" % \
- (str(uuid.uuid4())[1:8])
- cls.table_id = tables_utils.create_table(cls.server,
- cls.db_name,
- cls.schema_name,
- cls.table_name)
+ self.table_name = "table_indexconstraint_%s" % \
+ (str(uuid.uuid4())[1:8])
+ self.table_id = tables_utils.create_table(self.server,
+ self.db_name,
+ self.schema_name,
+ self.table_name)
def runTest(self):
"""This function will delete index constraint(primary key or
@@ -81,7 +80,6 @@ class IndexConstraintDeleteTestCase(BaseTestGenerator):
)
self.assertEquals(response.status_code, 200)
- @classmethod
- def tearDownClass(cls):
+ def tearDown(self):
# Disconnect the database
- database_utils.disconnect_database(cls, cls.server_id, cls.db_id)
+ database_utils.disconnect_database(self, self.server_id, self.db_id)
diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_get.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_get.py
index c08c2acb3..f838a2f73 100644
--- a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_get.py
+++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_get.py
@@ -38,31 +38,30 @@ class IndexConstraintGetTestCase(BaseTestGenerator):
type="UNIQUE"))
]
- @classmethod
- def setUpClass(cls):
- cls.db_name = parent_node_dict["database"][-1]["db_name"]
+ def setUp(self):
+ self.db_name = parent_node_dict["database"][-1]["db_name"]
schema_info = parent_node_dict["schema"][-1]
- cls.server_id = schema_info["server_id"]
- cls.db_id = schema_info["db_id"]
- db_con = database_utils.connect_database(cls, utils.SERVER_GROUP,
- cls.server_id, cls.db_id)
+ self.server_id = schema_info["server_id"]
+ self.db_id = schema_info["db_id"]
+ db_con = database_utils.connect_database(self, utils.SERVER_GROUP,
+ self.server_id, self.db_id)
if not db_con['data']["connected"]:
raise Exception("Could not connect to database to add a "
"index constraint(primary key or unique key).")
- cls.schema_id = schema_info["schema_id"]
- cls.schema_name = schema_info["schema_name"]
- schema_response = schema_utils.verify_schemas(cls.server,
- cls.db_name,
- cls.schema_name)
+ self.schema_id = schema_info["schema_id"]
+ self.schema_name = schema_info["schema_name"]
+ schema_response = schema_utils.verify_schemas(self.server,
+ self.db_name,
+ self.schema_name)
if not schema_response:
raise Exception("Could not find the schema to add a index "
"constraint(primary key or unique key).")
- cls.table_name = "table_indexconstraint_%s" % \
- (str(uuid.uuid4())[1:8])
- cls.table_id = tables_utils.create_table(cls.server,
- cls.db_name,
- cls.schema_name,
- cls.table_name)
+ self.table_name = "table_indexconstraint_%s" % \
+ (str(uuid.uuid4())[1:8])
+ self.table_id = tables_utils.create_table(self.server,
+ self.db_name,
+ self.schema_name,
+ self.table_name)
def runTest(self):
"""This function will fetch the index constraint(primary key or
@@ -81,7 +80,6 @@ class IndexConstraintGetTestCase(BaseTestGenerator):
)
self.assertEquals(response.status_code, 200)
- @classmethod
- def tearDownClass(cls):
+ def tearDown(self):
# Disconnect the database
- database_utils.disconnect_database(cls, cls.server_id, cls.db_id)
+ database_utils.disconnect_database(self, self.server_id, self.db_id)
diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_put.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_put.py
index 87289a620..6bdf70308 100644
--- a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_put.py
+++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/tests/test_index_constraint_put.py
@@ -40,31 +40,30 @@ class IndexConstraintUpdateTestCase(BaseTestGenerator):
type="UNIQUE", data=data))
]
- @classmethod
- def setUpClass(cls):
- cls.db_name = parent_node_dict["database"][-1]["db_name"]
+ def setUp(self):
+ self.db_name = parent_node_dict["database"][-1]["db_name"]
schema_info = parent_node_dict["schema"][-1]
- cls.server_id = schema_info["server_id"]
- cls.db_id = schema_info["db_id"]
- db_con = database_utils.connect_database(cls, utils.SERVER_GROUP,
- cls.server_id, cls.db_id)
+ self.server_id = schema_info["server_id"]
+ self.db_id = schema_info["db_id"]
+ db_con = database_utils.connect_database(self, utils.SERVER_GROUP,
+ self.server_id, self.db_id)
if not db_con['data']["connected"]:
raise Exception("Could not connect to database to add a "
"index constraint(primary key or unique key).")
- cls.schema_id = schema_info["schema_id"]
- cls.schema_name = schema_info["schema_name"]
- schema_response = schema_utils.verify_schemas(cls.server,
- cls.db_name,
- cls.schema_name)
+ self.schema_id = schema_info["schema_id"]
+ self.schema_name = schema_info["schema_name"]
+ schema_response = schema_utils.verify_schemas(self.server,
+ self.db_name,
+ self.schema_name)
if not schema_response:
raise Exception("Could not find the schema to add a index "
"constraint(primary key or unique key).")
- cls.table_name = "table_indexconstraint_%s" % \
- (str(uuid.uuid4())[1:8])
- cls.table_id = tables_utils.create_table(cls.server,
- cls.db_name,
- cls.schema_name,
- cls.table_name)
+ self.table_name = "table_indexconstraint_%s" % \
+ (str(uuid.uuid4())[1:8])
+ self.table_id = tables_utils.create_table(self.server,
+ self.db_name,
+ self.schema_name,
+ self.table_name)
def runTest(self):
"""This function will update index constraint(primary key or
@@ -84,7 +83,6 @@ class IndexConstraintUpdateTestCase(BaseTestGenerator):
follow_redirects=True)
self.assertEquals(response.status_code, 200)
- @classmethod
- def tearDownClass(cls):
+ def tearDown(self):
# Disconnect the database
- database_utils.disconnect_database(cls, cls.server_id, cls.db_id)
+ database_utils.disconnect_database(self, self.server_id, self.db_id)
diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/tests/test_table_parameters.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/tests/test_table_parameters.py
index bbed26998..02203d2f5 100644
--- a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/tests/test_table_parameters.py
+++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/tests/test_table_parameters.py
@@ -46,8 +46,9 @@ class TableUpdateParameterTestCase(BaseTestGenerator):
)
]
- @classmethod
- def setUpClass(self):
+ table_name = "test_table_parameters_%s" % (str(uuid.uuid4())[1:8])
+
+ def setUp(self):
self.db_name = parent_node_dict["database"][-1]["db_name"]
schema_info = parent_node_dict["schema"][-1]
self.server_id = schema_info["server_id"]
@@ -63,12 +64,14 @@ class TableUpdateParameterTestCase(BaseTestGenerator):
self.schema_name)
if not schema_response:
raise Exception("Could not find the schema to add a table.")
- self.table_name = "test_table_parameters_%s" % (str(uuid.uuid4())[1:8])
- self.table_id = tables_utils.create_table(
- self.server, self.db_name,
- self.schema_name,
- self.table_name)
+ self.table_id = tables_utils.get_table_id(self.server, self.db_name,
+ self.table_name)
+ if self.table_id is None:
+ self.table_id = tables_utils.create_table(
+ self.server, self.db_name,
+ self.schema_name,
+ self.table_name)
def runTest(self):
"""This function will fetch added table under schema node."""
@@ -130,7 +133,6 @@ class TableUpdateParameterTestCase(BaseTestGenerator):
follow_redirects=True)
self.assertEquals(response.status_code, 200)
- @classmethod
- def tearDownClass(self):
+ def tearDown(self):
# Disconnect the database
database_utils.disconnect_database(self, self.server_id, self.db_id)
diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/tests/utils.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/tests/utils.py
index 0a2963aab..8e5f665f3 100644
--- a/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/tests/utils.py
+++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/tables/tests/utils.py
@@ -483,3 +483,26 @@ def get_hash_partitions_data(data):
}]
data['partition_keys'] = \
[{'key_type': 'column', 'pt_column': 'empno'}]
+
+
+def get_table_id(server, db_name, table_name):
+ try:
+ connection = utils.get_db_connection(db_name,
+ server['username'],
+ server['db_password'],
+ server['host'],
+ server['port'],
+ server['sslmode'])
+ pg_cursor = connection.cursor()
+ pg_cursor.execute("select oid from pg_class where relname='%s'" %
+ table_name)
+ table = pg_cursor.fetchone()
+ if table:
+ table_id = table[0]
+ else:
+ table_id = None
+ connection.close()
+ return table_id
+ except Exception:
+ traceback.print_exc(file=sys.stderr)
+ raise
diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/views/tests/test_mviews_parameters.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/views/tests/test_mviews_parameters.py
index c84276ef6..1f63ef564 100644
--- a/web/pgadmin/browser/server_groups/servers/databases/schemas/views/tests/test_mviews_parameters.py
+++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/views/tests/test_mviews_parameters.py
@@ -45,8 +45,9 @@ class MViewsUpdateParameterTestCase(BaseTestGenerator):
)
]
- @classmethod
- def setUpClass(self):
+ m_view_name = "test_mview_put_%s" % (str(uuid.uuid4())[1:8])
+
+ def setUp(self):
self.db_name = parent_node_dict["database"][-1]["db_name"]
schema_info = parent_node_dict["schema"][-1]
self.server_id = schema_info["server_id"]
@@ -70,17 +71,19 @@ class MViewsUpdateParameterTestCase(BaseTestGenerator):
if not schema_response:
raise Exception("Could not find the schema to update a mview.")
- self.m_view_name = "test_mview_put_%s" % (str(uuid.uuid4())[1:8])
- m_view_sql = "CREATE MATERIALIZED VIEW %s.%s TABLESPACE pg_default " \
- "AS SELECT 'test_pgadmin' WITH NO DATA;ALTER TABLE " \
- "%s.%s OWNER TO %s"
-
- self.m_view_id = views_utils.create_view(self.server,
- self.db_name,
- self.schema_name,
- m_view_sql,
+ self.m_view_id = views_utils.get_view_id(self.server, self.db_name,
self.m_view_name)
+ if self.m_view_id is None:
+ m_view_sql = "CREATE MATERIALIZED VIEW %s.%s TABLESPACE " \
+ "pg_default AS SELECT 'test_pgadmin' WITH NO " \
+ "DATA;ALTER TABLE %s.%s OWNER TO %s"
+ self.m_view_id = views_utils.create_view(self.server,
+ self.db_name,
+ self.schema_name,
+ m_view_sql,
+ self.m_view_name)
+
def runTest(self):
"""This function will update the view/mview under schema node."""
mview_response = views_utils.verify_view(self.server, self.db_name,
@@ -141,7 +144,6 @@ class MViewsUpdateParameterTestCase(BaseTestGenerator):
follow_redirects=True)
self.assertEquals(response.status_code, 200)
- @classmethod
- def tearDownClass(self):
+ def tearDown(self):
# Disconnect the database
database_utils.disconnect_database(self, self.server_id, self.db_id)
diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/views/tests/utils.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/views/tests/utils.py
index d6180c4fc..7619ee673 100644
--- a/web/pgadmin/browser/server_groups/servers/databases/schemas/views/tests/utils.py
+++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/views/tests/utils.py
@@ -86,3 +86,28 @@ def verify_view(server, db_name, view_name):
except Exception:
traceback.print_exc(file=sys.stderr)
raise
+
+
+def get_view_id(server, db_name, view_name):
+ try:
+ connection = utils.get_db_connection(db_name,
+ server['username'],
+ server['db_password'],
+ server['host'],
+ server['port'],
+ server['sslmode'])
+ old_isolation_level = connection.isolation_level
+ connection.set_isolation_level(0)
+ pg_cursor = connection.cursor()
+ # Get 'oid' from newly created view
+ pg_cursor.execute("select oid from pg_class where relname='%s'" %
+ view_name)
+ view = pg_cursor.fetchone()
+ view_id = None
+ if view:
+ view_id = view[0]
+ connection.close()
+ return view_id
+ except Exception:
+ traceback.print_exc(file=sys.stderr)
+ raise
diff --git a/web/pgadmin/feature_tests/copy_selected_query_results_feature_test.py b/web/pgadmin/feature_tests/copy_selected_query_results_feature_test.py
index 7091ffa42..7475970ab 100644
--- a/web/pgadmin/feature_tests/copy_selected_query_results_feature_test.py
+++ b/web/pgadmin/feature_tests/copy_selected_query_results_feature_test.py
@@ -8,7 +8,6 @@
##########################################################################
from __future__ import print_function
-import pyperclip
import random
from selenium.webdriver import ActionChains
@@ -61,7 +60,6 @@ class CopySelectedQueryResultsFeatureTest(BaseFeatureTest):
self._copies_rows_with_header()
def _copies_rows(self):
- pyperclip.copy("old clipboard contents")
first_row = self.page.find_by_xpath(
QueryToolLocators.output_row_xpath.format(1))
first_row.click()
@@ -70,14 +68,23 @@ class CopySelectedQueryResultsFeatureTest(BaseFeatureTest):
QueryToolLocators.copy_button_css)
copy_button.click()
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
self.assertEqual('"Some-Name"\t6\t"some info"',
- pyperclip.paste())
+ clipboard_text)
+ scratch_pad_ele.clear()
def _copies_rows_with_header(self):
self.page.find_by_css_selector('#btn-copy-row-dropdown').click()
self.page.find_by_css_selector('a#btn-copy-with-header').click()
- pyperclip.copy("old clipboard contents")
select_all = self.page.find_by_xpath(
QueryToolLocators.select_all_column)
select_all.click()
@@ -86,13 +93,22 @@ class CopySelectedQueryResultsFeatureTest(BaseFeatureTest):
QueryToolLocators.copy_button_css)
copy_button.click()
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
self.assertEqual("""\"some_column"\t"value"\t"details"
\"Some-Name"\t6\t"some info"
\"Some-Other-Name"\t22\t"some other info"
-\"Yet-Another-Name"\t14\t"cool info\"""", pyperclip.paste())
+\"Yet-Another-Name"\t14\t"cool info\"""", clipboard_text)
+ scratch_pad_ele.clear()
def _copies_columns(self):
- pyperclip.copy("old clipboard contents")
column = self.page.find_by_css_selector(
QueryToolLocators.output_column_header_css.format('some_column'))
column.click()
@@ -101,14 +117,23 @@ class CopySelectedQueryResultsFeatureTest(BaseFeatureTest):
QueryToolLocators.copy_button_css)
copy_button.click()
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
self.assertEqual(
"""\"Some-Name"
"Some-Other-Name"
"Yet-Another-Name\"""",
- pyperclip.paste())
+ clipboard_text)
+ scratch_pad_ele.clear()
def _copies_row_using_keyboard_shortcut(self):
- pyperclip.copy("old clipboard contents")
first_row = self.page.find_by_xpath(
QueryToolLocators.output_row_xpath.format(1))
first_row.click()
@@ -116,27 +141,43 @@ class CopySelectedQueryResultsFeatureTest(BaseFeatureTest):
ActionChains(self.page.driver).key_down(
Keys.CONTROL).send_keys('c').key_up(Keys.CONTROL).perform()
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
self.assertEqual('"Some-Name"\t6\t"some info"',
- pyperclip.paste())
+ clipboard_text)
+ scratch_pad_ele.clear()
def _copies_column_using_keyboard_shortcut(self):
- pyperclip.copy("old clipboard contents")
column = self.page.find_by_css_selector(
QueryToolLocators.output_column_header_css.format('some_column'))
column.click()
ActionChains(self.page.driver).key_down(
Keys.CONTROL).send_keys('c').key_up(Keys.CONTROL).perform()
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
self.assertEqual(
"""\"Some-Name"
"Some-Other-Name"
"Yet-Another-Name\"""",
- pyperclip.paste())
+ clipboard_text)
+ scratch_pad_ele.clear()
def _copies_rectangular_selection(self):
- pyperclip.copy("old clipboard contents")
-
top_left_cell = \
self.page.find_by_xpath(
QueryToolLocators.output_column_data_xpath.
@@ -154,12 +195,20 @@ class CopySelectedQueryResultsFeatureTest(BaseFeatureTest):
self.page.driver
).key_down(Keys.CONTROL).send_keys('c').key_up(Keys.CONTROL).perform()
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
self.assertEqual(
- '"Some-Other-Name"\t22\n"Yet-Another-Name"\t14', pyperclip.paste())
+ '"Some-Other-Name"\t22\n"Yet-Another-Name"\t14', clipboard_text)
+ scratch_pad_ele.clear()
def _shift_resizes_rectangular_selection(self):
- pyperclip.copy("old clipboard contents")
-
top_left_cell = self.page.find_by_xpath(
QueryToolLocators.output_column_data_xpath.
format('Some-Other-Name')
@@ -180,12 +229,20 @@ class CopySelectedQueryResultsFeatureTest(BaseFeatureTest):
Keys.CONTROL
).send_keys('c').key_up(Keys.CONTROL).perform()
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
self.assertEqual("""\"Some-Other-Name"\t22\t"some other info"
-"Yet-Another-Name"\t14\t"cool info\"""", pyperclip.paste())
+"Yet-Another-Name"\t14\t"cool info\"""", clipboard_text)
+ scratch_pad_ele.clear()
def _shift_resizes_column_selection(self):
- pyperclip.copy("old clipboard contents")
-
column = self.page.find_by_css_selector(
QueryToolLocators.output_column_header_css.format('value')
)
@@ -197,13 +254,21 @@ class CopySelectedQueryResultsFeatureTest(BaseFeatureTest):
ActionChains(self.page.driver).key_down(
Keys.CONTROL).send_keys('c').key_up(Keys.CONTROL).perform()
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
self.assertEqual(
'"Some-Name"\t6\n"Some-Other-Name"\t22\n"Yet-Another-Name"\t14',
- pyperclip.paste())
+ clipboard_text)
+ scratch_pad_ele.clear()
def _mouseup_outside_grid_still_makes_a_selection(self):
- pyperclip.copy("old clipboard contents")
-
bottom_right_cell = self.page.find_by_xpath(
QueryToolLocators.output_column_data_xpath.format('cool info')
)
@@ -218,7 +283,17 @@ class CopySelectedQueryResultsFeatureTest(BaseFeatureTest):
ActionChains(self.page.driver).key_down(
Keys.CONTROL).send_keys('c').key_up(Keys.CONTROL).perform()
- self.assertIn('"cool info"', pyperclip.paste())
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
+ self.assertIn('"cool info"', clipboard_text)
+ scratch_pad_ele.clear()
def after(self):
self.page.close_query_tool()
diff --git a/web/pgadmin/feature_tests/file_manager_test.py b/web/pgadmin/feature_tests/file_manager_test.py
index 6d012869b..33f86ad86 100644
--- a/web/pgadmin/feature_tests/file_manager_test.py
+++ b/web/pgadmin/feature_tests/file_manager_test.py
@@ -38,7 +38,8 @@ class CheckFileManagerFeatureTest(BaseFeatureTest):
self.page.add_server(self.server)
self.wait = WebDriverWait(self.page.driver, 10)
- self.XSS_FILE = '/tmp/
.sql'
+ self.XSS_FILE = '/tmp/.sql'
# Remove any previous file
if os.path.isfile(self.XSS_FILE):
os.remove(self.XSS_FILE)
@@ -67,7 +68,7 @@ class CheckFileManagerFeatureTest(BaseFeatureTest):
self.page.open_query_tool()
def _create_new_file(self):
- self.page.find_by_css_selector(QueryToolLocators.btn_save_file)\
+ self.page.find_by_css_selector(QueryToolLocators.btn_save_file) \
.click()
# Set the XSS value in input
self.page.find_by_css_selector('.change_file_types')
@@ -112,8 +113,8 @@ class CheckFileManagerFeatureTest(BaseFeatureTest):
self.page.wait_for_query_tool_loading_indicator_to_disappear()
self._check_escaped_characters(
contents,
- '<img src=x onmouseover=alert("1")>.sql',
- 'File manager'
+ '<img src=x ' + self.server['name'][:13] +
+ '=alert("1")>.sql', 'File manager'
)
def _check_escaped_characters(self, source_code, string_to_find, source):
diff --git a/web/pgadmin/feature_tests/pg_utilities_backup_restore_test.py b/web/pgadmin/feature_tests/pg_utilities_backup_restore_test.py
index 1d6d7a43e..98b8a38b8 100644
--- a/web/pgadmin/feature_tests/pg_utilities_backup_restore_test.py
+++ b/web/pgadmin/feature_tests/pg_utilities_backup_restore_test.py
@@ -18,6 +18,7 @@ from regression.python_test_utils import test_utils
from regression.python_test_utils import test_gui_helper
from regression.feature_utils.locators import NavMenuLocators
from regression.feature_utils.tree_area_locators import TreeAreaLocators
+from selenium.webdriver import ActionChains
class PGUtilitiesBackupFeatureTest(BaseFeatureTest):
@@ -56,6 +57,7 @@ class PGUtilitiesBackupFeatureTest(BaseFeatureTest):
self.server['sslmode']
)
test_utils.drop_database(connection, self.database_name)
+ self._update_preferences()
db_id = test_utils.create_database(self.server, self.database_name)
if not db_id:
self.assertTrue(False, "Database {} is not "
@@ -130,7 +132,7 @@ class PGUtilitiesBackupFeatureTest(BaseFeatureTest):
self._check_detailed_window_for_xss('Backup')
else:
command = self.page.find_by_css_selector(
- NavMenuLocators.process_watcher_detailed_command_canvas_css).\
+ NavMenuLocators.process_watcher_detailed_command_canvas_css). \
text
self.assertIn(self.server['name'], str(command))
@@ -199,7 +201,7 @@ class PGUtilitiesBackupFeatureTest(BaseFeatureTest):
self._check_detailed_window_for_xss('Restore')
else:
command = self.page.find_by_css_selector(
- NavMenuLocators.process_watcher_detailed_command_canvas_css).\
+ NavMenuLocators.process_watcher_detailed_command_canvas_css). \
text
self.assertIn(self.server['name'], str(command))
@@ -242,3 +244,71 @@ class PGUtilitiesBackupFeatureTest(BaseFeatureTest):
# For XSS we need to search against element's html code
assert source_code.find(string_to_find) != - \
1, "{0} might be vulnerable to XSS ".format(source)
+
+ def _update_preferences(self):
+ file_menu = self.page.find_by_css_selector(
+ NavMenuLocators.file_menu_css)
+ file_menu.click()
+
+ pref_menu_item = self.page.find_by_css_selector(
+ NavMenuLocators.preference_menu_item_css)
+ pref_menu_item.click()
+
+ wait = WebDriverWait(self.page.driver, 10)
+
+ # Wait till the preference dialogue box is displayed by checking the
+ # visibility of Show System Object label
+ wait.until(EC.presence_of_element_located(
+ (By.XPATH, NavMenuLocators.show_system_objects_pref_label_xpath))
+ )
+
+ maximize_button = self.page.find_by_css_selector(
+ NavMenuLocators.maximize_pref_dialogue_css)
+ maximize_button.click()
+
+ path = self.page.find_by_xpath(
+ NavMenuLocators.specified_preference_tree_node.format('Paths'))
+ if self.page.find_by_xpath(
+ NavMenuLocators.specified_pref_node_exp_status.format('Paths')). \
+ get_attribute('aria-expanded') == 'false':
+ ActionChains(self.driver).double_click(path).perform()
+
+ binary_path = self.page.find_by_xpath(
+ NavMenuLocators.specified_sub_node_of_pref_tree_node.format(
+ 'Paths', 'Binary paths'))
+ binary_path.click()
+
+ default_binary_path = self.server['default_binary_paths']
+ if default_binary_path is not None:
+ server_types = default_binary_path.keys()
+ for serv in server_types:
+ if serv == 'pg':
+ path_input = self.page.find_by_xpath(
+ "//label[text()='PostgreSQL Binary "
+ "Path']/following-sibling::div//input")
+ path_input.clear()
+ path_input.click()
+ path_input.send_keys(default_binary_path['pg'])
+ elif serv == 'gpdb':
+ path_input = self.page.find_by_xpath(
+ "//label[text()='Greenplum Database Binary "
+ "Path']/following-sibling::div//input")
+ path_input.clear()
+ path_input.click()
+ path_input.send_keys(default_binary_path['gpdb'])
+ elif serv == 'ppas':
+ path_input = self.page.find_by_xpath(
+ "//label[text()='EDB Advanced Server Binary "
+ "Path']/following-sibling::div//input")
+ path_input.clear()
+ path_input.click()
+ path_input.send_keys(default_binary_path['ppas'])
+ else:
+ print('Binary path Key is Incorrect')
+
+ # save and close the preference dialog.
+ self.page.click_modal('Save')
+
+ self.page.wait_for_element_to_disappear(
+ lambda driver: driver.find_element_by_css_selector(".ajs-modal")
+ )
diff --git a/web/pgadmin/feature_tests/query_tool_journey_test.py b/web/pgadmin/feature_tests/query_tool_journey_test.py
index 2d7c4ef98..ddb4e3caf 100644
--- a/web/pgadmin/feature_tests/query_tool_journey_test.py
+++ b/web/pgadmin/feature_tests/query_tool_journey_test.py
@@ -9,7 +9,6 @@
from __future__ import print_function
import sys
-import pyperclip
import random
from selenium.webdriver import ActionChains
@@ -90,7 +89,6 @@ class QueryToolJourneyTest(BaseFeatureTest):
print(" OK.", file=sys.stderr)
def _test_copies_rows(self):
- pyperclip.copy("old clipboard contents")
self.page.driver.switch_to.default_content()
self.page.driver.switch_to_frame(
self.page.driver.find_element_by_tag_name("iframe"))
@@ -103,12 +101,21 @@ class QueryToolJourneyTest(BaseFeatureTest):
QueryToolLocators.copy_button_css)
copy_row.click()
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
self.assertEqual('"Some-Name"\t6\t"some info"',
- pyperclip.paste())
+ clipboard_text)
- def _test_copies_columns(self):
- pyperclip.copy("old clipboard contents")
+ scratch_pad_ele.clear()
+ def _test_copies_columns(self):
self.page.driver.switch_to.default_content()
self.page.driver.switch_to_frame(
self.page.driver.find_element_by_tag_name("iframe"))
@@ -121,9 +128,20 @@ class QueryToolJourneyTest(BaseFeatureTest):
QueryToolLocators.copy_button_css)
copy_btn.click()
- self.assertTrue('"Some-Name"' in pyperclip.paste())
- self.assertTrue('"Some-Other-Name"' in pyperclip.paste())
- self.assertTrue('"Yet-Another-Name"' in pyperclip.paste())
+ self.page.driver.switch_to.default_content()
+ self.page.driver.switch_to_frame(
+ self.page.driver.find_element_by_tag_name("iframe"))
+
+ scratch_pad_ele = self.page.find_by_css_selector(
+ QueryToolLocators.scratch_pad_css)
+ self.page.paste_values(scratch_pad_ele)
+
+ clipboard_text = scratch_pad_ele.get_attribute("value")
+
+ self.assertTrue('"Some-Name"' in clipboard_text)
+ self.assertTrue('"Some-Other-Name"' in clipboard_text)
+ self.assertTrue('"Yet-Another-Name"' in clipboard_text)
+ scratch_pad_ele.clear()
def _test_history_tab(self):
self.page.clear_query_tool()
@@ -370,10 +388,10 @@ class QueryToolJourneyTest(BaseFeatureTest):
self.page.find_by_css_selector(
QueryToolLocators.btn_clear_dropdown)
)
- ActionChains(self.driver)\
+ ActionChains(self.driver) \
.move_to_element(
- self.page.find_by_css_selector(
- QueryToolLocators.btn_clear_history)).perform()
+ self.page.find_by_css_selector(
+ QueryToolLocators.btn_clear_history)).perform()
self.page.click_element(
self.page.find_by_css_selector(QueryToolLocators.btn_clear_history)
)
diff --git a/web/pgadmin/feature_tests/view_data_dml_queries.py b/web/pgadmin/feature_tests/view_data_dml_queries.py
index 0155cdc32..db788bf24 100644
--- a/web/pgadmin/feature_tests/view_data_dml_queries.py
+++ b/web/pgadmin/feature_tests/view_data_dml_queries.py
@@ -131,13 +131,15 @@ CREATE TABLE public.nonintpkey
self.test_db, 'public')
self._load_config_data('table_insert_update_cases')
+ data_local = config_data
# iterate on both tables
for cnt in (1, 2):
- self._perform_test_for_table('defaults_{0}'.format(str(cnt)))
-
+ self._perform_test_for_table('defaults_{0}'.format(str(cnt)),
+ data_local)
# test nonint pkey table
self._load_config_data('table_insert_update_nonint')
- self._perform_test_for_table('nonintpkey')
+ data_local = config_data
+ self._perform_test_for_table('nonintpkey', data_local)
def after(self):
self.page.remove_server(self.server)
@@ -167,7 +169,7 @@ CREATE TABLE public.nonintpkey
global config_data
config_data = config_data_json[config_key]
- def _perform_test_for_table(self, table_name):
+ def _perform_test_for_table(self, table_name, config_data_local):
self.page.click_a_tree_node(
table_name,
TreeAreaLocators.sub_nodes_of_tables_node)
@@ -176,19 +178,20 @@ CREATE TABLE public.nonintpkey
self.page.wait_for_query_tool_loading_indicator_to_disappear()
# Run test to insert a new row in table with default values
- self._add_row()
- self._verify_row_data(True, config_data['add'])
+ self._add_row(config_data_local)
+ self._verify_row_data(True, config_data_local['add'])
# Run test to copy/paste a row
- self._copy_paste_row()
+ self._copy_paste_row(config_data_local)
- self._update_row()
+ self._update_row(config_data_local)
self.page.click_tab("Messages")
self._verify_messsages("")
self.page.click_tab("Data Output")
updated_row_data = {
- i: config_data['update'][i] if i in config_data['update'] else val
- for i, val in config_data['add'].items()
+ i: config_data_local['update'][i] if i in config_data_local[
+ 'update'] else val
+ for i, val in config_data_local['add'].items()
}
self._verify_row_data(False, updated_row_data)
@@ -219,7 +222,6 @@ CREATE TABLE public.nonintpkey
Returns: None
"""
-
self.wait.until(EC.visibility_of_element_located(
(By.XPATH, xpath)), CheckForViewDataTest.TIMEOUT_STRING
)
@@ -236,7 +238,7 @@ CREATE TABLE public.nonintpkey
if value == 'clear':
cell_el.find_element_by_css_selector('input').clear()
else:
- ActionChains(self.driver).send_keys(value).\
+ ActionChains(self.driver).send_keys(value). \
send_keys(Keys.ENTER).perform()
elif cell_type in ['text', 'json', 'text[]', 'boolean[]']:
text_area_ele = self.page.find_by_css_selector(
@@ -288,7 +290,7 @@ CREATE TABLE public.nonintpkey
self.page.driver.find_element_by_tag_name('iframe')
)
- def _copy_paste_row(self):
+ def _copy_paste_row(self, config_data_l):
row0_cell0_xpath = CheckForViewDataTest._get_cell_xpath("r0", 1)
self.page.find_by_xpath(row0_cell0_xpath).click()
@@ -298,12 +300,12 @@ CREATE TABLE public.nonintpkey
QueryToolLocators.paste_button_css).click()
# Update primary key of copied cell
- self._add_update_save_row(config_data['copy'], row=2)
+ self._add_update_save_row(config_data_l['copy'], row=2)
# Verify row 1 and row 2 data
updated_row_data = {
- i: config_data['copy'][i] if i in config_data['copy'] else val
- for i, val in config_data['add'].items()
+ i: config_data_l['copy'][i] if i in config_data_l['copy'] else val
+ for i, val in config_data_l['add'].items()
}
self._verify_row_data(False, updated_row_data)
@@ -326,11 +328,11 @@ CREATE TABLE public.nonintpkey
# save ajax is completed.
time.sleep(2)
- def _add_row(self):
- self._add_update_save_row(config_data['add'], 1)
+ def _add_row(self, config_data_l):
+ self._add_update_save_row(config_data_l['add'], 1)
- def _update_row(self):
- self._add_update_save_row(config_data['update'], 1)
+ def _update_row(self, config_data_l):
+ self._add_update_save_row(config_data_l['update'], 1)
def _verify_messsages(self, text):
messages_ele = self.page.find_by_css_selector(
diff --git a/web/pgadmin/feature_tests/xss_checks_panels_and_query_tool_test.py b/web/pgadmin/feature_tests/xss_checks_panels_and_query_tool_test.py
index 9efad8a36..69c3cba80 100644
--- a/web/pgadmin/feature_tests/xss_checks_panels_and_query_tool_test.py
+++ b/web/pgadmin/feature_tests/xss_checks_panels_and_query_tool_test.py
@@ -211,17 +211,23 @@ class CheckForXssFeatureTest(BaseFeatureTest):
"Query tool (History Entry)"
)
- # Check for history details message
- history_ele = self.driver\
- .find_element_by_css_selector(".query-detail .content-value")
-
- source_code = history_ele.get_attribute('innerHTML')
+ retry = 2
+ while retry > 0:
+ try:
+ history_ele = self.driver \
+ .find_element_by_css_selector(
+ ".query-detail .content-value")
+ source_code = history_ele.get_attribute('innerHTML')
+ break
+ except StaleElementReferenceException:
+ retry -= 1
self._check_escaped_characters(
source_code,
'<script>alert(1)</script>',
"Query tool (History Details-Message)"
)
+
retry = 2
while retry > 0:
try:
diff --git a/web/pgadmin/utils/route.py b/web/pgadmin/utils/route.py
index 35493e95d..98be31328 100644
--- a/web/pgadmin/utils/route.py
+++ b/web/pgadmin/utils/route.py
@@ -118,9 +118,8 @@ class BaseTestGenerator(unittest.TestCase):
self.skipTest('cannot run in: %s' %
server_con['data']['type'])
- @classmethod
- def setTestServer(cls, server):
- cls.server = server
+ def setTestServer(self, server):
+ self.server = server
@abstractmethod
def runTest(self):
@@ -135,17 +134,14 @@ class BaseTestGenerator(unittest.TestCase):
def setTestClient(cls, test_client):
cls.tester = test_client
- @classmethod
- def setDriver(cls, driver):
- cls.driver = driver
+ def setDriver(self, driver):
+ self.driver = driver
- @classmethod
- def setServerInformation(cls, server_information):
- cls.server_information = server_information
+ def setServerInformation(self, server_information):
+ self.server_information = server_information
- @classmethod
- def setTestDatabaseName(cls, database_name):
- cls.test_db = database_name
+ def setTestDatabaseName(self, database_name):
+ self.test_db = database_name
@classmethod
def setReSQLModuleList(cls, module_list):
diff --git a/web/regression/README b/web/regression/README
index a4591acb2..314e58da9 100644
--- a/web/regression/README
+++ b/web/regression/README
@@ -141,6 +141,28 @@ Python Tests:
and registered automatically by its module name in
'pgadmin4/web/pgadmin/utils/test.py' file.
+- To run Feature Tests in parallel using selenoid(grid + docker), selenoid
+ need to be installed. Steps to install selenoid -
+ - Install & Start docker
+ $yum -y install docker docker-registry
+ $vi /etc/sysconfig/docker # in OPTIONS add ‘--selinux-enabled=false’
+ $systemctl enable docker.service
+ $systemctl start docker.service
+ $systemctl status docker.service
+ - Install & Start Selenoid
+ $curl -s https://aerokube.com/cm/bash | bash
+ $./cm selenoid start --vnc -g -limit 3 -cpu 1.5
+ $./cm selenoid-ui start
+ Check selenoid status -
+ http://:4444/status
+ - Should show json with browsers details
+ http://:8080/#/
+ - Capabilities shows available browser
+ - Update config_local.py if exists else create new with 'DEFAULT_SERVER' value
+ DEFAULT_SERVER = ''
+ - Update 'test_config.json' with selenoid related info like selenoid url
+ & browser details where tests need to be executed
+
- Change to the regression test directory:
run 'cd web/regression'
@@ -190,6 +212,9 @@ Python Tests:
Example 2) Execute only reverse engineered SQL test framework for some modules
run 'python runtests.py --pkg resql --modules sequences,functions'
+- Execute ui selenium tests in parallel using selenoid(selenium grid + docker)
+ Example : --pkg feature_tests --parallel
+
Code Coverage:
---------------
diff --git a/web/regression/feature_utils/app_starter.py b/web/regression/feature_utils/app_starter.py
index fe4c441a1..7de10b25c 100644
--- a/web/regression/feature_utils/app_starter.py
+++ b/web/regression/feature_utils/app_starter.py
@@ -61,11 +61,16 @@ class AppStarter:
raise Exception('Unable to start python server even after '
'retrying 60 times.')
- launch_browser(0)
+ if self.driver is not None:
+ launch_browser(0)
+ else:
+ return "http://" + self.app_config.DEFAULT_SERVER + ":" \
+ + random_server_port
def stop_app(self):
""" This function stop the started app by killing process """
- self.driver.quit()
+ if self.driver is not None:
+ self.driver.quit()
# os.killpg supported in Mac and Unix as this function not supported in
# Windows
try:
diff --git a/web/regression/feature_utils/locators.py b/web/regression/feature_utils/locators.py
index b8169fa59..0df8e61dd 100644
--- a/web/regression/feature_utils/locators.py
+++ b/web/regression/feature_utils/locators.py
@@ -172,6 +172,8 @@ class QueryToolLocators:
new_row_xpath = "//div[contains(@class, 'new-row')]"
+ scratch_pad_css = ".sql-scratch > textarea"
+
copy_button_css = "#btn-copy-row"
paste_button_css = "#btn-paste-row"
diff --git a/web/regression/feature_utils/pgadmin_page.py b/web/regression/feature_utils/pgadmin_page.py
index aef031fd0..4d6f15e50 100644
--- a/web/regression/feature_utils/pgadmin_page.py
+++ b/web/regression/feature_utils/pgadmin_page.py
@@ -88,11 +88,17 @@ class PgadminPage:
(By.CSS_SELECTOR, "button[type='save'].btn.btn-primary")))
self.find_by_css_selector("button[type='save'].btn.btn-primary").\
click()
-
- WebDriverWait(self.driver, 10).until(
- EC.visibility_of_element_located(
- (By.XPATH,
- "//*[@id='tree']//*[.='" + server_config['name'] + "']")))
+ try:
+ WebDriverWait(self.driver, 10).until(
+ EC.visibility_of_element_located(
+ (By.XPATH,
+ "//*[@id='tree']//*[.='" + server_config['name'] + "']")))
+ except TimeoutException:
+ self.toggle_open_servers_group()
+ WebDriverWait(self.driver, 10).until(
+ EC.visibility_of_element_located(
+ (By.XPATH,
+ "//*[@id='tree']//*[.='" + server_config['name'] + "']")))
def open_query_tool(self):
self.driver.find_element_by_link_text("Tools").click()
@@ -910,7 +916,11 @@ class PgadminPage:
return element
except (NoSuchElementException, WebDriverException):
return False
-
+ time.sleep(1)
+ self.driver.switch_to.default_content()
+ self.driver.switch_to_frame(
+ self.driver.find_element_by_tag_name("iframe"))
+ self.find_by_xpath("//a[text()='Query Editor']").click()
codemirror_ele = WebDriverWait(
self.driver, timeout=self.timeout, poll_frequency=0.01)\
.until(find_codemirror,
@@ -1161,3 +1171,12 @@ class PgadminPage:
except Exception:
attempt += 1
return click_status
+
+ def paste_values(self, el=None):
+ actions = ActionChains(self.driver)
+ if el:
+ el.click()
+ actions.key_down(Keys.SHIFT)
+ actions.send_keys(Keys.INSERT)
+ actions.key_up(Keys.SHIFT)
+ actions.perform()
diff --git a/web/regression/python_test_utils/test_utils.py b/web/regression/python_test_utils/test_utils.py
index 971fc2403..71b75df7b 100644
--- a/web/regression/python_test_utils/test_utils.py
+++ b/web/regression/python_test_utils/test_utils.py
@@ -17,6 +17,13 @@ import sqlite3
import shutil
from functools import partial
from testtools.testcase import clone_test_with_new_id
+import re
+import time
+from selenium.common.exceptions import WebDriverException
+import urllib.request as urllib
+import json
+from selenium import webdriver
+from selenium.webdriver.chrome.options import Options
import config
import regression
@@ -1216,3 +1223,144 @@ def create_expected_output(parameters, actual_data):
actual_data.remove(value)
break
return expected_output
+
+
+def is_parallel_ui_tests(args):
+ """
+ This function checks for coverage args exists in command line args
+ :return: boolean
+ """
+ if "parallel" in args and args["parallel"]:
+ return True
+ return False
+
+
+def get_selenium_grid_status_and_browser_list(selrnoid_url):
+ """
+ This function checks selenoid status for given url
+ :param selrnoid_url:
+ :return: status of selenoid & list of browsers available with selenoid if
+ status is up
+ """
+ selenoid_status = False
+ browser_list = []
+ try:
+ selenoid_status = urllib.urlopen(
+ "http://" + re.split('/', (re.split('//', selrnoid_url, 1)[1]))[
+ 0] + "/status", timeout=10)
+ selenoid_status = json.load(selenoid_status)
+ if isinstance(selenoid_status, dict):
+ available_browsers = selenoid_status["browsers"]
+ selenoid_status = True
+
+ list_of_browsers = test_setup.config_data['selenoid_info'][
+ 'cross_Browsers']
+
+ for browser in list_of_browsers:
+ if browser["name"].lower() in available_browsers.keys():
+ versions = available_browsers[(browser["name"].lower())]
+ if browser["version"] in versions.keys():
+ browser_list.append(browser)
+ else:
+ print("Available {0} versions {1}".format(browser["name"],
+ versions.keys()))
+ print("Specified Version = {0}".format(browser["version"]))
+ else:
+ print("{0} is NOT available".format(browser["name"]))
+ except Exception as e:
+ print(str(e))
+ print("Unable to find Selenoid Status")
+
+ return selenoid_status, browser_list
+
+
+def is_feature_test_included(arguments):
+ """
+ :param arguments: his is command line arguments for module name to
+ which test suite will run
+ :return: boolean value whether to execute feature tests or NOT &
+ browser name if feature_test_tobe_included = True
+ """
+ exclude_pkgs = []
+ if arguments['exclude'] is not None:
+ exclude_pkgs += arguments['exclude'].split(',')
+
+ feature_test_tobe_included = 'feature_tests' not in exclude_pkgs and \
+ (arguments['pkg'] is None or arguments[
+ 'pkg'] == "all" or
+ arguments['pkg'] == "feature_tests")
+ return feature_test_tobe_included
+
+
+def launch_url_in_browser(driver_instance, url, retry_count):
+ """
+ Function launches urls in sepecided driver instance
+ :param driver_instance:browser instance
+ :param url:url to be launched
+ :param retry_count:
+ :return:
+ """
+ try:
+ driver_instance.get(url)
+ except WebDriverException as e:
+ # In case of WebDriverException sleep for 1 second and retry
+ # again. Retry 10 times and if still app will not start then
+ # raise exception.
+ time.sleep(1)
+ if retry_count < 60:
+ retry_count = retry_count + 1
+ launch_url_in_browser(driver_instance, url, retry_count)
+ else:
+ raise Exception('Unable to start python server even after '
+ 'retrying 60 times.')
+
+
+def get_remote_webdriver(browser, browser_ver, test_name):
+ hub_url = test_setup.config_data['selenoid_info']['selenoid_url']
+ test_name = browser + browser_ver + "_" + test_name + "-" + time.strftime(
+ "%m_%d_%y_%H_%M_%S", time.localtime())
+ driver_local = None
+
+ desired_capabilities = {
+ "version": browser_ver,
+ "enableVNC": True,
+ "enableVideo": True,
+ "enableLog": True,
+ "videoName": test_name + ".mp4",
+ "logName": test_name + ".log",
+ "name": test_name,
+ "timeZone": "Asia/Kolkata"
+ }
+
+ if browser == 'firefox':
+ profile = webdriver.FirefoxProfile()
+ profile.set_preference("dom.disable_beforeunload", True)
+ desired_capabilities["browserName"] = "firefox"
+ desired_capabilities["requireWindowFocus"] = True
+ desired_capabilities["enablePersistentHover"] = False
+ driver_local = webdriver.Remote(
+ command_executor=hub_url,
+ desired_capabilities=desired_capabilities, browser_profile=profile)
+ driver_local.implicitly_wait(1)
+
+ elif browser == 'chrome':
+ options = Options()
+ if test_setup.config_data:
+ if 'headless_chrome' in test_setup.config_data:
+ if test_setup.config_data['headless_chrome']:
+ options.add_argument("--headless")
+ options.add_argument("--no-sandbox")
+ options.add_argument("--disable-setuid-sandbox")
+ options.add_argument("--window-size=1280,1024")
+ options.add_argument("--disable-infobars")
+ desired_capabilities["browserName"] = "chrome"
+ driver_local = webdriver.Remote(
+ command_executor=hub_url,
+ desired_capabilities=desired_capabilities, options=options)
+ else:
+ print("Specified browser does not exist.")
+
+ # maximize browser window
+ driver_local.maximize_window()
+ driver_local.implicitly_wait(2)
+ return driver_local
diff --git a/web/regression/runtests.py b/web/regression/runtests.py
index fcf73a886..d41c962a2 100644
--- a/web/regression/runtests.py
+++ b/web/regression/runtests.py
@@ -21,7 +21,8 @@ import traceback
import json
import random
import coverage
-
+import threading
+import time
import unittest
if sys.version_info[0] >= 3:
@@ -136,7 +137,7 @@ scenarios.apply_scenario = test_utils.apply_scenario
def get_suite(module_list, test_server, test_app_client, server_information,
- test_db_name):
+ test_db_name, driver_passed):
"""
This function add the tests to test suite and return modified test suite
variable.
@@ -166,7 +167,7 @@ def get_suite(module_list, test_server, test_app_client, server_information,
obj.setApp(app)
obj.setTestClient(test_app_client)
obj.setTestServer(test_server)
- obj.setDriver(driver)
+ obj.setDriver(driver_passed)
obj.setServerInformation(server_information)
obj.setTestDatabaseName(test_db_name)
scenario = scenarios.generate_scenarios(obj)
@@ -207,57 +208,59 @@ def get_test_modules(arguments):
exclude_pkgs += arguments['exclude'].split(',')
if 'feature_tests' not in exclude_pkgs and \
- (arguments['pkg'] is None or arguments['pkg'] == "all" or
- arguments['pkg'] == "feature_tests"):
-
- from selenium import webdriver
- from selenium.webdriver.chrome.options import Options
- from selenium.webdriver.common.desired_capabilities import \
- DesiredCapabilities
-
- default_browser = 'chrome'
-
- # Check default browser provided through command line. If provided
- # then use that browser as default browser else check for the setting
- # provided in test_config.json file.
- if (
- 'default_browser' in arguments and
- arguments['default_browser'] is not None
- ):
- default_browser = arguments['default_browser'].lower()
- elif (
- test_setup.config_data and
- "default_browser" in test_setup.config_data
- ):
- default_browser = test_setup.config_data['default_browser'].lower()
-
- if default_browser == 'firefox':
- cap = DesiredCapabilities.FIREFOX
- cap['requireWindowFocus'] = True
- cap['enablePersistentHover'] = False
- profile = webdriver.FirefoxProfile()
- profile.set_preference("dom.disable_beforeunload", True)
- driver = webdriver.Firefox(capabilities=cap,
- firefox_profile=profile)
- driver.implicitly_wait(1)
- else:
- options = Options()
- if test_setup.config_data:
- if 'headless_chrome' in test_setup.config_data:
- if test_setup.config_data['headless_chrome']:
- options.add_argument("--headless")
- options.add_argument("--no-sandbox")
- options.add_argument("--disable-setuid-sandbox")
- options.add_argument("--window-size=1280,1024")
- options.add_argument("--disable-infobars")
- options.add_experimental_option('w3c', False)
- driver = webdriver.Chrome(chrome_options=options)
-
- # maximize browser window
- driver.maximize_window()
-
- app_starter = AppStarter(driver, config)
- app_starter.start_app()
+ (arguments['pkg'] is None or arguments['pkg'] == "all" or
+ arguments['pkg'] == "feature_tests"):
+
+ if not test_utils.is_parallel_ui_tests(args):
+ from selenium import webdriver
+ from selenium.webdriver.chrome.options import Options
+ from selenium.webdriver.common.desired_capabilities import \
+ DesiredCapabilities
+
+ default_browser = 'chrome'
+
+ # Check default browser provided through command line. If provided
+ # then use that browser as default browser else check for the
+ # settingn provided in test_config.json file.
+ if (
+ 'default_browser' in arguments and
+ arguments['default_browser'] is not None
+ ):
+ default_browser = arguments['default_browser'].lower()
+ elif (
+ test_setup.config_data and
+ "default_browser" in test_setup.config_data
+ ):
+ default_browser = test_setup.config_data[
+ 'default_browser'].lower()
+
+ if default_browser == 'firefox':
+ cap = DesiredCapabilities.FIREFOX
+ cap['requireWindowFocus'] = True
+ cap['enablePersistentHover'] = False
+ profile = webdriver.FirefoxProfile()
+ profile.set_preference("dom.disable_beforeunload", True)
+ driver = webdriver.Firefox(capabilities=cap,
+ firefox_profile=profile)
+ driver.implicitly_wait(1)
+ else:
+ options = Options()
+ if test_setup.config_data:
+ if 'headless_chrome' in test_setup.config_data:
+ if test_setup.config_data['headless_chrome']:
+ options.add_argument("--headless")
+ options.add_argument("--no-sandbox")
+ options.add_argument("--disable-setuid-sandbox")
+ options.add_argument("--window-size=1280,1024")
+ options.add_argument("--disable-infobars")
+ options.add_experimental_option('w3c', False)
+ driver = webdriver.Chrome(chrome_options=options)
+
+ # maximize browser window
+ driver.maximize_window()
+
+ app_starter = AppStarter(driver, config)
+ app_starter.start_app()
handle_cleanup = test_utils.get_cleanup_handler(test_client, app_starter)
# Register cleanup function to cleanup on exit
@@ -319,6 +322,9 @@ def add_arguments():
'--modules',
help='Executes the feature test for specific modules in pkg'
)
+ parser.add_argument('--parallel', nargs='?', const=True,
+ type=bool, default=False,
+ help='Enable parallel Feature Tests')
arg = parser.parse_args()
return arg
@@ -404,6 +410,115 @@ class StreamToLogger(object):
pass
+driver_info = []
+
+
+def run_test(test_module_list_passed, server_passed, driver_passed):
+ try:
+ print("\n=============Running the test cases for '%s' ============="
+ % server_passed['name'], file=sys.stderr)
+
+ # Store driver object info & thread id
+ driver_info.append((threading.current_thread().ident, driver_passed))
+
+ # Create test server
+ server_information = \
+ test_utils.create_parent_server_node(server_passed)
+
+ # Create test database with random number to avoid conflict in
+ # parallel execution on different platforms. This database will be
+ # used across all feature tests.
+ test_db_name = "acceptance_test_db" + \
+ str(random.randint(10000, 65535))
+ connection = test_utils.get_db_connection(
+ server_passed['db'],
+ server_passed['username'],
+ server_passed['db_password'],
+ server_passed['host'],
+ server_passed['port'],
+ server_passed['sslmode']
+ )
+
+ # Add the server version in server information
+ server_information['server_version'] = connection.server_version
+ server_information['type'] = server_passed['type']
+
+ # Drop the database if already exists.
+ test_utils.drop_database(connection, test_db_name)
+
+ # Create database
+ test_utils.create_database(server_passed, test_db_name)
+
+ # Configure preferences for the test cases
+ test_utils.configure_preferences(
+ default_binary_path=server_passed['default_binary_paths'])
+
+ # Get unit test suit
+ suite = get_suite(test_module_list_passed,
+ server_passed,
+ test_client,
+ server_information, test_db_name, driver_passed)
+
+ # Run unit test suit created
+ tests = unittest.TextTestRunner(stream=sys.stderr,
+ descriptions=True,
+ verbosity=2).run(suite)
+
+ # processing results
+ ran_tests, failed_cases, skipped_cases, passed_cases = \
+ get_tests_result(tests)
+
+ # This is required when some tests are running parallel
+ # & some sequential in case of parallel ui tests
+ if thread.getName() == "sequential_tests":
+ try:
+ if test_result[server_passed['name']][0] is not None:
+ ran_tests = test_result[server_passed['name']][0] + \
+ ran_tests
+ failed_cases.update(test_result[server_passed['name']][1])
+ skipped_cases.update(test_result[server_passed['name']][2])
+ passed_cases.update(test_result[server_passed['name']][3])
+ test_result[server_passed['name']] = [ran_tests, failed_cases,
+ skipped_cases,
+ passed_cases]
+ except KeyError:
+ pass
+
+ # Add final results server wise in test_result dict
+ test_result[server_passed['name']] = [ran_tests, failed_cases,
+ skipped_cases, passed_cases]
+
+ # Set empty list for 'passed' parameter for each testRun.
+ # So that it will not append same test case name
+ unittest.result.TestResult.passed = []
+
+ # Drop the testing database created initially
+ if connection:
+ test_utils.drop_database(connection, test_db_name)
+ connection.close()
+
+ # Delete test server
+ test_utils.delete_test_server(test_client)
+ except Exception:
+ traceback.print_exc(file=sys.stderr)
+ print("Exception in {0}".format(threading.current_thread().ident))
+ finally:
+ # Delete web-driver instance
+ thread_name = "parallel_tests" + server_passed['name']
+ if threading.currentThread().getName() == thread_name:
+ for d in driver_info:
+ if d[0] == threading.currentThread().ident:
+ d[1].quit()
+ driver_info.remove(d)
+ # Need time by selenoid to clean previous thread before new thread
+ time.sleep(10)
+
+ # Print info about completed tests
+ print(
+ "\n=============Completed the test cases for '%s'============="
+ % server_passed['name'], file=sys.stderr)
+
+
if __name__ == '__main__':
# Failure detected?
failure = False
@@ -423,7 +538,8 @@ if __name__ == '__main__':
fh = logging.FileHandler(CURRENT_PATH + '/' +
'regression.log', 'w', 'utf-8')
fh.setLevel(logging.DEBUG)
- fh.setFormatter(logging.Formatter(config.FILE_LOG_FORMAT))
+ fh.setFormatter(logging.Formatter('[%(thread)d] ' +
+ config.FILE_LOG_FORMAT))
logger = logging.getLogger()
logger.addHandler(fh)
@@ -451,69 +567,141 @@ if __name__ == '__main__':
cov = coverage.Coverage(config_file=COVERAGE_CONFIG_FILE)
cov.start()
- try:
- for server in servers_info:
- print("\n=============Running the test cases for '%s'============="
- % server['name'], file=sys.stderr)
- # Create test server
- server_information = test_utils.create_parent_server_node(server)
-
- # Create test database with random number to avoid conflict in
- # parallel execution on different platforms. This database will be
- # used across all feature tests.
- test_db_name = "acceptance_test_db" + \
- str(random.randint(10000, 65535))
- connection = test_utils.get_db_connection(
- server['db'],
- server['username'],
- server['db_password'],
- server['host'],
- server['port'],
- server['sslmode']
- )
-
- # Add the server version in server information
- server_information['server_version'] = connection.server_version
- server_information['type'] = server['type']
-
- # Drop the database if already exists.
- test_utils.drop_database(connection, test_db_name)
- # Create database
- test_utils.create_database(server, test_db_name)
- # Configure preferences for the test cases
- test_utils.configure_preferences(
- default_binary_path=server['default_binary_paths'])
-
- suite = get_suite(test_module_list,
- server,
- test_client,
- server_information, test_db_name)
- tests = unittest.TextTestRunner(stream=sys.stderr,
- descriptions=True,
- verbosity=2).run(suite)
-
- ran_tests, failed_cases, skipped_cases, passed_cases = \
- get_tests_result(tests)
- test_result[server['name']] = [ran_tests, failed_cases,
- skipped_cases, passed_cases]
-
- # Set empty list for 'passed' parameter for each testRun.
- # So that it will not append same test case name
- unittest.result.TestResult.passed = []
-
- if len(failed_cases) > 0:
- failure = True
-
- # Drop the testing database created initially
- if connection:
- test_utils.drop_database(connection, test_db_name)
- connection.close()
-
- # Delete test server
- test_utils.delete_test_server(test_client)
- except SystemExit:
- if handle_cleanup:
- handle_cleanup()
+ # Check if feature tests included & parallel tests switch passed
+ if test_utils.is_feature_test_included(args) and \
+ test_utils.is_parallel_ui_tests(args):
+
+ # Get selenium grid status & list of available browser out passed
+ selenium_grid_status, list_of_browsers \
+ = test_utils.get_selenium_grid_status_and_browser_list(
+ test_setup.config_data['selenoid_info']['selenoid_url'])
+ # list of files consisting tests that needs to be
+ # executed sequentially
+ sequential_tests_file = [
+ 'pgadmin.feature_tests.pg_utilities_backup_restore_test',
+ 'pgadmin.feature_tests.pg_utilities_maintenance_test',
+ 'pgadmin.feature_tests.keyboard_shortcut_test',
+ 'regression.re_sql.tests.test_resql']
+ # If selenium grid is up & browsers are available
+ if selenium_grid_status and len(list_of_browsers) > 0:
+
+ try:
+ # run across browsers
+ for browser_info in list_of_browsers:
+ # browser info
+ browser_name = browser_info["name"].lower()
+ browser_version = browser_info["version"]
+
+ # list of tests can be executed in parallel
+ parallel_tests = list(test_module_list)
+ for module in test_module_list:
+ if str(module[0]) in sequential_tests_file:
+ parallel_tests.remove(module)
+
+ # list of tests can be executed in sequentially
+ sequential_tests = list(
+ filter(lambda i: i not in parallel_tests,
+ test_module_list))
+
+ print(
+ "===================================================="
+ "==================\n",
+ file=sys.stderr
+ )
+ print(
+ "Total Tests # {0}\nParallel Tests # {1}, Sequential "
+ "Tests # {2}".format(
+ len(test_module_list), len(parallel_tests),
+ len(sequential_tests)),
+ file=sys.stderr)
+ print("Browser: [Name:{0}, Version: {1}]".format(
+ browser_name.capitalize(), browser_version),
+ file=sys.stderr)
+ print(
+ "===================================================="
+ "==================\n",
+ file=sys.stderr
+ )
+
+ # Running Parallel tests
+ if len(parallel_tests) > 0:
+ # Create app form source code
+ app_starter_local = AppStarter(None, config)
+ client_url = app_starter_local.start_app()
+
+ try:
+ for srv in servers_info:
+ # Get driver instance at server level
+ driver_obj = test_utils.get_remote_webdriver(
+ browser_name,
+ browser_version,
+ srv['name'])
+ test_utils.launch_url_in_browser(driver_obj,
+ client_url, 0)
+
+ threads = []
+ t_name = "parallel_tests" + srv['name']
+ thread = threading.Thread(target=run_test,
+ name=t_name,
+ args=(
+ parallel_tests,
+ srv, driver_obj))
+ threads.append(thread)
+ thread.start()
+ for thread in threads:
+ thread.join()
+ except Exception:
+ traceback.print_exc(file=sys.stderr)
+ print("Exception in thread {0}".format(
+ thread.ident()))
+
+ # Sequential Tests
+ if len(sequential_tests) > 0:
+ # Get driver instance
+ driver_obj2 = test_utils.get_remote_webdriver(
+ browser_name,
+ browser_version,
+ "Sequential_Tests")
+ test_utils.launch_url_in_browser(driver_obj2,
+ client_url, 0)
+ try:
+ t_name = "sequential_tests"
+ for srv in servers_info:
+ thread = threading.Thread(target=run_test,
+ name=t_name,
+ args=(
+ sequential_tests,
+ srv,
+ driver_obj2))
+ thread.start()
+ thread.join()
+ driver_obj2.quit()
+ except Exception:
+ traceback.print_exc(file=sys.stderr)
+ print("Exception in thread {0}".format(
+ thread.ident()))
+
+ # Clean up environmetn
+ if app_starter_local:
+ app_starter_local.stop_app()
+ except SystemExit:
+ if app_starter_local:
+ app_starter_local.stop_app()
+ if handle_cleanup:
+ handle_cleanup()
+ else:
+ try:
+ for server in servers_info:
+ thread = threading.Thread(target=run_test, args=(
+ test_module_list, server, driver))
+ thread.start()
+ thread.join()
+ except SystemExit:
+ if handle_cleanup:
+ handle_cleanup()
+
+ # Pause before printing result in order not to mix output
+ time.sleep(3)
print(
"\n==============================================================="
@@ -542,6 +730,8 @@ if __name__ == '__main__':
skipped_cases.items()).values())
total_passed_cases = int(
test_result[server_res][0]) - total_failed - total_skipped
+ if len(failed_cases) > 0:
+ failure = True
print(
"%s:\n\n\t%s test%s passed\n\t%s test%s failed%s%s"
diff --git a/web/regression/test_config.json.in b/web/regression/test_config.json.in
index 0a151e633..07bfacd7b 100644
--- a/web/regression/test_config.json.in
+++ b/web/regression/test_config.json.in
@@ -1,6 +1,8 @@
{
"headless_chrome": false,
"default_browser": "Chrome",
+ "selenoid_info":{"cross_Browsers": [{"name": "browser_name","version": "browserr_version"}],
+ "selenoid_url": "Selenoid Url"},
"pgAdmin4_login_credentials": {
"new_password": "NEWPASSWORD",
"login_password": "PASSWORD",