diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/foreign_servers_test_data.json b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/foreign_servers_test_data.json new file mode 100644 index 000000000..fd2eea9a3 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/foreign_servers_test_data.json @@ -0,0 +1,344 @@ +{ + "foreign_server_create": [ + { + "name": "Create foreign server: With valid data.", + "url": "/browser/foreign_server/obj/", + "is_positive_test": true, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while creating foreign server: With invalid data.", + "url": "/browser/foreign_server/obj/", + "is_positive_test": false, + "error_fdw_id": true, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 410 + } + }, + { + "name": "Error while creating foreign server: With insufficient data.", + "url": "/browser/foreign_server/obj/", + "is_positive_test": false, + "missing_parameter": true, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 410 + } + }, + { + "name": "Error while creating foreign server: internal server error.", + "url": "/browser/foreign_server/obj/", + "is_positive_test": false, + "internal_server_error": true, + "test_data": {}, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar", + "return_value": "(False, 'Mocked Internal Server Error')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error while creating foreign server: error in db.", + "url": "/browser/foreign_server/obj/", + "is_positive_test": false, + "error_in_db": true, + "test_data": {}, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server Error while creating a foreign server')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "foreign_server_delete": [ + { + "name": "Delete foreign server", + "url": "/browser/foreign_server/obj/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while deleting a foreign server - Internal server error", + "url": "/browser/foreign_server/obj/", + "internal_server_error": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar", + "return_value": "(False, 'Mocked Internal Server Error')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error while deleting a foreign server - wrong foreign server id", + "url": "/browser/foreign_server/obj/", + "wrong_foreign_server_id": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + }, + "expected_data": { + "status_code": 410 + } + } + ], + "foreign_server_multiple_delete": [ + { + "name": "Delete multiple foreign server", + "url": "/browser/foreign_server/obj/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + } + ], + "foreign_server_dependent_dependency": [ + { + "name": "Get foreign server dependents", + "url": "/browser/foreign_server/dependent/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Get foreign server dependencies", + "url": "/browser/foreign_server/dependency/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + } + ], + "foreign_server_get_nodes": [ + { + "name": "Get foreign server nodes", + "url": "/browser/foreign_server/nodes/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error fetching foreign server nodes", + "url": "/browser/foreign_server/nodes/", + "is_positive_test": false, + "internal_server_error": true, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray", + "return_value": "(False, 'Mocked Internal Server Error while fetching foreign server nodes')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Get foreign server node", + "url": "/browser/foreign_server/nodes/", + "is_positive_test": true, + "node": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error fetching foreign server node", + "url": "/browser/foreign_server/nodes/", + "is_positive_test": false, + "node": true, + "internal_server_error": true, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray", + "return_value": "(False, 'Mocked Internal Server Error while fetching foreign server nodes')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "foreign_server_update": [ + { + "name": "Update foreign server", + "url": "/browser/foreign_server/obj/", + "is_positive_test": true, + "mocking_required": false, + "test_data": { + "description": "This is foreign server update comment", + "id": "PLACE_HOLDER" + }, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while updating a foreign server - Internal server error", + "url": "/browser/foreign_server/obj/", + "error_in_db": true, + "test_data": { + "description": "This is foreign server update comment", + "id": "PLACE_HOLDER" + }, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar", + "return_value": "[(False, 'Mocked Internal Server Error while fetching a foreign server')]" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "foreign_server_get": [ + { + "name": "Get foreign server properties", + "url": "/browser/foreign_server/obj/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while fetching a foreign server properties", + "url": "/browser/foreign_server/obj/", + "error_fetching_fsrv": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server Error while fetching a foreign server')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error while fetching foreign server with wrong ID", + "url": "/browser/foreign_server/obj/", + "wrong_fsrv_id": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": {}, + "expected_data": { + "status_code": 410 + } + }, + { + "name": "Get foreign server list", + "url": "/browser/foreign_server/obj/", + "is_positive_test": true, + "foreign_server_list": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while fetching a foreign server list", + "url": "/browser/foreign_server/obj/", + "error_fetching_fsrv": true, + "is_positive_test": false, + "foreign_server_list": true, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server Error while fetching a foreign server')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "foreign_server_get_sql": [ + { + "name": "Get Foreign server SQL", + "url": "/browser/foreign_server/sql/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error - Get Foreign server SQL - Internal server error", + "url": "/browser/foreign_server/sql/", + "is_positive_test": false, + "internal_server_error": true, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server Error while fetching a foreign server')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error - Get Foreign server SQL using wrong fsrv id", + "url": "/browser/foreign_server/sql/", + "wrong_foreign_server_id": true, + "is_positive_test": false, + "mocking_required": false, + "mock_data": { + }, + "expected_data": { + "status_code": 410 + } + } + ], + "foreign_server_get_msql": [ + { + "name": "Get foreign server msql", + "is_positive_test": true, + "url": "/browser/foreign_server/msql/", + "test_data": { + "name": "modifying name", + "comment": "Comments to test update" + }, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + } + ] +} diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_add.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_add.py index 7b07e01f6..32c4edb16 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_add.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_add.py @@ -21,6 +21,8 @@ from pgadmin.browser.server_groups.servers.databases.tests import \ from pgadmin.utils.route import BaseTestGenerator from regression import parent_node_dict from regression.python_test_utils import test_utils as utils +from . import utils as fsrv_utils +from unittest.mock import patch class ForeignServerAddTestCase(BaseTestGenerator): @@ -28,10 +30,8 @@ class ForeignServerAddTestCase(BaseTestGenerator): This class will add foreign server under database node. """ skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign server node. - ('Check FSRV Node', dict(url='/browser/foreign_server/obj/')) - ] + scenarios = utils.generate_scenarios('foreign_server_create', + fsrv_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -48,6 +48,19 @@ class ForeignServerAddTestCase(BaseTestGenerator): self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, self.fdw_name) + def create_foreign_server(self): + """ + This function create a foreign server and returns the created + foreign server response + :return: created foreign server response + """ + return self.tester.post( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.fdw_id) + '/', + data=json.dumps(self.data), + content_type='html/json') + def runTest(self): """This function will fetch foreign data wrapper present under test database.""" @@ -62,45 +75,34 @@ class ForeignServerAddTestCase(BaseTestGenerator): if not fdw_response: raise Exception("Could not find FDW.") db_user = self.server["username"] - data = { - "fsrvacl": [ - { - "grantee": db_user, - "grantor": db_user, - "privileges": - [ - { - "privilege_type": "U", - "privilege": "true", - "with_grant": "false" - } - ] - } - ], - "fsrvoptions": [ - { - "fsrvoption": "host", - "fsrvvalue": self.server['host'] - }, - { - "fsrvoption": "port", - "fsrvvalue": str(self.server['port']) - }, - { - "fsrvoption": "dbname", - "fsrvvalue": self.db_name - } - ], - "fsrvowner": db_user, - "name": "test_fsrv_add_%s" % (str(uuid.uuid4())[1:8]) - } - response = self.tester.post( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + - '/' + str(self.fdw_id) + '/', - data=json.dumps(data), - content_type='html/json') - self.assertEquals(response.status_code, 200) + self.data = fsrv_utils.get_fs_data(db_user, self.server, self.db_name) + + if self.is_positive_test: + response = self.create_foreign_server() + else: + if hasattr(self, "error_fdw_id"): + self.fdw_id = 99999 + response = self.create_foreign_server() + + if hasattr(self, "missing_parameter"): + del self.data['name'] + response = self.create_foreign_server() + + if hasattr(self, "internal_server_error"): + return_value_object = eval(self.mock_data["return_value"]) + with patch(self.mock_data["function_name"], + side_effect=[return_value_object]): + response = self.create_foreign_server() + + if hasattr(self, "error_in_db"): + return_value_object = eval(self.mock_data["return_value"]) + with patch(self.mock_data["function_name"], + side_effect=[return_value_object]): + response = self.create_foreign_server() + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop added foreign diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete.py index e7ee2c90f..7cf29a091 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete.py @@ -19,15 +19,14 @@ from pgadmin.utils.route import BaseTestGenerator from regression import parent_node_dict from regression.python_test_utils import test_utils as utils from . import utils as fsrv_utils +from unittest.mock import patch class ForeignServerDeleteTestCase(BaseTestGenerator): - """This class will add foreign server under FDW node.""" + """This class will delete foreign server under FDW node.""" skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign server node. - ('Check FSRV Node', dict(url='/browser/foreign_server/obj/')) - ] + scenarios = utils.generate_scenarios('foreign_server_delete', + fsrv_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -45,6 +44,18 @@ class ForeignServerDeleteTestCase(BaseTestGenerator): self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, self.fsrv_name, self.fdw_name) + def delete_foreign_server(self): + """ + This function returns the foreign server delete response + :return: foreign server delete response + """ + return self.tester.delete(self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + '/' + + str(self.fsrv_id), + follow_redirects=True) + def runTest(self): """This function will fetch foreign server present under test database.""" @@ -62,13 +73,22 @@ class ForeignServerDeleteTestCase(BaseTestGenerator): self.fsrv_name) if not fsrv_response: raise Exception("Could not find FSRV.") - delete_response = self.tester.delete( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + - '/' + str(self.fdw_id) + '/' + - str(self.fsrv_id), - follow_redirects=True) - self.assertEquals(delete_response.status_code, 200) + + if self.is_positive_test: + delete_response = self.delete_foreign_server() + else: + if hasattr(self, "internal_server_error"): + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + delete_response = self.delete_foreign_server() + + if hasattr(self, "wrong_foreign_server_id"): + self.fsrv_id = 99999 + delete_response = self.delete_foreign_server() + + actual_response_code = delete_response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete_multiple.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete_multiple.py index a4074a430..481d8efc1 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete_multiple.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete_multiple.py @@ -23,12 +23,10 @@ from . import utils as fsrv_utils class ForeignServerDeleteMultipleTestCase(BaseTestGenerator): - """This class will add foreign server under FDW node.""" + """This class will delete foreign server under FDW node.""" skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign server node. - ('Check FSRV Node', dict(url='/browser/foreign_server/obj/')) - ] + scenarios = utils.generate_scenarios('foreign_server_multiple_delete', + fsrv_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -51,8 +49,22 @@ class ForeignServerDeleteMultipleTestCase(BaseTestGenerator): self.fsrv_names[1], self.fdw_name)] + def delete_multiple(self, data): + """ + This function returns multiple foreign server delete response + :param data: foreign server ids to delete + :return: foreign server delete response + """ + return self.tester.delete( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.fdw_id) + "/", + data=json.dumps(data), + content_type='html/json', + follow_redirects=True) + def runTest(self): - """This function will fetch foreign server present under test + """This function will delete foreign server present under test database.""" db_con = database_utils.connect_database(self, utils.SERVER_GROUP, @@ -73,14 +85,13 @@ class ForeignServerDeleteMultipleTestCase(BaseTestGenerator): if not fsrv_response: raise Exception("Could not find FSRV.") data = {'ids': self.fsrv_ids} - delete_response = self.tester.delete( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + - '/' + str(self.fdw_id) + "/", - data=json.dumps(data), - content_type='html/json', - follow_redirects=True) - self.assertEquals(delete_response.status_code, 200) + + if self.is_positive_test: + delete_response = self.delete_multiple(data) + + actual_response_code = delete_response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop added diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_dependent_dependency.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_dependent_dependency.py new file mode 100644 index 000000000..0af9a3c17 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_dependent_dependency.py @@ -0,0 +1,91 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from __future__ import print_function + +import uuid + +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\ + tests import utils as fdw_utils +from pgadmin.browser.server_groups.servers.databases.tests import \ + utils as database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from . import utils as fsrv_utils + + +class ForeignServerDependentsAndDependencyTestCase(BaseTestGenerator): + """ This class will fetch added foreign server dependency and dependent + under test database. """ + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('foreign_server_dependent_dependency', + fsrv_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(ForeignServerDependentsAndDependencyTestCase, self).setUp() + self.schema_data = parent_node_dict['schema'][-1] + self.server_id = self.schema_data['server_id'] + self.db_id = self.schema_data['db_id'] + self.db_name = parent_node_dict["database"][-1]["db_name"] + self.schema_name = self.schema_data['schema_name'] + self.fdw_name = "test_fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "test_fsrv_%s" % (str(uuid.uuid4())[1:8]) + + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, + self.fsrv_name, self.fdw_name) + + def dependents_foreign_server(self): + """ + This function returns the foreign server dependents response + :return: foreign server dependents response + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + '/' + + str(self.fsrv_id), + follow_redirects=True) + + def runTest(self): + """This function will fetch foreign server dependents and dependency + present under test database.""" + db_con = database_utils.connect_database(self, + utils.SERVER_GROUP, + self.server_id, + self.db_id) + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to database.") + fdw_response = fdw_utils.verify_fdw(self.server, self.db_name, + self.fdw_name) + if not fdw_response: + raise Exception("Could not find FDW.") + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + if not fsrv_response: + raise Exception("Could not find FSRV.") + + if self.is_positive_test: + response = self.dependents_foreign_server() + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) + + def tearDown(self): + """This function disconnect the test database and drop + added foreign data server and dependant objects.""" + fdw_utils.delete_fdw(self.server, self.db_name, + self.fdw_name) + database_utils.disconnect_database(self, self.server_id, + self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_get.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_get.py index ee19b536e..3bec68f66 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_get.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_get.py @@ -21,15 +21,14 @@ from pgadmin.utils.route import BaseTestGenerator from regression import parent_node_dict from regression.python_test_utils import test_utils as utils from . import utils as fsrv_utils +from unittest.mock import patch class ForeignServerGetTestCase(BaseTestGenerator): - """This class will add foreign server under FDW node.""" + """This class will get foreign server properties under FDW node.""" skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign server node. - ('Check FSRV Node', dict(url='/browser/foreign_server/obj/')) - ] + scenarios = utils.generate_scenarios('foreign_server_get', + fsrv_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -49,8 +48,31 @@ class ForeignServerGetTestCase(BaseTestGenerator): self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, self.fsrv_name, self.fdw_name) + def get_foreign_server(self): + """ + This function returns the foreign server get response + :return: foreign server get response + """ + return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + '/' + + str(self.fsrv_id), + content_type='html/json') + + def get_foreign_server_list(self): + """ + This functions returns the foreign server list + :return: foreign server list + """ + return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + "/", + content_type='html/json') + def runTest(self): - """This function will fetch foreign server present under test + """This function will get foreign server properties present under test database.""" db_con = database_utils.connect_database(self, utils.SERVER_GROUP, @@ -62,12 +84,28 @@ class ForeignServerGetTestCase(BaseTestGenerator): self.fdw_name) if not fdw_response: raise Exception("Could not find FDW.") - fsrv_response = self.tester.get( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + - '/' + str(self.fdw_id) + '/' + str(self.fsrv_id), - content_type='html/json') - self.assertEquals(fsrv_response.status_code, 200) + + if self.is_positive_test: + if hasattr(self, "foreign_server_list"): + fsrv_response = self.get_foreign_server_list() + else: + fsrv_response = self.get_foreign_server() + else: + if hasattr(self, "error_fetching_fsrv"): + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + if hasattr(self, "foreign_server_list"): + fsrv_response = self.get_foreign_server_list() + else: + fsrv_response = self.get_foreign_server() + + if hasattr(self, "wrong_fsrv_id"): + self.fsrv_id = 99999 + fsrv_response = self.get_foreign_server() + + actual_response_code = fsrv_response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop added extension diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_msql.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_msql.py new file mode 100644 index 000000000..a2874db28 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_msql.py @@ -0,0 +1,91 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from __future__ import print_function + +import uuid + +from pgadmin.browser.server_groups.servers.databases.extensions.tests import \ + utils as extension_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\ + tests import utils as fdw_utils +from pgadmin.browser.server_groups.servers.databases.tests import \ + utils as database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from . import utils as fsrv_utils + + +class ForeignServerGetSQLTestCase(BaseTestGenerator): + """This class will fetch foreign server msql under FDW node.""" + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('foreign_server_get_msql', + fsrv_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(ForeignServerGetSQLTestCase, self).setUp() + self.schema_data = parent_node_dict['schema'][-1] + self.server_id = self.schema_data['server_id'] + self.db_id = self.schema_data['db_id'] + self.db_name = parent_node_dict["database"][-1]["db_name"] + self.schema_name = self.schema_data['schema_name'] + self.extension_name = "cube" + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "test_fsrv_add_%s" % (str(uuid.uuid4())[1:8]) + self.extension_id = extension_utils.create_extension( + self.server, self.db_name, self.extension_name, self.schema_name) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, + self.fsrv_name, self.fdw_name) + + def msql_foreign_server(self): + """ + This function returns the foreign server msql response + :return: foreign server msql response + """ + return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + '/' + + str(self.fsrv_id), + follow_redirects=True) + + def runTest(self): + """This function will fetch foreign server msql present under test + database.""" + db_con = database_utils.connect_database(self, + utils.SERVER_GROUP, + self.server_id, + self.db_id) + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to database.") + fdw_response = fdw_utils.verify_fdw(self.server, self.db_name, + self.fdw_name) + if not fdw_response: + raise Exception("Could not find FDW.") + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + if not fsrv_response: + raise Exception("Could not find FSRV.") + + if self.is_positive_test: + response = self.msql_foreign_server() + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) + + def tearDown(self): + # Disconnect the database + extension_utils.drop_extension(self.server, self.db_name, + self.extension_name) + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_node.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_node.py new file mode 100644 index 000000000..884566171 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_node.py @@ -0,0 +1,102 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from __future__ import print_function + +import uuid +import json +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\ + tests import utils as fdw_utils +from pgadmin.browser.server_groups.servers.databases.tests import \ + utils as database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from . import utils as fsrv_utils +from unittest.mock import patch + + +class ForeignServerGetNodeTestCase(BaseTestGenerator): + """This class will update foreign server node under FDW node.""" + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('foreign_server_get_nodes', + fsrv_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(ForeignServerGetNodeTestCase, self).setUp() + self.schema_data = parent_node_dict['schema'][-1] + self.server_id = self.schema_data['server_id'] + self.db_id = self.schema_data['db_id'] + self.db_name = parent_node_dict["database"][-1]["db_name"] + self.schema_name = self.schema_data['schema_name'] + self.fdw_name = "test_fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "test_fsrv_%s" % (str(uuid.uuid4())[1:8]) + + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, + self.fsrv_name, self.fdw_name) + + def get_foreign_server_node(self, fsrv_id): + """ + This function returns the foreign server node response + :return: foreign server node response + """ + return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + '/' + + str(fsrv_id), + follow_redirects=True) + + def runTest(self): + """This function will fetch foreign server node present under test + database.""" + db_con = database_utils.connect_database(self, + utils.SERVER_GROUP, + self.server_id, + self.db_id) + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to database.") + fdw_response = fdw_utils.verify_fdw(self.server, self.db_name, + self.fdw_name) + if not fdw_response: + raise Exception("Could not find FDW.") + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + if not fsrv_response: + raise Exception("Could not find FSRV.") + + if self.is_positive_test: + if hasattr(self, "node"): + response = self.get_foreign_server_node(self.fsrv_id) + else: + response = self.get_foreign_server_node("") + + else: + if hasattr(self, "internal_server_error"): + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + if hasattr(self, "node"): + response = self.get_foreign_server_node(self.fsrv_id) + else: + response = self.get_foreign_server_node("") + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) + + def tearDown(self): + """This function disconnect the test database and drop + added foreign data server and dependant objects.""" + fdw_utils.delete_fdw(self.server, self.db_name, + self.fdw_name) + database_utils.disconnect_database(self, self.server_id, + self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_put.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_put.py index b7a1cc663..83ca0fc3c 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_put.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_put.py @@ -22,15 +22,14 @@ from pgadmin.utils.route import BaseTestGenerator from regression import parent_node_dict from regression.python_test_utils import test_utils as utils from . import utils as fsrv_utils +from unittest.mock import patch class ForeignServerPutTestCase(BaseTestGenerator): """This class will add foreign server under FDW node.""" skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign server node. - ('Check FSRV Node', dict(url='/browser/foreign_server/obj/')) - ] + scenarios = utils.generate_scenarios('foreign_server_update', + fsrv_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -50,6 +49,19 @@ class ForeignServerPutTestCase(BaseTestGenerator): self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, self.fsrv_name, self.fdw_name) + def update_fsrv(self): + """ + This functions update foreign servers + :return: foreign server update request details + """ + return self.tester.put(self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + '/' + + str(self.fsrv_id), + data=json.dumps(self.test_data), + follow_redirects=True) + def runTest(self): """This function will update foreign server present under test database.""" @@ -67,15 +79,20 @@ class ForeignServerPutTestCase(BaseTestGenerator): self.fsrv_name) if not fsrv_response: raise Exception("Could not find FSRV.") - data = {"description": "This is foreign server update comment", - "id": self.fsrv_id} - put_response = self.tester.put( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + - '/' + str(self.fdw_id) + '/' + - str(self.fsrv_id), data=json.dumps(data), - follow_redirects=True) - self.assertEquals(put_response.status_code, 200) + self.test_data["id"] = self.fsrv_id + + if self.is_positive_test: + put_response = self.update_fsrv() + + else: + if hasattr(self, "error_in_db"): + with patch(self.mock_data["function_name"], + side_effect=eval(self.mock_data["return_value"])): + put_response = self.update_fsrv() + + actual_response_code = put_response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop added extension diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_sql.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_sql.py new file mode 100644 index 000000000..59562d336 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_sql.py @@ -0,0 +1,102 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from __future__ import print_function + +import uuid + +from pgadmin.browser.server_groups.servers.databases.extensions.tests import \ + utils as extension_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\ + tests import utils as fdw_utils +from pgadmin.browser.server_groups.servers.databases.tests import \ + utils as database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from . import utils as fsrv_utils +from unittest.mock import patch + + +class ForeignServerGetSQLTestCase(BaseTestGenerator): + """This class will add foreign server under FDW node.""" + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('foreign_server_get_sql', + fsrv_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(ForeignServerGetSQLTestCase, self).setUp() + self.schema_data = parent_node_dict['schema'][-1] + self.server_id = self.schema_data['server_id'] + self.db_id = self.schema_data['db_id'] + self.db_name = parent_node_dict["database"][-1]["db_name"] + self.schema_name = self.schema_data['schema_name'] + self.extension_name = "cube" + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "test_fsrv_add_%s" % (str(uuid.uuid4())[1:8]) + self.extension_id = extension_utils.create_extension( + self.server, self.db_name, self.extension_name, self.schema_name) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, + self.fsrv_name, self.fdw_name) + + def sql_foreign_server(self): + """ + This function returns the foreign server sql response + :return: foreign server sql response + """ + return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + '/' + + str(self.fsrv_id), + follow_redirects=True) + + def runTest(self): + """This function will fetch foreign server present under test + database.""" + db_con = database_utils.connect_database(self, + utils.SERVER_GROUP, + self.server_id, + self.db_id) + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to database.") + fdw_response = fdw_utils.verify_fdw(self.server, self.db_name, + self.fdw_name) + if not fdw_response: + raise Exception("Could not find FDW.") + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + if not fsrv_response: + raise Exception("Could not find FSRV.") + + if self.is_positive_test: + response = self.sql_foreign_server() + else: + if hasattr(self, "internal_server_error"): + return_value_object = eval(self.mock_data["return_value"]) + with patch(self.mock_data["function_name"], + side_effect=[return_value_object]): + response = self.sql_foreign_server() + + if hasattr(self, "wrong_foreign_server_id"): + self.fsrv_id = 99999 + response = self.sql_foreign_server() + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) + + def tearDown(self): + # Disconnect the database + extension_utils.drop_extension(self.server, self.db_name, + self.extension_name) + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/utils.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/utils.py index 596c95792..d65b382fa 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/utils.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/utils.py @@ -11,10 +11,51 @@ from __future__ import print_function import os import sys +import uuid +import json from regression.python_test_utils.test_utils import get_db_connection file_name = os.path.basename(__file__) +CURRENT_PATH = os.path.dirname(os.path.realpath(__file__)) +with open(CURRENT_PATH + "/foreign_servers_test_data.json") as data_file: + test_cases = json.load(data_file) + + +def get_fs_data(db_user, server, db_name): + data = { + "fsrvacl": [ + { + "grantee": db_user, + "grantor": db_user, + "privileges": + [ + { + "privilege_type": "U", + "privilege": "true", + "with_grant": "false" + } + ] + } + ], + "fsrvoptions": [ + { + "fsrvoption": "host", + "fsrvvalue": server['host'] + }, + { + "fsrvoption": "port", + "fsrvvalue": str(server['port']) + }, + { + "fsrvoption": "dbname", + "fsrvvalue": db_name + } + ], + "fsrvowner": db_user, + "name": "test_fsrv_add_%s" % (str(uuid.uuid4())[1:8]) + } + return data def create_fsrv(server, db_name, fsrv_name, fdw_name): diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_add.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_add.py index a208f93e0..53ef8313a 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_add.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_add.py @@ -23,16 +23,15 @@ from pgadmin.browser.server_groups.servers.databases.tests import \ from pgadmin.utils.route import BaseTestGenerator from regression import parent_node_dict from regression.python_test_utils import test_utils as utils +from . import utils as user_utils +from unittest.mock import patch class UserMappingAddTestCase(BaseTestGenerator): """This class will add user mapping under foreign server node.""" skip_on_database = ['gpdb'] - - scenarios = [ - # Fetching default URL for user mapping node. - ('Check user mapping Node', dict(url='/browser/user_mapping/obj/')) - ] + scenarios = utils.generate_scenarios('user_mapping_create', + user_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -52,9 +51,23 @@ class UserMappingAddTestCase(BaseTestGenerator): self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, self.fsrv_name, self.fdw_name) + def create_user_mapping(self): + """ + This function create a user mapping and returns the created + user mapping response + :return: created user mapping response + """ + return self.tester.post( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str( + self.db_id) + + '/' + str(self.fdw_id) + '/' + str(self.fsrv_id) + '/', + data=json.dumps(self.data), + content_type='html/json') + def runTest(self): - """This function will update foreign server present under test - database. """ + """This function will fetch foreign server and add user mappping + present under test database. """ db_con = database_utils.connect_database(self, utils.SERVER_GROUP, self.server_id, @@ -70,26 +83,34 @@ class UserMappingAddTestCase(BaseTestGenerator): if not fsrv_response: raise Exception("Could not find FSRV.") db_user = self.server["username"] - data = {"name": db_user, - "um_options": [], - "umoptions": [ - { - "umoption": "user", - "umvalue": self.server["username"] - }, - { - "umoption": "password", - "umvalue": self.server["db_password"] - } - ]} - response = self.tester.post( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str( - self.db_id) + - '/' + str(self.fdw_id) + '/' + str(self.fsrv_id) + '/', - data=json.dumps(data), - content_type='html/json') - self.assertEquals(response.status_code, 200) + self.data = user_utils.get_fs_data(db_user, self.server) + if self.is_positive_test: + response = self.create_user_mapping() + + else: + if hasattr(self, "error_fsrv_id"): + self.fsrv_id = 99999 + response = self.create_user_mapping() + + if hasattr(self, "internal_server_error"): + return_value_object = eval(self.mock_data["return_value"]) + with patch(self.mock_data["function_name"], + side_effect=[return_value_object]): + response = self.create_user_mapping() + + if hasattr(self, "missing_parameter"): + del self.data['name'] + response = self.create_user_mapping() + + if hasattr(self, "error_in_db"): + return_value_object = eval(self.mock_data["return_value"]) + with patch(self.mock_data["function_name"], + side_effect=[return_value_object]): + response = self.create_user_mapping() + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop added extension diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete.py index fbd9f98dd..56e90bc0f 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete.py @@ -23,15 +23,14 @@ from pgadmin.utils.route import BaseTestGenerator from regression import parent_node_dict from regression.python_test_utils import test_utils as utils from . import utils as um_utils +from unittest.mock import patch class UserMappingDeleteTestCase(BaseTestGenerator): """This class will delete user mapping under foreign server node.""" skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for user mapping node. - ('Check user mapping Node', dict(url='/browser/user_mapping/obj/')) - ] + scenarios = utils.generate_scenarios('user_mapping_delete', + um_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -50,6 +49,18 @@ class UserMappingDeleteTestCase(BaseTestGenerator): self.um_id = um_utils.create_user_mapping(self.server, self.db_name, self.fsrv_name) + def delete_user_mapping(self): + """ + This function returns the user mapping delete response + :return: user mapping delete response + """ + return self.tester.delete( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.fdw_id) + '/' + + str(self.fsrv_id) + '/' + str(self.um_id), + follow_redirects=True) + def runTest(self): """This function will delete user mapping present under test database. """ @@ -71,13 +82,31 @@ class UserMappingDeleteTestCase(BaseTestGenerator): self.fsrv_name) if not um_response: raise Exception("Could not find user mapping.") - delete_response = self.tester.delete( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + - '/' + str(self.fdw_id) + '/' + - str(self.fsrv_id) + '/' + str(self.um_id), - follow_redirects=True) - self.assertEquals(delete_response.status_code, 200) + if self.is_positive_test: + delete_response = self.delete_user_mapping() + else: + if hasattr(self, "internal_server_error"): + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + delete_response = self.delete_user_mapping() + + if hasattr(self, "error_in_db"): + return_value_object = eval(self.mock_data["return_value"]) + with patch(self.mock_data["function_name"], + side_effect=[return_value_object]): + delete_response = self.delete_user_mapping() + + if hasattr(self, "wrong_foreign_server_id"): + self.fsrv_id = 99999 + delete_response = self.delete_user_mapping() + + if hasattr(self, "wrong_user_mapping_id"): + self.um_id = 99999 + delete_response = self.delete_user_mapping() + + actual_response_code = delete_response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete_multiple.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete_multiple.py index 726a9ba0d..343fbfa47 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete_multiple.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete_multiple.py @@ -27,12 +27,10 @@ from . import utils as um_utils class UserMappingDeleteMultipleTestCase(BaseTestGenerator): - """This class will delete user mapping under foreign server node.""" + """This class will delete user mappings under foreign server node.""" skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for user mapping node. - ('Check user mapping Node', dict(url='/browser/user_mapping/obj/')) - ] + scenarios = utils.generate_scenarios('user_mapping_multiple_delete', + um_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -51,6 +49,21 @@ class UserMappingDeleteMultipleTestCase(BaseTestGenerator): self.um_id = um_utils.create_user_mapping(self.server, self.db_name, self.fsrv_name) + def delete_multiple(self): + """ + This function returns multiple user mapping delete response + :param data: user mapping ids to delete + :return: user mapping delete response + """ + return self.tester.delete( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.fdw_id) + '/' + + str(self.fsrv_id) + "/", + follow_redirects=True, + data=json.dumps(self.data), + content_type='html/json') + def runTest(self): """This function will delete user mapping present under test database. """ @@ -72,16 +85,13 @@ class UserMappingDeleteMultipleTestCase(BaseTestGenerator): self.fsrv_name) if not um_response: raise Exception("Could not find user mapping.") - data = {'ids': [self.um_id]} - delete_response = self.tester.delete( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + - '/' + str(self.fdw_id) + '/' + - str(self.fsrv_id) + "/", - follow_redirects=True, - data=json.dumps(data), - content_type='html/json') - self.assertEquals(delete_response.status_code, 200) + self.data = {'ids': [self.um_id]} + if self.is_positive_test: + delete_response = self.delete_multiple() + + actual_response_code = delete_response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_dependent_dependency.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_dependent_dependency.py new file mode 100644 index 000000000..86ece52f2 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_dependent_dependency.py @@ -0,0 +1,100 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from __future__ import print_function + +import uuid + +from pgadmin.browser.server_groups.servers.databases.extensions.tests import \ + utils as extension_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \ + foreign_servers.tests import utils as fsrv_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\ + tests import utils as fdw_utils +from pgadmin.browser.server_groups.servers.databases.tests import \ + utils as database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from . import utils as um_utils +from unittest.mock import patch + + +class UserMappingDependentsAndDependencyTestCase(BaseTestGenerator): + """ This class will fetch added user mapping dependency and dependent + under test database. """ + + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('user_mapping_dependent_dependency', + um_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(UserMappingDependentsAndDependencyTestCase, self).setUp() + self.schema_data = parent_node_dict['schema'][-1] + self.server_id = self.schema_data['server_id'] + self.db_id = self.schema_data['db_id'] + self.db_name = parent_node_dict["database"][-1]["db_name"] + self.schema_name = self.schema_data['schema_name'] + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, + self.fsrv_name, self.fdw_name) + self.um_id = um_utils.create_user_mapping(self.server, self.db_name, + self.fsrv_name) + + def dependents_user_mapping(self): + """ + This function returns the user mapping dependents response + :return: user mapping dependents response + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.fdw_id) + '/' + + str(self.fsrv_id) + '/' + str(self.um_id), + follow_redirects=True) + + def runTest(self): + """This function will fetch user mapping present under test + database. """ + db_con = database_utils.connect_database(self, + utils.SERVER_GROUP, + self.server_id, + self.db_id) + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to database.") + fdw_response = fdw_utils.verify_fdw(self.server, self.db_name, + self.fdw_name) + if not fdw_response: + raise Exception("Could not find FDW.") + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + if not fsrv_response: + raise Exception("Could not find FSRV.") + um_response = um_utils.verify_user_mapping(self.server, self.db_name, + self.fsrv_name) + if not um_response: + raise Exception("Could not find user mapping.") + if self.is_positive_test: + response = self.dependents_user_mapping() + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) + + def tearDown(self): + """This function disconnect the test database and drop + foreign data wrapper and dependant objects.""" + fdw_utils.delete_fdw(self.server, self.db_name, + self.fdw_name) + database_utils.disconnect_database(self, self.server_id, + self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_get.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_get.py index 980d58f0f..af376de89 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_get.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_get.py @@ -23,15 +23,14 @@ from pgadmin.utils.route import BaseTestGenerator from regression import parent_node_dict from regression.python_test_utils import test_utils as utils from . import utils as um_utils +from unittest.mock import patch class UserMappingGetTestCase(BaseTestGenerator): """This class will add user mapping under foreign server node.""" skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for user mapping node. - ('Check user mapping Node', dict(url='/browser/user_mapping/obj/')) - ] + scenarios = utils.generate_scenarios('user_mapping_get', + um_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -53,6 +52,31 @@ class UserMappingGetTestCase(BaseTestGenerator): self.um_id = um_utils.create_user_mapping(self.server, self.db_name, self.fsrv_name) + def get_user_mapping(self): + """ + This function returns the user mapping get response + :return: user mapping get response + """ + return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + '/' + + str(self.fsrv_id) + '/' + + str(self.um_id), + content_type='html/json') + + def get_user_mapping_list(self): + """ + This functions returns the user mapping list + :return: user mapping list + """ + return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id) + "/" + + str(self.fsrv_id) + '/', + content_type='html/json') + def runTest(self): """This function will update foreign server present under test database.""" @@ -70,12 +94,29 @@ class UserMappingGetTestCase(BaseTestGenerator): self.fsrv_name) if not fsrv_response: raise Exception("Could not find FSRV.") - response = self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str( - self.db_id) + '/' + str(self.fdw_id) + '/' + str( - self.fsrv_id) + '/' + str( - self.um_id), content_type='html/json') - self.assertEquals(response.status_code, 200) + + if self.is_positive_test: + if hasattr(self, "um_list"): + response = self.get_user_mapping_list() + else: + response = self.get_user_mapping() + + else: + if hasattr(self, "error_fetching_um"): + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + if hasattr(self, "um_list"): + response = self.get_user_mapping_list() + else: + response = self.get_user_mapping() + + if hasattr(self, "wrong_um_id"): + self.um_id = 99999 + response = self.get_user_mapping() + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop added extension diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_msql.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_msql.py new file mode 100644 index 000000000..8b506eb10 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_msql.py @@ -0,0 +1,101 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from __future__ import print_function + +import uuid + +from pgadmin.browser.server_groups.servers.databases.extensions.tests import \ + utils as extension_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \ + foreign_servers.tests import utils as fsrv_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\ + tests import utils as fdw_utils +from pgadmin.browser.server_groups.servers.databases.tests import \ + utils as database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from . import utils as um_utils + + +class UserMappingGetMSQLTestCase(BaseTestGenerator): + """This class will add user mapping under foreign server node.""" + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('user_mapping_get_msql', + um_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(UserMappingGetMSQLTestCase, self).setUp() + self.schema_data = parent_node_dict['schema'][-1] + self.server_id = self.schema_data['server_id'] + self.db_id = self.schema_data['db_id'] + self.db_name = parent_node_dict["database"][-1]["db_name"] + self.schema_name = self.schema_data['schema_name'] + self.extension_name = "cube" + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) + self.extension_id = extension_utils.create_extension( + self.server, self.db_name, self.extension_name, self.schema_name) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, + self.fsrv_name, self.fdw_name) + self.um_id = um_utils.create_user_mapping(self.server, self.db_name, + self.fsrv_name) + + def msql_user_mapping(self): + """ + This function returns the user mapping msql response + :return: user mapping msql response + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.fdw_id) + '/' + + str(self.fsrv_id) + '/' + str(self.um_id), + follow_redirects=True) + + def runTest(self): + """This function will fetch foreign server present under test + database""" + db_con = database_utils.connect_database(self, + utils.SERVER_GROUP, + self.server_id, + self.db_id) + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to database.") + fdw_response = fdw_utils.verify_fdw(self.server, self.db_name, + self.fdw_name) + if not fdw_response: + raise Exception("Could not find FDW.") + fsrv_response = fsrv_utils.verify_fsrv(self.server, + self.db_name, + self.fsrv_name) + if not fsrv_response: + raise Exception("Could not find FSRV.") + um_response = um_utils.verify_user_mapping(self.server, self.db_name, + self.fsrv_name) + if not um_response: + raise Exception("Could not find user mapping.") + + if self.is_positive_test: + response = self.msql_user_mapping() + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) + + def tearDown(self): + """This function disconnect the test database and drop added + extension and dependant objects.""" + extension_utils.drop_extension(self.server, self.db_name, + self.extension_name) + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_node.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_node.py new file mode 100644 index 000000000..8717b918f --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_node.py @@ -0,0 +1,110 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from __future__ import print_function + +import uuid + +from pgadmin.browser.server_groups.servers.databases.extensions.tests import \ + utils as extension_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \ + foreign_servers.tests import utils as fsrv_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\ + tests import utils as fdw_utils +from pgadmin.browser.server_groups.servers.databases.tests import \ + utils as database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from . import utils as um_utils +from unittest.mock import patch + + +class UserMappingNodesTestCase(BaseTestGenerator): + """This class will delete user mapping under foreign server node.""" + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('user_mapping_get_nodes', + um_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(UserMappingNodesTestCase, self).setUp() + self.schema_data = parent_node_dict['schema'][-1] + self.server_id = self.schema_data['server_id'] + self.db_id = self.schema_data['db_id'] + self.db_name = parent_node_dict["database"][-1]["db_name"] + self.schema_name = self.schema_data['schema_name'] + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, + self.fsrv_name, self.fdw_name) + self.um_id = um_utils.create_user_mapping(self.server, self.db_name, + self.fsrv_name) + + def get_user_mapping_node(self, um_id): + """ + This function returns the user mapping node response + :return: user mapping node response + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.fdw_id) + '/' + + str(self.fsrv_id) + '/' + str(um_id), + follow_redirects=True) + + def runTest(self): + """This function will fetch user mapping present under test + database. """ + db_con = database_utils.connect_database(self, + utils.SERVER_GROUP, + self.server_id, + self.db_id) + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to database.") + fdw_response = fdw_utils.verify_fdw(self.server, self.db_name, + self.fdw_name) + if not fdw_response: + raise Exception("Could not find FDW.") + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + if not fsrv_response: + raise Exception("Could not find FSRV.") + um_response = um_utils.verify_user_mapping(self.server, self.db_name, + self.fsrv_name) + if not um_response: + raise Exception("Could not find user mapping.") + + if self.is_positive_test: + if hasattr(self, "node"): + response = self.get_user_mapping_node(self.um_id) + else: + response = self.get_user_mapping_node("") + else: + if hasattr(self, "internal_server_error"): + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + if hasattr(self, "node"): + response = self.get_user_mapping_node(self.um_id) + else: + response = self.get_user_mapping_node("") + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) + + def tearDown(self): + """This function disconnect the test database and drop + foreign data wrapper and dependant objects.""" + fdw_utils.delete_fdw(self.server, self.db_name, + self.fdw_name) + database_utils.disconnect_database(self, self.server_id, + self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_put.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_put.py index f008442fc..12a16c088 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_put.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_put.py @@ -24,15 +24,14 @@ from pgadmin.utils.route import BaseTestGenerator from regression import parent_node_dict from regression.python_test_utils import test_utils as utils from . import utils as um_utils +from unittest.mock import patch class UserMappingPutTestCase(BaseTestGenerator): """This class will update user mapping under foreign server node.""" skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for user mapping node. - ('Check user mapping Node', dict(url='/browser/user_mapping/obj/')) - ] + scenarios = utils.generate_scenarios('user_mapping_update', + um_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -54,8 +53,21 @@ class UserMappingPutTestCase(BaseTestGenerator): self.um_id = um_utils.create_user_mapping(self.server, self.db_name, self.fsrv_name) + def update_user_mapping(self): + """ + This functions update user mapping + :return: user mapping update request details + """ + return self.tester.put( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.fdw_id) + '/' + + str(self.fsrv_id) + '/' + str(self.um_id), + data=json.dumps(self.test_data), + follow_redirects=True) + def runTest(self): - """This function will update foreign server present under test + """This function will update user mapping present under test database""" db_con = database_utils.connect_database(self, utils.SERVER_GROUP, @@ -76,26 +88,20 @@ class UserMappingPutTestCase(BaseTestGenerator): self.fsrv_name) if not um_response: raise Exception("Could not find user mapping.") - data = { - "id": self.um_id, - "umoptions": - { - "changed": - [ - {"umoption": "user", - "umvalue": "public" - } - ] - } - } - put_response = self.tester.put( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + - '/' + str(self.fdw_id) + '/' + - str(self.fsrv_id) + '/' + str(self.um_id), - data=json.dumps(data), - follow_redirects=True) - self.assertEquals(put_response.status_code, 200) + self.test_data["id"] = self.um_id + + if self.is_positive_test: + put_response = self.update_user_mapping() + + else: + if hasattr(self, "error_in_db"): + with patch(self.mock_data["function_name"], + side_effect=eval(self.mock_data["return_value"])): + put_response = self.update_user_mapping() + + actual_response_code = put_response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) def tearDown(self): """This function disconnect the test database and drop added diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_sql.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_sql.py new file mode 100644 index 000000000..f28da4822 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_sql.py @@ -0,0 +1,112 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +from __future__ import print_function + +import uuid + +from pgadmin.browser.server_groups.servers.databases.extensions.tests import \ + utils as extension_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \ + foreign_servers.tests import utils as fsrv_utils +from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\ + tests import utils as fdw_utils +from pgadmin.browser.server_groups.servers.databases.tests import \ + utils as database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from . import utils as um_utils +from unittest.mock import patch + + +class UserMappingGetSQLTestCase(BaseTestGenerator): + """This class will add user mapping under foreign server node.""" + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('user_mapping_get_sql', + um_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(UserMappingGetSQLTestCase, self).setUp() + self.schema_data = parent_node_dict['schema'][-1] + self.server_id = self.schema_data['server_id'] + self.db_id = self.schema_data['db_id'] + self.db_name = parent_node_dict["database"][-1]["db_name"] + self.schema_name = self.schema_data['schema_name'] + self.extension_name = "cube" + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) + self.extension_id = extension_utils.create_extension( + self.server, self.db_name, self.extension_name, self.schema_name) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, + self.fsrv_name, self.fdw_name) + self.um_id = um_utils.create_user_mapping(self.server, self.db_name, + self.fsrv_name) + + def sql_user_mapping(self): + """ + This function returns the user mapping sql response + :return: user mapping sql response + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.fdw_id) + '/' + + str(self.fsrv_id) + '/' + str(self.um_id), + follow_redirects=True) + + def runTest(self): + """This function will fetch user mapping present under test + database""" + db_con = database_utils.connect_database(self, + utils.SERVER_GROUP, + self.server_id, + self.db_id) + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to database.") + fdw_response = fdw_utils.verify_fdw(self.server, self.db_name, + self.fdw_name) + if not fdw_response: + raise Exception("Could not find FDW.") + fsrv_response = fsrv_utils.verify_fsrv(self.server, + self.db_name, + self.fsrv_name) + if not fsrv_response: + raise Exception("Could not find FSRV.") + um_response = um_utils.verify_user_mapping(self.server, self.db_name, + self.fsrv_name) + if not um_response: + raise Exception("Could not find user mapping.") + + if self.is_positive_test: + response = self.sql_user_mapping() + else: + if hasattr(self, "internal_server_error"): + return_value_object = eval(self.mock_data["return_value"]) + with patch(self.mock_data["function_name"], + side_effect=[return_value_object]): + response = self.sql_user_mapping() + + if hasattr(self, "wrong_um_id"): + self.um_id = 99999 + response = self.sql_user_mapping() + + actual_response_code = response.status_code + expected_response_code = self.expected_data['status_code'] + self.assertEquals(actual_response_code, expected_response_code) + + def tearDown(self): + """This function disconnect the test database and drop added + extension and dependant objects.""" + extension_utils.drop_extension(self.server, self.db_name, + self.extension_name) + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/user_mapping_test_data.json b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/user_mapping_test_data.json new file mode 100644 index 000000000..b90ac63c3 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/user_mapping_test_data.json @@ -0,0 +1,388 @@ +{ + "user_mapping_create": [ + { + "name": "Create user mapping: With valid data.", + "url": "/browser/user_mapping/obj/", + "is_positive_test": true, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error creating user mapping: Internal server error.", + "url": "/browser/user_mapping/obj/", + "is_positive_test": false, + "internal_server_error": true, + "test_data": {}, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server Error while creating a user mapping')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error creating user mapping - wrong fsrv_id", + "url": "/browser/user_mapping/obj/", + "is_positive_test": false, + "error_fsrv_id": true, + "test_data": {}, + "mocking_required": false, + "mock_data": { + }, + "expected_data": { + "status_code": 410 + } + }, + { + "name": "Error while creating user mapping: With insufficient data.", + "url": "/browser/user_mapping/obj/", + "is_positive_test": false, + "missing_parameter": true, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 410 + } + }, + { + "name": "Error creating user mapping: Internal server error 2.", + "url": "/browser/user_mapping/obj/", + "is_positive_test": false, + "error_in_db": true, + "test_data": {}, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar", + "return_value": "(False, 'Mocked Internal Server Error')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "user_mapping_delete": [ + { + "name": "delete user mapping", + "url": "/browser/user_mapping/obj/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while deleting a user mapping - Internal server error 1", + "url": "/browser/user_mapping/obj/", + "internal_server_error": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar", + "return_value": "(False, 'Mocked Internal Server Error')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error while deleting a user mapping - wrong foreign server id", + "url": "/browser/user_mapping/obj/", + "wrong_foreign_server_id": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + }, + "expected_data": { + "status_code": 410 + } + }, + { + "name": "Error while deleting a user mapping - wrong user mapping id", + "url": "/browser/user_mapping/obj/", + "wrong_user_mapping_id": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + }, + "expected_data": { + "status_code": 410 + } + }, + { + "name": "Error creating user mapping: Error in db.", + "url": "/browser/user_mapping/obj/", + "is_positive_test": false, + "error_in_db": true, + "test_data": {}, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server Error while deleting a user mapping')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "user_mapping_multiple_delete": [ + { + "name": "Delete multiple user mapping", + "url": "/browser/user_mapping/obj/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + } + ], + "user_mapping_dependent_dependency": [ + { + "name": "Get user mapping dependents", + "url": "/browser/user_mapping/dependent/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Get user mapping dependencies", + "url": "/browser/user_mapping/dependency/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + } + ], + "user_mapping_get_nodes": [ + { + "name": "Get user mapping nodes", + "url": "/browser/user_mapping/nodes/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error fetching user mapping nodes", + "url": "/browser/user_mapping/nodes/", + "is_positive_test": false, + "internal_server_error": true, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray", + "return_value": "(False, 'Mocked Internal Server Error while fetching user mapping nodes')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Get user mapping node", + "url": "/browser/user_mapping/nodes/", + "is_positive_test": true, + "node": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error fetching user mapping node", + "url": "/browser/user_mapping/nodes/", + "is_positive_test": false, + "node": true, + "internal_server_error": true, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray", + "return_value": "(False, 'Mocked Internal Server Error while fetching user mapping nodes')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "user_mapping_update": [ + { + "name": "Update user mapping", + "url": "/browser/user_mapping/obj/", + "is_positive_test": true, + "mocking_required": false, + "test_data": { + "umoptions": + { + "changed": + [ + {"umoption": "user", + "umvalue": "public" + } + ] + }, + "id": "PLACE_HOLDER" + }, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while updating a user mapping - Internal server error", + "url": "/browser/user_mapping/obj/", + "error_in_db": true, + "test_data": { + "umoptions": + { + "changed": + [ + {"umoption": "user", + "umvalue": "public" + } + ] + }, + "id": "PLACE_HOLDER" + }, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar", + "return_value": "[(False, 'Mocked Internal Server Error while fetching a user mapping')]" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "user_mapping_get": [ + { + "name": "Get user mapping properties", + "url": "/browser/user_mapping/obj/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while fetching a user mapping properties", + "url": "/browser/user_mapping/obj/", + "error_fetching_um": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server Error while fetching a user mapping')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error - Fetching user mapping with wrong ID", + "url": "/browser/user_mapping/obj/", + "wrong_um_id": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": {}, + "expected_data": { + "status_code": 410 + } + }, + { + "name": "Get user mapping list", + "url": "/browser/user_mapping/obj/", + "is_positive_test": true, + "um_list": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while fetching a user mapping list", + "url": "/browser/user_mapping/obj/", + "error_fetching_um": true, + "is_positive_test": false, + "um_list": true, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server Error while fetching a user mapping')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "user_mapping_get_sql": [ + { + "name": "Get user mapping SQL", + "url": "/browser/user_mapping/sql/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error - Get user mapping SQL - Internal server error", + "url": "/browser/user_mapping/sql/", + "is_positive_test": false, + "internal_server_error": true, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server Error while fetching a user mapping')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error - Get user mapping SQL using wrong um id", + "url": "/browser/user_mapping/sql/", + "wrong_um_id": true, + "is_positive_test": false, + "mocking_required": false, + "mock_data": { + }, + "expected_data": { + "status_code": 410 + } + } + ], + "user_mapping_get_msql": [ + { + "name": "Get user mapping msql", + "is_positive_test": true, + "url": "/browser/user_mapping/msql/", + "test_data": { + "name": "modifying name", + "comment": "Comments to test update" + }, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + } + ] +} diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/utils.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/utils.py index ef767fc8a..05187bf0f 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/utils.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/utils.py @@ -9,11 +9,34 @@ from __future__ import print_function +import os import sys import traceback +import json from regression.python_test_utils.test_utils import get_db_connection +CURRENT_PATH = os.path.dirname(os.path.realpath(__file__)) +with open(CURRENT_PATH + "/user_mapping_test_data.json") as data_file: + test_cases = json.load(data_file) + + +def get_fs_data(db_user, server): + + data = {"name": db_user, + "um_options": [], + "umoptions": [ + { + "umoption": "user", + "umvalue": server["username"] + }, + { + "umoption": "password", + "umvalue": server["db_password"] + } + ]} + return data + def create_user_mapping(server, db_name, fsrv_name): """