diff --git a/web/pgadmin/browser/server_groups/servers/tests/test_server_add.py b/web/pgadmin/browser/server_groups/servers/tests/test_server_add.py index 5ad0469..c3be213 100644 --- a/web/pgadmin/browser/server_groups/servers/tests/test_server_add.py +++ b/web/pgadmin/browser/server_groups/servers/tests/test_server_add.py @@ -36,5 +36,5 @@ class ServersAddTestCase(BaseTestGenerator): utils.write_node_info("sid", server_dict) def tearDown(self): - """This function delete the server from SQLite """ + """This function delete the server from SQLite""" utils.delete_server_with_api(self.tester, self.server_id) diff --git a/web/pgadmin/browser/server_groups/servers/types.py b/web/pgadmin/browser/server_groups/servers/types.py index d0bea4f..159c860 100644 --- a/web/pgadmin/browser/server_groups/servers/types.py +++ b/web/pgadmin/browser/server_groups/servers/types.py @@ -61,7 +61,6 @@ class ServerType(object): for key in cls.registry: st = cls.registry[key] default_path = config.DEFAULT_BINARY_PATHS.get(st.stype, "") - st.utility_path = paths.register( 'bin_paths', st.stype + '_bin_dir', st.UTILITY_PATH_LABEL, diff --git a/web/pgadmin/feature_tests/pg_utilities_backup_restore_test.py b/web/pgadmin/feature_tests/pg_utilities_backup_restore_test.py new file mode 100644 index 0000000..c8cbe28 --- /dev/null +++ b/web/pgadmin/feature_tests/pg_utilities_backup_restore_test.py @@ -0,0 +1,127 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import time +from selenium.webdriver.support.ui import WebDriverWait +from selenium.webdriver.common.by import By +from selenium.webdriver.support import expected_conditions as EC +from regression.feature_utils.base_feature_test import BaseFeatureTest +from regression.python_test_utils import test_utils + +import config + + +class PGUtilitiesBackupFeatureTest(BaseFeatureTest): + """ This class test PG utilities - Backup and Restore test scenarios """ + + scenarios = [ + ("Test for PG utilities - Backup and Restore", dict()) + ] + + def before(self): + if self.server['DEFAULT_BINARY_PATHS'] is None: + self.skipTest( + "DEFAULT_BINARY_PATHS is not set for the server {0}".format( + self.server['name'] + ) + ) + connection = test_utils.get_db_connection( + self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port'], + self.server['sslmode'] + ) + test_utils.drop_database(connection, "pg_utility_test_db") + + test_utils.create_database(self.server, "pg_utility_test_db") + self.page.add_server(self.server) + + self.wait = WebDriverWait(self.page.driver, 10) + + def runTest(self): + self.page.toggle_open_server(self.server['name']) + self.page.toggle_open_tree_item('Databases') + self.page.toggle_open_tree_item('pg_utility_test_db') + self.driver.find_element_by_link_text("Tools").click() + + self.page.find_by_partial_link_text("Backup...").click() + + self.wait.until( + EC.element_to_be_clickable((By.NAME, 'file')) + ) + + self.page.fill_input_by_field_name("file", "test_backup_file") + self.page.find_by_xpath("//button[contains(@class,'fa-save') " + "and contains(.,'Backup')]").click() + + self.page.wait_for_element_to_disappear( + lambda driver: driver.find_element_by_css_selector(".ajs-modal") + ) + + status = self.page.find_by_xpath("//div[contains(@class," + "'bg-success')]").text + + self.assertEquals(status, "Successfully completed.") + self.page.find_by_xpath("//span[contains(string(), " + "'Click here for details.')]").click() + command = self.page.find_by_xpath("//p[contains(@class, " + "'bg-detailed-desc')]").text + + self.assertIn(self.server['name'], str(command)) + self.assertIn("from database 'pg_utility_test_db'", str(command)) + self.assertIn("test_backup_file", str(command)) + self.assertIn("pg_dump", str(command)) + + self.page.find_by_xpath("//div[contains(@class,'wcFloatingFocus')" + "]//div[contains(@class,'fa-close')]").click() + + self.driver.find_element_by_link_text("Tools").click() + self.page.find_by_partial_link_text("Restore...").click() + + self.wait.until( + EC.element_to_be_clickable((By.NAME, 'file')) + ) + + self.page.fill_input_by_field_name("file", "test_backup_file") + self.page.find_by_xpath("//button[contains(@class,'fa-upload')" + " and contains(.,'Restore')]").click() + + self.page.wait_for_element_to_disappear( + lambda driver: driver.find_element_by_css_selector(".ajs-modal") + ) + + status = self.page.find_by_xpath("//div[contains(@class," + "'bg-success')]").text + + self.assertEquals(status, "Successfully completed.") + self.page.find_by_xpath("//span[contains(string()," + " 'Click here for details.')]").click() + command = self.page.find_by_xpath("//p[contains(@class," + " 'bg-detailed-desc')]").text + + self.assertIn(self.server['name'], str(command)) + self.assertIn("test_backup_file", str(command)) + self.assertIn("pg_restore", str(command)) + + self.page.find_by_xpath("//div[contains(@class,'wcFloatingFocus')]" + "//div[contains(@class,'fa-close')]").click() + + 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'], + self.server['sslmode'] + ) + test_utils.drop_database(connection, "pg_utility_test_db") diff --git a/web/pgadmin/feature_tests/pg_utilities_maintenance_test.py b/web/pgadmin/feature_tests/pg_utilities_maintenance_test.py new file mode 100644 index 0000000..83f3736 --- /dev/null +++ b/web/pgadmin/feature_tests/pg_utilities_maintenance_test.py @@ -0,0 +1,112 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## +# _*_ coding: utf-8 _*_ +import time +from selenium.webdriver.support.ui import WebDriverWait +from selenium.webdriver.common.by import By +from selenium.webdriver.support import expected_conditions as EC +from regression.feature_utils.base_feature_test import BaseFeatureTest +from regression.python_test_utils import test_utils + + +class PGUtilitiesMaintenanceFeatureTest(BaseFeatureTest): + """ This class test PG utilities test scenarios """ + + scenarios = [ + ("Test for PG maintenance: database pg_maintenance", dict( + database_name='pg_maintenance', + table_name='pg_maintenance_table', + test_level='database' + )), + ("Test for PG maintenance: database", dict( + database_name='pg_maintenance', + table_name='pg_maintenance_table', + test_level='table' + )), + ] + + def before(self): + if self.server['DEFAULT_BINARY_PATHS'] is None: + self.skipTest( + "DEFAULT_BINARY_PATHS is not set for the server {0}".format( + self.server['name'] + ) + ) + connection = test_utils.get_db_connection( + self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port'], + self.server['sslmode'] + ) + test_utils.create_database(self.server, self.database_name) + test_utils.create_table(self.server, self.database_name, + self.table_name) + self.page.add_server(self.server) + self.wait = WebDriverWait(self.page.driver, 10) + + def runTest(self): + self._open_maintenance_dialogue() + # time.sleep + self.page.find_by_xpath("//button[contains(@class,'fa-save') and" + " contains(.,'OK')]").click() + self.page.wait_for_element_to_disappear( + lambda driver: driver.find_element_by_css_selector(".ajs-modal") + ) + self._verify_command() + + def _open_maintenance_dialogue(self): + self.page.toggle_open_server(self.server['name']) + self.page.toggle_open_tree_item('Databases') + self.page.toggle_open_tree_item(self.database_name) + if self.test_level == 'table': + self.page.toggle_open_tree_item('Schemas') + self.page.toggle_open_tree_item('public') + self.page.toggle_open_tree_item('Tables') + self.page.find_by_xpath( + "//*[@id='tree']//" + "*[.='" + self.table_name + "']/../*[@class='aciTreeItem'" + "]").click() + self.driver.find_element_by_link_text("Tools").click() + self.page.find_by_partial_link_text("Maintenance...").click() + time.sleep(0.5) + + def _verify_command(self): + status = self.page.find_by_xpath("//div[contains(@class," + "'bg-success')]").text + self.assertEquals(status, "Successfully completed.") + self.page.find_by_xpath("//span[contains(string()," + " 'Click here for details.')]").click() + command = self.page.find_by_xpath("//p[contains(@class," + " 'bg-detailed-desc')]").text + if self.test_level == 'database': + self.assertEquals(command, "VACUUM " + "(VERBOSE)\nRunning Query:" + "\nVACUUM VERBOSE;") + else: + self.assertEquals(command, "VACUUM " + "(VERBOSE)\nRunning Query:" + "\nVACUUM VERBOSE" + " public." + self.table_name + ";") + + self.page.find_by_xpath("//div[contains(@class,'wcFloatingFocus')]//" + "div[contains(@class,'fa-close')]").click() + + 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'], + self.server['sslmode'] + ) + test_utils.drop_database(connection, self.database_name) diff --git a/web/pgadmin/tools/backup/tests/__init__.py b/web/pgadmin/tools/backup/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/web/pgadmin/tools/backup/tests/test_backup_create_job_unit_test.py b/web/pgadmin/tools/backup/tests/test_backup_create_job_unit_test.py new file mode 100644 index 0000000..1ff608b --- /dev/null +++ b/web/pgadmin/tools/backup/tests/test_backup_create_job_unit_test.py @@ -0,0 +1,458 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + + +import sys +import simplejson as json + +from pgadmin.misc.bgprocess.processes import BatchProcess +from pgadmin.tools.backup import BackupMessage +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from pgadmin.utils import server_utils as server_utils +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils + + +if sys.version_info < (3, 3): + from mock import patch, MagicMock +else: + from unittest.mock import patch, MagicMock + + +class BackupCreateJobTest(BaseTestGenerator): + """Test the BackupCreateJob class""" + scenarios = [ + ('When backup object with default options', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='custom', + verbose=True, + blobs=True, + schemas=[], + tables=[], + database='postgres' + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--verbose', '--format=c', '--blobs'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup the object with option sections to all data', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='custom', + verbose=True, + schemas=[], + tables=[], + database='postgres', + data=True, + pre_data=True, + post_data=True + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--verbose', '--format=c', + '--section=pre-data', '--section=data', + '--section=post-data'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup the object with option only_data', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='plain', + verbose=True, + schemas=[], + tables=[], + database='postgres', + only_data=True, + only_schema=False + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--verbose', '--format=p', '--data-only'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup the object with option only_data', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='plain', + verbose=True, + schemas=[], + tables=[], + database='postgres', + only_data=True, + only_schema=True, + dns_owner=True + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--verbose', '--format=p', '--data-only'], + not_expected_cmd_opts=['--schema-only', '--no-owner'], + expected_exit_code=[0, None] + )), + ('When backup the object with option only_schema', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='plain', + verbose=True, + schemas=[], + tables=[], + database='postgres', + only_data=False, + only_schema=True + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--verbose', '--format=p', '--schema-only'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup the object with option - format plain and dns_owner', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='plain', + verbose=True, + schemas=[], + tables=[], + database='postgres', + dns_owner=True + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--verbose', '--format=p', '--no-owner'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup the object with option - Do not save privilege,' + ' tablespace, unlogged table data', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='custom', + verbose=True, + schemas=[], + tables=[], + database='postgres', + dns_privilege=True, + dns_unlogged_tbl_data=True, + dns_tablespace=True + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--no-privileges', + '--no-tablespaces', + '--no-unlogged-table-data'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup the object with option - all queries', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='plain', + verbose=True, + schemas=[], + tables=[], + database='postgres', + use_column_inserts=True, + include_create_database=True, + use_insert_commands=True, + include_drop_database=True + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--create', '--clean', '--inserts', + '--column-inserts'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup the object with option - all queries and format custom', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='custom', + verbose=True, + schemas=[], + tables=[], + database='postgres', + use_column_inserts=True, + include_create_database=True, + use_insert_commands=True, + include_drop_database=True + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--inserts', + '--column-inserts'], + not_expected_cmd_opts=['--create', '--clean'], + expected_exit_code=[0, None] + )), + ('When backup the object with option - miscellaneous', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='custom', + verbose=True, + schemas=[], + tables=[], + database='postgres', + disable_quoting=True, + use_set_session_auth=True, + with_oids=True, + dqoute=True + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--verbose', '--quote-all-identifiers', + '--disable-dollar-quoting', '--oids', + '--use-set-session-authorization'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup the object with format tar', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='tar', + verbose=True, + schemas=[], + tables=[], + database='postgres', + blobs=True, + ), + url='/backup/job/{0}/object', + expected_cmd_opts=['--verbose', + '--blobs', + '--format=t'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup the server', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_server_file', + dqoute=False, + verbose=True, + type='server' + ), + url='/backup/job/{0}', + expected_cmd_opts=['--verbose'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When backup globals', + dict( + class_params=dict( + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + username='postgres' + ), + params=dict( + file='test_backup_global_file', + dqoute=False, + verbose=True, + type='globals' + ), + url='/backup/job/{0}', + expected_cmd_opts=['--verbose'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )) + ] + + def setUp(self): + if self.server['DEFAULT_BINARY_PATHS'] is None: + self.skipTest( + "DEFAULT_BINARY_PATHS is not set for the server {0}".format( + self.server['name'] + ) + ) + + @patch('pgadmin.tools.backup.Server') + @patch('pgadmin.tools.backup.current_user') + @patch('pgadmin.tools.backup.BackupMessage') + @patch('pgadmin.tools.backup.filename_with_file_manager_path') + @patch('pgadmin.tools.backup.BatchProcess') + def runTest(self, batch_process_mock, filename_mock, + backup_message_mock, current_user_mock, server_mock): + class TestMockServer(): + def __init__(self, name, host, port, id, username, + maintenance_db): + self.name = name + self.host = host + self.port = port + self.id = id + self.username = username + self.maintenance_db = maintenance_db + + self.db_name = '' + self.server_id = parent_node_dict["server"][-1]["server_id"] + mock_obj = TestMockServer(self.class_params['name'], + self.class_params['host'], + self.class_params['port'], + self.server_id, + self.class_params['username'], + self.class_params['database'] + ) + mock_result = server_mock.query.filter_by.return_value + mock_result.first.return_value = mock_obj + + filename_mock.return_value = self.params['file'] + + batch_process_mock.set_env_variables = MagicMock( + return_value=True + ) + batch_process_mock.start = MagicMock( + return_value=True + ) + + server_response = server_utils.connect_server(self, self.server_id) + if server_response["info"] == "Server connected.": + db_owner = server_response['data']['user']['name'] + self.data = database_utils.get_db_data(db_owner) + self.db_name = self.data['name'] + + url = self.url.format(self.server_id) + + # Create the backup job + response = self.tester.post(url, + data=json.dumps(self.params), + content_type='html/json') + self.assertEquals(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + job_id = response_data['data']['job_id'] + + assert backup_message_mock.called + assert batch_process_mock.called + + if self.expected_cmd_opts: + for opt in self.expected_cmd_opts: + self.assertIn( + opt, + batch_process_mock.call_args_list[0][1]['args'] + ) + if self.not_expected_cmd_opts: + for opt in self.not_expected_cmd_opts: + self.assertNotIn( + opt, + batch_process_mock.call_args_list[0][1]['args'] + ) diff --git a/web/pgadmin/tools/backup/tests/test_backup_message.py b/web/pgadmin/tools/backup/tests/test_backup_message.py new file mode 100644 index 0000000..761d286 --- /dev/null +++ b/web/pgadmin/tools/backup/tests/test_backup_message.py @@ -0,0 +1,158 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## +import sys + +from flask import Response +import simplejson as json + +from pgadmin.tools.backup import BackupMessage, BACKUP +from pgadmin.utils.route import BaseTestGenerator + +if sys.version_info < (3, 3): + from mock import patch, MagicMock +else: + from unittest.mock import patch, MagicMock + + +class BackupMessageTest(BaseTestGenerator): + """Test the BackupMessage class""" + scenarios = [ + ('When Backup server', + dict( + class_params=dict( + type=BACKUP.SERVER, + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_restore', + args=[ + '--file', + "backup_file", + '--host', + "localhost", + '--port', + "5444", + '--username', + "postgres", + '--no-password', + '--database', + "postgres" + ], + cmd="/test_path/pg_dump" + ), + extected_msg="Backing up the server" + " 'test_backup_server (localhost:5444)'...", + expetced_details_cmd='/test_path/pg_dump --file ' + '"backup_file" --host "localhost" ' + '--port "5444" --username "postgres" ' + '--no-password --database "postgres"' + + )), + ('When Backup global', + dict( + class_params=dict( + type=BACKUP.GLOBALS, + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + args=[ + '--file', + 'backup_file', + '--host', + 'localhost', + '--port', + '5444', + '--username', + 'postgres', + '--no-password', + '--database', + 'postgres' + ], + cmd="/test_path/pg_dump" + ), + extected_msg="Backing up the global objects on the server " + "'test_backup_server (localhost:5444)'...", + expetced_details_cmd='/test_path/pg_dump --file "backup_file" ' + '--host "localhost"' + ' --port "5444" --username "postgres" ' + '--no-password --database "postgres"' + + )), + ('When backup object', + dict( + class_params=dict( + type=BACKUP.OBJECT, + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_backup', + args=[ + '--file', + 'backup_file', + '--host', + 'localhost', + '--port', + '5444', + '--username', + 'postgres', + '--no-password', + '--database', + 'postgres' + ], + cmd="/test_path/pg_dump" + ), + extected_msg="Backing up an object on the server " + "'test_backup_server (localhost:5444)'" + " from database 'postgres'...", + expetced_details_cmd='/test_path/pg_dump --file "backup_file" ' + '--host "localhost" ' + '--port "5444" --username "postgres" ' + '--no-password --database "postgres"' + + )) + ] + + @patch('pgadmin.tools.backup.Server') + @patch('pgadmin.tools.backup.current_user') + def runTest(self, current_user_mock, server_mock): + current_user_mock = 1 + + class TestMockServer(): + def __init__(self, name, host, port): + self.name = name + self.host = host + self.port = port + mock_obj = TestMockServer(self.class_params['name'], + self.class_params['host'], + self.class_params['port']) + mock_result = server_mock.query.filter_by.return_value + mock_result.first.return_value = mock_obj + + backup_obj = BackupMessage( + self.class_params['type'], + self.class_params['sid'], + self.class_params['bfile'], + *self.class_params['args'], + **{'database': self.class_params['database']} + ) + + # Check the expected message returned + assert backup_obj.message == self.extected_msg + + # Check the command + obj_details = backup_obj.details(self.class_params['cmd'], + self.class_params['args']) + self.assertIn(self.expetced_details_cmd, obj_details) diff --git a/web/pgadmin/tools/backup/tests/test_backup_utils.py b/web/pgadmin/tools/backup/tests/test_backup_utils.py new file mode 100644 index 0000000..3c0c98c --- /dev/null +++ b/web/pgadmin/tools/backup/tests/test_backup_utils.py @@ -0,0 +1,111 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import time +import random +import simplejson as json + + +def create_backup_job(tester, url, params): + # Create the backup job + response = tester.post(url, + data=json.dumps(params), + content_type='html/json') + assert response.status_code == 200 + response_data = json.loads(response.data.decode('utf-8')) + job_id = response_data['data']['job_id'] + return job_id + + +def run_backup_job(tester, job_id, expected_params, assertIn, assertNotIn): + cnt = 0 + while 1: + if cnt > 1: + break + # Check the process list + response1 = tester.get('/misc/bgprocess/?_='.format( + random.randint(1, 9999999))) + assert response1.status_code == 200 + process_list = json.loads(response1.data.decode('utf-8')) + + if len(process_list) > 0 and 'execution_time' in process_list[0]: + break + time.sleep(0.5) + cnt += 1 + + assert 'execution_time' in process_list[0] + assert 'stime' in process_list[0] + assert 'exit_code' in process_list[0] + assert process_list[0]['exit_code'] in expected_params[ + 'expected_exit_code' + ] + + if expected_params['expected_cmd_opts']: + for opt in expected_params['expected_cmd_opts']: + assertIn(opt, process_list[0]['details']) + if expected_params['not_expected_cmd_opts']: + for opt in expected_params['not_expected_cmd_opts']: + assertNotIn(opt, process_list[0]['details']) + + # Check the process details + p_details = tester.get('/misc/bgprocess/{0}?_='.format( + job_id, random.randint(1, 9999999)) + ) + assert p_details.status_code == 200 + p_details_data = json.loads(p_details.data.decode('utf-8')) + + p_details = tester.get('/misc/bgprocess/{0}/{1}/{2}/?_='.format( + job_id, 0, 0, random.randint(1, 9999999)) + ) + assert p_details.status_code == 200 + p_details_data = json.loads(p_details.data.decode('utf-8')) + + cnt = 0 + # Retrieve the backup job process logs + while 1: + out, err, status = get_params(p_details_data) + if status or cnt >= 10: + break + + p_details = tester.get( + '/misc/bgprocess/{0}/{1}/{2}/?_={3}'.format( + job_id, out, err, random.randint(1, 9999999)) + ) + assert p_details.status_code == 200 + p_details_data = json.loads(p_details.data.decode('utf-8')) + + cnt += 1 + time.sleep(1) + + # Check the job is complete. + backup_ack = tester.put('/misc/bgprocess/{0}'.format(job_id)) + assert backup_ack.status_code == 200 + backup_ack_res = json.loads(backup_ack.data.decode('utf-8')) + + assert backup_ack_res['success'] == 1 + + +def get_params(data): + out = 0 + out_done = False + err = 0 + err_done = False + if 'out' in data: + out = data['out'] and data['out']['pos'] + + if 'done' in data['out']: + out_done = data['out']['done'] + + if 'err' in data: + err = data['err'] and data['err']['pos'] + + if 'done' in data['err']: + err_done = data['err']['done'] + + return out, err, (out_done and err_done) diff --git a/web/pgadmin/tools/backup/tests/test_batch_process.py b/web/pgadmin/tools/backup/tests/test_batch_process.py new file mode 100644 index 0000000..8cba9be --- /dev/null +++ b/web/pgadmin/tools/backup/tests/test_batch_process.py @@ -0,0 +1,219 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## +import sys +import simplejson as json + +from pgadmin.misc.bgprocess.processes import BatchProcess, IProcessDesc +from pgadmin.tools.backup import BackupMessage, BACKUP +from pgadmin.utils.route import BaseTestGenerator +from pickle import dumps, loads + +if sys.version_info < (3, 3): + from mock import patch +else: + from unittest.mock import patch + + +class BatchProcessTest(BaseTestGenerator): + """Test the BatchProcess class""" + scenarios = [ + ('When backup server', + dict( + class_params=dict( + type=BACKUP.SERVER, + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + username='postgres', + bfile='test_backup', + args=[ + '--file', + "backup_file", + '--host', + "localhost", + '--port', + "5444", + '--username', + "postgres", + '--no-password', + '--database', + "postgres" + ], + cmd='backup_server' + ) + )), + ('When backup globals', + dict( + class_params=dict( + type=BACKUP.GLOBALS, + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + username='postgres', + bfile='test_backup', + args=[ + '--file', + "backup_file", + '--host', + "localhost", + '--port', + "5444", + '--username', + "postgres", + '--no-password', + '--database', + "postgres" + ], + cmd='backup' + ) + )), + ('When backup object', + dict( + class_params=dict( + type=BACKUP.OBJECT, + sid=1, + name='test_backup_server', + port=5444, + host='localhost', + database='postgres', + username='postgres', + bfile='test_backup', + args=[ + '--file', + "backup_file", + '--host', + "localhost", + '--port', + "5444", + '--username', + "postgres", + '--no-password', + '--database', + "postgres" + ], + cmd='backup' + ) + )) + ] + + @patch('pgadmin.misc.bgprocess.processes.Popen') + @patch('pgadmin.misc.bgprocess.processes.current_app') + @patch('pgadmin.misc.bgprocess.processes.db') + @patch('pgadmin.tools.backup.Server') + @patch('pgadmin.tools.backup.current_user') + @patch('pgadmin.misc.bgprocess.processes.current_user') + def runTest(self, current_user_mock, current_user, server_mock, db_mock, + current_app_mock, popen_mock): + current_user.id = 1 + current_user_mock.id = 1 + current_app_mock.PGADMIN_RUNTIME = False + + class TestMockServer(): + def __init__(self, name, host, port): + self.name = name + self.host = host + self.port = port + + def db_session_add_mock(j): + cmd_obj = loads(j.desc) + assert isinstance(cmd_obj, IProcessDesc) + self.assertEquals(cmd_obj.backup_type, self.class_params['type']) + self.assertEquals(cmd_obj.bfile, self.class_params['bfile']) + self.assertEquals(cmd_obj.database, self.class_params['database']) + self.assertEquals(cmd_obj.cmd, + ' --file "backup_file" ' + '--host "{0}" ' + '--port "{1}" ' + '--username "{2}" ' + '--no-password ' + '--database "{3}"'.format( + self.class_params['host'], + self.class_params['port'], + self.class_params['username'], + self.class_params['database'] + )) + + mock_obj = TestMockServer(self.class_params['name'], + self.class_params['host'], + self.class_params['port']) + mock_result = server_mock.query.filter_by.return_value + mock_result.first.return_value = mock_obj + + db_mock.session.add.side_effect = db_session_add_mock + + backup_obj = BackupMessage( + self.class_params['type'], + self.class_params['sid'], + self.class_params['bfile'], + *self.class_params['args'], + **{'database': self.class_params['database']} + ) + + p = BatchProcess( + desc=backup_obj, + cmd=self.class_params['cmd'], + args=self.class_params['args'] + ) + + # Check that _create_process has been called + assert db_mock.session.add.called + + # Check start method + self._check_start(popen_mock, p) + + # Check list method + self._check_list(p, backup_obj) + + def _check_start(self, popen_mock, p): + cmd_test = self.class_params['cmd'] + + class popenMockSideEffect(): + def __init__(self, cmd, **kwargs): + assert cmd_test in cmd + assert 'env' in kwargs + + def poll(self): + pass + + popen_mock.side_effect = popenMockSideEffect + p.start() + + assert popen_mock.called + + @patch('pgadmin.misc.bgprocess.processes.Process') + @patch('pgadmin.misc.bgprocess.processes.BatchProcess.' + 'update_process_info') + def _check_list(self, p, backup_obj, update_process_info_mock, + process_mock): + class TestMockProcess(): + def __init__(self, desc, args, cmd): + self.pid = 1 + self.exit_code = 1 + self.start_time = '2018-04-17 06:18:56.315445 +0000' + self.end_time = None + self.desc = dumps(desc) + self.arguments = " ".join(args) + self.command = cmd + self.acknowledge = None + + process_mock.query.filter_by.return_value = [ + TestMockProcess(backup_obj, + self.class_params['args'], + self.class_params['cmd'])] + + update_process_info_mock.return_value = [True, True] + + ret_value = p.list() + self.assertEqual(1, len(ret_value)) + assert 'details' in ret_value[0] + assert 'desc' in ret_value[0] diff --git a/web/pgadmin/tools/backup/tests/test_create_backup_job.py b/web/pgadmin/tools/backup/tests/test_create_backup_job.py new file mode 100644 index 0000000..d9f0531 --- /dev/null +++ b/web/pgadmin/tools/backup/tests/test_create_backup_job.py @@ -0,0 +1,58 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +import pgadmin.tools.backup.tests.test_backup_utils as backup_utils + + +class BackupJobTest(BaseTestGenerator): + """Backup api test cases""" + scenarios = [ + ('When backup the object with the default options', + dict( + params=dict( + file='test_backup_file', + format='custom', + verbose=True, + blobs=True, + schemas=[], + tables=[], + database='postgres' + ), + url='/backup/job/{0}/object', + expected_params=dict( + expected_cmd_opts=['--verbose', '--format=c', '--blobs'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + ) + )) + ] + + def setUp(self): + if self.server['DEFAULT_BINARY_PATHS'] is None: + self.skipTest( + "DEFAULT_BINARY_PATHS is not set for the server {0}".format( + self.server['name'] + ) + ) + + def runTest(self): + self.server_id = parent_node_dict["server"][-1]["server_id"] + url = self.url.format(self.server_id) + + # Create the backup job + job_id = backup_utils.create_backup_job(self.tester, url, self.params) + backup_utils.run_backup_job(self.tester, + job_id, + self.expected_params, + self.assertIn, + self.assertNotIn + ) diff --git a/web/pgadmin/tools/maintenance/tests/__init__.py b/web/pgadmin/tools/maintenance/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/web/pgadmin/tools/maintenance/tests/test_batch_process_maintenance.py b/web/pgadmin/tools/maintenance/tests/test_batch_process_maintenance.py new file mode 100644 index 0000000..55c3db8 --- /dev/null +++ b/web/pgadmin/tools/maintenance/tests/test_batch_process_maintenance.py @@ -0,0 +1,153 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## +import sys + +from pgadmin.misc.bgprocess.processes import BatchProcess, IProcessDesc +from pgadmin.tools.maintenance import Message +from pgadmin.utils.route import BaseTestGenerator +from pickle import dumps, loads + +if sys.version_info < (3, 3): + from mock import patch +else: + from unittest.mock import patch + + +class BatchProcessTest(BaseTestGenerator): + """Test the BatchProcess class""" + scenarios = [ + ('When maintained server', + dict( + class_params=dict( + sid=1, + host='localhost', + port=5444, + username='postgres', + args=[ + '--host', + "localhost", + '--port', + "5444", + '--username', + '--dbname', + "postgres", + '--command', + "VACUUM VERBOSE;\n" + ], + data={ + 'database': 'postgres', + 'op': 'VACUUM', + 'vacuum_analyze': False, + 'vacuum_freeze': False, + 'vacuum_full': False, + 'verbose': True + }, + cmd="VACUUM VERBOSE;\n" + ), + expected_msg="Maintenance (Vacuum)", + expetced_details_cmd='VACUUM VERBOSE;' + )) + ] + + @patch('pgadmin.misc.bgprocess.processes.Popen') + @patch('pgadmin.misc.bgprocess.processes.current_app') + @patch('pgadmin.misc.bgprocess.processes.db') + @patch('pgadmin.tools.maintenance.Server') + @patch('pgadmin.misc.bgprocess.processes.current_user') + def runTest(self, current_user_mock, server_mock, db_mock, + current_app_mock, popen_mock): + current_user_mock.id = 1 + current_app_mock.PGADMIN_RUNTIME = False + + class TestMockServer(): + def __init__(self, name, host, port): + self.name = name + self.host = host + self.port = port + + def db_session_add_mock(j): + cmd_obj = loads(j.desc) + assert isinstance(cmd_obj, IProcessDesc) + self.assertEquals(cmd_obj.query, self.class_params['cmd']) + self.assertEquals(cmd_obj.message, self.expected_msg) + self.assertEquals(cmd_obj.data, self.class_params['data']) + + mock_obj = TestMockServer(self.class_params['username'], + self.class_params['host'], + self.class_params['port']) + mock_result = server_mock.query.filter_by.return_value + mock_result.first.return_value = mock_obj + + db_mock.session.add.side_effect = db_session_add_mock + + maintenance_obj = Message( + self.class_params['sid'], + self.class_params['data'], + self.class_params['cmd'] + ) + + p = BatchProcess( + desc=maintenance_obj, + cmd=self.class_params['cmd'], + args=self.class_params['args'] + ) + + # Check that _create_process has been called + assert db_mock.session.add.called + + # Check start method + self._check_start(popen_mock, p) + + # Check list method + self._check_list(p, maintenance_obj) + + def _check_start(self, popen_mock, p): + cmd_test = self.class_params['cmd'] + + class popenMockSideEffect(): + def __init__(self, cmd, **kwargs): + assert cmd_test in cmd + assert 'env' in kwargs + + def poll(self): + pass + + popen_mock.side_effect = popenMockSideEffect + p.start() + + assert popen_mock.called + + @patch('pgadmin.misc.bgprocess.processes.Process') + @patch('pgadmin.misc.bgprocess.processes.BatchProcess.' + 'update_process_info') + def _check_list(self, p, maintenance_obj, update_process_info_mock, + process_mock): + class TestMockProcess(): + def __init__(self, desc, args, cmd): + self.pid = 1 + self.exit_code = 1 + self.start_time = '2018-04-17 06:18:56.315445 +0000' + self.end_time = None + self.desc = dumps(desc) + self.arguments = " ".join(args) + self.command = cmd + self.acknowledge = None + + process_mock.query.filter_by.return_value = [ + TestMockProcess(maintenance_obj, + self.class_params['args'], + self.class_params['cmd']) + ] + + update_process_info_mock.return_value = [True, True] + + ret_value = p.list() + self.assertEqual(1, len(ret_value)) + assert 'details' in ret_value[0] + assert 'desc' in ret_value[0] diff --git a/web/pgadmin/tools/maintenance/tests/test_create_maintenance_job.py b/web/pgadmin/tools/maintenance/tests/test_create_maintenance_job.py new file mode 100644 index 0000000..a424920 --- /dev/null +++ b/web/pgadmin/tools/maintenance/tests/test_create_maintenance_job.py @@ -0,0 +1,142 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## +import time +import random +import simplejson as json + +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from pgadmin.utils import server_utils as server_utils +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils + + +class MaintenanceJobTest(BaseTestGenerator): + """Maintenance api test cases""" + scenarios = [ + ('When maintenance the object with the default options', + dict( + params=dict( + data={ + 'database': 'postgres', + 'op': 'VACUUM', + 'vacuum_analyze': False, + 'vacuum_freeze': False, + 'vacuum_full': False, + 'verbose': True + }, + cmd="VACUUM VERBOSE;\n" + ), + url='/maintenance/job/{0}/{1}', + expected_cmd='VACUUM VERBOSE', + expected_exit_code=[0, None] + )) + ] + + def setUp(self): + if self.server['DEFAULT_BINARY_PATHS'] is None: + self.skipTest( + "DEFAULT_BINARY_PATHS is not set for the server {0}".format( + self.server['name'] + ) + ) + + def runTest(self): + self.db_name = '' + self.server_id = parent_node_dict["database"][-1]["server_id"] + self.db_id = parent_node_dict["database"][-1]["db_id"] + url = self.url.format(self.server_id, self.db_id) + + utils.set_preference(self.server['DEFAULT_BINARY_PATHS']) + + # Create the backup job + response = self.tester.post(url, + data=json.dumps(self.params['data']), + content_type='html/json') + self.assertEquals(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + job_id = response_data['data']['job_id'] + + cnt = 0 + while 1: + if cnt > 1: + break + # Check the process list + response1 = self.tester.get('/misc/bgprocess/?_='.format( + random.randint(1, 9999999))) + self.assertEquals(response1.status_code, 200) + process_list = json.loads(response1.data.decode('utf-8')) + + if len(process_list) > 0 and 'execution_time' in process_list[0]: + break + time.sleep(0.5) + cnt += 1 + + assert 'execution_time' in process_list[0] + assert 'stime' in process_list[0] + assert 'exit_code' in process_list[0] + assert process_list[0]['exit_code'] in self.expected_exit_code + + self.assertIn(self.expected_cmd, process_list[0]['details']) + + # Check the process details + p_details = self.tester.get('/misc/bgprocess/{0}?_='.format( + job_id, random.randint(1, 9999999)) + ) + self.assertEquals(p_details.status_code, 200) + p_details_data = json.loads(p_details.data.decode('utf-8')) + + p_details = self.tester.get('/misc/bgprocess/{0}/{1}/{2}/?_='.format( + job_id, 0, 0, random.randint(1, 9999999)) + ) + self.assertEquals(p_details.status_code, 200) + p_details_data = json.loads(p_details.data.decode('utf-8')) + + # Retrieve the backup job process logs + while 1: + out, err, status = MaintenanceJobTest.get_params(p_details_data) + if status: + break + + p_details = self.tester.get( + '/misc/bgprocess/{0}/{1}/{2}/?_={3}'.format( + job_id, out, err, random.randint(1, 9999999)) + ) + self.assertEquals(p_details.status_code, 200) + p_details_data = json.loads(p_details.data.decode('utf-8')) + + time.sleep(1) + + # Check the job is complete. + backup_ack = self.tester.put('/misc/bgprocess/{0}'.format(job_id)) + self.assertEquals(backup_ack.status_code, 200) + backup_ack_res = json.loads(backup_ack.data.decode('utf-8')) + + self.assertEquals(backup_ack_res['success'], 1) + + @staticmethod + def get_params(data): + out = 0 + out_done = False + err = 0 + err_done = False + if 'out' in data: + out = data['out'] and data['out']['pos'] + + if 'done' in data['out']: + out_done = data['out']['done'] + + if 'err' in data: + err = data['err'] and data['err']['pos'] + + if 'done' in data['err']: + err_done = data['err']['done'] + + return out, err, (out_done and err_done) diff --git a/web/pgadmin/tools/maintenance/tests/test_maintenance_create_job_unit_test.py b/web/pgadmin/tools/maintenance/tests/test_maintenance_create_job_unit_test.py new file mode 100644 index 0000000..6cf0465 --- /dev/null +++ b/web/pgadmin/tools/maintenance/tests/test_maintenance_create_job_unit_test.py @@ -0,0 +1,195 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + + +import sys +import simplejson as json + + +from pgadmin.misc.bgprocess.processes import BatchProcess +from pgadmin.tools.maintenance import Message +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from pgadmin.utils import server_utils as server_utils +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils + + +if sys.version_info < (3, 3): + from mock import patch, MagicMock +else: + from unittest.mock import patch, MagicMock + + +class MaintenanceCreateJobTest(BaseTestGenerator): + """Test the BackupCreateJob class""" + scenarios = [ + ('When maintenance object with default options', + dict( + class_params=dict( + sid=1, + name='test_maintenance_server', + port=5444, + host='localhost', + username='postgres' + ), + params=dict( + database='postgres', + op='VACUUM', + vacuum_analyze=False, + vacuum_freeze=False, + vacuum_full=False, + verbose=True + ), + url='/maintenance/job/{0}/{1}', + expected_cmd_opts=['VACUUM VERBOSE;\n'], + )), + ('When maintenance object with VACUUM FULL', + dict( + class_params=dict( + sid=1, + name='test_maintenance_server', + port=5444, + host='localhost', + username='postgres' + ), + params=dict( + database='postgres', + op='VACUUM', + vacuum_analyze=False, + vacuum_freeze=False, + vacuum_full=True, + verbose=True + ), + url='/maintenance/job/{0}/{1}', + expected_cmd_opts=['VACUUM FULL VERBOSE;\n'], + )), + ('When maintenance object with the ANALYZE', + dict( + class_params=dict( + sid=1, + name='test_maintenance_server', + port=5444, + host='localhost', + username='postgres' + ), + params=dict( + database='postgres', + op='ANALYZE', + vacuum_analyze=True, + vacuum_freeze=False, + vacuum_full=False, + verbose=True + ), + url='/maintenance/job/{0}/{1}', + expected_cmd_opts=['ANALYZE VERBOSE;\n'], + )), + ('When maintenance the object with the REINDEX', + dict( + class_params=dict( + sid=1, + name='test_maintenance_server', + port=5444, + host='localhost', + username='postgres' + ), + params=dict( + database='postgres', + op='REINDEX', + vacuum_analyze=False, + vacuum_freeze=False, + vacuum_full=False, + verbose=False + ), + url='/maintenance/job/{0}/{1}', + expected_cmd_opts=['REINDEX DATABASE postgres;\n'], + )), + ('When maintenance the object with the CLUSTER', + dict( + class_params=dict( + sid=1, + name='test_maintenance_server', + port=5444, + host='localhost', + username='postgres' + ), + params=dict( + database='postgres', + op='CLUSTER', + vacuum_analyze=False, + vacuum_freeze=False, + vacuum_full=False, + verbose=False + ), + url='/maintenance/job/{0}/{1}', + expected_cmd_opts=['CLUSTER;\n'], + )) + ] + + def setUp(self): + if self.server['DEFAULT_BINARY_PATHS'] is None: + self.skipTest( + "DEFAULT_BINARY_PATHS is not set for the server {0}".format( + self.server['name'] + ) + ) + + @patch('pgadmin.tools.maintenance.Server') + @patch('pgadmin.tools.maintenance.Message') + @patch('pgadmin.tools.maintenance.BatchProcess') + def runTest(self, batch_process_mock, message_mock, server_mock): + self.server_id = parent_node_dict["database"][-1]["server_id"] + self.db_id = parent_node_dict["database"][-1]["db_id"] + url = self.url.format(self.server_id, self.db_id) + + class TestMockServer(): + def __init__(self, host, port, id, username): + self.host = host + self.port = port + self.id = id + self.username = username + + mock_obj = TestMockServer(self.class_params['host'], + self.class_params['port'], + self.server_id, + self.class_params['username'] + ) + mock_result = server_mock.query.filter_by.return_value + mock_result.first.return_value = mock_obj + + batch_process_mock.set_env_variables = MagicMock( + return_value=True + ) + batch_process_mock.start = MagicMock( + return_value=True + ) + + server_response = server_utils.connect_server(self, self.server_id) + if server_response["info"] == "Server connected.": + db_owner = server_response['data']['user']['name'] + self.data = database_utils.get_db_data(db_owner) + self.db_name = self.data['name'] + + # Create the backup job + response = self.tester.post(url, + data=json.dumps(self.params), + content_type='html/json') + self.assertEquals(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + job_id = response_data['data']['job_id'] + + assert message_mock.called + assert batch_process_mock.called + + print(batch_process_mock.call_args_list[0][1]['args']) + if self.expected_cmd_opts: + for opt in self.expected_cmd_opts: + self.assertIn(opt, + batch_process_mock.call_args_list[0][1]['args']) diff --git a/web/pgadmin/tools/maintenance/tests/test_maintenance_message.py b/web/pgadmin/tools/maintenance/tests/test_maintenance_message.py new file mode 100644 index 0000000..4cb89ed --- /dev/null +++ b/web/pgadmin/tools/maintenance/tests/test_maintenance_message.py @@ -0,0 +1,124 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from flask import Response +import simplejson as json + +from pgadmin.tools.maintenance import Message +from pgadmin.utils.route import BaseTestGenerator + + +class MaintenanceMessageTest(BaseTestGenerator): + """Test the Maintenance Message class""" + scenarios = [ + ('When maintained the server', + dict( + class_params=dict( + sid=1, + data={ + 'database': 'postgres', + 'op': 'VACUUM', + 'vacuum_analyze': False, + 'vacuum_freeze': False, + 'vacuum_full': False, + 'verbose': True + }, + cmd="VACUUM VERBOSE;\n" + ), + extected_msg="Maintenance (Vacuum)", + expetced_details_cmd='VACUUM VERBOSE;' + + )), + ('When maintained the server with FULL VERBOSE options', + dict( + class_params=dict( + sid=1, + data={ + 'database': 'postgres', + 'op': 'VACUUM', + 'vacuum_analyze': False, + 'vacuum_freeze': False, + 'vacuum_full': True, + 'verbose': True + }, + cmd="VACUUM FULL VERBOSE;\n" + ), + extected_msg="Maintenance (Vacuum)", + expetced_details_cmd='VACUUM FULL VERBOSE;' + + )), + ('When maintained the server with ANALYZE', + dict( + class_params=dict( + sid=1, + data={ + 'database': 'postgres', + 'op': 'ANALYZE', + 'vacuum_analyze': False, + 'vacuum_freeze': False, + 'vacuum_full': False, + 'verbose': True + }, + cmd="ANALYZE VERBOSE;\n" + ), + extected_msg="Maintenance (Analyze)", + expetced_details_cmd='ANALYZE VERBOSE;' + + )), + ('When maintained the server with REINDEX', + dict( + class_params=dict( + sid=1, + data={ + 'database': 'postgres', + 'op': 'REINDEX', + 'vacuum_analyze': False, + 'vacuum_freeze': False, + 'vacuum_full': False, + 'verbose': False + }, + cmd="REINDEX;\n" + ), + extected_msg="Maintenance (Reindex)", + expetced_details_cmd='REINDEX;' + + )), + ('When maintained the server with CLUSTER', + dict( + class_params=dict( + sid=1, + data={ + 'database': 'postgres', + 'op': 'CLUSTER', + 'vacuum_analyze': False, + 'vacuum_freeze': False, + 'vacuum_full': False, + 'verbose': True + }, + cmd="CLUSTER VERBOSE;\n" + ), + extected_msg="Maintenance (Cluster)", + expetced_details_cmd='CLUSTER VERBOSE;' + + )), + ] + + def runTest(self): + maintenance_obj = Message( + self.class_params['sid'], + self.class_params['data'], + self.class_params['cmd'] + ) + + # Check the expected message returned + assert maintenance_obj.message == self.extected_msg + + # Check the command + obj_details = maintenance_obj.details(self.class_params['cmd'], None) + self.assertIn(self.expetced_details_cmd, obj_details) diff --git a/web/pgadmin/tools/restore/__init__.py b/web/pgadmin/tools/restore/__init__.py index f6f92b9..460d01f 100644 --- a/web/pgadmin/tools/restore/__init__.py +++ b/web/pgadmin/tools/restore/__init__.py @@ -192,6 +192,7 @@ def create_restore_job(sid): if _file is None: return make_json_response( + status=410, success=0, errormsg=_("File could not be found.") ) diff --git a/web/pgadmin/tools/restore/tests/__init__.py b/web/pgadmin/tools/restore/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/web/pgadmin/tools/restore/tests/test_batch_process.py b/web/pgadmin/tools/restore/tests/test_batch_process.py new file mode 100644 index 0000000..3c674e1 --- /dev/null +++ b/web/pgadmin/tools/restore/tests/test_batch_process.py @@ -0,0 +1,161 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## +import sys + +from pgadmin.misc.bgprocess.processes import BatchProcess, IProcessDesc +from pgadmin.tools.restore import RestoreMessage +from pgadmin.utils.route import BaseTestGenerator +from pickle import dumps, loads + +if sys.version_info < (3, 3): + from mock import patch +else: + from unittest.mock import patch + + +class BatchProcessTest(BaseTestGenerator): + """Test the BatchProcess class""" + scenarios = [ + ('When restore server', + dict( + class_params=dict( + sid=1, + name='test_restore_server', + port=5444, + host='localhost', + database='postgres', + username='postgres', + bfile='test_restore', + args=[ + '--file', + "restore_file", + '--host', + "localhost", + '--port', + "5444", + '--username', + "postgres", + '--no-password', + '--database', + "postgres" + ], + cmd='restore_server' + ) + )) + ] + + @patch('pgadmin.misc.bgprocess.processes.Popen') + @patch('pgadmin.misc.bgprocess.processes.current_app') + @patch('pgadmin.misc.bgprocess.processes.db') + @patch('pgadmin.tools.restore.Server') + @patch('pgadmin.tools.restore.current_user') + @patch('pgadmin.misc.bgprocess.processes.current_user') + def runTest(self, current_user_mock, current_user, server_mock, db_mock, + current_app_mock, popen_mock): + current_user.id = 1 + current_user_mock.id = 1 + current_app_mock.PGADMIN_RUNTIME = False + + class TestMockServer(): + def __init__(self, name, host, port): + self.name = name + self.host = host + self.port = port + + def db_session_add_mock(j): + cmd_obj = loads(j.desc) + assert isinstance(cmd_obj, IProcessDesc) + print(cmd_obj) + self.assertEquals(cmd_obj.bfile, self.class_params['bfile']) + self.assertEquals(cmd_obj.cmd, + ' --file "restore_file" ' + '--host "{0}" ' + '--port "{1}" ' + '--username "{2}" ' + '--no-password ' + '--database "{3}"'.format( + self.class_params['host'], + self.class_params['port'], + self.class_params['username'], + self.class_params['database'] + )) + + mock_obj = TestMockServer(self.class_params['name'], + self.class_params['host'], + self.class_params['port']) + mock_result = server_mock.query.filter_by.return_value + mock_result.first.return_value = mock_obj + + db_mock.session.add.side_effect = db_session_add_mock + + restore_obj = RestoreMessage( + self.class_params['sid'], + self.class_params['bfile'], + *self.class_params['args'] + ) + + p = BatchProcess( + desc=restore_obj, + cmd=self.class_params['cmd'], + args=self.class_params['args'] + ) + + # Check that _create_process has been called + assert db_mock.session.add.called + + # Check start method + self._check_start(popen_mock, p) + + # Check list method + self._check_list(p, restore_obj) + + def _check_start(self, popen_mock, p): + cmd_test = self.class_params['cmd'] + + class popenMockSideEffect(): + def __init__(self, cmd, **kwargs): + assert cmd_test in cmd + assert 'env' in kwargs + + def poll(self): + pass + + popen_mock.side_effect = popenMockSideEffect + p.start() + + assert popen_mock.called + + @patch('pgadmin.misc.bgprocess.processes.Process') + @patch('pgadmin.misc.bgprocess.processes.BatchProcess.' + 'update_process_info') + def _check_list(self, p, restore_obj, update_process_info_mock, + process_mock): + class TestMockProcess(): + def __init__(self, desc, args, cmd): + self.pid = 1 + self.exit_code = 1 + self.start_time = '2018-04-17 06:18:56.315445 +0000' + self.end_time = None + self.desc = dumps(desc) + self.arguments = " ".join(args) + self.command = cmd + self.acknowledge = None + + process_mock.query.filter_by.return_value = [ + TestMockProcess(restore_obj, + self.class_params['args'], + self.class_params['cmd']) + ] + + update_process_info_mock.return_value = [True, True] + + ret_value = p.list() + self.assertEqual(1, len(ret_value)) + assert 'details' in ret_value[0] + assert 'desc' in ret_value[0] diff --git a/web/pgadmin/tools/restore/tests/test_create_restore_job.py b/web/pgadmin/tools/restore/tests/test_create_restore_job.py new file mode 100644 index 0000000..64b12b6 --- /dev/null +++ b/web/pgadmin/tools/restore/tests/test_create_restore_job.py @@ -0,0 +1,196 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## +import sys +import time +import random + +import simplejson as json + +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from pgadmin.utils import server_utils as server_utils +import pgadmin.tools.backup.tests.test_backup_utils as backup_utils + + +if sys.version_info < (3, 3): + from mock import patch, MagicMock +else: + from unittest.mock import patch, MagicMock + + +class RestoreJobTest(BaseTestGenerator): + """Backup api test cases""" + scenarios = [ + ('When restore the object with the default options', + dict( + params=dict( + file='test_restore_file', + format='custom', + custom=False, + verbose=True, + blobs=True, + schemas=[], + tables=[], + database='test_restore_database' + ), + url='/restore/job/{0}', + expected_cmd_opts=['--verbose'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None], + backup_options=dict( + params=dict( + file='test_restore_file', + format='custom', + verbose=True, + blobs=True, + schemas=[], + tables=[], + database='test_restore_database' + ), + url='/backup/job/{0}/object', + expected_params=dict( + expected_cmd_opts=['--verbose', '--format=c', '--blobs'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + ) + + ) + )) + ] + + def setUp(self): + if self.server['DEFAULT_BINARY_PATHS'] is None: + self.skipTest( + "DEFAULT_BINARY_PATHS is not set for the server {0}".format( + self.server['name'] + ) + ) + + def create_backup(self): + url = self.backup_options['url'].format(self.server_id) + job_id = backup_utils.create_backup_job(self.tester, url, + self.backup_options['params']) + backup_utils.run_backup_job(self.tester, job_id, + self.backup_options['expected_params'], + self.assertIn, self.assertNotIn) + + def runTest(self): + self.db_name = '' + self.server_id = parent_node_dict["server"][-1]["server_id"] + server_response = server_utils.connect_server(self, self.server_id) + db_id = utils.create_database(self.server, self.params['database']) + + self.create_backup() + url = self.url.format(self.server_id) + + # Create the restore job + response = self.tester.post(url, + data=json.dumps(self.params), + content_type='html/json') + self.assertEquals(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + job_id = response_data['data']['job_id'] + + cnt = 0 + while 1: + if cnt > 1: + break + # Check the process list + response1 = self.tester.get('/misc/bgprocess/?_='.format( + random.randint(1, 9999999))) + self.assertEquals(response1.status_code, 200) + process_list = json.loads(response1.data.decode('utf-8')) + + if len(process_list) > 0 and 'execution_time' in process_list[0]: + break + time.sleep(0.5) + cnt += 1 + + print(process_list[0]) + + assert 'execution_time' in process_list[0] + assert 'stime' in process_list[0] + assert 'exit_code' in process_list[0] + assert process_list[0]['exit_code'] in self.expected_exit_code + + if self.expected_cmd_opts: + for opt in self.expected_cmd_opts: + self.assertIn(opt, process_list[0]['details']) + if self.not_expected_cmd_opts: + for opt in self.not_expected_cmd_opts: + self.assertNotIn(opt, process_list[0]['details']) + + # Check the process details + p_details = self.tester.get('/misc/bgprocess/{0}?_='.format( + job_id, random.randint(1, 9999999)) + ) + self.assertEquals(p_details.status_code, 200) + p_details_data = json.loads(p_details.data.decode('utf-8')) + + p_details = self.tester.get('/misc/bgprocess/{0}/{1}/{2}/?_='.format( + job_id, 0, 0, random.randint(1, 9999999)) + ) + self.assertEquals(p_details.status_code, 200) + p_details_data = json.loads(p_details.data.decode('utf-8')) + + # Retrieve the restore job process logs + cnt = 0 + while 1: + out, err, status = RestoreJobTest.get_params(p_details_data) + if status or cnt >= 10: + break + + p_details = self.tester.get( + '/misc/bgprocess/{0}/{1}/{2}/?_={3}'.format( + job_id, out, err, random.randint(1, 9999999)) + ) + self.assertEquals(p_details.status_code, 200) + p_details_data = json.loads(p_details.data.decode('utf-8')) + + cnt += 1 + time.sleep(1) + + # Check the job is complete. + restore_ack = self.tester.put('/misc/bgprocess/{0}'.format(job_id)) + self.assertEquals(restore_ack.status_code, 200) + restore_ack_res = json.loads(restore_ack.data.decode('utf-8')) + + self.assertEquals(restore_ack_res['success'], 1) + + @staticmethod + def get_params(data): + out = 0 + out_done = False + err = 0 + err_done = False + if 'out' in data: + out = data['out'] and data['out']['pos'] + + if 'done' in data['out']: + out_done = data['out']['done'] + + if 'err' in data: + err = data['err'] and data['err']['pos'] + + if 'done' in data['err']: + err_done = data['err']['done'] + + return out, err, (out_done and err_done) + + def after(self): + connection = utils.get_db_connection( + self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port'], + self.server['sslmode'] + ) + utils.drop_database(connection, self.params['database']) diff --git a/web/pgadmin/tools/restore/tests/test_restore_create_job_unit_test.py b/web/pgadmin/tools/restore/tests/test_restore_create_job_unit_test.py new file mode 100644 index 0000000..0cf315f --- /dev/null +++ b/web/pgadmin/tools/restore/tests/test_restore_create_job_unit_test.py @@ -0,0 +1,313 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import sys +import simplejson as json + +from pgadmin.tools.restore import RestoreMessage +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from pgadmin.utils import server_utils as server_utils +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils + +if sys.version_info < (3, 3): + from mock import patch, MagicMock +else: + from unittest.mock import patch, MagicMock + + +class RestoreCreateJobTest(BaseTestGenerator): + """Test the RestoreCreateJob class""" + scenarios = [ + ('When restore object with default options', + dict( + class_params=dict( + sid=1, + name='test_restore_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_restore', + username='postgres' + ), + params=dict( + file='test_restore_file', + format='custom', + custom=False, + verbose=True, + blobs=True, + schemas=[], + tables=[], + database='postgres' + ), + url='/restore/job/{0}', + expected_cmd_opts=['--verbose'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When restore object with the sections options', + dict( + class_params=dict( + sid=1, + name='test_restore_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_restore', + username='postgres' + ), + params=dict( + file='test_restore_file', + format='custom', + no_of_jobs='2', + custom=False, + verbose=True, + schemas=[], + tables=[], + database='postgres', + data=True, + pre_data=True, + post_data=True, + only_data=True, + only_schema=True + ), + url='/restore/job/{0}', + # Please include sections data here, right now this is a bug + expected_cmd_opts=['--verbose', '--jobs', '2'], + not_expected_cmd_opts=[], + # Below options should be enabled once we fix the issue #3368 + # not_expected_cmd_opts=['--data-only', '--schema-only'], + expected_exit_code=[0, None], + )), + ('When restore the object with Type of objects', + dict( + class_params=dict( + sid=1, + name='test_restore_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_restore', + username='postgres' + ), + params=dict( + file='test_restore_file', + format='custom', + no_of_jobs='2', + custom=False, + verbose=True, + schemas=[], + tables=[], + database='postgres', + only_data=True, + only_schema=True, + dns_owner=True + ), + url='/restore/job/{0}', + expected_cmd_opts=['--verbose', '--data-only'], + not_expected_cmd_opts=[], + # Below options should be enabled once we fix the issue #3368 + # not_expected_cmd_opts=['--schema-only', '--no-owner'], + expected_exit_code=[0, None], + )), + ('When restore object with option - Do not save', + dict( + class_params=dict( + sid=1, + name='test_restore_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_restore', + username='postgres' + ), + params=dict( + file='test_restore_file', + format='custom', + verbose=True, + custom=False, + schemas=[], + tables=[], + database='postgres', + dns_owner=True, + dns_privilege=True, + dns_tablespace=True, + only_data=False + ), + url='/restore/job/{0}', + # Add '--no-privileges' to the expected_cmd once #3363 fixed + expected_cmd_opts=['--no-owner', + '--no-tablespaces'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When restore object with option - Queries', + dict( + class_params=dict( + sid=1, + name='test_restore_file', + port=5444, + host='localhost', + database='postgres', + bfile='test_restore', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='custom', + verbose=True, + schemas=[], + tables=[], + database='postgres', + clean=True, + include_create_database=True, + single_transaction=True, + ), + url='/restore/job/{0}', + expected_cmd_opts=['--create', '--clean', + '--single-transaction'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When restore object with option - Disbale', + dict( + class_params=dict( + sid=1, + name='test_restore_file', + port=5444, + host='localhost', + database='postgres', + bfile='test_restore', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='custom', + verbose=True, + schemas=[], + tables=[], + database='postgres', + disable_trigger=True, + no_data_fail_table=True, + only_schema=False + ), + url='/restore/job/{0}', + # Add '--no-data-for-failed-tables' into + # expected_cmd_opts once #3363 fixed + expected_cmd_opts=['--disable-triggers'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ('When restore object with option - Miscellaneous', + dict( + class_params=dict( + sid=1, + name='test_restore_file', + port=5444, + host='localhost', + database='postgres', + bfile='test_restore', + username='postgres' + ), + params=dict( + file='test_backup_file', + format='custom', + verbose=True, + schemas=[], + tables=[], + database='postgres', + use_set_session_auth=True, + exit_on_error=True, + ), + url='/restore/job/{0}', + # Add '--use_set_session_auth' into + # expected_cmd_opts once #3363 fixed + expected_cmd_opts=['--exit-on-error'], + not_expected_cmd_opts=[], + expected_exit_code=[0, None] + )), + ] + + def setUp(self): + if self.server['DEFAULT_BINARY_PATHS'] is None: + self.skipTest( + "DEFAULT_BINARY_PATHS is not set for the server {0}".format( + self.server['name'] + ) + ) + + @patch('pgadmin.tools.restore.Server') + @patch('pgadmin.tools.restore.current_user') + @patch('pgadmin.tools.restore.RestoreMessage') + @patch('pgadmin.tools.restore.filename_with_file_manager_path') + @patch('pgadmin.tools.restore.BatchProcess') + def runTest(self, batch_process_mock, filename_mock, + restore_message_mock, current_user_mock, server_mock): + class TestMockServer(): + def __init__(self, name, host, port, id, username): + self.name = name + self.host = host + self.port = port + self.id = id + self.username = username + + self.db_name = '' + self.server_id = parent_node_dict["server"][-1]["server_id"] + + mock_obj = TestMockServer(self.class_params['name'], + self.class_params['host'], + self.class_params['port'], + self.server_id, + self.class_params['username'] + ) + mock_result = server_mock.query.filter_by.return_value + mock_result.first.return_value = mock_obj + + filename_mock.return_value = self.params['file'] + + batch_process_mock.set_env_variables = MagicMock( + return_value=True + ) + batch_process_mock.start = MagicMock( + return_value=True + ) + + server_response = server_utils.connect_server(self, self.server_id) + if server_response["info"] == "Server connected.": + db_owner = server_response['data']['user']['name'] + self.data = database_utils.get_db_data(db_owner) + self.db_name = self.data['name'] + + url = self.url.format(self.server_id) + + # Create the restore job + response = self.tester.post(url, + data=json.dumps(self.params), + content_type='html/json') + self.assertEquals(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + job_id = response_data['data']['job_id'] + + assert restore_message_mock.called + assert batch_process_mock.called + + if self.expected_cmd_opts: + for opt in self.expected_cmd_opts: + self.assertIn( + opt, + batch_process_mock.call_args_list[0][1]['args'] + ) + if self.not_expected_cmd_opts: + for opt in self.not_expected_cmd_opts: + self.assertNotIn( + opt, + batch_process_mock.call_args_list[0][1]['args'] + ) diff --git a/web/pgadmin/tools/restore/tests/test_restore_message.py b/web/pgadmin/tools/restore/tests/test_restore_message.py new file mode 100644 index 0000000..0b3e09d --- /dev/null +++ b/web/pgadmin/tools/restore/tests/test_restore_message.py @@ -0,0 +1,85 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2018, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## +import sys + +from pgadmin.tools.restore import RestoreMessage +from pgadmin.utils.route import BaseTestGenerator + +if sys.version_info < (3, 3): + from mock import patch +else: + from unittest.mock import patch + + +class RestoreMessageTest(BaseTestGenerator): + """Test the RestoreMessage class""" + scenarios = [ + ('When restore object', + dict( + class_params=dict( + sid=1, + name='test_restore_server', + port=5444, + host='localhost', + database='postgres', + bfile='test_restore', + args=[ + '--file', + 'restore_file', + '--host', + 'localhost', + '--port', + '5444', + '--username', + 'postgres', + '--no-password', + '--database', + 'postgres' + ], + cmd="/test_path/pg_restore" + ), + extected_msg="Restoring backup on the server " + "'test_restore_server (localhost:5444)'...", + expetced_details_cmd='/test_path/pg_restore --file ' + '"restore_file" --host "localhost"' + ' --port "5444" --username "postgres" ' + '--no-password --database "postgres"' + + )) + ] + + @patch('pgadmin.tools.restore.Server') + @patch('pgadmin.tools.restore.current_user') + def runTest(self, current_user_mock, server_mock): + current_user_mock = 1 + + class TestMockServer(): + def __init__(self, name, host, port): + self.name = name + self.host = host + self.port = port + mock_obj = TestMockServer(self.class_params['name'], + self.class_params['host'], + self.class_params['port']) + mock_result = server_mock.query.filter_by.return_value + mock_result.first.return_value = mock_obj + + restore_obj = RestoreMessage( + self.class_params['sid'], + self.class_params['bfile'], + *self.class_params['args'] + ) + + # Check the expected message returned + assert restore_obj.message == self.extected_msg + + # Check the command + obj_details = restore_obj.details(self.class_params['cmd'], + self.class_params['args']) + self.assertIn(self.expetced_details_cmd, obj_details) diff --git a/web/regression/python_test_utils/test_utils.py b/web/regression/python_test_utils/test_utils.py index 3e517b6..d61c5d8 100644 --- a/web/regression/python_test_utils/test_utils.py +++ b/web/regression/python_test_utils/test_utils.py @@ -21,6 +21,8 @@ import config import regression from regression import test_setup +from pgadmin.utils.preferences import Preferences + SERVER_GROUP = test_setup.config_data['server_group'] file_name = os.path.realpath(__file__) @@ -86,7 +88,8 @@ def get_config_data(): "db_password": srv['db_password'], "role": "", "sslmode": srv['sslmode'], - "tablespace_path": srv.get('tablespace_path', None) + "tablespace_path": srv.get('tablespace_path', None), + "DEFAULT_BINARY_PATHS": srv.get('DEFAULT_BINARY_PATHS', None) } if 'db_type' in srv: data['db_type'] = srv['db_type'] @@ -445,6 +448,13 @@ def delete_server_with_api(tester, sid): url = '/browser/server/obj/' + str(SERVER_GROUP) + "/" # Call API to delete the server response = tester.delete(url + str(sid)) + + cnt = 0 + for s in regression.parent_node_dict["server"]: + if s['server_id'] == int(sid): + del regression.parent_node_dict["server"][cnt] + cnt += 1 + except Exception: traceback.print_exc(file=sys.stderr) @@ -596,6 +606,47 @@ def get_db_server(sid): return connection +def set_preference(default_binary_path): + conn = sqlite3.connect(config.TEST_SQLITE_PATH) + cur = conn.cursor() + + perf = Preferences.module('paths') + pg_path_pref = perf.preference('pg_bin_dir') + + user_pref = cur.execute( + 'SELECT pid, uid FROM user_preferences where pid=%s' % pg_path_pref.pid + ) + user_pref = user_pref.fetchone() + + if user_pref: + cur.execute('UPDATE user_preferences SET value = ? WHERE pid = ?', + (default_binary_path['pg'], pg_path_pref.pid)) + else: + pg_pref_details = (pg_path_pref.pid, 1, + default_binary_path['pg']) + cur.execute('INSERT INTO user_preferences(pid, uid, value)' + ' VALUES (?,?,?)', pg_pref_details) + + ppas_path_pref = perf.preference('ppas_bin_dir') + + user_pref = cur.execute( + 'SELECT pid, uid FROM user_preferences where pid=%s' % + ppas_path_pref.pid + ) + user_pref = user_pref.fetchone() + + if user_pref: + cur.execute('UPDATE user_preferences SET value = ? WHERE pid = ? ', + (default_binary_path['ppas'], ppas_path_pref.pid)) + else: + ppas_pref_details = (ppas_path_pref.pid, 1, + default_binary_path['ppas']) + cur.execute('INSERT INTO user_preferences(pid, uid, value)' + ' VALUES (?,?,?)', ppas_pref_details) + + conn.commit() + + def remove_db_file(): """This function use to remove SQLite DB file""" if os.path.isfile(config.TEST_SQLITE_PATH): diff --git a/web/regression/runtests.py b/web/regression/runtests.py index d786692..21e4e35 100644 --- a/web/regression/runtests.py +++ b/web/regression/runtests.py @@ -114,6 +114,9 @@ test_client = app.test_client() driver = None app_starter = None handle_cleanup = None +app.PGADMIN_RUNTIME = True +if config.SERVER_MODE is True: + app.PGADMIN_RUNTIME = False setattr(unit_test.result.TestResult, "passed", []) @@ -234,7 +237,6 @@ def get_test_modules(arguments): # Sort module list so that test suite executes the test cases sequentially module_list = TestsGeneratorRegistry.registry.items() module_list = sorted(module_list, key=lambda module_tuple: module_tuple[0]) - return module_list @@ -393,6 +395,15 @@ if __name__ == '__main__': # Create test server server_information = test_utils.create_parent_server_node(server) + if server['DEFAULT_BINARY_PATHS'] is not None: + test_utils.set_preference(server['DEFAULT_BINARY_PATHS']) + + config.DEFAULT_BINARY_PATHS = { + "pg": str(server['DEFAULT_BINARY_PATHS']['pg']), + "ppas": str(server['DEFAULT_BINARY_PATHS']['ppas']), + "gpdb": "" + } + suite = get_suite(test_module_list, server, test_client, diff --git a/web/regression/test_config.json.in b/web/regression/test_config.json.in index ebc1466..cb6b98c 100644 --- a/web/regression/test_config.json.in +++ b/web/regression/test_config.json.in @@ -23,7 +23,12 @@ "maintenance_db": "postgres", "sslmode": "prefer", "tablespace_path": "", - "enabled": true + "enabled": true, + "DEFAULT_BINARY_PATHS": { + "pg": "/opt/PostgreSQL/9.4/bin/", + "ppas": "/opt/edb/as10/bin/", + "gpdb": "" + } } ], "server_update_data": [