mirror of
https://github.com/pgadmin-org/pgadmin4.git
synced 2025-02-25 18:55:31 -06:00
Improve code coverage and API test cases for Grant Wizard. Fixes #5344
This commit is contained in:
Nikhil Mohite
committed by
Akshay Joshi
Akshay Joshi
parent
79e6480513
commit
6415294782
@@ -183,7 +183,6 @@ def properties(sid, did, node_id, node_type):
|
||||
"""It fetches the properties of object types
|
||||
and render into selection page of wizard
|
||||
"""
|
||||
|
||||
# unquote encoded url parameter
|
||||
node_type = unquote(node_type)
|
||||
|
||||
@@ -197,23 +196,22 @@ def properties(sid, did, node_id, node_type):
|
||||
node_types = []
|
||||
show_sysobj = blueprint.show_system_objects().get()
|
||||
if node_type == 'database':
|
||||
|
||||
# Fetch list of schemas
|
||||
# Get sys_obj_values and get list of schemas
|
||||
ntype = 'schema'
|
||||
SQL = render_template("/".join(
|
||||
sql = render_template("/".join(
|
||||
[server_prop['template_path'], '/sql/get_schemas.sql']),
|
||||
show_sysobj=show_sysobj)
|
||||
status, res = conn.execute_dict(SQL)
|
||||
status, res = conn.execute_dict(sql)
|
||||
|
||||
if not status:
|
||||
return internal_server_error(errormsg=res)
|
||||
node_types = res['rows']
|
||||
else:
|
||||
SQL = render_template("/".join(
|
||||
sql = render_template("/".join(
|
||||
[server_prop['template_path'], '/sql/get_schemas.sql']),
|
||||
nspid=node_id, show_sysobj=False)
|
||||
status, res = conn.execute_dict(SQL)
|
||||
status, res = conn.execute_dict(sql)
|
||||
|
||||
if not status:
|
||||
return internal_server_error(errormsg=res)
|
||||
@@ -226,11 +224,11 @@ def properties(sid, did, node_id, node_type):
|
||||
|
||||
# Fetch functions against schema
|
||||
if ntype in ['schema', 'function']:
|
||||
SQL = render_template("/".join(
|
||||
sql = render_template("/".join(
|
||||
[server_prop['template_path'], '/sql/function.sql']),
|
||||
node_id=node_id, type='function')
|
||||
|
||||
status, res = conn.execute_dict(SQL)
|
||||
status, res = conn.execute_dict(sql)
|
||||
if not status:
|
||||
current_app.logger.error(res)
|
||||
failed_objects.append('function')
|
||||
@@ -243,11 +241,11 @@ def properties(sid, did, node_id, node_type):
|
||||
(server_prop['server_type'] == 'pg' and
|
||||
server_prop['version'] >= 11000)) and
|
||||
ntype in ['schema', 'procedure']):
|
||||
SQL = render_template("/".join(
|
||||
sql = render_template("/".join(
|
||||
[server_prop['template_path'], '/sql/function.sql']),
|
||||
node_id=node_id, type='procedure')
|
||||
|
||||
status, res = conn.execute_dict(SQL)
|
||||
status, res = conn.execute_dict(sql)
|
||||
|
||||
if not status:
|
||||
current_app.logger.error(res)
|
||||
@@ -257,10 +255,10 @@ def properties(sid, did, node_id, node_type):
|
||||
|
||||
# Fetch trigger functions
|
||||
if ntype in ['schema', 'trigger_function']:
|
||||
SQL = render_template("/".join(
|
||||
sql = render_template("/".join(
|
||||
[server_prop['template_path'], '/sql/function.sql']),
|
||||
node_id=node_id, type='trigger_function')
|
||||
status, res = conn.execute_dict(SQL)
|
||||
status, res = conn.execute_dict(sql)
|
||||
|
||||
if not status:
|
||||
current_app.logger.error(res)
|
||||
@@ -270,11 +268,11 @@ def properties(sid, did, node_id, node_type):
|
||||
|
||||
# Fetch Sequences against schema
|
||||
if ntype in ['schema', 'sequence']:
|
||||
SQL = render_template("/".join(
|
||||
sql = render_template("/".join(
|
||||
[server_prop['template_path'], '/sql/sequence.sql']),
|
||||
node_id=node_id)
|
||||
|
||||
status, res = conn.execute_dict(SQL)
|
||||
status, res = conn.execute_dict(sql)
|
||||
if not status:
|
||||
current_app.logger.error(res)
|
||||
failed_objects.append('sequence')
|
||||
@@ -283,11 +281,11 @@ def properties(sid, did, node_id, node_type):
|
||||
|
||||
# Fetch Tables against schema
|
||||
if ntype in ['schema', 'table']:
|
||||
SQL = render_template("/".join(
|
||||
sql = render_template("/".join(
|
||||
[server_prop['template_path'], '/sql/table.sql']),
|
||||
node_id=node_id)
|
||||
|
||||
status, res = conn.execute_dict(SQL)
|
||||
status, res = conn.execute_dict(sql)
|
||||
if not status:
|
||||
current_app.logger.error(res)
|
||||
failed_objects.append('table')
|
||||
@@ -296,11 +294,11 @@ def properties(sid, did, node_id, node_type):
|
||||
|
||||
# Fetch Views against schema
|
||||
if ntype in ['schema', 'view']:
|
||||
SQL = render_template("/".join(
|
||||
sql = render_template("/".join(
|
||||
[server_prop['template_path'], '/sql/view.sql']),
|
||||
node_id=node_id, node_type='v')
|
||||
|
||||
status, res = conn.execute_dict(SQL)
|
||||
status, res = conn.execute_dict(sql)
|
||||
if not status:
|
||||
current_app.logger.error(res)
|
||||
failed_objects.append('view')
|
||||
@@ -309,11 +307,11 @@ def properties(sid, did, node_id, node_type):
|
||||
|
||||
# Fetch Materialzed Views against schema
|
||||
if ntype in ['schema', 'mview']:
|
||||
SQL = render_template("/".join(
|
||||
sql = render_template("/".join(
|
||||
[server_prop['template_path'], '/sql/view.sql']),
|
||||
node_id=node_id, node_type='m')
|
||||
|
||||
status, res = conn.execute_dict(SQL)
|
||||
status, res = conn.execute_dict(sql)
|
||||
if not status:
|
||||
current_app.logger.error(res)
|
||||
failed_objects.append('materialized view')
|
||||
@@ -343,7 +341,6 @@ def msql(sid, did):
|
||||
"""
|
||||
This function will return modified SQL
|
||||
"""
|
||||
|
||||
server_prop = server_info
|
||||
data = request.form if request.form else json.loads(request.data.decode())
|
||||
# Form db connection
|
||||
@@ -360,7 +357,6 @@ def msql(sid, did):
|
||||
current_app.logger.exception(e)
|
||||
|
||||
try:
|
||||
|
||||
# Parse privileges
|
||||
data['priv'] = {}
|
||||
if 'acl' in data:
|
||||
@@ -381,29 +377,29 @@ def msql(sid, did):
|
||||
sql_data = ''
|
||||
data_func = {'objects': data['objects'],
|
||||
'priv': data['priv']['function']}
|
||||
SQL = render_template(
|
||||
sql = render_template(
|
||||
"/".join([server_prop['template_path'],
|
||||
'/sql/grant_function.sql']),
|
||||
data=data_func, conn=conn)
|
||||
if SQL and SQL.strip('\n') != '':
|
||||
sql_data += SQL
|
||||
if sql and sql.strip('\n') != '':
|
||||
sql_data += sql
|
||||
|
||||
data_seq = {'objects': data['objects'],
|
||||
'priv': data['priv']['sequence']}
|
||||
SQL = render_template(
|
||||
sql = render_template(
|
||||
"/".join([server_prop['template_path'],
|
||||
'/sql/grant_sequence.sql']),
|
||||
data=data_seq, conn=conn)
|
||||
if SQL and SQL.strip('\n') != '':
|
||||
sql_data += SQL
|
||||
if sql and sql.strip('\n') != '':
|
||||
sql_data += sql
|
||||
|
||||
data_table = {'objects': data['objects'],
|
||||
'priv': data['priv']['table']}
|
||||
SQL = render_template(
|
||||
sql = render_template(
|
||||
"/".join([server_prop['template_path'], '/sql/grant_table.sql']),
|
||||
data=data_table, conn=conn)
|
||||
if SQL and SQL.strip('\n') != '':
|
||||
sql_data += SQL
|
||||
if sql and sql.strip('\n') != '':
|
||||
sql_data += sql
|
||||
|
||||
res = {'data': sql_data}
|
||||
|
||||
@@ -469,29 +465,29 @@ def save(sid, did):
|
||||
sql_data = ''
|
||||
data_func = {'objects': data['objects'],
|
||||
'priv': data['priv']['function']}
|
||||
SQL = render_template(
|
||||
sql = render_template(
|
||||
"/".join([server_prop['template_path'],
|
||||
'/sql/grant_function.sql']),
|
||||
data=data_func, conn=conn)
|
||||
if SQL and SQL.strip('\n') != '':
|
||||
sql_data += SQL
|
||||
if sql and sql.strip('\n') != '':
|
||||
sql_data += sql
|
||||
|
||||
data_seq = {'objects': data['objects'],
|
||||
'priv': data['priv']['sequence']}
|
||||
SQL = render_template(
|
||||
sql = render_template(
|
||||
"/".join([server_prop['template_path'],
|
||||
'/sql/grant_sequence.sql']),
|
||||
data=data_seq, conn=conn)
|
||||
if SQL and SQL.strip('\n') != '':
|
||||
sql_data += SQL
|
||||
if sql and sql.strip('\n') != '':
|
||||
sql_data += sql
|
||||
|
||||
data_table = {'objects': data['objects'],
|
||||
'priv': data['priv']['table']}
|
||||
SQL = render_template(
|
||||
sql = render_template(
|
||||
"/".join([server_prop['template_path'], '/sql/grant_table.sql']),
|
||||
data=data_table, conn=conn)
|
||||
if SQL and SQL.strip('\n') != '':
|
||||
sql_data += SQL
|
||||
if sql and sql.strip('\n') != '':
|
||||
sql_data += sql
|
||||
|
||||
status, res = conn.execute_dict(sql_data)
|
||||
if not status:
|
||||
|
||||
16
web/pgadmin/tools/grant_wizard/tests/__init__.py
Normal file
16
web/pgadmin/tools/grant_wizard/tests/__init__.py
Normal file
@@ -0,0 +1,16 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
|
||||
|
||||
class GrantWizardTestGenerator(BaseTestGenerator):
|
||||
|
||||
def runTest(self):
|
||||
return []
|
||||
234
web/pgadmin/tools/grant_wizard/tests/grant_wizard_test_data.json
Normal file
234
web/pgadmin/tools/grant_wizard/tests/grant_wizard_test_data.json
Normal file
@@ -0,0 +1,234 @@
|
||||
{
|
||||
"grant_wizard_save_permissions": [
|
||||
{
|
||||
"name": "Grant permissions to user for table.",
|
||||
"url": "/grant_wizard/",
|
||||
"is_positive_test": true,
|
||||
"mocking_required": false,
|
||||
"test_data": {
|
||||
"objects": [
|
||||
{
|
||||
"selected": true,
|
||||
"name": "test_tablespace",
|
||||
"name_with_args": "test_tablespace",
|
||||
"nspname": "public",
|
||||
"object_type": "Table"
|
||||
}
|
||||
],
|
||||
"acl": [
|
||||
{
|
||||
"privileges": [
|
||||
{
|
||||
"privilege_type": "a",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "r",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "w",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "d",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "D",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "x",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
"mock_data": {},
|
||||
"expected_data": {
|
||||
"status_code": 200
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "Error while: Grant permissions to user for table.",
|
||||
"url": "/grant_wizard/",
|
||||
"is_positive_test": false,
|
||||
"mocking_required": true,
|
||||
"test_data": {
|
||||
"objects": [
|
||||
{
|
||||
"selected": true,
|
||||
"name": "test_tablespace",
|
||||
"name_with_args": "test_tablespace",
|
||||
"nspname": "public",
|
||||
"object_type": "Table"
|
||||
}
|
||||
],
|
||||
"acl": [
|
||||
{
|
||||
"privileges": [
|
||||
{
|
||||
"privilege_type": "a",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "r",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "w",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "d",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "D",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
},
|
||||
{
|
||||
"privilege_type": "x",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
"mock_data": {
|
||||
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
|
||||
"return_value": "(False, 'Mocked Internal Server Error while grant permissions.')"
|
||||
},
|
||||
"expected_data": {
|
||||
"status_code": 500
|
||||
}
|
||||
}
|
||||
],
|
||||
"grant_wizard_get_properties": [
|
||||
{
|
||||
"name": "Get properties for grant permissions to user.",
|
||||
"url": "/grant_wizard/",
|
||||
"id_database": true,
|
||||
"is_positive_test": true,
|
||||
"mocking_required": false,
|
||||
"mock_data": {},
|
||||
"expected_data": {
|
||||
"status_code": 200
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "Error while: Get properties for grant permissions to user.",
|
||||
"url": "/grant_wizard/",
|
||||
"id_database": true,
|
||||
"is_positive_test": false,
|
||||
"mocking_required": true,
|
||||
"mock_data": {
|
||||
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
|
||||
"return_value": "(False, 'Mocked Internal Server Error while getting list of object for grant wizard.')"
|
||||
},
|
||||
"expected_data": {
|
||||
"status_code": 500
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "Get properties for grant permissions to user.",
|
||||
"url": "/grant_wizard/",
|
||||
"id_database": false,
|
||||
"node_type": "table",
|
||||
"is_positive_test": true,
|
||||
"mocking_required": false,
|
||||
"mock_data": {
|
||||
},
|
||||
"expected_data": {
|
||||
"status_code": 200
|
||||
}
|
||||
},{
|
||||
"name": "Get properties for grant permissions to user.",
|
||||
"url": "/grant_wizard/",
|
||||
"node_type": "table",
|
||||
"id_database": false,
|
||||
"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 getting list of object for grant wizard.')"
|
||||
},
|
||||
"expected_data": {
|
||||
"status_code": 500
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "Get properties for grant permissions to node type.",
|
||||
"url": "/grant_wizard/",
|
||||
"node_type": "schema",
|
||||
"id_database": false,
|
||||
"is_positive_test": true,
|
||||
"mocking_required": false,
|
||||
"mock_data": {
|
||||
},
|
||||
"expected_data": {
|
||||
"status_code": 200
|
||||
}
|
||||
}
|
||||
],
|
||||
"grant_wizard_get_sql": [
|
||||
{
|
||||
"name": "Grant wizard SQL statement.",
|
||||
"url": "/grant_wizard/sql/",
|
||||
"is_positive_test": true,
|
||||
"mocking_required": false,
|
||||
"test_data": {
|
||||
"objects":[
|
||||
{
|
||||
"selected": true,
|
||||
"name": "auth_group",
|
||||
"name_with_args": "auth_group",
|
||||
"nspname": "public",
|
||||
"object_type": "Table"
|
||||
}
|
||||
],
|
||||
"acl":[
|
||||
{
|
||||
"privileges":[
|
||||
{
|
||||
"privilege_type": "a",
|
||||
"privilege": true,
|
||||
"with_grant": false
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
"mock_data": {},
|
||||
"expected_data": {
|
||||
"status_code": 200
|
||||
}
|
||||
}
|
||||
],
|
||||
"grant_wizard_get_acl_list": [
|
||||
{
|
||||
"name": "Grant wizard acl list",
|
||||
"url": "/grant_wizard/acl/",
|
||||
"is_positive_test": true,
|
||||
"mocking_required": false,
|
||||
"mock_data": {},
|
||||
"expected_data": {
|
||||
"status_code": 200
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
##########################################################################
|
||||
#
|
||||
# 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 import parent_node_dict
|
||||
from regression.python_test_utils import test_utils as utils
|
||||
from . import utils as grant_wizard_utils
|
||||
|
||||
|
||||
class GrantWizardSaveGetACLTestCase(BaseTestGenerator):
|
||||
"""
|
||||
This will Get acl list for grant permissions.
|
||||
"""
|
||||
|
||||
scenarios = utils.generate_scenarios(
|
||||
'grant_wizard_get_acl_list',
|
||||
grant_wizard_utils.test_cases
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
self.database_info = parent_node_dict["database"][-1]
|
||||
self.db_name = self.database_info["db_name"]
|
||||
self.did = self.database_info["db_id"]
|
||||
self.sid = parent_node_dict["server"][-1]["server_id"]
|
||||
|
||||
db_con = database_utils.connect_database(self, utils.SERVER_GROUP,
|
||||
self.sid, self.did)
|
||||
if not db_con['data']["connected"]:
|
||||
raise Exception("Could not connect to database to add a table.")
|
||||
|
||||
def grant_permissions_acl(self):
|
||||
response = self.tester.get(
|
||||
self.url + str(self.sid) + '/' + str(self.did) + '/',
|
||||
content_type='html/json'
|
||||
)
|
||||
return response
|
||||
|
||||
def runTest(self):
|
||||
""" This function will alc list for grant permission """
|
||||
response = self.grant_permissions_acl()
|
||||
actual_response_code = response.status_code
|
||||
expected_response_code = self.expected_data['status_code']
|
||||
|
||||
self.assertEquals(actual_response_code, expected_response_code)
|
||||
|
||||
def tearDown(self):
|
||||
"""This function disconnect database."""
|
||||
database_utils.disconnect_database(self, self.sid,
|
||||
self.did)
|
||||
@@ -0,0 +1,93 @@
|
||||
##########################################################################
|
||||
#
|
||||
# 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 import parent_node_dict
|
||||
from regression.python_test_utils import test_utils as utils
|
||||
from . import utils as grant_wizard_utils
|
||||
|
||||
|
||||
class GrantWizardSaveGetPropertiesTestCase(BaseTestGenerator):
|
||||
"""
|
||||
This will Get the properties as per selected node of database.
|
||||
For test we are testing for table node only.
|
||||
"""
|
||||
|
||||
scenarios = utils.generate_scenarios(
|
||||
'grant_wizard_get_properties',
|
||||
grant_wizard_utils.test_cases
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
self.database_info = parent_node_dict["database"][-1]
|
||||
self.db_name = self.database_info["db_name"]
|
||||
self.did = self.database_info["db_id"]
|
||||
self.sid = parent_node_dict["server"][-1]["server_id"]
|
||||
|
||||
db_con = database_utils.connect_database(self, utils.SERVER_GROUP,
|
||||
self.sid, self.did)
|
||||
if not db_con['data']["connected"]:
|
||||
raise Exception("Could not connect to database to add a table.")
|
||||
|
||||
def grant_permissions_database(self):
|
||||
response = self.tester.get(
|
||||
self.url + str(self.sid) + '/' + str(self.did) + '/' +
|
||||
str(self.did) + '/database/',
|
||||
content_type='html/json'
|
||||
)
|
||||
return response
|
||||
|
||||
def grant_permissions_non_database(self):
|
||||
response = self.tester.get(
|
||||
self.url + str(self.sid) + '/' + str(self.did) + '/' +
|
||||
str(self.did) + '/' + self.node_type + '/',
|
||||
content_type='html/json'
|
||||
)
|
||||
return response
|
||||
|
||||
def runTest(self):
|
||||
""" This function will grant permission for user under database
|
||||
object. """
|
||||
if self.id_database:
|
||||
if self.is_positive_test:
|
||||
response = self.grant_permissions_database()
|
||||
actual_response_code = response.status_code
|
||||
expected_response_code = self.expected_data['status_code']
|
||||
else:
|
||||
with patch(self.mock_data["function_name"],
|
||||
return_value=eval(self.mock_data["return_value"])):
|
||||
response = self.grant_permissions_database()
|
||||
actual_response_code = response.status_code
|
||||
expected_response_code = self.expected_data['status_code']
|
||||
else:
|
||||
if self.is_positive_test:
|
||||
response = self.grant_permissions_non_database()
|
||||
actual_response_code = response.status_code
|
||||
expected_response_code = self.expected_data['status_code']
|
||||
else:
|
||||
with patch(self.mock_data["function_name"],
|
||||
return_value=eval(self.mock_data["return_value"])):
|
||||
response = self.grant_permissions_non_database()
|
||||
actual_response_code = response.status_code
|
||||
expected_response_code = self.expected_data['status_code']
|
||||
|
||||
self.assertEquals(actual_response_code, expected_response_code)
|
||||
|
||||
def tearDown(self):
|
||||
"""This function disconnect database."""
|
||||
database_utils.disconnect_database(self, self.sid,
|
||||
self.did)
|
||||
@@ -0,0 +1,97 @@
|
||||
##########################################################################
|
||||
#
|
||||
# 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.browser.server_groups.servers.databases.schemas.tests import \
|
||||
utils as schema_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.schemas.tables.tests \
|
||||
import utils as tables_utils
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from regression import parent_node_dict
|
||||
from regression.python_test_utils import test_utils as utils
|
||||
from . import utils as grant_wizard_utils
|
||||
|
||||
|
||||
class GrantWizardSaveGetSQLTestCase(BaseTestGenerator):
|
||||
"""
|
||||
This will Get SQL for grant permissions.
|
||||
"""
|
||||
|
||||
scenarios = utils.generate_scenarios(
|
||||
'grant_wizard_get_sql',
|
||||
grant_wizard_utils.test_cases
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
self.database_info = parent_node_dict["database"][-1]
|
||||
self.db_name = self.database_info["db_name"]
|
||||
self.did = self.database_info["db_id"]
|
||||
self.sid = parent_node_dict["server"][-1]["server_id"]
|
||||
|
||||
db_con = database_utils.connect_database(self, utils.SERVER_GROUP,
|
||||
self.sid, self.did)
|
||||
if not db_con['data']["connected"]:
|
||||
raise Exception("Could not connect to database to add a table.")
|
||||
|
||||
self.schema_id = parent_node_dict['schema'][-1]["schema_id"]
|
||||
self.schema_name = parent_node_dict['schema'][-1]["schema_name"]
|
||||
schema_response = schema_utils.verify_schemas(self.server,
|
||||
self.db_name,
|
||||
self.schema_name)
|
||||
if not schema_response:
|
||||
raise Exception("Could not find the schema to add a table.")
|
||||
self.table_name = "table_for_wizard%s" % (str(uuid.uuid4())[1:8])
|
||||
self.table_id = tables_utils.create_table(self.server, self.db_name,
|
||||
self.schema_name,
|
||||
self.table_name)
|
||||
self.test_data['objects'][-1]['name'] = self.table_name
|
||||
self.test_data['objects'][-1]['name_with_args'] = self.table_name
|
||||
self.test_data['objects'][-1]['nspname'] = self.schema_name
|
||||
|
||||
if self.server_information['type'] == 'ppas':
|
||||
self.test_data['acl'][-1]['grantee'] = 'enterprisedb'
|
||||
self.test_data['acl'][-1]['grantor'] = 'enterprisedb'
|
||||
else:
|
||||
self.test_data['acl'][-1]['grantee'] = 'postgres'
|
||||
self.test_data['acl'][-1]['grantor'] = 'postgres'
|
||||
|
||||
def grant_permissions_sql(self):
|
||||
response = self.tester.post(
|
||||
self.url + str(self.sid) + '/' + str(self.did) + '/',
|
||||
data=json.dumps(self.test_data),
|
||||
content_type='html/json'
|
||||
)
|
||||
return response
|
||||
|
||||
def runTest(self):
|
||||
""" This function will grant permission for user under database
|
||||
object. """
|
||||
if self.is_positive_test:
|
||||
response = self.grant_permissions_sql()
|
||||
actual_response_code = response.status_code
|
||||
expected_response_code = self.expected_data['status_code']
|
||||
else:
|
||||
with patch(self.mock_data["function_name"],
|
||||
return_value=eval(self.mock_data["return_value"])):
|
||||
response = self.grant_permissions_sql()
|
||||
actual_response_code = response.status_code
|
||||
expected_response_code = self.expected_data['status_code']
|
||||
|
||||
self.assertEquals(actual_response_code, expected_response_code)
|
||||
|
||||
def tearDown(self):
|
||||
"""This function disconnect database."""
|
||||
database_utils.disconnect_database(self, self.sid,
|
||||
self.did)
|
||||
@@ -0,0 +1,94 @@
|
||||
##########################################################################
|
||||
#
|
||||
# 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.browser.server_groups.servers.databases.schemas.tests import \
|
||||
utils as schema_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.schemas.tables.tests \
|
||||
import utils as tables_utils
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from regression import parent_node_dict
|
||||
from regression.python_test_utils import test_utils as utils
|
||||
from . import utils as grant_wizard_utils
|
||||
|
||||
|
||||
class GrantWizardSavePermissionsTestCase(BaseTestGenerator):
|
||||
""" This will grant permission to user for specified database object. """
|
||||
|
||||
scenarios = utils.generate_scenarios(
|
||||
'grant_wizard_save_permissions',
|
||||
grant_wizard_utils.test_cases
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
self.database_info = parent_node_dict["database"][-1]
|
||||
self.db_name = self.database_info["db_name"]
|
||||
self.did = self.database_info["db_id"]
|
||||
self.sid = parent_node_dict["server"][-1]["server_id"]
|
||||
|
||||
db_con = database_utils.connect_database(self, utils.SERVER_GROUP,
|
||||
self.sid, self.did)
|
||||
if not db_con['data']["connected"]:
|
||||
raise Exception("Could not connect to database to add a table.")
|
||||
|
||||
self.schema_id = parent_node_dict['schema'][-1]["schema_id"]
|
||||
self.schema_name = parent_node_dict['schema'][-1]["schema_name"]
|
||||
schema_response = schema_utils.verify_schemas(self.server,
|
||||
self.db_name,
|
||||
self.schema_name)
|
||||
if not schema_response:
|
||||
raise Exception("Could not find the schema to add a table.")
|
||||
self.table_name = "table_for_wizard%s" % (str(uuid.uuid4())[1:8])
|
||||
self.table_id = tables_utils.create_table(self.server, self.db_name,
|
||||
self.schema_name,
|
||||
self.table_name)
|
||||
self.test_data['objects'][-1]['name'] = self.table_name
|
||||
self.test_data['objects'][-1]['name_with_args'] = self.table_name
|
||||
self.test_data['objects'][-1]['nspname'] = self.schema_name
|
||||
|
||||
if self.server_information['type'] == 'ppas':
|
||||
self.test_data['acl'][-1]['grantee'] = 'enterprisedb'
|
||||
self.test_data['acl'][-1]['grantor'] = 'enterprisedb'
|
||||
else:
|
||||
self.test_data['acl'][-1]['grantee'] = 'postgres'
|
||||
self.test_data['acl'][-1]['grantor'] = 'postgres'
|
||||
|
||||
def grant_permissions(self):
|
||||
response = self.tester.post(
|
||||
self.url + str(self.sid) + '/' + str(self.did) + '/',
|
||||
data=json.dumps(self.test_data),
|
||||
content_type='html/json',
|
||||
follow_redirects=True
|
||||
)
|
||||
return response
|
||||
|
||||
def runTest(self):
|
||||
""" This function will grant permission for user under database
|
||||
object. """
|
||||
|
||||
if self.is_positive_test:
|
||||
response = self.grant_permissions()
|
||||
actual_response_code = response.status_code
|
||||
expected_response_code = self.expected_data['status_code']
|
||||
else:
|
||||
with patch(self.mock_data["function_name"],
|
||||
return_value=eval(self.mock_data["return_value"])):
|
||||
response = self.grant_permissions()
|
||||
actual_response_code = response.status_code
|
||||
expected_response_code = self.expected_data['status_code']
|
||||
self.assertEquals(actual_response_code, expected_response_code)
|
||||
|
||||
def tearDown(self):
|
||||
"""This function disconnect database."""
|
||||
database_utils.disconnect_database(self, self.sid,
|
||||
self.did)
|
||||
15
web/pgadmin/tools/grant_wizard/tests/utils.py
Normal file
15
web/pgadmin/tools/grant_wizard/tests/utils.py
Normal file
@@ -0,0 +1,15 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
import os
|
||||
import json
|
||||
|
||||
file_name = os.path.basename(__file__)
|
||||
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
|
||||
with open(CURRENT_PATH + "/grant_wizard_test_data.json") as data_file:
|
||||
test_cases = json.load(data_file)
|
||||
Reference in New Issue
Block a user