diff --git a/web/pgadmin/feature_tests/test_data.json b/web/pgadmin/feature_tests/test_data.json new file mode 100644 index 0000000..2f60de6 --- /dev/null +++ b/web/pgadmin/feature_tests/test_data.json @@ -0,0 +1,18 @@ +{ + "table_insert_update_cases": { + "add_update": { + "1": [1, 1, "int", "Value at 0 index is actual value(to be inserted), Value at index 1 is expected value, int is column type"], + "2": ["", 1, "int"], + "3": ["", "[null]", "int"], + "4": ["", "Hello World", "text"], + "5": ["\"\"", "", "text", "Two double quotes"], + "6": ["\\\"\\\"", "\"\"", "text", "double backslash followed by a double quote"], + "7": ["\\\\\"\\\\\"", "\\\"\\\"", "text", "double backslash followed by a double quote"], + "8": ["", "[null]", "text"], + "9": ["", "[51,52]", "json"], + "10": ["[61,62]", "[61,62]", "json"], + "11": ["", "true", "bool"], + "12": ["", "[null]", "bool"] + } + } +} \ No newline at end of file diff --git a/web/pgadmin/feature_tests/view_data_dml_queries.py b/web/pgadmin/feature_tests/view_data_dml_queries.py new file mode 100644 index 0000000..a7ebd17 --- /dev/null +++ b/web/pgadmin/feature_tests/view_data_dml_queries.py @@ -0,0 +1,277 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2017, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import os +from selenium.webdriver import ActionChains +from regression.python_test_utils import test_utils +from regression.feature_utils.base_feature_test import BaseFeatureTest +import time +from selenium.webdriver.common.by import By +from selenium.webdriver.support import expected_conditions as EC +from selenium.webdriver.support.ui import WebDriverWait + + +CURRENT_PATH = os.path.dirname(os.path.realpath(__file__)) + +try: + with open(CURRENT_PATH + '/test_data.json') as data_file: + config_data = json.load(data_file)['table_insert_update_cases']['add_update'] +except Exception as e: + print(str(e)) + + +class CheckForViewDataTest(BaseFeatureTest): + """ + Test cases to validate insert, update operations in table + with input test data + + First of all, the test data is inserted/updated into table and then + inserted data is compared with original data to check if expected data + is returned from table or not. + + We will cover test cases for, + 1) Insert with default values + 2) Update with null values + 3) Update with blank string + 4) Copy/Paste row + """ + + scenarios = [ + ("Validate Insert, Update operations in View data with given test " + "data", + dict()) + ] + + TIMEOUT_STRING = "Timed out waiting for div element to appear" + + # query for creating 'defaults_text' table + defaults_query = """ +CREATE TABLE public.defaults +( + id serial NOT NULL, + number_defaults numeric(100) DEFAULT 1, + number_null numeric(100), + text_defaults text COLLATE pg_catalog."default" DEFAULT 'Hello World'::text, + text_null1 text COLLATE pg_catalog."default", + text_null2 text COLLATE pg_catalog."default", + text_null3 text COLLATE pg_catalog."default", + text_null4 text COLLATE pg_catalog."default", + json_defaults json DEFAULT '[51, 52]'::json, + json_null json, + boolean_defaults boolean DEFAULT true, + boolean_null boolean, + CONSTRAINT defaults_pkey PRIMARY KEY (id) +) +""" + + def before(self): + connection = test_utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + test_utils.drop_database(connection, "acceptance_test_db") + test_utils.create_database(self.server, "acceptance_test_db") + + # Create pre-requisite table + test_utils.create_table_with_query( + self.server, + "acceptance_test_db", + CheckForViewDataTest.defaults_query) + # Initialize an instance of WebDriverWait with timeout of 3 seconds + self.wait = WebDriverWait(self.driver, 3) + + def runTest(self): + self.page.wait_for_spinner_to_disappear() + self.page.add_server(self.server) + self._tables_node_expandable() + self.driver.switch_to.default_content() + self.page.select_tree_item('defaults') + # Open Object -> View data + self._view_data_grid() + + # Run test to insert a new row in table with default values + self._add_row() + self._verify_row_data(True) + + # Run test to copy/paste a row + self._copy_paste_row() + self.page.close_query_tool() + + def after(self): + self.page.remove_server(self.server) + connection = test_utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + test_utils.drop_database(connection, "acceptance_test_db") + + @staticmethod + def _get_cell_xpath(cell, row): + if str(row) == 'new-row' or str(row) == 'active': + xpath_grid_row = '//*[@id="0"]//*[@id="datagrid"]' \ + '//div[contains(@class, "'+row+'")]' + else: + xpath_grid_row = '//*[@id="0"]//*[@id="datagrid"]//div[' \ + 'contains(@class, "grid-canvas")]/div['+row+']' + xpath_row_cell = '//div[contains(@class, "'+cell+'")]' + + xpath_cell = '{0}{1}'.format(xpath_grid_row, xpath_row_cell) + + return xpath_cell + + def _compare_cell_value(self, xpath, value): + # Initialize an instance of WebDriverWait with timeout of 0.8 seconds + wait = WebDriverWait(self.driver, 0.8) + try: + wait.until(EC.text_to_be_present_in_element( + (By.XPATH, xpath+"//span"), str(value)), + CheckForViewDataTest.TIMEOUT_STRING + ) + except Exception: + wait.until(EC.text_to_be_present_in_element( + (By.XPATH, xpath), str(value)), + CheckForViewDataTest.TIMEOUT_STRING + ) + + def _update_cell(self, xpath, data): + """ + This function updates the given cell(xpath) with + given value + Args: + xpath: xpath of cell element + data: list with cell related data + + Returns: None + + """ + self.wait.until(EC.visibility_of_element_located( + (By.XPATH, xpath)), CheckForViewDataTest.TIMEOUT_STRING + ) + cell_el = self.page.find_by_xpath(xpath) + ActionChains(self.driver).move_to_element(cell_el).double_click( + cell_el + ).perform() + + cell_type = data[2] + value = data[0] + + if cell_type == 'int': + if value == 'clear': + cell_el.find_element_by_css_selector('input').clear() + else: + ActionChains(self.driver).send_keys(value).perform() + + elif cell_type in ['text', 'json']: + self.page.driver.find_element_by_css_selector( + "div[style*='z-index: 1000'] textarea" + ).click() + ActionChains(self.driver).send_keys(value).perform() + save_btn_xpath = "div[style*='z-index: 1000'] " \ + "div button:first-child" + self.wait.until( + EC.visibility_of_element_located( + (By.CSS_SELECTOR, save_btn_xpath) + ), CheckForViewDataTest.TIMEOUT_STRING + ) + self.page.driver.find_element_by_css_selector( + save_btn_xpath + ).click() # Click on editor's Save button + else: + if data[1] == 'true': + checkbox_el = cell_el.find_element_by_xpath(".//input") + checkbox_el.click() + ActionChains(self.driver).move_to_element(checkbox_el).double_click( + checkbox_el + ).perform() + + def _tables_node_expandable(self): + self.page.toggle_open_tree_item(self.server['name']) + self.page.toggle_open_tree_item('Databases') + self.page.toggle_open_tree_item('acceptance_test_db') + self.page.toggle_open_tree_item('Schemas') + self.page.toggle_open_tree_item('public') + self.page.toggle_open_tree_item('Tables') + + def _view_data_grid(self): + self.page.driver.find_element_by_link_text("Object").click() + ActionChains(self.page.driver) \ + .move_to_element( + self.page.driver.find_element_by_link_text("View Data")) \ + .perform() + self.page.find_by_partial_link_text("View All Rows").click() + self.wait.until( + EC.visibility_of_element_located( + (By.CSS_SELECTOR, 'iframe') + ), CheckForViewDataTest.TIMEOUT_STRING + ) + self.page.driver.switch_to.frame( + self.page.driver.find_element_by_tag_name('iframe') + ) + + def _copy_paste_row(self): + row0_cell0_xpath = CheckForViewDataTest._get_cell_xpath("r0", "1") + row1_cell1_xpath = CheckForViewDataTest._get_cell_xpath("r1", "2") + row1_cell2_xpath = CheckForViewDataTest._get_cell_xpath("r2", "2") + + self.page.find_by_xpath(row0_cell0_xpath).click() + self.page.find_by_xpath("//*[@id='btn-copy-row']").click() + self.page.find_by_xpath("//*[@id='btn-paste-row']").click() + + # Update primary key of copied cell + self._update_cell(row1_cell1_xpath, [2, "", "int"]) + self.page.find_by_xpath( + CheckForViewDataTest._get_cell_xpath("r1", "3") + ).click() + + # Check if removing a cell value with default value sets + # markup to [default] if cell is cleared + self._update_cell(row1_cell2_xpath, ["clear", "", "int"]) + # click outside + self.page.find_by_xpath( + CheckForViewDataTest._get_cell_xpath("r1", "3") + ).click() + + self._compare_cell_value(row1_cell2_xpath, "[default]") + # reset cell value to previous one + self._update_cell(row1_cell2_xpath, ["1", "", "int"]) + + self.page.find_by_id("btn-save").click() # Save data + + # Verify row 1 and row 2 data + self._verify_row_data(False) + + def _add_row(self): + last_row = 'new-row' + for idx in range(1, len(config_data.keys())): + cell_xpath = CheckForViewDataTest._get_cell_xpath( + 'r'+str(idx), last_row + ) + self._update_cell(cell_xpath, config_data[str(idx)]) + last_row = 'active' if last_row == 'new-row' else last_row + + self.page.find_by_id("btn-save").click() # Save data + + def _verify_row_data(self, is_new_row): + self.page.find_by_id("btn-flash").click() + row_id = "1" if is_new_row else "2" + + for idx in range(1, len(config_data.keys())): + cell_xpath = CheckForViewDataTest._get_cell_xpath( + 'r'+str(idx), row_id + ) + # after copy & paste row, the first cell of row 1 and + # row 2(being primary keys) won't match + # see if cell values matched to actual value + if idx != 1 and not is_new_row: + self._compare_cell_value(cell_xpath, config_data[str(idx)][1]) + elif is_new_row: + self._compare_cell_value(cell_xpath, config_data[str(idx)][1]) \ No newline at end of file diff --git a/web/regression/python_test_utils/test_utils.py b/web/regression/python_test_utils/test_utils.py index 2b7c695..c50dd31 100644 --- a/web/regression/python_test_utils/test_utils.py +++ b/web/regression/python_test_utils/test_utils.py @@ -172,6 +172,35 @@ def create_table(server, db_name, table_name): except Exception: traceback.print_exc(file=sys.stderr) + +def create_table_with_query(server, db_name, query): + """ + This function create the table in given database name + :param server: server details + :type server: dict + :param db_name: database name + :type db_name: str + :param query: create table query + :type query: str + :return: None + """ + try: + connection = get_db_connection(db_name, + server['username'], + server['db_password'], + server['host'], + server['port']) + old_isolation_level = connection.isolation_level + connection.set_isolation_level(0) + pg_cursor = connection.cursor() + pg_cursor.execute(query) + connection.set_isolation_level(old_isolation_level) + connection.commit() + + except Exception: + traceback.print_exc(file=sys.stderr) + + def create_constraint( server, db_name, table_name, constraint_type="unique", constraint_name="test_unique"): @@ -274,7 +303,6 @@ def drop_database(connection, database_name): connection.commit() connection.close() - def drop_tablespace(connection): """This function used to drop the tablespace""" pg_cursor = connection.cursor()