diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/fdw_test_data.json b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/fdw_test_data.json new file mode 100644 index 000000000..6137b5762 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/fdw_test_data.json @@ -0,0 +1,409 @@ +{ + "fdw_create": [ + { + "name": "Create fdw: With valid data.", + "url": "/browser/foreign_data_wrapper/obj/", + "is_positive_test": true, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while creating fdw while server down", + "url": "/browser/foreign_data_wrapper/obj/", + "is_positive_test": false, + "error_creating_fdw": 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 data wrapper')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error while creating fdw - internal server error", + "url": "/browser/foreign_data_wrapper/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": "(True, True), (False, 'Mocked Internal Server Error')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error while creating fdw - invalid data", + "url": "/browser/foreign_data_wrapper/obj/", + "is_positive_test": false, + "invalid_data": true, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 410 + } + } + ], + "fdw_dependency_dependent": [ + { + "name": "Get fdw dependency", + "url": "/browser/foreign_data_wrapper/dependency/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Get fdw dependent", + "url": "/browser/foreign_data_wrapper/dependent/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + } + ], + "fdw_delete": [ + { + "name": "delete fdw", + "url": "/browser/foreign_data_wrapper/obj/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while deleting a fdw", + "url": "/browser/foreign_data_wrapper/obj/", + "error_deleting_fdw": 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 while deleting a fdw')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error while deleting a fdw - internal server error", + "url": "/browser/foreign_data_wrapper/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": "(True, True), (False, 'Mocked Internal Server Error while deleting a fdw')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "delete fdw using wrong fdw id", + "url": "/browser/foreign_data_wrapper/obj/", + "wrong_fdw_id": true, + "is_positive_test": false, + "mocking_required": false, + "mock_data": { + }, + "expected_data": { + "status_code": 410 + } + } + ], + "fdw_delete_multiple": [ + { + "name": "Delete multiple fdw", + "url": "/browser/foreign_data_wrapper/obj/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + } + ], + "fdw_get_nodes_and_node": [ + { + "name": "Get fdw nodes", + "url": "/browser/foreign_data_wrapper/nodes/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while fetching fdw nodes", + "url": "/browser/foreign_data_wrapper/nodes/", + "error_fetching_fdw": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray", + "return_value": "(False, 'Mocked Internal Server Error while fetching fdw nodes')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Get fdw node", + "url": "/browser/foreign_data_wrapper/nodes/", + "is_positive_test": true, + "node": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while fetching fdw node", + "url": "/browser/foreign_data_wrapper/nodes/", + "error_fetching_fdw": true, + "is_positive_test": false, + "node": 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 fdw nodes')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Fetching fdw node with wrong id", + "url": "/browser/foreign_data_wrapper/nodes/", + "wrong_id": true, + "is_positive_test": false, + "node": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 410 + } + } + ], + "fdw_get_handlers": [ + { + "name": "Get get_handlers", + "url": "/browser/foreign_data_wrapper/get_handlers/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while getting handlers", + "url": "/browser/foreign_data_wrapper/get_handlers/", + "internal_server_error": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray", + "return_value": "(False, 'Mocked Internal Server Error while fetching handlers')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "fdw_get_validators": [ + { + "name": "Get get_validators", + "url": "/browser/foreign_data_wrapper/get_validators/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while getting validtors", + "url": "/browser/foreign_data_wrapper/get_validators/", + "internal_server_error": true, + "is_positive_test": false, + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray", + "return_value": "(False, 'Mocked Internal Server Error while fetching handlers')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "fdw_update": [ + { + "name": "Update fdw", + "url": "/browser/foreign_data_wrapper/obj/", + "is_positive_test": true, + "mocking_required": false, + "test_data": { + "description": "This is fdw update comment", + "id": "PLACE_HOLDER" + }, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while updating a fdw", + "url": "/browser/foreign_data_wrapper/obj/", + "internal_server_error": true, + "test_data": { + "description": "This is fdw 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 updating fdw')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "fdw_get": [ + { + "name": "Get fdw", + "url": "/browser/foreign_data_wrapper/obj/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while fetching a fdw", + "url": "/browser/foreign_data_wrapper/obj/", + "error_fetching_fdw": 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 fdw')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Fetch fdw using wrong fdw id", + "url": "/browser/foreign_data_wrapper/obj/", + "wrong_fdw_id": true, + "is_positive_test": false, + "mocking_required": false, + "mock_data": { + }, + "expected_data": { + "status_code": 410 + } + }, + { + "name": "Get fdw list", + "url": "/browser/foreign_data_wrapper/obj/", + "is_positive_test": true, + "fdw_list": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while fetching fdw list", + "url": "/browser/foreign_data_wrapper/obj/", + "error_fetching_fdw": true, + "is_positive_test": false, + "fdw_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 fdw')" + }, + "expected_data": { + "status_code": 500 + } + } + ], + "fdw_get_sql": [ + { + "name": "Get fdw SQL", + "url": "/browser/foreign_data_wrapper/sql/", + "is_positive_test": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200 + } + }, + { + "name": "Error while fetching SQL generated - Internal server error", + "url": "/browser/foreign_data_wrapper/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')" + }, + "expected_data": { + "status_code": 500 + } + }, + { + "name": "Error while fetching SQL generated - Internal server error", + "url": "/browser/foreign_data_wrapper/sql/", + "is_positive_test": false, + "wrong_fdw_id": true, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 410 + } + } + ], + "fdw_get_msql": [ + { + "name": "Get fdw msql", + "is_positive_test": true, + "url": "/browser/foreign_data_wrapper/msql/", + "mocking_required": true, + "mock_data": { + "function_name": "pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.ForeignDataWrapperView.get_sql", + "return_value": "('', 'Mocked get_sql function')" + }, + "expected_data": { + "status_code": 200 + } + } + ] +} diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_add.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_add.py index ec6f09216..786293184 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_add.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_add.py @@ -17,17 +17,15 @@ 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 fdw_utils +from unittest.mock import patch class FDWDAddTestCase(BaseTestGenerator): """ This class will add foreign data wrappers under database node. """ skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign_data_wrapper node. - ('Check FDW Node', - dict(url='/browser/foreign_data_wrapper/obj/')) - ] + scenarios = utils.generate_scenarios('fdw_create', + fdw_utils.test_cases) def setUp(self): """ This function will create extension.""" @@ -39,6 +37,18 @@ class FDWDAddTestCase(BaseTestGenerator): self.schema_name = self.schema_data['schema_name'] self.db_name = parent_node_dict["database"][-1]["db_name"] + def create_foreign_data_wrapper(self): + """ + This function create a foreign data wrapper and returns the created + foreign data wrapper response + :return: created foreign data wrapper response + """ + return self.tester.post( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + '/', + data=json.dumps(self.data), + content_type='html/json') + def runTest(self): """This function will add foreign data wrapper under test database.""" db_con = database_utils.connect_database(self, @@ -49,15 +59,28 @@ class FDWDAddTestCase(BaseTestGenerator): raise Exception("Could not connect to database.") self.data = fdw_utils.get_fdw_data(self.schema_name, self.server['username']) - response = self.tester.post( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + '/', - data=json.dumps(self.data), - content_type='html/json') - self.assertEquals(response.status_code, 200) + if self.is_positive_test: + response = self.create_foreign_data_wrapper() + else: + if hasattr(self, "error_creating_fdw"): + 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_data_wrapper() + + if hasattr(self, "internal_server_error"): + with patch(self.mock_data["function_name"], + side_effect=eval(self.mock_data["return_value"])): + response = self.create_foreign_data_wrapper() + + if hasattr(self, "invalid_data"): + del self.data['name'] + response = self.create_foreign_data_wrapper() + + self.assertEquals(response.status_code, + self.expected_data['status_code']) def tearDown(self): """This function delete the FDW and disconnect the test database """ - fdw_utils.delete_fdw(self.server, self.db_name, self.data["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/tests/test_fdw_delete.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_delete.py index f4fea98de..2c511438e 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_delete.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_delete.py @@ -17,18 +17,18 @@ 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 fdw_utils +from unittest.mock import patch -class FDWDDeleteTestCase(BaseTestGenerator): +class FDWDeleteTestCase(BaseTestGenerator): """This class will delete foreign data wrappers under test database.""" skip_on_database = ['gpdb'] - scenarios = [ # Fetching default URL for foreign_data_wrapper node. - ('Check FDW Node', - dict(url='/browser/foreign_data_wrapper/obj/'))] + scenarios = utils.generate_scenarios('fdw_delete', + fdw_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" - super(FDWDDeleteTestCase, self).setUp() + super(FDWDeleteTestCase, 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'] @@ -38,6 +38,18 @@ class FDWDDeleteTestCase(BaseTestGenerator): self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, self.fdw_name) + def delete_fdw(self): + """ + This function deletes fdw + :return: fdw delete + """ + return self.tester.delete(self.url + + str(utils.SERVER_GROUP) + + '/' + str(self.server_id) + '/' + + str(self.db_id) + + '/' + str(self.fdw_id), + follow_redirects=True) + def runTest(self): """This function will fetch foreign data wrapper present under test database.""" @@ -51,13 +63,28 @@ class FDWDDeleteTestCase(BaseTestGenerator): self.fdw_name) if not fdw_response: raise Exception("Could not find FDW.") - delete_response = self.tester.delete(self.url + - str(utils.SERVER_GROUP) + - '/' + str(self.server_id) + '/' + - str(self.db_id) + - '/' + str(self.fdw_id), - follow_redirects=True) - self.assertEquals(delete_response.status_code, 200) + + if self.is_positive_test: + response = self.delete_fdw() + + else: + if hasattr(self, "error_deleting_fdw"): + return_value_object = eval(self.mock_data["return_value"]) + with patch(self.mock_data["function_name"], + side_effect=[return_value_object]): + response = self.delete_fdw() + + if hasattr(self, "internal_server_error"): + with patch(self.mock_data["function_name"], + side_effect=eval(self.mock_data["return_value"])): + response = self.delete_fdw() + + if hasattr(self, "wrong_fdw_id"): + self.fdw_id = 99999 + response = self.delete_fdw() + + self.assertEquals(response.status_code, + self.expected_data['status_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/tests/test_fdw_delete_multiple.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_delete_multiple.py index f707f0123..54c453f37 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_delete_multiple.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_delete_multiple.py @@ -23,9 +23,8 @@ from . import utils as fdw_utils class FDWDDeleteMultipleTestCase(BaseTestGenerator): """This class will delete foreign data wrappers under test database.""" skip_on_database = ['gpdb'] - scenarios = [ # Fetching default URL for foreign_data_wrapper node. - ('Check FDW Node', - dict(url='/browser/foreign_data_wrapper/obj/'))] + scenarios = utils.generate_scenarios('fdw_delete_multiple', + fdw_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -42,6 +41,19 @@ class FDWDDeleteMultipleTestCase(BaseTestGenerator): fdw_utils.create_fdw(self.server, self.db_name, self.fdw_names[1])] + def delete_multiple(self, data): + """ + This function returns multiple fdw delete response + :param data: fdw ids to delete + :return: fdw delete response + """ + return self.tester.delete(self.url + str(utils.SERVER_GROUP) + + '/' + str(self.server_id) + + '/' + str(self.db_id) + '/', + follow_redirects=True, + data=json.dumps(data), + content_type='html/json') + def runTest(self): """This function will fetch foreign data wrapper present under test database.""" @@ -60,14 +72,10 @@ class FDWDDeleteMultipleTestCase(BaseTestGenerator): if not fdw_response: raise Exception("Could not find FDW.") data = {'ids': self.fdw_ids} - delete_response = self.tester.delete(self.url + - str(utils.SERVER_GROUP) + - '/' + str(self.server_id) + '/' + - str(self.db_id) + '/', - follow_redirects=True, - data=json.dumps(data), - content_type='html/json') - self.assertEquals(delete_response.status_code, 200) + delete_response = self.delete_multiple(data) + + self.assertEquals(delete_response.status_code, + self.expected_data['status_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/tests/test_fdw_dependency_dependent.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_dependency_dependent.py new file mode 100644 index 000000000..eac310867 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_dependency_dependent.py @@ -0,0 +1,77 @@ +########################################################################## +# +# 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.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 fdw_utils + + +class FDWDependencyAndDependentTestCase(BaseTestGenerator): + """This class will test fdw dependents under test database.""" + skip_on_database = ['gpdb'] + + scenarios = utils.generate_scenarios('fdw_dependency_dependent', + fdw_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(FDWDependencyAndDependentTestCase, 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_{0}".format(str(uuid.uuid4())[1:8]) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + + def get_dependency_dependent(self): + """ + This function returns the fdw dependency and dependent + :return: fdw dependency and dependent + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.fdw_id), + follow_redirects=True) + + def runTest(self): + """This function will fetch foreign data wrapper 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.") + + if self.is_positive_test: + response = self.get_dependency_dependent() + expected_response_code = self.expected_data['status_code'] + + self.assertEquals(response.status_code, expected_response_code) + + def tearDown(self): + """This function disconnect the test database and drop added extension + and dependant objects.""" + database_utils.disconnect_database(self, self.server_id, + self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get-handlers.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get-handlers.py new file mode 100644 index 000000000..0fa54aa8e --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get-handlers.py @@ -0,0 +1,83 @@ +########################################################################## +# +# 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.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 fdw_utils +from unittest.mock import patch + + +class FDWHandlersTestCase(BaseTestGenerator): + """This class will test fdw handler under test database.""" + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('fdw_get_handlers', + fdw_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(FDWHandlersTestCase, 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_{0}".format(str(uuid.uuid4())[1:8]) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + + def get_handlers_fdw(self): + """ + This function gets fdw handlers + :return: fdw handlers + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/', + content_type='html/json') + + def runTest(self): + """This function will fetch foreign data wrapper 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.") + + if self.is_positive_test: + response = self.get_handlers_fdw() + + 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.get_handlers_fdw() + + self.assertEquals(response.status_code, + self.expected_data['status_code']) + + def tearDown(self): + """This function disconnect the test database and drop added extension + and dependant objects.""" + database_utils.disconnect_database(self, self.server_id, + self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get.py index cd0212fab..ff4ebeae7 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get.py @@ -17,16 +17,15 @@ 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 fdw_utils +from unittest.mock import patch class FDWDGetTestCase(BaseTestGenerator): - """ This class will add foreign data wrappers under test database. """ + """ This class will test fdw properties + and list API under test database. """ skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign_data_wrapper node. - ('Check FDW Node', - dict(url='/browser/foreign_data_wrapper/obj/')) - ] + scenarios = utils.generate_scenarios('fdw_get', + fdw_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -40,6 +39,28 @@ class FDWDGetTestCase(BaseTestGenerator): self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, self.fdw_name) + def get_fdw(self): + """ + This functions returns the fdw properties + :return: fdw properties + """ + 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 get_fdw_list(self): + """ + This functions returns the fdw list + :return: fdw list + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + str( + self.server_id) + '/' + + str(self.db_id) + '/', + content_type='html/json') + def runTest(self): """This function will fetch foreign data wrapper present under test database.""" @@ -49,12 +70,28 @@ class FDWDGetTestCase(BaseTestGenerator): self.db_id) if not db_con["info"] == "Database connected.": raise Exception("Could not connect to database.") - response = 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') - self.assertEquals(response.status_code, 200) + + if self.is_positive_test: + if hasattr(self, "fdw_list"): + response = self.get_fdw_list() + else: + response = self.get_fdw() + else: + if hasattr(self, "error_fetching_fdw"): + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + if hasattr(self, "fdw_list"): + response = self.get_fdw_list() + else: + response = self.get_fdw() + + if hasattr(self, "wrong_fdw_id"): + self.fdw_id = 99999 + response = self.get_fdw() + + 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 delete the FDW and disconnect the test database """ diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get_nodes.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get_nodes.py new file mode 100644 index 000000000..d80d5f039 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get_nodes.py @@ -0,0 +1,105 @@ +########################################################################## +# +# 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.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 fdw_utils +from unittest.mock import patch + + +class FDWDNodesTestCase(BaseTestGenerator): + """This class will delete foreign data wrappers under test database.""" + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('fdw_get_nodes_and_node', + fdw_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(FDWDNodesTestCase, 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_{0}".format(str(uuid.uuid4())[1:8]) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + + def get_fdw_nodes(self): + """ + This function returns fdw nodes + :return: fdw nodes + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/', + content_type='html/json') + + def get_fdw_node(self): + """ + This functions returns the fdw node + :return: fdw node + """ + 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 data wrapper 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.") + + if self.is_positive_test: + if hasattr(self, "node"): + response = self.get_fdw_node() + else: + response = self.get_fdw_nodes() + + else: + if hasattr(self, "error_fetching_fdw"): + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + if hasattr(self, "node"): + response = self.get_fdw_node() + else: + response = self.get_fdw_nodes() + + if hasattr(self, "wrong_id"): + self.fdw_id = 99999 + response = self.get_fdw_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 extension + and dependant objects.""" + database_utils.disconnect_database(self, self.server_id, + self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get_validators.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get_validators.py new file mode 100644 index 000000000..d251d8e99 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get_validators.py @@ -0,0 +1,81 @@ +########################################################################## +# +# 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.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 fdw_utils +from unittest.mock import patch + + +class FDWValidatorsTestCase(BaseTestGenerator): + """This class will test fdw validators under test database.""" + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('fdw_get_validators', + fdw_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(FDWValidatorsTestCase, 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_{0}".format(str(uuid.uuid4())[1:8]) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + + def get_validators_fdw(self): + """ + This function gets fdw validators + :return: fdw validators + """ + return self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/', + content_type='html/json') + + def runTest(self): + """This function will fetch foreign data wrapper 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.") + + if self.is_positive_test: + response = self.get_validators_fdw() + 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.get_validators_fdw() + + self.assertEquals(response.status_code, + self.expected_data['status_code']) + + def tearDown(self): + """This function delete the FDW and disconnect the test database """ + database_utils.disconnect_database(self, self.server_id, + self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_msql.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_msql.py new file mode 100644 index 000000000..c77357405 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_msql.py @@ -0,0 +1,72 @@ +########################################################################## +# +# 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.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 fdw_utils +from unittest.mock import patch + + +class FDWDGetMSQLTestCase(BaseTestGenerator): + """ This class will test fdw msql under test database. """ + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('fdw_get_msql', + fdw_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(FDWDGetMSQLTestCase, 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_{0}".format(str(uuid.uuid4())[1:8]) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + + def get_fdw_msql(self): + """ + This functions returns the fdw msql + :return: fdw sql + """ + 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 data wrapper 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.") + + if self.is_positive_test: + response = self.get_fdw_msql() + + 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 database to delete it + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_put.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_put.py index 238d98550..93e72d2d9 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_put.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_put.py @@ -18,17 +18,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 fdw_utils +from unittest.mock import patch class FDWDPutTestCase(BaseTestGenerator): """This class will update foreign data wrappers under test database.""" skip_on_database = ['gpdb'] - - scenarios = [ - # Fetching default URL for foreign_data_wrapper node. - ('Check FDW Node', - dict(url='/browser/foreign_data_wrapper/obj/')) - ] + scenarios = utils.generate_scenarios('fdw_update', + fdw_utils.test_cases) def setUp(self): """ This function will create extension and foreign data wrapper.""" @@ -42,6 +39,18 @@ class FDWDPutTestCase(BaseTestGenerator): self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, self.fdw_name) + def update_fdw(self): + """ + This function returns the fdw update response + :return: fdw update response + """ + return self.tester.put( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + str(self.fdw_id), + data=json.dumps(self.test_data), + follow_redirects=True) + def runTest(self): """ This function will fetch foreign data wrapper present under test database. """ @@ -55,20 +64,21 @@ class FDWDPutTestCase(BaseTestGenerator): self.fdw_name) if not fdw_response: raise Exception("Could not find FDW.") - data = { - "description": "This is FDW update comment", - "id": self.fdw_id - } - put_response = self.tester.put( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + - str(self.db_id) + '/' + str(self.fdw_id), - data=json.dumps(data), - follow_redirects=True) - self.assertEquals(put_response.status_code, 200) + self.test_data['id'] = self.fdw_id + + if self.is_positive_test: + put_response = self.update_fdw() + + else: + if hasattr(self, "internal_server_error"): + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + put_response = self.update_fdw() + + self.assertEquals(put_response.status_code, + self.expected_data['status_code']) def tearDown(self): """This function delete the FDW and disconnect the test database """ - 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/tests/test_fdw_sql.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_sql.py new file mode 100644 index 000000000..2515a514e --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_sql.py @@ -0,0 +1,82 @@ +########################################################################## +# +# 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.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 fdw_utils +from unittest.mock import patch + + +class FDWDGetSQLTestCase(BaseTestGenerator): + """ This class will add foreign data wrappers under test database. """ + skip_on_database = ['gpdb'] + scenarios = utils.generate_scenarios('fdw_get_sql', + fdw_utils.test_cases) + + def setUp(self): + """ This function will create extension and foreign data wrapper.""" + super(FDWDGetSQLTestCase, 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_{0}".format(str(uuid.uuid4())[1:8]) + self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name, + self.fdw_name) + + def get_fdw_sql(self): + """ + This functions returns the collation sql + :return: collation sql + """ + 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 data wrapper 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.") + + if self.is_positive_test: + response = self.get_fdw_sql() + 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.get_fdw_sql() + + if hasattr(self, "wrong_fdw_id"): + self.fdw_id = 99999 + response = self.get_fdw_sql() + + 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 database to delete it + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/utils.py b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/utils.py index dac2ba209..1bf659290 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/utils.py +++ b/web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/utils.py @@ -12,9 +12,15 @@ from __future__ import print_function import sys import traceback import uuid +import json +import os from regression.python_test_utils.test_utils import get_db_connection +CURRENT_PATH = os.path.dirname(os.path.realpath(__file__)) +with open(CURRENT_PATH + "/fdw_test_data.json") as data_file: + test_cases = json.load(data_file) + def get_fdw_data(schema_name, db_user): data = { @@ -63,7 +69,8 @@ def create_fdw(server, db_name, fdw_name): old_isolation_level = connection.isolation_level connection.set_isolation_level(0) pg_cursor = connection.cursor() - pg_cursor.execute('''CREATE FOREIGN DATA WRAPPER "%s"''' % fdw_name) + pg_cursor.execute('''CREATE FOREIGN DATA WRAPPER "%s" + OPTIONS (op1 '5')''' % fdw_name) connection.set_isolation_level(old_isolation_level) connection.commit() # Get 'oid' from newly created foreign data wrapper