public inbox for [email protected]
help / color / mirror / Atom feedFrom: Nikhil Mohite <[email protected]>
To: Akshay Joshi <[email protected]>
Cc: pgadmin-hackers <[email protected]>
Subject: Re: [pgAdmin][RM5330]: [Code Coverage] Improve API test cases for Functions
Date: Tue, 22 Sep 2020 12:44:31 +0530
Message-ID: <CAOBg0AN==k1y7fj=LDOzBj2pbbBg1vwM3qWtf9kNxgKBKjhHmg@mail.gmail.com> (raw)
In-Reply-To: <CANxoLDevEYsKrtRPTrYhgqAdPi69sFyQaFbW9e8ZyF-cBOs-4A@mail.gmail.com>
References: <CAOBg0AMPMqnuORm9Qj3un7MKBf9CM4EYZFMF1q_yrHR4pVK0dA@mail.gmail.com>
<CANxoLDevEYsKrtRPTrYhgqAdPi69sFyQaFbW9e8ZyF-cBOs-4A@mail.gmail.com>
Hi Akshay/Team,
I checked the code and found an issue in the "msql" function, due to which
test cases were failing.
I have fixed the issue, PFA updated patch for the same.
Sorry for the inconvenience.
Regards,
Nikhil Mohite.
On Tue, Sep 22, 2020 at 11:41 AM Akshay Joshi <[email protected]>
wrote:
> Hi Nikhil
>
> Test cases are failing for all the database server with the following
> error:
> return True, sql.strip('\n')
> AttributeError: 'tuple' object has no attribute 'strip'
>
> Please check and resend the patch.
>
> On Tue, Sep 22, 2020 at 10:56 AM Nikhil Mohite <
> [email protected]> wrote:
>
>> Hi Team,
>>
>> I have added few more test cases for the "Functions" module as per
>> RM-5330 <https://redmine.postgresql.org/issues/5330;
>> PFA patch.
>>
>> --
>> *Thanks & Regards,*
>> *Nikhil Mohite*
>> *Software Engineer.*
>> *EDB Postgres* <https://www.enterprisedb.com/;
>> *Mob.No: +91-7798364578.*
>>
>
>
> --
> *Thanks & Regards*
> *Akshay Joshi*
> *pgAdmin Hacker | Sr. Software Architect*
> *EDB Postgres <http://edbpostgres.com>*
>
> *Mobile: +91 976-788-8246*
>
Attachments:
[application/octet-stream] RM_5330_v2.patch (74.0K, 3-RM_5330_v2.patch)
download | inline diff:
diff --git a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/__init__.py b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/__init__.py
index 1edc46f..5f84b28 100644
--- a/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/__init__.py
+++ b/web/pgadmin/browser/server_groups/servers/databases/schemas/functions/__init__.py
@@ -1219,15 +1219,15 @@ 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 +1359,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 +1423,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 +1464,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)
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..e717a18
--- /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": "",
+ "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)
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
view thread (15+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected]
Subject: Re: [pgAdmin][RM5330]: [Code Coverage] Improve API test cases for Functions
In-Reply-To: <CAOBg0AN==k1y7fj=LDOzBj2pbbBg1vwM3qWtf9kNxgKBKjhHmg@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox