diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/foreign_tables_test_data.json b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/foreign_tables_test_data.json new file mode 100644 index 000000000..e9160e372 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/foreign_tables_test_data.json @@ -0,0 +1,602 @@ +{ + "foreign_table_create": [ + { + "name": "Create Foreign Table: With valid data.", + "is_positive_test": true, + "inventory_data": {}, + "test_data": { + "acl": [], + "basensp": "schema_name", + "columns": [ + { + "attname": "ename", + "datatype": "text", + "coloptions": [] + } + ], + "constraints": [], + "ftoptions": [], + "inherits": [], + "ftsrvname": "fsrv_name", + "name": "ft_name", + "owner": "server user name", + "relacl": [], + "seclabels": [], + "stracl": [] + }, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + } + }, + { + "name": "Create Foreign Table: With insufficient data.", + "is_positive_test": false, + "inventory_data": {}, + "test_data": { + "acl": [], + "basensp": "schema_name", + "columns": [ + { + "attname": "ename", + "datatype": "text", + "coloptions": [] + } + ], + "constraints": [], + "ftoptions": [], + "inherits": [], + "name": "ft_name", + "owner": "server user name", + "relacl": [], + "seclabels": [], + "stracl": [] + }, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 410, + "error_msg": "Could not find the required parameter (ftsrvname).", + "test_result_data": {} + } + }, + { + "name": "Create Foreign Table: With valid data while server down.", + "is_positive_test": false, + "inventory_data": {}, + "test_data": { + "acl": [], + "basensp": "schema_name", + "columns": [ + { + "attname": "ename", + "datatype": "text", + "coloptions": [] + } + ], + "constraints": [], + "ftoptions": [], + "inherits": [], + "ftsrvname": "fsrv_name", + "name": "ft_name", + "owner": "server user name", + "relacl": [], + "seclabels": [], + "stracl": [] + }, + "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, + "error_msg": "Mocked Internal Server Error", + "test_result_data": {} + } + } + ], + "foreign_table_get": [ + { + "name": "Get Foreign Table: With existing foreign table having precision data type column.", + "is_positive_test": true, + "inventory_data": { + "query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; \" %(schema_name, foreign_table_name, fsrv_name)" + }, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table list: With existing foreign table.", + "is_positive_test": true, + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": true + }, + { + "name": "Get Foreign Table: With Non-existing foreign table.", + "is_positive_test": false, + "inventory_data": {}, + "test_data": { + "ft_id": 999999 + }, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 410, + "error_msg": "could not find the specified foreign table.", + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table: With existing foreign table while server is down.", + "is_positive_test": false, + "inventory_data": {}, + "test_data": {}, + "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, + "error_msg": "Mocked Internal Server Error", + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table list: With existing foreign table while server is down.", + "is_positive_test": false, + "inventory_data": {}, + "test_data": {}, + "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, + "error_msg": "Mocked Internal Server Error", + "test_result_data": {} + }, + "is_list": true + } + ], + "foreign_table_delete": [ + { + "name": "Delete Foreign Table: With existing foreign table.", + "is_positive_test": true, + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Delete Multiple Foreign Tables: With existing foreign tables.", + "is_positive_test": true, + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": true + }, + { + "name": "Delete Foreign Table: With existing foreign table while server is down.", + "is_positive_test": false, + "inventory_data": {}, + "test_data": {}, + "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, + "error_msg": "Mocked Internal Server Error", + "test_result_data": {} + }, + "is_list": false + } + ], + "foreign_table_put": [ + { + "name": "Update Foreign Table: With existing foreign table.", + "is_positive_test": true, + "inventory_data": { + "query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) OPTIONS (columnoptionkey 'columnOptionVal') NULL ) SERVER %s;\" %(schema_name, foreign_table_name, fsrv_name)" + }, + "test_data": { + "description": "This is foreign table update comment", + "fsrvacl": { + "added": [ + { + "grantee": "PUBLIC", + "grantor": "postgres", + "privileges": [ + { + "privilege_type": "U", + "privilege": true, + "with_grant": false + } + ] + } + ] + }, + "columns": { + "added": [ + { + "attname": "col2", + "datatype": "character varying[]", + "coloptions": [] + } + ], + "changed": [ + { + "coloptions": [ + { + "option": "OptionKey", + "value": "OptionValue" + } + ], + "attname": "emp", + "datatype": "\"char\"", + "typdefault": null, + "attnotnull": false, + "collname": "pg_catalog.\"default\"", + "attnum": 1, + "inheritedfrom": null, + "inheritedid": null, + "attstattarget": -1, + "attoptions": null, + "attfdwoptions": [ + "ColOptionKey=ColOptionValue" + ], + "attndims": 0, + "atttypmod": -1, + "fulltype": "text", + "isdup": false + } + ] + } + }, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Update Foreign Table: With existing foreign table while server is down.", + "is_positive_test": false, + "inventory_data": {}, + "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, + "error_msg": "Mocked Internal Server Error", + "test_result_data": {} + }, + "is_list": false + } + ], + "foreign_table_sql": [ + { + "name": "Get Foreign Table Sql: With existing foreign table.", + "is_positive_test": true, + "url": "/browser/foreign_table/sql/", + "inventory_data": { + "query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, foreign_table_name, fsrv_name, schema_name, foreign_table_name)" + }, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table Sql: With existing foreign table having inheritted table.", + "is_positive_test": true, + "url": "/browser/foreign_table/sql/", + "inventory_data": { + "query": "\"CREATE TABLE %s.toinherittable (col1 character varying(10) NULL); CREATE FOREIGN TABLE %s.%s () INHERITS (%s.toinherittable) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, schema_name, foreign_table_name, schema_name, fsrv_name, schema_name, foreign_table_name)" + }, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table Sql: With existing foreign table while server is down.", + "is_positive_test": false, + "url": "/browser/foreign_table/sql/", + "inventory_data": {}, + "test_data": {}, + "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, + "error_msg": "Mocked Internal Server Error", + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table Sql: With Non-existing foreign table.", + "is_positive_test": false, + "url": "/browser/foreign_table/sql/", + "inventory_data": {}, + "test_data": { + "ft_id": 999999 + }, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 410, + "error_msg": "could not find the specified foreign table.", + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table Insert Sql: With existing foreign table.", + "is_positive_test": true, + "url": "/browser/foreign_table/insert_sql/", + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table Update Sql: With existing foreign table.", + "is_positive_test": true, + "url": "/browser/foreign_table/update_sql/", + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table Delete Sql: With existing foreign table.", + "is_positive_test": true, + "url": "/browser/foreign_table/delete_sql/", + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table Insert Sql: With existing foreign table.", + "is_positive_test": true, + "url": "/browser/foreign_table/select_sql/", + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + } + ], + "foreign_table_dependencies_dependents": [ + { + "name": "Get Foreign Table Dependents: With existing foreign table.", + "is_positive_test": true, + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_dependent": true + }, + { + "name": "Get Foreign Table Dependency: With existing foreign table.", + "is_positive_test": true, + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_dependent": false + } + ], + "foreign_table_get_nodes": [ + { + "name": "Get Foreign Table Nodes: With existing foreign table.", + "is_positive_test": true, + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table list node: With existing foreign table.", + "is_positive_test": true, + "inventory_data": {}, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": true + } + ], + "foreign_table_get_msql": [ + { + "name": "Get Foreign Table msql: With existing foreign table.", + "is_positive_test": true, + "inventory_data": {}, + "test_data": { + "description": "Testing Msql API" + }, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table msql: Modify existing foreign table by inheriting.", + "is_positive_test": true, + "inventory_data": { + "query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; \" %(schema_name, foreign_table_name, fsrv_name)" + }, + "test_data": { + "description": "Testing Msql API", + "inherits": [ + "table_id" + ] + }, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + } + ], + "foreign_table_get_creation_parameters": [ + { + "name": "Get Foreign Table Pre Creation parameters: Get collation.", + "is_positive_test": true, + "url": "/browser/foreign_table/get_collations/", + "inventory_data": { + "query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, foreign_table_name, fsrv_name, schema_name, foreign_table_name)" + }, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table Pre Creation parameters: Get tables.", + "is_positive_test": true, + "url": "/browser/foreign_table/get_tables/", + "inventory_data": { + "query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, foreign_table_name, fsrv_name, schema_name, foreign_table_name)" + }, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + }, + { + "name": "Get Foreign Table Pre Creation parameters: Get types.", + "is_positive_test": true, + "url": "/browser/foreign_table/get_types/", + "inventory_data": { + "query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, foreign_table_name, fsrv_name, schema_name, foreign_table_name)" + }, + "test_data": {}, + "mocking_required": false, + "mock_data": {}, + "expected_data": { + "status_code": 200, + "error_msg": null, + "test_result_data": {} + }, + "is_list": false + } + ] +} diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_add.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_add.py index 67ea5cd1c..054aa0fed 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_add.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_add.py @@ -8,8 +8,8 @@ ########################################################################## -import json import uuid +from unittest.mock import patch from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \ foreign_servers.tests import utils as fsrv_utils @@ -29,22 +29,32 @@ class ForeignTableAddTestCase(BaseTestGenerator): """ skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign server node. - ('Check foreign table Node', dict(url='/browser/foreign_table/obj/')) - ] + # url + url = '/browser/foreign_table/obj/' + + # Generates scenarios + scenarios = utils.generate_scenarios("foreign_table_create", + ft_utils.test_cases) def setUp(self): """ This function will create foreign data wrapper and foreign server. """ super(ForeignTableAddTestCase, self).setUp() + # Load test data + self.data = self.test_data + + # Get parent schema info 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"] + + # Create schema self.schema_name = self.schema_data['schema_name'] self.schema_id = self.schema_data['schema_id'] + + # Create FDW & server 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, @@ -53,9 +63,14 @@ class ForeignTableAddTestCase(BaseTestGenerator): self.fsrv_name, self.fdw_name) self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:8]) + self.data['basensp'] = self.schema_name + self.data['ftsrvname'] = self.fsrv_name + self.data['name'] = self.ft_name + self.data['owner'] = self.server["username"] + def runTest(self): """This function will add foreign table under test database.""" - + # Create db connection db_con = database_utils.connect_database(self, utils.SERVER_GROUP, self.server_id, @@ -69,40 +84,40 @@ class ForeignTableAddTestCase(BaseTestGenerator): if not fsrv_response: raise Exception("Could not find Foreign Server.") - data = { - "acl": [], - "basensp": self.schema_name, - "columns": [ - { - "attname": "ename", - "datatype": "text", - "coloptions": [] - } - ], - "constraints": [], - "ftoptions": [], - "inherits": [], - "ftsrvname": self.fsrv_name, - "name": self.ft_name, - "owner": self.server["username"], - "relacl": [], - "seclabels": [], - "stracl": [] - } - - response = self.tester.post( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + str(self.db_id) + '/' + - str(self.schema_id) + '/', data=json.dumps(data), - content_type='html/json') - - self.assertEqual(response.status_code, 200) + if self.is_positive_test: + response = ft_utils.api_create(self) + + # Assert response + utils.assert_status_code(self, response) + + # Verify in backend + cross_check_res = ft_utils.verify_foreign_table(self.server, + self.db_name, + self.fsrv_name) + + self.assertIsNotNone(cross_check_res, "Could not find the newly" + " created foreign table.") + else: + if self.mocking_required: + with patch(self.mock_data["function_name"], + side_effect=eval(self.mock_data["return_value"])): + response = ft_utils.api_create(self) + + # Assert response + utils.assert_status_code(self, response) + utils.assert_error_message(self, response) + else: + del self.data['ftsrvname'] + response = ft_utils.api_create(self) + # Assert response + utils.assert_status_code(self, response) + utils.assert_error_message(self, response) def tearDown(self): """ This function disconnect the test database and delete test foreign table object. """ - ft_utils.delete_foregin_table(self.server, self.db_name, - self.schema_name, self.ft_name - ) + if self.is_positive_test: + ft_utils.delete_foregin_table(self.server, self.db_name, + self.schema_name, self.ft_name) database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_delete.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_delete.py index dd51aa970..3a67ee166 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_delete.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_delete.py @@ -9,6 +9,7 @@ import uuid +from unittest.mock import patch from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \ foreign_servers.tests import utils as fsrv_utils @@ -28,22 +29,30 @@ class ForeignTableDeleteTestCase(BaseTestGenerator): """ skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign table node. - ('Check foreign table Node', dict(url='/browser/foreign_table/obj/')) - ] + # url + url = '/browser/foreign_table/obj/' + + # Generates scenarios + scenarios = utils.generate_scenarios("foreign_table_delete", + ft_utils.test_cases) def setUp(self): """ This function will create foreign data wrapper, foreign server and foreign table. """ super(ForeignTableDeleteTestCase, self).setUp() + # Load test data + self.data = self.test_data + + # Get parent schema info 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.schema_id = self.schema_data['schema_id'] + + # Create FDW, server & table self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:8]) @@ -56,6 +65,16 @@ class ForeignTableDeleteTestCase(BaseTestGenerator): self.fsrv_name, self.ft_name) + # In case of multiple foreign tables + if self.is_list: + self.ft_name_2 = "ft_%s" % (str(uuid.uuid4())[1:8]) + self.ft_id_2 = ft_utils.create_foreign_table(self.server, + self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name_2) + self.data['ids'] = [self.ft_id, self.ft_id_2] + def runTest(self): """This function will delete foreign table under test database.""" @@ -78,15 +97,15 @@ class ForeignTableDeleteTestCase(BaseTestGenerator): if not ft_response: raise Exception("Could not find Foreign Table.") - delete_response = self.tester.delete( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + - str(self.db_id) + '/' + - str(self.schema_id) + '/' + - str(self.ft_id), - follow_redirects=True) + if self.is_positive_test: + + if self.is_list: + response = ft_utils.api_delete(self) + else: + response = ft_utils.api_delete(self, self.ft_id) - self.assertEqual(delete_response.status_code, 200) + # Assert response + utils.assert_status_code(self, response) def tearDown(self): """ This function disconnect the test database. """ diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_dependencies_dependents.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_dependencies_dependents.py new file mode 100644 index 000000000..c4dc8bf38 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_dependencies_dependents.py @@ -0,0 +1,104 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + + +import uuid +from unittest.mock import patch + +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 ft_utils + + +class ForeignTableDependenciesDependentsTestCase(BaseTestGenerator): + """ + This class will fetch foreign table under database node. + """ + skip_on_database = ['gpdb'] + + # url + url = '/browser/foreign_table/' + + # Generates scenarios + scenarios = utils.generate_scenarios( + "foreign_table_dependencies_dependents", ft_utils.test_cases) + + def setUp(self): + """ This function will create foreign data wrapper, foreign server + and foreign table. """ + super(ForeignTableDependenciesDependentsTestCase, self).setUp() + + # Load test data + self.data = self.test_data + + # Get parent schema info + 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"] + + # Create schema + self.schema_name = self.schema_data['schema_name'] + self.schema_id = self.schema_data['schema_id'] + + # Create FDW, server & table + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) + self.ft_name = "ft_%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.ft_id = ft_utils.create_foreign_table(self.server, self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name) + + def runTest(self): + """This function will fetch foreign table under test database.""" + # Create db connection + 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.") + + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + + if not fsrv_response: + raise Exception("Could not find Foreign Server.") + + if self.is_positive_test: + if self.is_dependent: + self.url = self.url + 'dependent/' + response = ft_utils.api_get(self) + else: + self.url = self.url + 'dependency/' + response = ft_utils.api_get(self) + + utils.assert_status_code(self, response) + + def tearDown(self): + """ This function disconnect the test database and delete test + foreign table object. """ + ft_utils.delete_foregin_table(self.server, self.db_name, + self.schema_name, self.ft_name) + + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get.py index 623950cc5..32eff6f72 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get.py @@ -9,6 +9,7 @@ import uuid +from unittest.mock import patch from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \ foreign_servers.tests import utils as fsrv_utils @@ -28,22 +29,32 @@ class ForeignTableGetTestCase(BaseTestGenerator): """ skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign server node. - ('Check foreign table Node', dict(url='/browser/foreign_table/obj/')) - ] + # url + url = '/browser/foreign_table/obj/' + + # Generates scenarios + scenarios = utils.generate_scenarios("foreign_table_get", + ft_utils.test_cases) def setUp(self): """ This function will create foreign data wrapper, foreign server and foreign table. """ super(ForeignTableGetTestCase, self).setUp() + # Load test data + self.data = self.test_data + + # Get parent schema info 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"] + + # Create schema self.schema_name = self.schema_data['schema_name'] self.schema_id = self.schema_data['schema_id'] + + # Create FDW, server & table self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:8]) @@ -52,14 +63,35 @@ class ForeignTableGetTestCase(BaseTestGenerator): self.fdw_name) self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, self.fsrv_name, self.fdw_name) - self.ft_id = ft_utils.create_foreign_table(self.server, self.db_name, - self.schema_name, - self.fsrv_name, - self.ft_name) + + if bool(self.inventory_data): + query = self.inventory_data['query'] + self.ft_id = ft_utils.create_foreign_table(self.server, + self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name, query) + else: + self.ft_id = ft_utils.create_foreign_table(self.server, + self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name) + + # In case of multiple foreign tables + if self.is_list: + self.ft_name_2 = "ft_%s" % (str(uuid.uuid4())[1:8]) + self.ft_id_2 = ft_utils.create_foreign_table(self.server, + self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name_2) + + self.data['ids'] = [self.ft_id, self.ft_id_2] def runTest(self): """This function will fetch foreign table under test database.""" - + # Create db connection db_con = database_utils.connect_database(self, utils.SERVER_GROUP, self.server_id, @@ -74,20 +106,35 @@ class ForeignTableGetTestCase(BaseTestGenerator): if not fsrv_response: raise Exception("Could not find Foreign Server.") - response = self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + - str(self.db_id) + '/' + - str(self.schema_id) + '/' + - str(self.ft_id), - content_type='html/json') - - self.assertEqual(response.status_code, 200) + if self.is_positive_test: + if self.is_list: + response = ft_utils.api_get(self, '') + else: + response = ft_utils.api_get(self) + + # Assert response + utils.assert_status_code(self, response) + else: + if self.mocking_required: + with patch(self.mock_data["function_name"], + side_effect=[eval(self.mock_data["return_value"])]): + if self.is_list: + response = ft_utils.api_get(self, '') + else: + response = ft_utils.api_get(self) + elif 'ft_id' in self.data: + # Non-existing ft_id id + self.ft_id = self.data["ft_id"] + response = ft_utils.api_get(self) + + # Assert response + utils.assert_status_code(self, response) + utils.assert_error_message(self, response) def tearDown(self): """ This function disconnect the test database and delete test foreign table object. """ ft_utils.delete_foregin_table(self.server, self.db_name, - self.schema_name, self.ft_name - ) + self.schema_name, self.ft_name) database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_msql.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_msql.py new file mode 100644 index 000000000..d9ad4ae5d --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_msql.py @@ -0,0 +1,111 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + + +import uuid + +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 ft_utils +from pgadmin.browser.server_groups.servers.databases.schemas.tables.tests \ + import utils as table_utils + + +class ForeignTableGetMsqlTestCase(BaseTestGenerator): + """ + This class will fetch foreign table msql under database node. + """ + skip_on_database = ['gpdb'] + + # url + url = '/browser/foreign_table/msql/' + + # Generates scenarios + scenarios = utils.generate_scenarios("foreign_table_get_msql", + ft_utils.test_cases) + + def setUp(self): + """ This function will create foreign data wrapper, foreign server + and foreign table. """ + super(ForeignTableGetMsqlTestCase, self).setUp() + + # Load test data + self.data = self.test_data + + # Get parent schema info + 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"] + + # Create schema + self.schema_name = self.schema_data['schema_name'] + self.schema_id = self.schema_data['schema_id'] + + # Create FDW, server & table + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) + self.ft_name = "ft_%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.ft_id = ft_utils.create_foreign_table(self.server, self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name) + + def runTest(self): + """This function will fetch foreign table msql under test database.""" + # Create db connection + 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.") + + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + + if not fsrv_response: + raise Exception("Could not find Foreign Server.") + + if self.is_positive_test: + self.data['oid'] = self.ft_id + + if 'inherits' in self.data: + tid = table_utils.create_table(self.server, self.db_name, + self.schema_name, 'toinherit', + "'CREATE TABLE %s.%s ()' " + "%(schema_name, table_name)") + self.data['inherits'] = [tid] + + url_encode_data = self.data + response = ft_utils.api_get_msql(self, url_encode_data) + + # Assert response + utils.assert_status_code(self, response) + + def tearDown(self): + """ This function disconnect the test database and delete test + foreign table object. """ + ft_utils.delete_foregin_table(self.server, self.db_name, + self.schema_name, self.ft_name) + + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_nodes.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_nodes.py new file mode 100644 index 000000000..c7c7b39ca --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_nodes.py @@ -0,0 +1,130 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + + +import uuid +from unittest.mock import patch + +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 ft_utils + + +class ForeignTableGetNodesTestCase(BaseTestGenerator): + """ + This class will fetch foreign table nodes under database node. + """ + skip_on_database = ['gpdb'] + + # url + url = '/browser/foreign_table/nodes/' + + # Generates scenarios + scenarios = utils.generate_scenarios("foreign_table_get_nodes", + ft_utils.test_cases) + + def setUp(self): + """ This function will create foreign data wrapper, foreign server + and foreign table. """ + super(ForeignTableGetNodesTestCase, self).setUp() + + # Load test data + self.data = self.test_data + + # Get parent schema info + 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"] + + # Create schema + self.schema_name = self.schema_data['schema_name'] + self.schema_id = self.schema_data['schema_id'] + + # Create FDW, server & table + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) + self.ft_name = "ft_%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.ft_id = ft_utils.create_foreign_table(self.server, self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name) + + # In case of multiple foreign tables + if self.is_list: + self.ft_name_2 = "ft_%s" % (str(uuid.uuid4())[1:8]) + self.ft_id_2 = ft_utils.create_foreign_table(self.server, + self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name_2) + + self.data['ids'] = [self.ft_id, self.ft_id_2] + + def runTest(self): + """This function will fetch foreign table nodes under test database.""" + # Create db connection + 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.") + + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + + if not fsrv_response: + raise Exception("Could not find Foreign Server.") + + if self.is_positive_test: + if self.is_list: + response = ft_utils.api_get(self, '') + else: + response = ft_utils.api_get(self) + + # Assert response + utils.assert_status_code(self, response) + else: + if self.mocking_required: + with patch(self.mock_data["function_name"], + side_effect=[eval(self.mock_data["return_value"])]): + if self.is_list: + response = ft_utils.api_get(self, '') + else: + response = ft_utils.api_get(self) + elif 'ft_id' in self.data: + # Non-existing ft_id id + self.ft_id = self.data["ft_id"] + response = ft_utils.api_get(self) + + # Assert response + utils.assert_status_code(self, response) + utils.assert_error_message(self, response) + + def tearDown(self): + """ This function disconnect the test database and delete test + foreign table object. """ + ft_utils.delete_foregin_table(self.server, self.db_name, + self.schema_name, self.ft_name) + + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_delete_multiple.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_precreation_parameters.py similarity index 66% rename from web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_delete_multiple.py rename to web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_precreation_parameters.py index 432cec063..a4bbb2630 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_delete_multiple.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_precreation_parameters.py @@ -9,7 +9,7 @@ import uuid -import json +from unittest.mock import patch from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \ foreign_servers.tests import utils as fsrv_utils @@ -23,48 +23,52 @@ from regression.python_test_utils import test_utils as utils from . import utils as ft_utils -class ForeignTableDeleteMultipleTestCase(BaseTestGenerator): +class ForeignTableGetTestCase(BaseTestGenerator): """ - This class will delete foreign table under database node. + This class will fetch foreign table pre creation parameters under + database node. """ skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign table node. - ('Check foreign table Node', dict(url='/browser/foreign_table/obj/')) - ] + # url + url = '/browser/foreign_table/obj/' + + # Generates scenarios + scenarios = utils.generate_scenarios( + "foreign_table_get_creation_parameters", ft_utils.test_cases) def setUp(self): """ This function will create foreign data wrapper, foreign server and foreign table. """ - super(ForeignTableDeleteMultipleTestCase, self).setUp() + super(ForeignTableGetTestCase, self).setUp() + + # Load test data + self.data = self.test_data + # Get parent schema info 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"] + + # Create schema self.schema_name = self.schema_data['schema_name'] self.schema_id = self.schema_data['schema_id'] + + # Create FDW, server & table self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) self.ft_name = "ft_%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.ft_ids = [ft_utils.create_foreign_table( - self.server, self.db_name, - self.schema_name, self.fsrv_name, - "ft_%s" % (str(uuid.uuid4())[1:8])), - ft_utils.create_foreign_table( - self.server, self.db_name, - self.schema_name, self.fsrv_name, - "ft_%s" % (str(uuid.uuid4())[1:8])) - ] def runTest(self): - """This function will delete foreign table under test database.""" - + """This function will fetch foreign table pre creation + under test database.""" + # Create db connection db_con = database_utils.connect_database(self, utils.SERVER_GROUP, self.server_id, @@ -79,20 +83,12 @@ class ForeignTableDeleteMultipleTestCase(BaseTestGenerator): if not fsrv_response: raise Exception("Could not find Foreign Server.") - data = {'ids': self.ft_ids} + if self.is_positive_test: + response = ft_utils.api_get(self, '') - delete_response = self.tester.delete( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + - str(self.db_id) + '/' + - str(self.schema_id) + '/', - content_type='html/json', - follow_redirects=True, - data=json.dumps(data)) - - self.assertEqual(delete_response.status_code, 200) + # Assert response + utils.assert_status_code(self, response) def tearDown(self): - """ This function disconnect the test database. """ - + """ This function 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/schemas/foreign_tables/tests/test_foreign_table_put.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_put.py index 99f9552f7..9bcb924d2 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_put.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_put.py @@ -10,6 +10,7 @@ import json import uuid +from unittest.mock import patch from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \ foreign_servers.tests import utils as fsrv_utils @@ -29,22 +30,30 @@ class ForeignTablePutTestCase(BaseTestGenerator): """ skip_on_database = ['gpdb'] - scenarios = [ - # Fetching default URL for foreign server node. - ('Check foreign table Node', dict(url='/browser/foreign_table/obj/')) - ] + # url + url = '/browser/foreign_table/obj/' + + # Generates scenarios + scenarios = utils.generate_scenarios("foreign_table_put", + ft_utils.test_cases) def setUp(self): """ This function will create foreign data wrapper, foreign server and foreign table. """ super(ForeignTablePutTestCase, self).setUp() + # Load test data + self.data = self.test_data + + # Get parent schema info 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.schema_id = self.schema_data['schema_id'] + + # Create FDW, server & table self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:8]) @@ -53,14 +62,23 @@ class ForeignTablePutTestCase(BaseTestGenerator): self.fdw_name) self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name, self.fsrv_name, self.fdw_name) - self.ft_id = ft_utils.create_foreign_table(self.server, self.db_name, - self.schema_name, - self.fsrv_name, - self.ft_name) + + if bool(self.inventory_data): + query = self.inventory_data['query'] + self.ft_id = ft_utils.create_foreign_table(self.server, + self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name, query) + else: + self.ft_id = ft_utils.create_foreign_table(self.server, + self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name) def runTest(self): """This function will update foreign table under test database.""" - db_con = database_utils.connect_database(self, utils.SERVER_GROUP, self.server_id, @@ -80,28 +98,27 @@ class ForeignTablePutTestCase(BaseTestGenerator): if not ft_response: raise Exception("Could not find Foreign Table.") - data = \ - { - "description": "This is foreign table update comment", - "id": self.ft_id, - } + self.data['id'] = self.ft_id + + if self.is_positive_test: + response = ft_utils.api_put(self) - put_response = self.tester.put( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + - str(self.db_id) + '/' + - str(self.schema_id) + '/' + - str(self.ft_id), - data=json.dumps(data), - follow_redirects=True) + # Assert response + utils.assert_status_code(self, response) + else: + if self.mocking_required: + with patch(self.mock_data["function_name"], + side_effect=[eval(self.mock_data["return_value"])]): + response = ft_utils.api_put(self) - self.assertEqual(put_response.status_code, 200) + # Assert response + utils.assert_status_code(self, response) + utils.assert_error_message(self, response) def tearDown(self): """ This function disconnect the test database and delete test foreign table object. """ ft_utils.delete_foregin_table(self.server, self.db_name, - self.schema_name, self.ft_name - ) + self.schema_name, self.ft_name) database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_sql.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_sql.py new file mode 100644 index 000000000..b1783be16 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_sql.py @@ -0,0 +1,123 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + + +import uuid +from unittest.mock import patch + +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 ft_utils + + +class ForeignTableSqlTestCase(BaseTestGenerator): + """ + This class will fetch foreign table under database node. + """ + skip_on_database = ['gpdb'] + + # Generates scenarios + scenarios = utils.generate_scenarios("foreign_table_sql", + ft_utils.test_cases) + + def setUp(self): + """ This function will create foreign data wrapper, foreign server + and foreign table. """ + super(ForeignTableSqlTestCase, self).setUp() + + # Load test data + self.data = self.test_data + + # Get parent schema info + 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"] + + # Create schema + self.schema_name = self.schema_data['schema_name'] + self.schema_id = self.schema_data['schema_id'] + + # Create FDW, server & table + self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8]) + self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8]) + self.ft_name = "ft_%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) + + if bool(self.inventory_data): + query = self.inventory_data['query'] + self.ft_id = ft_utils.create_foreign_table(self.server, + self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name, query) + else: + self.ft_id = ft_utils.create_foreign_table(self.server, + self.db_name, + self.schema_name, + self.fsrv_name, + self.ft_name) + + def runTest(self): + """This function will fetch foreign table under test database.""" + # Create db connection + 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.") + + fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name, + self.fsrv_name) + + if not fsrv_response: + raise Exception("Could not find Foreign Server.") + + if self.is_positive_test: + response = ft_utils.api_get(self) + + # Assert response + utils.assert_status_code(self, response) + else: + if self.mocking_required: + with patch(self.mock_data["function_name"], + side_effect=[eval(self.mock_data["return_value"])]): + if self.is_list: + response = ft_utils.api_get(self, '') + else: + response = ft_utils.api_get(self) + elif 'ft_id' in self.data: + # Non-existing ft_id id + self.ft_id = self.data["ft_id"] + response = ft_utils.api_get(self) + + # Assert response + utils.assert_status_code(self, response) + utils.assert_error_message(self, response) + + def tearDown(self): + """ This function disconnect the test database and delete test + foreign table object. """ + ft_utils.delete_foregin_table(self.server, self.db_name, + self.schema_name, self.ft_name) + + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/utils.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/utils.py index d659fce5e..c679bfdd9 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/utils.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/utils.py @@ -6,19 +6,67 @@ # This software is released under the PostgreSQL Licence # ########################################################################## - - +import json import os import sys import traceback +from urllib.parse import urlencode from regression.python_test_utils.test_utils import get_db_connection +from regression.python_test_utils import test_utils as utils + +# Load test data from json file. +CURRENT_PATH = os.path.dirname(os.path.realpath(__file__)) +with open(CURRENT_PATH + "/foreign_tables_test_data.json") as data_file: + test_cases = json.load(data_file) file_name = os.path.basename(__file__) +def api_create(self): + return self.tester.post("{0}{1}/{2}/{3}/{4}/". + format(self.url, utils.SERVER_GROUP, + self.server_id, self.db_id, self.schema_id), + data=json.dumps(self.data), + content_type='html/json') + + +def api_get(self, ft_table_id=None): + if ft_table_id is None: + ft_table_id = self.ft_id + return self.tester.get("{0}{1}/{2}/{3}/{4}/{5}". + format(self.url, utils.SERVER_GROUP, self.server_id, + self.db_id, self.schema_id, ft_table_id), + content_type='html/json') + + +def api_delete(self, ft_tbl_id=''): + return self.tester.delete("{0}{1}/{2}/{3}/{4}/{5}". + format(self.url, utils.SERVER_GROUP, + self.server_id, self.db_id, + self.schema_id, ft_tbl_id), + data=json.dumps(self.data), + follow_redirects=True) + + +def api_put(self): + return self.tester.put("{0}{1}/{2}/{3}/{4}/{5}". + format(self.url, utils.SERVER_GROUP, self.server_id, + self.db_id, self.schema_id, self.ft_id), + data=json.dumps(self.data), + follow_redirects=True) + + +def api_get_msql(self, url_encode_data): + return self.tester.get("{0}{1}/{2}/{3}/{4}/{5}?{6}". + format(self.url, utils.SERVER_GROUP, self.server_id, + self.db_id, self.schema_id, self.ft_id, + urlencode(url_encode_data)), + follow_redirects=True) + + def create_foreign_table(server, db_name, schema_name, fsrv_name, - foreign_table_name): + foreign_table_name, sql_query=None): """ This function will create foreign table under the existing dummy schema. @@ -28,6 +76,12 @@ def create_foreign_table(server, db_name, schema_name, fsrv_name, """ try: + query = "CREATE FOREIGN TABLE " + schema_name + "." + \ + foreign_table_name + "(emp_name text NULL) SERVER %s" % \ + fsrv_name + + if sql_query is not None: + query = eval(sql_query) connection = get_db_connection(db_name, server['username'], server['db_password'], @@ -36,11 +90,7 @@ def create_foreign_table(server, db_name, schema_name, fsrv_name, old_isolation_level = connection.isolation_level connection.set_isolation_level(0) pg_cursor = connection.cursor() - - pg_cursor.execute( - "CREATE FOREIGN TABLE " + schema_name + "." + foreign_table_name + - "(emp_name text NULL) SERVER %s" % fsrv_name) - + pg_cursor.execute(query) connection.set_isolation_level(old_isolation_level) connection.commit()