diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_add.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_add.py index 2eb6139..7301cc8 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_add.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_add.py @@ -9,23 +9,85 @@ import json import uuid - +from unittest.mock import patch from pgadmin.browser.server_groups.servers.databases.tests import utils as \ database_utils from pgadmin.utils.route import BaseTestGenerator from regression.python_test_utils import test_utils as utils from . import utils as funcs_utils +from .. import FunctionView class FunctionAddTestCase(BaseTestGenerator): """ This class will add new function under schema node. """ scenarios = [ # Fetching default URL for function node. - ('Fetch Function Node URL', dict( - url='/browser/function/obj/')) + ( + 'Fetch Function Node URL', + dict( + url='/browser/function/obj/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ('Create Function Get Sql Fail', dict( + url='/browser/function/obj/', + is_positive_test=False, + mocking_required=True, + is_mock_function=True, + mock_data={ + "function_name": '_get_sql', + "return_value": "(False, '')" + }, + expected_data={ + "status_code": 500 + } + )), + ('Create Function Get Sql Execution Fail', dict( + url='/browser/function/obj/', + 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 create new function get sql.')" + }, + expected_data={ + "status_code": 500 + } + )), + ('Create Function Fail', dict( + url='/browser/function/obj/', + 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 create new function.')" + }, + expected_data={ + "status_code": 500 + } + )), ] + def create_function(self, data): + 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' + ) + return response + def runTest(self): """ This function will add function under schema node. """ super(FunctionAddTestCase, self).runTest() @@ -82,14 +144,23 @@ class FunctionAddTestCase(BaseTestGenerator): data['prosupportfuc'] = support_function_name - 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' - ) + if self.is_positive_test: + response = self.create_function(data) + else: + if hasattr(self, 'is_mock_function'): + def _get_sql(self, **kwargs): + return False, '' + with patch.object(FunctionView, + self.mock_data["function_name"], + new=_get_sql): + response = self.create_function(data) + + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.create_function(data) - self.assertEqual(response.status_code, 200) + self.assertEqual(response.status_code, + self.expected_data['status_code']) # Disconnect the database database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_delete.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_delete.py index 4244aaa..8fa5161 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_delete.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_delete.py @@ -7,7 +7,9 @@ # ########################################################################## +import json import uuid +from unittest.mock import patch from pgadmin.browser.server_groups.servers.databases.tests import utils as \ database_utils @@ -20,10 +22,80 @@ class FunctionDeleteTestCase(BaseTestGenerator): """ This class will delete the function under schema node. """ scenarios = [ # Fetching default URL for function node. - ('Fetch Function Node URL', - dict(url='/browser/function/obj/')) + ('Fetch Function Node URL', dict( + url='/browser/function/obj/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + )), + ('Delete Function without function id.', dict( + url='/browser/function/obj/', + is_positive_test=True, + mocking_required=False, + without_functions_id=True, + test_data={}, + mock_data={}, + expected_data={ + "status_code": 200 + } + )), + ('Delete Function Fail', dict( + url='/browser/function/obj/', + is_positive_test=False, + mocking_required=True, + without_functions_id=False, + test_data={}, + mock_data={ + "function_name": "pgadmin.utils.driver.psycopg2." + "connection.Connection.execute_2darray", + "return_value": "(False, 'Mocked Internal Server " + "Error while delete function.')" + }, + expected_data={ + "status_code": 500 + } + )), + ('Delete Function with no object found', dict( + url='/browser/function/obj/', + is_positive_test=True, + mocking_required=True, + mock_empty_result=True, + test_data={}, + mock_data={ + "function_name": "pgadmin.utils.driver.psycopg2." + "connection.Connection.execute_2darray", + "return_value": "(True, {'rows': []})" + }, + expected_data={ + "status_code": 200 + } + )), ] + def delete_function(self, func_id): + if hasattr(self, 'without_functions_id'): + + response = self.tester.delete( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.schema_id) + '/' + str(func_id), + data=json.dumps(self.test_data), + content_type='html/json' + ) + else: + response = self.tester.delete( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.schema_id) + '/' + str(func_id), + content_type='html/json' + ) + return response + def runTest(self): """ This function will delete function under database node. """ super(FunctionDeleteTestCase, self).setUp() @@ -34,14 +106,25 @@ class FunctionDeleteTestCase(BaseTestGenerator): self.server, self.db_name, self.schema_name, func_name) func_id = function_info[0] - response = self.tester.delete( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + - str(self.db_id) + '/' + - str(self.schema_id) + '/' + str(func_id), - content_type='html/json' - ) - self.assertEqual(response.status_code, 200) + if self.is_positive_test: + if hasattr(self, 'without_functions_id'): + func_id = '' + self.test_data = { + "ids": [function_info[0]] + } + if hasattr(self, 'mock_empty_result') and self.mock_empty_result: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.delete_function(func_id) + else: + response = self.delete_function(func_id) + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.delete_function(func_id) + + self.assertEqual(response.status_code, + self.expected_data['status_code']) # Disconnect the database database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get.py index 17ba115..124c272 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get.py @@ -8,6 +8,7 @@ ########################################################################## import uuid +from unittest.mock import patch from pgadmin.browser.server_groups.servers.databases.tests import utils as \ database_utils @@ -21,10 +22,62 @@ class FunctionGetTestCase(BaseTestGenerator): skip_on_database = ['gpdb'] scenarios = [ # Fetching default URL for function node. - ('Fetch Function Node URL', - dict(url='/browser/function/obj/')) + ( + 'Fetch Function Node URL', + dict( + url='/browser/function/obj/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ) + ), + ( + 'Fetch Function properties fail.', + dict( + url='/browser/function/obj/', + 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 get properties.')" + }, + expected_data={ + "status_code": 500 + } + ) + ), + ( + 'Fetch Function properties no function found.', + dict( + url='/browser/function/obj/', + is_positive_test=False, + mocking_required=True, + mock_data={ + "function_name": "pgadmin.utils.driver.psycopg2." + "connection.Connection.execute_dict", + "return_value": "(False, {'rows': []})" + }, + expected_data={ + "status_code": 500 + } + ) + ) ] + def get_properties(self, trigger_func_id): + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + + str(self.db_id) + '/' + + str(self.schema_id) + '/' + str(trigger_func_id), + content_type='html/json') + return response + def runTest(self): """ This function will delete function under database node. """ super(FunctionGetTestCase, self).setUp() @@ -35,13 +88,15 @@ class FunctionGetTestCase(BaseTestGenerator): self.server, self.db_name, self.schema_name, func_name) trigger_func_id = function_info[0] - response = self.tester.get( - self.url + str(utils.SERVER_GROUP) + '/' + - str(self.server_id) + '/' + - str(self.db_id) + '/' + - str(self.schema_id) + '/' + str(trigger_func_id), - content_type='html/json') - self.assertEqual(response.status_code, 200) + if self.is_positive_test: + response = self.get_properties(trigger_func_id) + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_properties(trigger_func_id) + + self.assertEqual(response.status_code, + self.expected_data['status_code']) # Disconnect the database database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_languages.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_languages.py new file mode 100644 index 0000000..b43b663 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_languages.py @@ -0,0 +1,80 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionGetLanguagesTestCase(BaseTestGenerator): + """ This class get languages. """ + scenarios = [ + ( + 'Fetch Function languages', + dict( + url='/browser/function/get_languages/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function languages fail', + dict( + url='/browser/function/get_languages/', + 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 get languages for functions.')" + }, + expected_data={ + "status_code": 500 + } + ), + ) + ] + + def get_languages(self): + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/', + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetLanguagesTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_languages() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_languages() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_msql.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_msql.py new file mode 100644 index 0000000..5a2dec2 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_msql.py @@ -0,0 +1,224 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from urllib.parse import urlencode +from .. import FunctionView + + +class FunctionGetmsqlTestCase(BaseTestGenerator): + """ This class get SQL. """ + scenarios = [ + ( + 'Fetch Function msql', + dict( + url='/browser/function/msql/', + is_positive_test=True, + mocking_required=False, + is_mock_local_function=False, + test_data={ + "name": "Test Function", + "funcowner": "postgres", + "pronamespace": 2200, + "prorettypename": "character varying", + "lanname": "sql", + "arguments": [], + "prosrc": "select '1'", + "probin": "$libdir/", + "variables": [], + "seclabels": [], + "acl": [] + }, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function msql fail', + dict( + url='/browser/function/msql/', + is_positive_test=False, + mocking_required=True, + is_mock_local_function=True, + test_data={ + "name": "Test Function", + "funcowner": "postgres", + "pronamespace": 2200, + "prorettypename": "character varying", + "lanname": "sql", + "arguments": [], + "prosrc": "select '1'", + "probin": "$libdir/", + "variables": [], + "seclabels": [], + "acl": [] + }, + mock_data={ + "function_name": '_get_sql', + "return_value": "(False, '')" + }, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function msql with function id', + dict( + url='/browser/function/msql/', + is_positive_test=True, + mocking_required=True, + with_function_id=True, + is_mock_local_function=True, + test_data={ + "name": "Test Function", + "funcowner": "postgres", + "pronamespace": 2200, + "prorettypename": "character varying", + "lanname": "sql", + "arguments": [], + "prosrc": "select '1'", + "probin": "$libdir/", + "variables": [], + "seclabels": [], + "acl": [] + }, + mock_data={ + "function_name": '_get_sql', + "return_value": "(False, '')" + }, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function msql fetch properties fail', + dict( + url='/browser/function/msql/', + is_positive_test=False, + mocking_required=True, + with_function_id=True, + is_mock_local_function=False, + test_data={ + "name": "Test Function", + "funcowner": "postgres", + "pronamespace": 2200, + "prorettypename": "character varying", + "lanname": "sql", + "arguments": [], + "prosrc": "select '1'", + "probin": "$libdir/", + "variables": [], + "seclabels": [], + "acl": [] + }, + mock_data={ + "function_name": "pgadmin.utils.driver.psycopg2." + "connection.Connection.execute_dict", + "return_value": "(False, 'Mocked Internal Server " + "Error while get msq fetch properties.')" + }, + expected_data={ + "status_code": 500 + } + ) + ), + ( + 'Fetch Function msql fetch properties not found', + dict( + url='/browser/function/msql/', + is_positive_test=False, + mocking_required=True, + with_function_id=True, + is_mock_local_function=False, + test_data={ + "name": "Test Function", + "funcowner": "postgres", + "pronamespace": 2200, + "prorettypename": "character varying", + "lanname": "sql", + "arguments": [], + "prosrc": "select '1'", + "probin": "$libdir/", + "variables": [], + "seclabels": [], + "acl": [] + }, + mock_data={ + "function_name": "pgadmin.utils.driver.psycopg2." + "connection.Connection.execute_dict", + "return_value": "(True, {'rows': []})" + }, + expected_data={ + "status_code": 500 + } + ), + ), + ] + + def get_sql(self): + if hasattr(self, "with_function_id") and self.with_function_id: + func_name = "test_function_delete_%s" % str(uuid.uuid4())[1:8] + function_info = funcs_utils.create_function( + self.server, self.db_name, self.schema_name, func_name) + + func_id = function_info[0] + self.test_data['oid'] = func_id + self.test_data['name'] = func_name + url = self.url + str(utils.SERVER_GROUP) + '/' + str( + self.server_id) + '/' + str(self.db_id) + '/' + str( + self.schema_id) + '/' + str(func_id) + '?' + ( + urlencode(self.test_data)) + else: + url = self.url + str(utils.SERVER_GROUP) + '/' + str( + self.server_id) + '/' + str(self.db_id) + '/' + str( + self.schema_id) + '/?' + (urlencode(self.test_data)) + response = self.tester.get( + url, + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetmsqlTestCase, self).runTest() + self = funcs_utils.set_up(self) + db_user = self.server["username"] + self.test_data["funcowner"] = db_user + + if self.is_positive_test: + response = self.get_sql() + else: + def _get_sql(self, **kwargs): + return False, '' + if self.is_mock_local_function: + with patch.object(FunctionView, + self.mock_data["function_name"], + new=_get_sql): + response = self.get_sql() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_sql() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_nodes.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_nodes.py new file mode 100644 index 0000000..2a46468 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_nodes.py @@ -0,0 +1,121 @@ +########################################################################## +# +# 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.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils + + +class FunctionGetNodesTestCase(BaseTestGenerator): + """ This class get function types """ + scenarios = [ + ( + 'Fetch Function Nodes', + dict( + url='/browser/function/nodes/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function Nodes Fail.', + dict( + url='/browser/function/nodes/', + 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 get function nodes.')" + }, + expected_data={ + "status_code": 500 + } + ), + ), + ( + 'Fetch Function Nodes with function id.', + dict( + url='/browser/function/nodes/', + is_positive_test=True, + mocking_required=False, + with_function_id=True, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function Nodes with function id fail.', + dict( + url='/browser/function/nodes/', + is_positive_test=False, + mocking_required=True, + with_function_id=True, + mock_data={ + "function_name": "pgadmin.utils.driver.psycopg2." + "connection.Connection.execute_2darray", + "return_value": "(True, {'rows':[]})" + }, + expected_data={ + "status_code": 410 + } + ), + ) + ] + + def get_nodes(self): + if hasattr(self, 'with_function_id') and self.with_function_id: + func_name = "test_function_delete_%s" % str(uuid.uuid4())[1:8] + function_info = funcs_utils.create_function( + self.server, self.db_name, self.schema_name, func_name) + + func_id = function_info[0] + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/' + str(func_id), + content_type='html/json' + ) + else: + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/', + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetNodesTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_nodes() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_nodes() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_sql.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_sql.py new file mode 100644 index 0000000..0f5a1e2 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_sql.py @@ -0,0 +1,85 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionSqlTestCase(BaseTestGenerator): + """ This class for function SQL """ + scenarios = [ + ( + 'Fetch Function SQL', + dict( + url='/browser/function/sql/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function SQL fail', + dict( + url='/browser/function/sql/', + 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 SQL for functions.')" + }, + expected_data={ + "status_code": 500 + } + ), + ) + ] + + def sql(self, func_id): + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/' + str(func_id), + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionSqlTestCase, self).runTest() + self = funcs_utils.set_up(self) + func_name = "test_function_delete_%s" % str(uuid.uuid4())[1:8] + function_info = funcs_utils.create_function( + self.server, self.db_name, self.schema_name, func_name) + + func_id = function_info[0] + + if self.is_positive_test: + response = self.sql(func_id) + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.sql(func_id) + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_types.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_types.py new file mode 100644 index 0000000..dd38406 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_types.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 +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionGetTypesTestCase(BaseTestGenerator): + """ This class get types """ + scenarios = [ + ( + 'Fetch Function types', + dict( + url='/browser/function/get_types/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function types fail', + dict( + url='/browser/function/get_types/', + is_positive_test=False, + mocking_required=True, + mock_data={ + "function_name": 'get_types', + "return_value": "(False, [])" + }, + expected_data={ + "status_code": 500 + } + ), + ) + ] + + def get_types(self): + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/', + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetTypesTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_types() + else: + def _get_types(self, conn, condition, add_serials=False, + schema_oid=''): + return False, [] + + with patch.object(FunctionView, self.mock_data["function_name"], + new=_get_types): + response = self.get_types() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_variable_options.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_variable_options.py new file mode 100644 index 0000000..4523202 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_variable_options.py @@ -0,0 +1,80 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionGetLanguagesTestCase(BaseTestGenerator): + """ This class get variable options. """ + scenarios = [ + ( + 'Fetch Function variable options', + dict( + url='/browser/function/vopts/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function variable options fail', + dict( + url='/browser/function/vopts/', + 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 get languages for functions.')" + }, + expected_data={ + "status_code": 500 + } + ), + ) + ] + + def get_languages(self): + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/', + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetLanguagesTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_languages() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_languages() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_put.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_put.py index 1db46a3..076656a 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_put.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_put.py @@ -9,12 +9,14 @@ import json import uuid +from unittest.mock import patch from pgadmin.browser.server_groups.servers.databases.tests import utils as \ database_utils from pgadmin.utils.route import BaseTestGenerator from regression.python_test_utils import test_utils as utils from . import utils as funcs_utils +from .. import FunctionView class FunctionPutTestCase(BaseTestGenerator): @@ -22,18 +24,133 @@ class FunctionPutTestCase(BaseTestGenerator): skip_on_database = ['gpdb'] scenarios = [ # Fetching default URL for function node. - ('Fetch Function Node URL', - dict(url='/browser/function/obj/')) + ('Fetch Function Node URL', dict( + url='/browser/function/obj/', + is_positive_test=True, + mocking_required=False, + is_mock_function=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + )), + ('Fetch Function update fail', dict( + url='/browser/function/obj/', + is_positive_test=False, + mocking_required=True, + is_mock_function=False, + mock_data={ + "function_name": "pgadmin.utils.driver.psycopg2." + "connection.Connection.execute_scalar", + "return_value": "(False, 'Mocked Internal Server " + "Error while create new function get sql.')" + }, + expected_data={ + "status_code": 500 + } + )), + ('Fetch Function update get sql fail', dict( + url='/browser/function/obj/', + is_positive_test=False, + mocking_required=True, + is_mock_function=True, + mock_data={ + "function_name": '_get_sql', + "return_value": [False, ''] + }, + expected_data={ + "status_code": 500 + } + )), + ('Fetch Function update get sql with no result.', dict( + url='/browser/function/obj/', + is_positive_test=False, + mocking_required=True, + is_mock_function=True, + mock_data={ + "function_name": '_get_sql', + "return_value": [True, ''] + }, + expected_data={ + "status_code": 200 + } + )), + ( + 'Fetch Function update with arguments', + dict( + url='/browser/function/obj/', + is_positive_test=True, + mocking_required=False, + with_function_id=True, + is_mock_local_function=False, + is_add_argument=True, + test_data={ + "arguments": { + "changed": [{ + "argdefval": "2", + "argid": 0, + "argmode": "IN", + "argname": "test", + "argtype": "integer", + }] + } + }, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function update with arguments fail', + dict( + url='/browser/function/obj/', + is_positive_test=True, + mocking_required=False, + with_function_id=True, + is_mock_local_function=False, + is_add_argument=True, + test_data={ + "arguments": { + "changed": [{ + "argdefval": "2", + "argid": 0, + "argmode": "IN", + "argname": "param", + "argtype": "integer", + }] + } + }, + mock_data={}, + expected_data={ + "status_code": 500 + } + ), + ), ] + def update_function(self, func_id, data): + put_response = self.tester.put( + self.url + str(utils.SERVER_GROUP) + + '/' + str(self.server_id) + '/' + str(self.db_id) + '/' + + str(self.schema_id) + '/' + + str(func_id), + data=json.dumps(data), + follow_redirects=True) + return put_response + def runTest(self): """ This function will update function under database node. """ super(FunctionPutTestCase, self).setUp() self = funcs_utils.set_up(self) - func_name = "test_event_delete_%s" % str(uuid.uuid4())[1:8] + + if hasattr(self, "is_add_argument") and self.is_add_argument: + args = "IN test integer DEFAULT 1" + else: + args = '' function_info = funcs_utils.create_function( - self.server, self.db_name, self.schema_name, func_name) + self.server, self.db_name, self.schema_name, func_name, args=args) func_id = function_info[0] @@ -42,6 +159,9 @@ class FunctionPutTestCase(BaseTestGenerator): "id": func_id } + if hasattr(self, "is_add_argument") and self.is_add_argument: + data['arguments'] = self.test_data['arguments'] + if self.server_version >= 120000: support_function_name = 'supportfunc_%s' % str(uuid.uuid4())[1:8] funcs_utils.create_support_internal_function( @@ -53,14 +173,29 @@ class FunctionPutTestCase(BaseTestGenerator): data['prosupportfuc'] = support_function_name - put_response = self.tester.put( - self.url + str(utils.SERVER_GROUP) + - '/' + str(self.server_id) + '/' + str(self.db_id) + '/' + - str(self.schema_id) + '/' + - str(func_id), - data=json.dumps(data), - follow_redirects=True) - self.assertEqual(put_response.status_code, 200) + if self.is_positive_test: + response = self.update_function(func_id, data) + else: + + if hasattr(self, 'is_mock_function') and self.is_mock_function: + local_ref = self + + def _get_sql(self, **kwargs): + status = local_ref.mock_data["return_value"][0] + result = local_ref.mock_data['return_value'][1] + return status, result + + with patch.object(FunctionView, + self.mock_data["function_name"], + new=_get_sql): + response = self.update_function(func_id, data) + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.update_function(func_id, data) + + self.assertEqual(response.status_code, + self.expected_data['status_code']) # Disconnect the database database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_dependencies.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_dependencies.py new file mode 100644 index 0000000..439e45d --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_dependencies.py @@ -0,0 +1,85 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionGetDependenciesTestCase(BaseTestGenerator): + """ This class get dependencies for functions. """ + scenarios = [ + ( + 'Fetch Function dependencies.', + dict( + url='/browser/function/dependency/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function dependencies fail', + dict( + url='/browser/function/dependency/', + is_positive_test=False, + mocking_required=True, + mock_data={ + "function_name": 'pgadmin.utils.driver.psycopg2.' + 'connection.Connection.execute_dict', + "return_value": "(False, {'rows': []})" + }, + expected_data={ + "status_code": 200 + } + ), + ) + ] + + def get_dependency(self): + func_name = "test_function_delete_%s" % str(uuid.uuid4())[1:8] + function_info = funcs_utils.create_function( + self.server, self.db_name, self.schema_name, func_name) + + func_id = function_info[0] + + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/' + str(func_id), + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetDependenciesTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_dependency() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_dependency() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_dependents.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_dependents.py new file mode 100644 index 0000000..d62ecda --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_dependents.py @@ -0,0 +1,85 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionGetDependentsTestCase(BaseTestGenerator): + """ This class get dependents for functions. """ + scenarios = [ + ( + 'Fetch Function dependents.', + dict( + url='/browser/function/dependent/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function dependents fail', + dict( + url='/browser/function/dependent/', + is_positive_test=False, + mocking_required=True, + mock_data={ + "function_name": 'pgadmin.utils.driver.psycopg2.' + 'connection.Connection.execute_dict', + "return_value": "(False, {'rows': []})" + }, + expected_data={ + "status_code": 200 + } + ), + ) + ] + + def get_dependents(self): + func_name = "test_function_delete_%s" % str(uuid.uuid4())[1:8] + function_info = funcs_utils.create_function( + self.server, self.db_name, self.schema_name, func_name) + + func_id = function_info[0] + + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/' + str(func_id), + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetDependentsTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_dependents() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_dependents() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_function_statistics.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_function_statistics.py new file mode 100644 index 0000000..ff74106 --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_function_statistics.py @@ -0,0 +1,120 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionGetFunctionStatisticsTestCase(BaseTestGenerator): + """ This class get functions statistics. """ + scenarios = [ + ( + 'Fetch Function get statistics', + dict( + url='/browser/function/stats/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function get statistics fail', + dict( + url='/browser/function/stats/', + 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 get function statistics.')" + }, + expected_data={ + "status_code": 500 + } + ), + ), + ( + 'Fetch Function get statistics without function id', + dict( + url='/browser/function/stats/', + is_positive_test=True, + mocking_required=True, + without_function_id=True, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function get statistics without function id fail', + dict( + url='/browser/function/stats/', + is_positive_test=False, + mocking_required=True, + without_function_id=True, + mock_data={ + "function_name": 'pgadmin.utils.driver.psycopg2.' + 'connection.Connection.execute_scalar', + "return_value": "(False, 'Mocked Internal Server Error " + "while get function statistics.')" + }, + expected_data={ + "status_code": 500 + } + ), + ), + ] + + def get_function_statistics(self): + if hasattr(self, "without_function_id") and self.without_function_id: + func_id = '' + else: + func_name = "test_function_delete_%s" % str(uuid.uuid4())[1:8] + function_info = funcs_utils.create_function( + self.server, self.db_name, self.schema_name, func_name) + + func_id = function_info[0] + + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/' + str(func_id), + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetFunctionStatisticsTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_function_statistics() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_function_statistics() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_list.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_list.py new file mode 100644 index 0000000..5fead0a --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_list.py @@ -0,0 +1,80 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionGetListTestCase(BaseTestGenerator): + """ This class get list of functions. """ + scenarios = [ + ( + 'Fetch Function list.', + dict( + url='/browser/function/obj/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function dependents fail', + dict( + url='/browser/function/obj/', + 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 get function list.')" + }, + expected_data={ + "status_code": 500 + } + ), + ) + ] + + def get_list(self): + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/', + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetListTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_list() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_list() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_select_sql.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_select_sql.py new file mode 100644 index 0000000..3eef04a --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_select_sql.py @@ -0,0 +1,122 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionGetSelectSqlTestCase(BaseTestGenerator): + """ This class get select SQL for functions. """ + scenarios = [ + ( + 'Fetch Function select sql.', + dict( + url='/browser/function/select_sql/', + is_positive_test=True, + mocking_required=False, + is_add_argument=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function select sql fail', + dict( + url='/browser/function/select_sql/', + is_positive_test=False, + mocking_required=True, + is_add_argument=False, + mock_data={ + "function_name": 'pgadmin.utils.driver.psycopg2.' + 'connection.Connection.execute_2darray', + "return_value": "(False, 'Mocked Internal Server Error " + "while get select sql.')" + }, + expected_data={ + "status_code": 500 + } + ), + ), + ( + 'Fetch Function select sql not found', + dict( + url='/browser/function/select_sql/', + is_positive_test=False, + mocking_required=True, + is_add_argument=False, + mock_data={ + "function_name": 'pgadmin.utils.driver.psycopg2.' + 'connection.Connection.execute_2darray', + "return_value": "(True, {'rows': []})" + }, + expected_data={ + "status_code": 410 + } + ), + ), + ( + 'Fetch Function select sql with arguments', + dict( + url='/browser/function/select_sql/', + is_positive_test=True, + mocking_required=False, + is_add_argument=True, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ] + + def get_select_sql(self): + func_name = "test_function_delete_%s" % str(uuid.uuid4())[1:8] + if self.is_add_argument: + args = "IN test integer DEFAULT 1" + else: + args = None + function_info = funcs_utils.create_function( + self.server, self.db_name, self.schema_name, func_name, args=args) + + func_id = function_info[0] + + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/' + str(func_id), + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetSelectSqlTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_select_sql() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_select_sql() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_supported_functions.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_supported_functions.py new file mode 100644 index 0000000..573bf8e --- /dev/null +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_supported_functions.py @@ -0,0 +1,80 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2020, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +import json +import uuid +from unittest.mock import patch + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression.python_test_utils import test_utils as utils +from . import utils as funcs_utils +from .. import FunctionView + + +class FunctionGetSupportedFunctionsTestCase(BaseTestGenerator): + """ This class get supported functions. """ + scenarios = [ + ( + 'Fetch Function supported functions', + dict( + url='/browser/function/get_support_functions/', + is_positive_test=True, + mocking_required=False, + mock_data={}, + expected_data={ + "status_code": 200 + } + ), + ), + ( + 'Fetch Function support functions fail', + dict( + url='/browser/function/get_support_functions/', + 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 get supported function')" + }, + expected_data={ + "status_code": 500 + } + ), + ) + ] + + def get_supported_functions(self): + response = self.tester.get( + self.url + str(utils.SERVER_GROUP) + '/' + + str(self.server_id) + '/' + str(self.db_id) + + '/' + str(self.schema_id) + '/', + content_type='html/json' + ) + return response + + def runTest(self): + """ This function will get function nodes under schema. """ + super(FunctionGetSupportedFunctionsTestCase, self).runTest() + self = funcs_utils.set_up(self) + + if self.is_positive_test: + response = self.get_supported_functions() + else: + with patch(self.mock_data["function_name"], + return_value=eval(self.mock_data["return_value"])): + response = self.get_supported_functions() + + self.assertEqual(response.status_code, + self.expected_data['status_code']) + # Disconnect the database + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/utils.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/utils.py index 09ae7ee..a73572d 100644 --- a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/utils.py +++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/utils.py @@ -142,7 +142,7 @@ def create_procedure(server, db_name, schema_name, func_name, s_type, traceback.print_exc(file=sys.stderr) -def create_function(server, db_name, schema_name, func_name): +def create_function(server, db_name, schema_name, func_name, args=None): """This function add the procedure to schema""" try: connection = utils.get_db_connection(db_name, @@ -152,11 +152,15 @@ def create_function(server, db_name, schema_name, func_name): server['port'], server['sslmode']) pg_cursor = connection.cursor() + if args: + args = "{0}".format(args) + else: + args = '' query = "CREATE FUNCTION " + schema_name + "." + func_name + \ - "()" \ + "({0})" \ " RETURNS integer LANGUAGE 'sql' STABLE" \ " SECURITY DEFINER AS $$" \ - " SELECT 1; $$;" + " SELECT 1; $$;".format(args) pg_cursor.execute(query) connection.commit() # Get 'oid' from newly created function