diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_add.py b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_add.py index 9d8f949..11b9fa7 100644 --- a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_add.py +++ b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_add.py @@ -6,9 +6,11 @@ # This software is released under the PostgreSQL Licence # # ################################################################## +import json from pgadmin.utils.route import BaseTestGenerator from regression import test_utils as utils +from regression import test_server_dict from pgadmin.browser.server_groups.servers.tests import utils as server_utils from . import utils as roles_utils @@ -21,35 +23,33 @@ class LoginRoleAddTestCase(BaseTestGenerator): ('Check Role Node', dict(url='/browser/role/obj/')) ] - @classmethod - def setUpClass(cls): - """ - This function used to add the sever - - :return: None - """ - - # Add the server - server_utils.add_server(cls.tester) - - # Connect to server - cls.server_connect_response, cls.server_group, cls.server_ids = \ - server_utils.connect_server(cls.tester) - - if len(cls.server_connect_response) == 0: - raise Exception("No Server(s) connected to add the roles!!!") + def setUp(self): + pass def runTest(self): """This function test the add role scenario""" - - roles_utils.add_role(self.tester, self.server_connect_response, - self.server_group, self.server_ids) - - @classmethod - def tearDownClass(cls): - """This function deletes the role,server and parent id file""" - - roles_utils.delete_role(cls.tester) - server_utils.delete_server(cls.tester) - utils.delete_parent_id_file() - + server_id = test_server_dict["server"][0]["server_id"] + server_response = server_utils.connect_server(self, server_id) + if not server_response['data']['connected']: + raise Exception("Server not found to add the role.") + + data = roles_utils.get_role_data(self.server['db_password']) + self.role_name = data['rolname'] + response = self.tester.post(self.url + str(utils.SERVER_GROUP) + '/' + + str(server_id) + '/', + data=json.dumps(data), + content_type='html/json') + self.assertEquals(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + role_id = response_data['node']['_id'] + role_dict = {"server_id": server_id, "role_id": role_id} + utils.write_node_info(role_id, "lrid", role_dict) + + def tearDown(self): + """This function delete the role from added server""" + connection = utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + roles_utils.delete_role(connection, self.role_name) diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_delete.py b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_delete.py index f3204cd..9a495ff 100644 --- a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_delete.py +++ b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_delete.py @@ -6,53 +6,40 @@ # This software is released under the PostgreSQL Licence # # ################################################################## - +import uuid from pgadmin.utils.route import BaseTestGenerator from regression import test_utils as utils -from pgadmin.browser.server_groups.servers.tests import utils as server_utils +from regression import test_server_dict from . import utils as roles_utils class LoginRoleDeleteTestCase(BaseTestGenerator): """This class has delete role scenario""" - scenarios = [ # Fetching default URL for roles node. ('Check Role Node', dict(url='/browser/role/obj/')) ] - @classmethod - def setUpClass(cls): - """ - This function used to add the sever and roles - - :return: None - """ - - # Add the server - server_utils.add_server(cls.tester) - - # Connect to server - cls.server_connect_response, cls.server_group, cls.server_ids = \ - server_utils.connect_server(cls.tester) - - if len(cls.server_connect_response) == 0: - raise Exception("No Server(s) connected to add the roles!!!") - - # Add the role - roles_utils.add_role(cls.tester, cls.server_connect_response, - cls.server_group, cls.server_ids) + def setUp(self): + self.role_name = "role_delete_%s" % str(uuid.uuid4())[1:6] + self.role_id = roles_utils.create_role(self.server, self.role_name) def runTest(self): - """This function tests the delete role scenario""" - - roles_utils.delete_role(self.tester) - - @classmethod - def tearDownClass(self): - """This function deletes the role,server and parent id file""" - - server_utils.delete_server(self.tester) - utils.delete_parent_id_file() + """This function test the delete role scenario""" + server_id = test_server_dict["server"][0]["server_id"] + response = self.tester.delete( + self.url + str(utils.SERVER_GROUP) + '/' + + str(server_id) + '/' + str(self.role_id), + follow_redirects=True) + self.assertEquals(response.status_code, 200) + + def tearDown(self): + """This function delete the role from added server""" + connection = utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + roles_utils.delete_role(connection, self.role_name) diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_get.py b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_get.py index 4187fca..736ae3c 100644 --- a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_get.py +++ b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_get.py @@ -6,62 +6,39 @@ # This software is released under the PostgreSQL Licence # # ################################################################## +import uuid from pgadmin.utils.route import BaseTestGenerator from regression import test_utils as utils -from pgadmin.browser.server_groups.servers.tests import utils as server_utils +from regression import test_server_dict from . import utils as roles_utils class LoginRoleGetTestCase(BaseTestGenerator): """This class tests the get role scenario""" - scenarios = [ # Fetching default URL for roles node. ('Check Role Node', dict(url='/browser/role/obj/')) ] - @classmethod - def setUpClass(cls): - """ - This function used to add the sever and roles - - :return: None - """ - - # Add the server - server_utils.add_server(cls.tester) - - # Connect to server - cls.server_connect_response, cls.server_group, cls.server_ids = \ - server_utils.connect_server(cls.tester) - - if len(cls.server_connect_response) == 0: - raise Exception("No Server(s) connected to get the roles!!!") - - # Add the role - roles_utils.add_role(cls.tester, cls.server_connect_response, - cls.server_group, cls.server_ids) + def setUp(self): + self.role_name = "role_get_%s" % str(uuid.uuid4())[1:6] + self.role_id = roles_utils.create_role(self.server, self.role_name) def runTest(self): """This function test the get role scenario""" - - all_id = utils.get_ids() - server_ids = all_id["sid"] - role_ids_dict = all_id["lrid"][0] - - for server_id in server_ids: - role_id = role_ids_dict[int(server_id)] - response = self.tester.get( - self.url + str(utils.SERVER_GROUP) + '/' + - str(server_id) + '/' + str(role_id), - follow_redirects=True) - self.assertEquals(response.status_code, 200) - - @classmethod - def tearDownClass(cls): - """This function deletes the role,server and parent id file""" - - roles_utils.delete_role(cls.tester) - server_utils.delete_server(cls.tester) - utils.delete_parent_id_file() + server_id = test_server_dict["server"][0]["server_id"] + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(server_id) + '/' + str(self.role_id), + follow_redirects=True) + self.assertEquals(response.status_code, 200) + + def tearDown(self): + """This function delete the role from added server""" + connection = utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + roles_utils.delete_role(connection, self.role_name) diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_put.py b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_put.py index f8c353e..8b6a6aa 100644 --- a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_put.py +++ b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_put.py @@ -8,77 +8,48 @@ # ################################################################## import json +import uuid from pgadmin.utils.route import BaseTestGenerator from regression import test_utils as utils -from pgadmin.browser.server_groups.servers.tests import utils as server_utils -from regression.test_setup import advanced_config_data +from regression import test_server_dict from . import utils as roles_utils class LoginRolePutTestCase(BaseTestGenerator): """This class has update role scenario""" - scenarios = [ # Fetching default URL for roles node. ('Check Role Node', dict(url='/browser/role/obj/')) ] - @classmethod - def setUpClass(cls): - """ - This function used to add the sever and roles - - :return: None - """ - - # Add the server - server_utils.add_server(cls.tester) - - # Connect to server - cls.server_connect_response, cls.server_group, cls.server_ids = \ - server_utils.connect_server(cls.tester) - - if len(cls.server_connect_response) == 0: - raise Exception("No Server(s) connected to get the roles!!!") - - # Add the role - roles_utils.add_role(cls.tester, cls.server_connect_response, - cls.server_group, cls.server_ids) + def setUp(self): + self.role_name = "role_put_%s" % str(uuid.uuid4())[1:6] + self.role_id = roles_utils.create_role(self.server, self.role_name) def runTest(self): """This function tests the update role data scenario""" + server_id = test_server_dict["server"][0]["server_id"] + role_response = roles_utils.verify_role(self.server, self.role_name) + if len(role_response) == 0: + raise Exception("No roles(s) to update!!!") + data = { + "description": "This is the test description for cast", + "lrid": self.role_id + } + put_response = self.tester.put( + self.url + str(utils.SERVER_GROUP) + '/' + + str(server_id) + '/' + str(self.role_id), + data=json.dumps(data), + follow_redirects=True) + self.assertEquals(put_response.status_code, 200) + + def tearDown(self): + """This function delete the role from added server""" + connection = utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + roles_utils.delete_role(connection, self.role_name) - all_id = utils.get_ids() - server_ids = all_id["sid"] - role_ids_dict = all_id["lrid"][0] - - for server_id in server_ids: - role_id = role_ids_dict[int(server_id)] - role_response = roles_utils.verify_role(self.tester, - utils.SERVER_GROUP, - server_id, - role_id) - if len(role_response) == 0: - raise Exception("No roles(s) to update!!!") - - data = { - "description": advanced_config_data["lr_update_data"] - ["comment"], - "lrid": role_id - } - put_response = self.tester.put( - self.url + str(utils.SERVER_GROUP) + '/' + - str(server_id) + '/' + str(role_id), - data=json.dumps(data), - follow_redirects=True) - - self.assertEquals(put_response.status_code, 200) - - @classmethod - def tearDownClass(cls): - """This function deletes the role,server and parent id file""" - - roles_utils.delete_role(cls.tester) - server_utils.delete_server(cls.tester) - utils.delete_parent_id_file() diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/utils.py b/web/pgadmin/browser/server_groups/servers/roles/tests/utils.py index 85e0990..491ba2e 100644 --- a/web/pgadmin/browser/server_groups/servers/roles/tests/utils.py +++ b/web/pgadmin/browser/server_groups/servers/roles/tests/utils.py @@ -6,45 +6,51 @@ # This software is released under the PostgreSQL Licence # # ################################################################## +from __future__ import print_function import json import os import pickle import uuid +import sys -from regression.test_setup import pickle_path, config_data, advanced_config_data +from regression.test_setup import config_data from pgadmin.browser.server_groups.servers.tests import utils as server_utils from regression import test_utils as utils ROLE_URL = '/browser/role/obj/' +file_name = os.path.basename(__file__) -def verify_role(tester, server_group, server_id, role_id): +def verify_role(server, role_name): """ This function calls the GET API for role to verify - - :param tester: test client - :type tester: flask test client object - :param server_group: server group id - :type server_group: int - :param server_id: server id - :type server_id: str - :param role_id: role id - :type role_id: int - :return: dict/None + :param server: server details + :type server: dict + :param role_name: role name + :type role_name: str + :return role: role record from db + :rtype role: dict """ - - srv_connect = server_utils.verify_server(tester, server_group, server_id) - if srv_connect['data']['connected']: - response = tester.get( - '{0}{1}/{2}/{3}'.format(ROLE_URL, server_group, server_id, - role_id), - content_type='html/json') - temp_response = json.loads(response.data.decode('utf-8')) - return temp_response - else: - return None + try: + connection = utils.get_db_connection(server['db'], + server['username'], + server['db_password'], + server['host'], + server['port']) + pg_cursor = connection.cursor() + pg_cursor.execute( + "SELECT * from pg_catalog.pg_roles pr WHERE pr.rolname='%s'" % + role_name) + connection.commit() + role = pg_cursor.fetchone() + connection.close() + return role + except Exception as exception: + exception = "Error while getting role: %s: line:%s %s" % ( + file_name, sys.exc_traceback.tb_lineno, exception) + print(exception, file=sys.stderr) def test_getrole(tester): @@ -65,58 +71,55 @@ def test_getrole(tester): return role_response_data -def get_role_data(): - """This function returns the role data from config file""" - +def get_role_data(lr_pwd): + """This function returns the role data""" data = { - "rolcanlogin": advanced_config_data['lr_credentials'] - ['can_login'], - "rolconnlimit": advanced_config_data['lr_credentials'] - ['conn_limit'], - "rolcreaterole": advanced_config_data['lr_credentials'] - ['create_role'], - "rolinherit": advanced_config_data['lr_credentials'] - ['role_inherit'], - "rolmembership": advanced_config_data['lr_credentials'] - ['role_membership'], - "rolname": str(uuid.uuid4())[1:8], - "rolpassword": advanced_config_data['lr_credentials'] - ['lr_password'], - "rolvaliduntil": advanced_config_data['lr_credentials'] - ['lr_validity'], - "seclabels": advanced_config_data['lr_credentials'] - ['sec_lable'], - "variables": advanced_config_data['lr_credentials'] - ['variable'] + "rolcanlogin": "true", + "rolconnlimit": -1, + "rolcreaterole": "true", + "rolinherit": "true", + "rolmembership": [], + "rolname": "test_role_%s" % str(uuid.uuid4())[1:6], + "rolpassword": lr_pwd, + "rolvaliduntil": "12/27/2016", + "seclabels": [], + "variables": [{"name": "work_mem", + "database": "postgres", + "value": 65}] } return data -def add_role(tester, server_connect_response, server_group, server_ids): +def create_role(server, role_name): """ - This function is used to add the roles to server - - :param tester: flask test client - :type tester: flask test client object - :param server_connect_response: server connect API response - :type server_connect_response: dict - :param server_group: server group - :type server_group: int - :param server_ids: list of server id - :type server_ids: list - :return: None + This function create the role. + :param server: + :param role_name: + :return: """ - - for server_connect, server_id in zip(server_connect_response, - server_ids): - if server_connect['data']['connected']: - data = get_role_data() - response = tester.post(ROLE_URL + str(server_group) + '/' - + server_id + '/', data=json.dumps(data), - content_type='html/json') - assert response.status_code == 200 - response_data = json.loads(response.data.decode('utf-8')) - write_role_id(response_data) + try: + connection = utils.get_db_connection(server['db'], + server['username'], + server['db_password'], + server['host'], + server['port']) + pg_cursor = connection.cursor() + pg_cursor.execute("CREATE ROLE %s LOGIN" % role_name) + connection.commit() + # Get 'oid' from newly created tablespace + pg_cursor.execute( + "SELECT pr.oid from pg_catalog.pg_roles pr WHERE pr.rolname='%s'" % + role_name) + oid = pg_cursor.fetchone() + role_id = '' + if oid: + role_id = oid[0] + connection.close() + return role_id + except Exception as exception: + exception = "Error while deleting role: %s: line:%s %s" % ( + file_name, sys.exc_traceback.tb_lineno, exception) + print(exception, file=sys.stderr) def write_role_id(response_data): @@ -147,41 +150,30 @@ def write_role_id(response_data): db_output.close() -def delete_role(tester): +def delete_role(connection, role_name): """ This function use to delete the existing roles in the servers - :param tester: flask test client - :type tester: flask test object + :param connection: db connection + :type connection: connection object + :param role_name: role name + :type role_name: str :return: None """ - - server_ids = None - role_ids_dict = None - - all_id = utils.get_ids() - if "sid" and "lrid" in all_id.keys(): - server_ids = all_id["sid"] - if all_id['lrid']: - role_ids_dict = all_id['lrid'][0] - else: - raise Exception("Keys are not found: {}".format(["sid", "lrid"])) - - if server_ids and role_ids_dict is not None: - for server_id in server_ids: - server_response = server_utils.verify_server(tester, - utils.SERVER_GROUP, - server_id) - if server_response["data"]["connected"]: - role_id = role_ids_dict[int(server_id)] - response = tester.delete( - ROLE_URL + str(utils.SERVER_GROUP) + '/' + - str(server_id) + '/' + str(role_id), - follow_redirects=True) - assert response.status_code == 200 - response_data = json.loads(response.data.decode('utf-8')) - assert response_data['success'] == 1 - else: - raise Exception("No servers/roles found.") - - + try: + pg_cursor = connection.cursor() + pg_cursor.execute( + "SELECT * FROM pg_catalog.pg_roles WHERE rolname='%s'" % role_name) + role_count = pg_cursor.fetchone() + if role_count: + old_isolation_level = connection.isolation_level + connection.set_isolation_level(0) + pg_cursor = connection.cursor() + pg_cursor.execute("DROP ROLE %s" % role_name) + connection.set_isolation_level(old_isolation_level) + connection.commit() + connection.close() + except Exception as exception: + exception = "Error while deleting role: %s: line:%s %s" % ( + file_name, sys.exc_traceback.tb_lineno, exception) + print(exception, file=sys.stderr) diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_add.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_add.py index caa6966..09e9e0e 100644 --- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_add.py +++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_add.py @@ -6,50 +6,68 @@ # This software is released under the PostgreSQL Licence # # ################################################################## +from __future__ import print_function + +import os +import sys +import json from pgadmin.utils.route import BaseTestGenerator from regression import test_utils as utils +from regression import test_server_dict from pgadmin.browser.server_groups.servers.tests import utils as server_utils from . import utils as tablespace_utils class TableSpaceAddTestCase(BaseTestGenerator): """This class will add tablespace node under server""" - scenarios = [ # Fetching default URL for tablespace node. ('Check Tablespace Node', dict(url='/browser/tablespace/obj/')) ] - @classmethod - def setUpClass(cls): - """ - This function used to add the sever - - :return: None - """ - - # Add the server - server_utils.add_server(cls.tester) - - # Connect to server - cls.server_connect_response, cls.server_group, cls.server_ids = \ - server_utils.connect_server(cls.tester) - - if len(cls.server_connect_response) == 0: - raise Exception("No Server(s) connected to add the roles!!!") + def setUp(self): + self.tablespace_name = '' + if not self.server['tablespace_path']\ + or self.server['tablespace_path'] is None: + message = "Skipped tablespace add test case. Tablespace path" \ + " not configured for server: %s" % self.server['name'] + # Skip the test case if tablespace_path not found. + self.skipTest(message) def runTest(self): - """This function test the add tablespace scenario""" - - tablespace_status = tablespace_utils.add_table_space( - self.tester, self.server_connect_response, self.server_group, - self.server_ids) + """This function test the add tablespace API""" + server_id = test_server_dict["server"][0]["server_id"] + server_response = server_utils.connect_server(self, server_id) + if not server_response['data']['connected']: + raise Exception("Unable to connect server to get tablespace.") - @classmethod - def tearDownClass(cls): - """This function deletes the tablespaces,server and parent_id file""" + db_owner = server_response['data']['user']['name'] + table_space_path = self.server['tablespace_path'] + data = tablespace_utils.get_tablespace_data( + table_space_path, db_owner) + self.tablespace_name = data['name'] + response = self.tester.post( + self.url + str(utils.SERVER_GROUP) + '/' + + str(server_id) + '/', + data=json.dumps(data), + content_type='html/json') + self.assertEquals(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + tablespace_id = response_data['node']['_id'] + tablespace_dict = {"tablespace_id": tablespace_id, + "tablespace_name": self.tablespace_name, + "server_id": server_id} + utils.write_node_info(tablespace_id, "tsid", tablespace_dict) - tablespace_utils.delete_table_space(cls.tester) - server_utils.delete_server(cls.tester) - utils.delete_parent_id_file() + def tearDown(self): + """ + This function delete the tablespace from server added in SQLite and + clears the node_info_dict + """ + connection = utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + tablespace_utils.delete_tablespace(connection, self.tablespace_name) diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_delete.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_delete.py index f280ee6..605b6af 100644 --- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_delete.py +++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_delete.py @@ -6,11 +6,13 @@ # This software is released under the PostgreSQL Licence # # ################################################################## - +from __future__ import print_function +import uuid +import json from pgadmin.utils.route import BaseTestGenerator from regression import test_utils as utils -from pgadmin.browser.server_groups.servers.tests import utils as server_utils +from regression import test_server_dict from . import utils as tablespace_utils @@ -22,37 +24,39 @@ class TableSpaceDeleteTestCase(BaseTestGenerator): ('Check Tablespace Node', dict(url='/browser/tablespace/obj/')) ] - @classmethod - def setUpClass(cls): - """ - This function used to add the sever - - :return: None - """ - - # Add the server - server_utils.add_server(cls.tester) - - # Connect to server - cls.server_connect_response, cls.server_group, cls.server_ids = \ - server_utils.connect_server(cls.tester) - - if len(cls.server_connect_response) == 0: - raise Exception("No Server(s) connected to add the roles!!!") - - # Add tablespace - tablespace_utils.add_table_space(cls.tester, - cls.server_connect_response, - cls.server_group, cls.server_ids) + def setUp(self): + if not self.server['tablespace_path']\ + or self.server['tablespace_path'] is None: + message = "Skipped tablespace delete test case. Tablespace path" \ + " not configured for server: %s" % self.server['name'] + # Skip the test case if tablespace_path not found. + self.skipTest(message) + self.tablespace_name = "tablespace_delete_%s" % str(uuid.uuid4())[1:6] + self.tablespace_id = tablespace_utils.create_tablespace( + self.server, self.tablespace_name) def runTest(self): - """This function tests the delete table space scenario""" - - tablespace_utils.delete_table_space(self.tester) - - @classmethod - def tearDownClass(cls): - """This function deletes the server and parent id file""" + """This function tests the delete table space api""" + server_id = test_server_dict["server"][0]["server_id"] + tablespace_count = tablespace_utils.verify_table_space( + self.server, self.tablespace_name) + if tablespace_count == 0: + raise Exception("No tablespace(s) to delete!!!") + + response = self.tester.delete(self.url + str(utils.SERVER_GROUP) + + '/' + str(server_id) + '/' + + str(self.tablespace_id), + follow_redirects=True) + self.assertEquals(response.status_code, 200) + delete_response_data = json.loads(response.data.decode('utf-8')) + self.assertEquals(delete_response_data['success'], 1) + + def tearDown(self): + """This function deletes the tablespace""" + connection = utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + tablespace_utils.delete_tablespace(connection, self.tablespace_name) - server_utils.delete_server(cls.tester) - utils.delete_parent_id_file() diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_get.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_get.py index b26d71d..4f30ed6 100644 --- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_get.py +++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_get.py @@ -6,9 +6,11 @@ # This software is released under the PostgreSQL Licence # # ################################################################## +import uuid from pgadmin.utils.route import BaseTestGenerator from regression import test_utils as utils +from regression import test_server_dict from pgadmin.browser.server_groups.servers.tests import utils as server_utils from . import utils as tablespace_utils @@ -21,58 +23,39 @@ class TablespaceGetTestCase(BaseTestGenerator): ('Check Tablespace Node', dict(url='/browser/tablespace/obj/')) ] - @classmethod - def setUpClass(cls): - """ - This function used to add the sever - - :return: None - """ - - # Add the server - server_utils.add_server(cls.tester) - - # Connect to server - cls.server_connect_response, cls.server_group, cls.server_ids = \ - server_utils.connect_server(cls.tester) - - if len(cls.server_connect_response) == 0: - raise Exception("No Server(s) connected to add the roles!!!") - - # Add tablespace - tablespace_utils.add_table_space(cls.tester, - cls.server_connect_response, - cls.server_group, cls.server_ids) + def setUp(self): + if not self.server['tablespace_path']\ + or self.server['tablespace_path'] is None: + message = "Skipped tablespace get test case. Tablespace path" \ + " not configured for server: %s" % self.server['name'] + # Skip the test case if tablespace_path not found. + self.skipTest(message) + self.tablespace_name = "tablespace_delete_%s" % str(uuid.uuid4())[1:6] + self.tablespace_id = tablespace_utils.create_tablespace( + self.server, self.tablespace_name) def runTest(self): """This function test the get table space scenario""" - - tablespace_ids_dict = None - all_id = utils.get_ids() - server_ids = all_id["sid"] - if "tsid" in all_id and all_id["tsid"]: - tablespace_ids_dict = all_id["tsid"][0] - - if tablespace_ids_dict: - for server_id in server_ids: - tablespace_id = tablespace_ids_dict[int(server_id)] - server_response = server_utils.verify_server(self.tester, - utils.SERVER_GROUP, - server_id) - if server_response['data']['connected']: - tablespace_utils.verify_table_space( - self.tester, utils.SERVER_GROUP, server_id, tablespace_id) - response = self.tester.get( - self.url + str(utils.SERVER_GROUP) + '/' + - str(server_id) + '/' + str( - tablespace_id), - follow_redirects=True) - self.assertEquals(response.status_code, 200) - - @classmethod - def tearDownClass(cls): - """This function deletes the tablespaces,server and parent id file""" - - tablespace_utils.delete_table_space(cls.tester) - server_utils.delete_server(cls.tester) - utils.delete_parent_id_file() + server_id = test_server_dict["server"][0]["server_id"] + server_response = server_utils.connect_server(self, server_id) + if not server_response['data']['connected']: + raise Exception("Unable to connect server to get tablespace.") + + tablespace_count = tablespace_utils.verify_table_space( + self.server, self.tablespace_name) + if tablespace_count == 0: + raise Exception("No tablespace(s) to update!!!") + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(server_id) + '/' + str(self.tablespace_id), + follow_redirects=True) + self.assertEquals(response.status_code, 200) + + def tearDown(self): + """This function delete the tablespace from added server""" + connection = utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + tablespace_utils.delete_tablespace(connection, self.tablespace_name) diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_put.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_put.py index 22257f9..b5e1986 100644 --- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_put.py +++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_put.py @@ -8,11 +8,11 @@ # ################################################################## import json +import uuid from pgadmin.utils.route import BaseTestGenerator from regression import test_utils as utils -from pgadmin.browser.server_groups.servers.tests import utils as server_utils -from test_setup import advanced_config_data +from regression import test_server_dict from . import utils as tablespace_utils @@ -24,67 +24,40 @@ class TableSpaceUpdateTestCase(BaseTestGenerator): ('Check Tablespace Node', dict(url='/browser/tablespace/obj/')) ] - @classmethod - def setUpClass(cls): - """ - This function used to add the sever - - :return: None - """ - - # Add the server - server_utils.add_server(cls.tester) - - # Connect to server - cls.server_connect_response, cls.server_group, cls.server_ids = \ - server_utils.connect_server(cls.tester) - - if len(cls.server_connect_response) == 0: - raise Exception("No Server(s) connected to add the roles!!!") - - # Add tablespace - tablespace_utils.add_table_space(cls.tester, - cls.server_connect_response, - cls.server_group, cls.server_ids) + def setUp(self): + if not self.server['tablespace_path']\ + or self.server['tablespace_path'] is None: + message = "Skipped tablespace delete test case. Tablespace path" \ + " not configured for server: %s" % self.server['name'] + # Skip the test case if tablespace_path not found. + self.skipTest(message) + self.tablespace_name = "tablespace_delete_%s" % str(uuid.uuid4())[1:6] + self.tablespace_id = tablespace_utils.create_tablespace( + self.server, self.tablespace_name) def runTest(self): """This function tests the update tablespace data scenario""" - - tablespace_ids_dict = None - all_id = utils.get_ids() - server_ids = all_id["sid"] - if "tsid" in all_id and all_id["tsid"]: - tablespace_ids_dict = all_id["tsid"][0] - - if tablespace_ids_dict: - for server_id in server_ids: - tablespace_id = tablespace_ids_dict[int(server_id)] - tablespace_response = tablespace_utils.verify_table_space( - self.tester, - utils.SERVER_GROUP, server_id, - tablespace_id) - if len(tablespace_response) == 0: - raise Exception("No tablespace(s) to update!!!") - - data = { - "description": advanced_config_data["tbspc_update_data"] - ["comment"], - "table_space_id": tablespace_id - } - - put_response = self.tester.put( - self.url + str(utils.SERVER_GROUP) + '/' + - str(server_id) + '/' + str( - tablespace_id), - data=json.dumps(data), - follow_redirects=True) - - self.assertEquals(put_response.status_code, 200) - - @classmethod - def tearDownClass(cls): - """This function deletes the tablespaces,server and parent id file""" - - tablespace_utils.delete_table_space(cls.tester) - server_utils.delete_server(cls.tester) - utils.delete_parent_id_file() + server_id = test_server_dict["server"][0]["server_id"] + tablespace_count = tablespace_utils.verify_table_space( + self.server, self.tablespace_name) + if tablespace_count == 0: + raise Exception("No tablespace(s) to update!!!") + + data = { + "description": "This is test description.", + "table_space_id": self.tablespace_id + } + put_response = self.tester.put( + self.url + str(utils.SERVER_GROUP) + '/' + str(server_id) + '/' + + str(self.tablespace_id), data=json.dumps(data), + follow_redirects=True) + self.assertEquals(put_response.status_code, 200) + + def tearDown(self): + """This function deletes the tablespace""" + connection = utils.get_db_connection(self.server['db'], + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port']) + tablespace_utils.delete_tablespace(connection, self.tablespace_name) diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/utils.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/utils.py index 4fd21bd..c9b6a8a 100644 --- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/utils.py +++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/utils.py @@ -9,187 +9,114 @@ from __future__ import print_function -import json import os -import pickle import uuid import sys -from regression.test_setup import pickle_path, config_data, \ - advanced_config_data -from pgadmin.browser.server_groups.servers.tests import utils as server_utils + from regression import test_utils as utils TABLE_SPACE_URL = '/browser/tablespace/obj/' - - -# Tablespace utility -def get_tablespace_data(server_connect): - """This function returns the tablespace data from config file""" - - adv_config_data = None - data = None - db_user = server_connect['data']['user']['name'] - server_config_data = config_data['server_credentials'] - - # Get the config data of appropriate db user - for config_test_data in advanced_config_data['tablespc_credentials']: - if db_user == config_test_data['spc_user']: - # Add the tablespace path from server config - server_config = (item for item in server_config_data if - item["db_username"] == db_user).next() - if "tablespace_path" in server_config: - config_test_data['spc_location'] = server_config['tablespace_path'] - adv_config_data = config_test_data - else: - config_test_data['spc_location'] = None - - if adv_config_data is not None: - data = { - "name": str(uuid.uuid4())[1:8], - "seclabels": adv_config_data["spc_seclable"], - "spcacl": adv_config_data["spc_acl"], - "spclocation": adv_config_data["spc_location"], - "spcoptions": adv_config_data["spc_opts"], - "spcuser": adv_config_data["spc_user"] - } +file_name = os.path.basename(__file__) + + +def get_tablespace_data(tablespace_path, db_owner): + """This function returns the tablespace data""" + data = { + "name": "test_%s" % str(uuid.uuid4())[1:8], + "seclabels": [], + "spcacl": [ + { + "grantee": db_owner, + "grantor": db_owner, + "privileges": [ + { + "privilege_type": "C", + "privilege": True, + "with_grant": False + } + ] + } + ], + "spclocation": tablespace_path, + "spcoptions": [], + "spcuser": db_owner + } return data -def write_tablespace_id(response_data): - """ - This function write the table space id to parent_id.pkl - - :param response_data: create table space API response data - :type response_data: dict - :return: None - """ - - table_space_id = response_data['node']['_id'] - server_id = response_data['node']['_pid'] - pickle_id_dict = utils.get_pickle_id_dict() - if os.path.isfile(pickle_path): - existing_server_id = open(pickle_path, 'rb') - tol_server_id = pickle.load(existing_server_id) - pickle_id_dict = tol_server_id - if 'tsid' in pickle_id_dict: - if pickle_id_dict['tsid']: - # Add the db_id as value in dict - pickle_id_dict["tsid"][0].update( - {server_id: table_space_id}) - else: - # Create new dict with server_id and db_id - pickle_id_dict["tsid"].append( - {server_id: table_space_id}) - db_output = open(pickle_path, 'wb') - pickle.dump(pickle_id_dict, db_output) - db_output.close() - - -def add_table_space(tester, server_connect_response, server_group, server_ids): - """ - This function is used to add the roles to server - - :param tester: flask test client - :type tester: flask test client object - :param server_connect_response: server connect API response - :type server_connect_response: dict - :param server_group: server group - :type server_group: int - :param server_ids: list of server id - :type server_ids: list - :return: None - """ - - total_servers_count = len(server_ids) - servers_without_tablespace_path = [] - - for server_connect, server_id in zip(server_connect_response, - server_ids): - tablespace_path = server_connect['tablespace_path'] - # Skip the test case if tablespace_path does not exist - if not tablespace_path or tablespace_path is None: - file_name = os.path.basename( - sys._getframe().f_back.f_code.co_filename) - servers_without_tablespace_path.append(server_id) - if total_servers_count == len(servers_without_tablespace_path): - print("Skipping tablespaces test cases for the file <{0}>, " - "Tablespace path not configured for the servers: " - "{1}".format(file_name, server_ids), file=sys.stderr) - return - else: - print("Skipping tablespaces test case for the file <{0}>: " - "Tablespace path not configured for server: {1}".format( - file_name, server_id), file=sys.stderr) - continue - - if server_connect['data']['connected']: - data = get_tablespace_data(server_connect) - response = tester.post(TABLE_SPACE_URL + str(server_group) + '/' - + server_id + '/', - data=json.dumps(data), - content_type='html/json') - assert response.status_code == 200 - response_data = json.loads(response.data.decode('utf-8')) - write_tablespace_id(response_data) - - -def verify_table_space(tester, server_group, server_id, tablespace_id): +def create_tablespace(server, test_tablespace_name): + try: + connection = utils.get_db_connection(server['db'], + 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("CREATE TABLESPACE %s LOCATION '%s'" % + (test_tablespace_name, server['tablespace_path'])) + connection.set_isolation_level(old_isolation_level) + connection.commit() + + # Get 'oid' from newly created tablespace + pg_cursor.execute( + "SELECT ts.oid from pg_tablespace ts WHERE ts.spcname='%s'" % + test_tablespace_name) + oid = pg_cursor.fetchone() + tspc_id = '' + if oid: + tspc_id = oid[0] + connection.close() + return tspc_id + except Exception as exception: + raise Exception("Error while creating tablespace. %s" % exception) + + +def verify_table_space(server, test_tablespace_name): """ This function calls the GET API for role to verify - - :param tester: test client - :type tester: flask test client object - :param server_group: server group id - :type server_group: int - :param server_id: server id - :type server_id: str - :param tablespace_id: table space id - :type tablespace_id: int - :return: dict/None - """ - - srv_connect = server_utils.verify_server(tester, server_group, server_id) - if srv_connect['data']['connected']: - response = tester.get( - '{0}{1}/{2}/{3}'.format(TABLE_SPACE_URL, server_group, server_id, - tablespace_id), - content_type='html/json') - assert response.status_code == 200 - temp_response = json.loads(response.data.decode('utf-8')) - return temp_response - else: - return None - - -def delete_table_space(tester): + :param server: server info + :type server: dict + :param test_tablespace_name: tablespace name + :type test_tablespace_name: str + :return tablespace_count: tablespace count + :rtype: int """ - This function use to delete the existing tablespace in the servers - - :param tester: flask test client - :type tester: flask test object - :return: None - """ - - all_id = utils.get_ids() - server_ids = all_id["sid"] - if "tsid" in all_id and all_id["tsid"]: - tablespace_ids_dict = all_id["tsid"][0] - else: - tablespace_ids_dict = None - - if tablespace_ids_dict is not None: - for server_id in server_ids: - tablespace_id = tablespace_ids_dict[int(server_id)] - role_response = verify_table_space(tester, utils.SERVER_GROUP, - server_id, - tablespace_id) - if len(role_response) == 0: - raise Exception("No tablespace(s) to delete!!!") - response = tester.delete( - TABLE_SPACE_URL + str(utils.SERVER_GROUP) + '/' + - str(server_id) + '/' + str(tablespace_id), - follow_redirects=True) - assert response.status_code == 200 - delete_response_data = json.loads(response.data.decode('utf-8')) - assert delete_response_data['success'] == 1 + try: + connection = utils.get_db_connection(server['db'], + server['username'], + server['db_password'], + server['host'], + server['port']) + pg_cursor = connection.cursor() + pg_cursor.execute("SELECT * FROM pg_tablespace ts WHERE" + " ts.spcname='%s'" % test_tablespace_name) + tablespace_count = len(pg_cursor.fetchall()) + connection.close() + return tablespace_count + except Exception as exception: + exception = "%s: line:%s %s" % ( + file_name, sys.exc_traceback.tb_lineno, exception) + print(exception, file=sys.stderr) + + +def delete_tablespace(connection, test_tablespace_name): + try: + pg_cursor = connection.cursor() + pg_cursor.execute("SELECT * FROM pg_tablespace ts WHERE" + " ts.spcname='%s'" % test_tablespace_name) + tablespace_count = len(pg_cursor.fetchall()) + if tablespace_count: + old_isolation_level = connection.isolation_level + connection.set_isolation_level(0) + pg_cursor.execute("DROP TABLESPACE %s" % test_tablespace_name) + connection.set_isolation_level(old_isolation_level) + connection.commit() + connection.close() + except Exception as exception: + exception = "%s: line:%s %s" % ( + file_name, sys.exc_traceback.tb_lineno, exception) + print(exception, file=sys.stderr) + raise Exception(exception) diff --git a/web/regression/__init__.py b/web/regression/__init__.py index c001be6..6b7cb2f 100644 --- a/web/regression/__init__.py +++ b/web/regression/__init__.py @@ -6,29 +6,32 @@ # #This software is released under the PostgreSQL Licence # # ########################################################################## +import pgadmin.browser.server_groups.servers.roles.tests.utils as roles_utils +import pgadmin.browser.server_groups.servers.tablespaces.tests.utils as \ + tablespace_utils global node_info_dict node_info_dict = { - "sid": [], # server - "did": [], # database - "lrid": [], # role - "tsid": [], # tablespace - "scid": [], # schema - "tfnid": [], # trigger functions - "coid": [], # collation - "cid": [], # casts - "etid": [], # event_trigger - "eid": [], # extension - "fid": [], # FDW - "fsid": [], # FRS - "umid": [], # user_mapping - "seid": [] # sequence + "sid": [], # server + "did": [], # database + "lrid": [], # role + "tsid": [], # tablespace + "scid": [], # schema + "tfnid": [], # trigger functions + "coid": [], # collation + "cid": [], # casts + "etid": [], # event_trigger + "eid": [], # extension + "fid": [], # FDW + "fsid": [], # FRS + "umid": [], # user_mapping + "seid": [] # sequence } global test_server_dict test_server_dict = { - "server": [], - "database": [], - "tablespace": [], - "role": [] -} \ No newline at end of file + "server": [], + "database": [], + "tablespace": [], + "role": [] +} diff --git a/web/regression/runtests.py b/web/regression/runtests.py index 678adc8..8655ba0 100644 --- a/web/regression/runtests.py +++ b/web/regression/runtests.py @@ -182,14 +182,12 @@ def get_tests_result(test_suite): error_case[0]).split('.')[-1].split()[0].strip(')') if class_name not in failed_cases_result: failed_cases_result.append(class_name) - # TODO: Code remaining to count the skipped test cases - # if test_suite.skipped: - # for skip_test in test_suite.skipped: - # class_name = str( - # skip_test[0]).split('.')[-1].split()[0].strip(')') - # if class_name not in failed_cases_result: - # skipped_cases_result.append(class_name) - # print(class_name) + if test_suite.skipped: + for skip_test in test_suite.skipped: + class_name = str( + skip_test[0]).split('.')[-1].split()[0].strip(')') + if class_name not in failed_cases_result: + skipped_cases_result.append(class_name) return total_ran, failed_cases_result, skipped_cases_result except Exception as exception: exception = "Exception: %s: line:%s %s" % ( @@ -275,15 +273,22 @@ if __name__ == '__main__': print("============================", file=sys.stderr) for server_res in test_result: failed_cases = "\n\t".join(test_result[server_res][1]) + skipped_cases = "\n\t".join(test_result[server_res][2]) total_failed = len(test_result[server_res][1]) - # TODO : code remaining to handle '-1' condition - total_passed = int(test_result[server_res][0]) - total_failed - - print("%s: %s test%s passed, %s test%s failed %s%s" % - (server_res, total_passed, (total_passed != 1 and "s" or ""), - total_failed, (total_failed != 1 and "s" or ""), - (total_failed != 0 and ":\n\t" or ""), failed_cases), - file=sys.stderr) + total_skipped = len(test_result[server_res][2]) + total_passed_cases = int( + test_result[server_res][0]) - total_failed - total_skipped + + print( + "%s: %s test%s passed, %s test%s failed %s%s," + " %s test%s skipped %s%s" % + (server_res, total_passed_cases, + (total_passed_cases != 1 and "s" or ""), + total_failed, (total_failed != 1 and "s" or ""), + (total_failed != 0 and ":\n\t" or ""), failed_cases, + total_skipped, (total_skipped != 1 and "s" or ""), + (total_skipped != 0 and ":\n\t" or ""), skipped_cases), + file=sys.stderr) print("============================", file=sys.stderr) diff --git a/web/regression/test_utils.py b/web/regression/test_utils.py index 73c45a6..731f1b6 100644 --- a/web/regression/test_utils.py +++ b/web/regression/test_utils.py @@ -13,11 +13,11 @@ import sys import uuid import psycopg2 import sqlite3 +from functools import partial import config import test_setup import regression -from functools import partial SERVER_GROUP = test_setup.config_data['server_group'] file_name = os.path.basename(__file__) @@ -78,7 +78,7 @@ def get_config_data(): "db_password": srv['db_password'], "role": "", "sslmode": srv['sslmode'], - "tablespace_path": srv['tablespace_path']} + "tablespace_path": srv.get('tablespace_path', None)} server_data.append(data) return server_data @@ -138,7 +138,9 @@ def create_database(server, db_name): connection.close() return db_id except Exception as exception: - raise Exception("Error while creating database. %s" % exception) + exception = "Error while creating database: %s: line:%s %s" % ( + file_name, sys.exc_traceback.tb_lineno, exception) + print(exception, file=sys.stderr) def drop_database(connection, database_name): @@ -217,54 +219,6 @@ def delete_server_from_sqlite(sid): print(exception, file=sys.stderr) -def create_tablespace(server, test_tablespace_name): - try: - connection = get_db_connection(server['db'], - 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("CREATE TABLESPACE %s LOCATION '%s'" % - (test_tablespace_name, server['tablespace_path'])) - connection.set_isolation_level(old_isolation_level) - connection.commit() - - # Get 'oid' from newly created tablespace - pg_cursor.execute( - "SELECT ts.oid from pg_tablespace ts WHERE ts.spcname='%s'" % - test_tablespace_name) - oid = pg_cursor.fetchone() - tspc_id = '' - if oid: - tspc_id = oid[0] - connection.close() - return tspc_id - except Exception as exception: - raise Exception("Error while creating tablespace. %s" % exception) - - -def delete_tablespace(connection, test_tablespace_name): - try: - pg_cursor = connection.cursor() - pg_cursor.execute("SELECT * FROM pg_tablespace ts WHERE" - " ts.spcname='%s'" % test_tablespace_name) - tablespace_count = len(pg_cursor.fetchall()) - if tablespace_count: - old_isolation_level = connection.isolation_level - connection.set_isolation_level(0) - pg_cursor.execute("DROP TABLESPACE %s" % test_tablespace_name) - connection.set_isolation_level(old_isolation_level) - connection.commit() - connection.close() - except Exception as exception: - exception = "%s: line:%s %s" % ( - file_name, sys.exc_traceback.tb_lineno, exception) - print(exception, file=sys.stderr) - - def create_test_server(server_info): """ This function create the test server which will act as parent server, @@ -280,11 +234,6 @@ def create_test_server(server_info): test_db_name = "test_db_%s" % str(uuid.uuid4())[1:8] db_id = create_database(server_info, test_db_name) - # TODO: Need to decide about test tablespace creation - # Create tablespace - # test_tablespace_name = "test_tablespace" - # tablespace_id = create_tablespace(server, test_tablespace_name) - # Add server info to test_server_dict regression.test_server_dict["server"].append({"server_id": srv_id, "server": server_info}) @@ -390,7 +339,32 @@ def _drop_objects(tester): "pg_global"]: tablespace_name = tablespace[0] # Delete tablespace - delete_tablespace(connection, tablespace_name) + connection = get_db_connection(server_info[3], + server_info[4], + db_password, + server_info[1], + server_info[2]) + regression.tablespace_utils.delete_tablespace( + connection, tablespace_name) + + # Delete role + connection = get_db_connection(server_info[3], + server_info[4], + db_password, + server_info[1], + server_info[2]) + pg_cursor = connection.cursor() + pg_cursor.execute("SELECT * FROM pg_catalog.pg_roles") + roles = pg_cursor.fetchall() + if roles: + for role_name in roles: + # Do not delete default table spaces + if role_name[0] not in ["postgres", + "enterprisedb"]: + role_name = role_name[0] + # Delete role + regression.roles_utils.delete_role(connection, + role_name) for server_info in all_servers: server_id = server_info[5]