IntenseWebs/pgadmin4
3
0
Fork 0
mirror of https://github.com/pgadmin-org/pgadmin4.git synced 2025-02-25 18:55:31 -06:00
Code Issues Packages Projects Releases Wiki Activity

Improve code coverage and API test cases for Functions. Fixes #5330

Browse Source
This commit is contained in:
Nikhil Mohite 2020-09-22 14:43:46 +05:30 committed by Akshay Joshi
parent 458036ad6b
commit 43d97b0e6f
19 changed files with 1659 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/en_US/release_notes_4_27.rst
View File

@ -13,6 +13,7 @@ New features
Housekeeping
************
| `Issue #5330 <https://redmine.postgresql.org/issues/5330>`_ - Improve code coverage and API test cases for Functions.
Bug fixes
*********

29
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/__init__.py
View File

@ -1219,15 +1219,14 @@ class FunctionView(PGChildNodeView, DataTypeReader, SchemaDiffObjectCompare):
status, sql = self._get_sql(gid=gid, sid=sid, did=did, scid=scid,
data=self.request, fnid=fnid)
if status:
sql = re.sub('\n{2,}', '\n\n', sql)
return make_json_response(
data=sql,
status=200
)
else:
sql = re.sub('\n{2,}', '\n\n', sql)
return sql
if not status:
return internal_server_error(errormsg=sql)
sql = re.sub('\n{2,}', '\n\n', sql)
return make_json_response(
data=sql,
status=200
)
@staticmethod
def _update_arguments_for_get_sql(data, old_data):
@ -1359,7 +1358,7 @@ class FunctionView(PGChildNodeView, DataTypeReader, SchemaDiffObjectCompare):
if not isinstance(old_data, dict):
return False, gettext(
"Could not find the function in the database."
)
), ''
# Get Schema Name
old_data['pronamespace'] = self._get_schema(
@ -1423,7 +1422,7 @@ class FunctionView(PGChildNodeView, DataTypeReader, SchemaDiffObjectCompare):
"/".join([self.sql_template_path, self._UPDATE_SQL]),
data=data, o_data=old_data
)
return sql
return True, '', sql
def _get_sql(self, **kwargs):
"""
@ -1464,8 +1463,12 @@ class FunctionView(PGChildNodeView, DataTypeReader, SchemaDiffObjectCompare):
'is_sql': is_sql,
'is_schema_diff': is_schema_diff,
}
sql = self._get_sql_for_edit_mode(data, parallel_dict,
all_ids_dict, vol_dict)
status, errmsg, sql = self._get_sql_for_edit_mode(
data, parallel_dict, all_ids_dict, vol_dict)
if not status:
return False, errmsg
else:
# Parse Privileges
self._parse_privilege_data(data)

93
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_add.py
View File

@ -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)
self.assertEqual(response.status_code, 200)
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,
self.expected_data['status_code'])
# Disconnect the database
database_utils.disconnect_database(self, self.server_id, self.db_id)

103
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_delete.py
View File

@ -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)

73
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get.py
View File

@ -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)

80
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_languages.py Normal file
View File

@ -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)

224
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_msql.py Normal file
View File

@ -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": "",
"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": "",
"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": 500
}
),
),
(
'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": "",
"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": "",
"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": "",
"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)

121
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_nodes.py Normal file
View File

@ -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)

85
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_sql.py Normal file
View File

@ -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)

82
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_types.py Normal file
View File

@ -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)

80
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_get_variable_options.py Normal file
View File

@ -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)

159
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_function_put.py
View File

@ -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)

85
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_dependencies.py Normal file
View File

@ -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)

85
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_dependents.py Normal file
View File

@ -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)

120
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_function_statistics.py Normal file
View File

@ -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)

80
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_list.py Normal file
View File

@ -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)

122
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_select_sql.py Normal file
View File

@ -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)

85
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/test_get_supported_functions.py Normal file
View File

@ -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 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. """
if self.server_information['server_version'] < 120000:
message = "Supported functions are not supported by PG/EPAS " \
"< 120000."
self.skipTest(message)
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)

10
web/pgadmin/browser/server_groups/servers/databases/schemas/functions/tests/utils.py
View File

@ -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