IntenseWebs/pgadmin4
3
0 Fork 0
mirror of https://github.com/pgadmin-org/pgadmin4.git synced 2026-10-13 16:41:59 -05:00
Code Issues Packages Projects Releases Wiki Activity

Improve code coverage and API test cases for Foreign Servers and User Mappings. Fixes #5324

Browse Source
This commit is contained in:
Satish V authored and Akshay Joshi committed 2020-07-24 14:14:13 +05:30
1 parent 937877e841
commit 4cf8c2f300
23 files changed
+1983 -182
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/en_US/release_notes_4_25.rst
+1
View File
@@ -13,6 +13,7 @@ New features
Housekeeping
************
| `Issue #5324 <https://redmine.postgresql.org/issues/5324>`_ - Improve code coverage and API test cases for Foreign Servers and User Mappings.
Bug fixes
*********
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/foreign_servers_test_data.json
+344
View File
@@ -0,0 +1,344 @@
{
"foreign_server_create": [
{
"name": "Create foreign server: With valid data.",
"url": "/browser/foreign_server/obj/",
"is_positive_test": true,
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while creating foreign server: With invalid data.",
"url": "/browser/foreign_server/obj/",
"is_positive_test": false,
"error_fdw_id": true,
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error while creating foreign server: With insufficient data.",
"url": "/browser/foreign_server/obj/",
"is_positive_test": false,
"missing_parameter": true,
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error while creating foreign server: internal server error.",
"url": "/browser/foreign_server/obj/",
"is_positive_test": false,
"internal_server_error": true,
"test_data": {},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while creating foreign server: error in db.",
"url": "/browser/foreign_server/obj/",
"is_positive_test": false,
"error_in_db": true,
"test_data": {},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while creating a foreign server')"
},
"expected_data": {
"status_code": 500
}
}
],
"foreign_server_delete": [
{
"name": "Delete foreign server",
"url": "/browser/foreign_server/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while deleting a foreign server - Internal server error",
"url": "/browser/foreign_server/obj/",
"internal_server_error": true,
"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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while deleting a foreign server - wrong foreign server id",
"url": "/browser/foreign_server/obj/",
"wrong_foreign_server_id": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
},
"expected_data": {
"status_code": 410
}
}
],
"foreign_server_multiple_delete": [
{
"name": "Delete multiple foreign server",
"url": "/browser/foreign_server/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"foreign_server_dependent_dependency": [
{
"name": "Get foreign server dependents",
"url": "/browser/foreign_server/dependent/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get foreign server dependencies",
"url": "/browser/foreign_server/dependency/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"foreign_server_get_nodes": [
{
"name": "Get foreign server nodes",
"url": "/browser/foreign_server/nodes/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error fetching foreign server nodes",
"url": "/browser/foreign_server/nodes/",
"is_positive_test": false,
"internal_server_error": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching foreign server nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get foreign server node",
"url": "/browser/foreign_server/nodes/",
"is_positive_test": true,
"node": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error fetching foreign server node",
"url": "/browser/foreign_server/nodes/",
"is_positive_test": false,
"node": true,
"internal_server_error": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching foreign server nodes')"
},
"expected_data": {
"status_code": 500
}
}
],
"foreign_server_update": [
{
"name": "Update foreign server",
"url": "/browser/foreign_server/obj/",
"is_positive_test": true,
"mocking_required": false,
"test_data": {
"description": "This is foreign server update comment",
"id": "PLACE_HOLDER"
},
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while updating a foreign server - Internal server error",
"url": "/browser/foreign_server/obj/",
"error_in_db": true,
"test_data": {
"description": "This is foreign server update comment",
"id": "PLACE_HOLDER"
},
"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 fetching a foreign server')]"
},
"expected_data": {
"status_code": 500
}
}
],
"foreign_server_get": [
{
"name": "Get foreign server properties",
"url": "/browser/foreign_server/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a foreign server properties",
"url": "/browser/foreign_server/obj/",
"error_fetching_fsrv": 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 fetching a foreign server')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while fetching foreign server with wrong ID",
"url": "/browser/foreign_server/obj/",
"wrong_fsrv_id": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get foreign server list",
"url": "/browser/foreign_server/obj/",
"is_positive_test": true,
"foreign_server_list": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a foreign server list",
"url": "/browser/foreign_server/obj/",
"error_fetching_fsrv": true,
"is_positive_test": false,
"foreign_server_list": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching a foreign server')"
},
"expected_data": {
"status_code": 500
}
}
],
"foreign_server_get_sql": [
{
"name": "Get Foreign server SQL",
"url": "/browser/foreign_server/sql/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error - Get Foreign server SQL - Internal server error",
"url": "/browser/foreign_server/sql/",
"is_positive_test": false,
"internal_server_error": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching a foreign server')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error - Get Foreign server SQL using wrong fsrv id",
"url": "/browser/foreign_server/sql/",
"wrong_foreign_server_id": true,
"is_positive_test": false,
"mocking_required": false,
"mock_data": {
},
"expected_data": {
"status_code": 410
}
}
],
"foreign_server_get_msql": [
{
"name": "Get foreign server msql",
"is_positive_test": true,
"url": "/browser/foreign_server/msql/",
"test_data": {
"name": "modifying name",
"comment": "Comments to test update"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
]
}
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_add.py
+45 -43
View File
@@ -21,6 +21,8 @@ from pgadmin.browser.server_groups.servers.databases.tests import \
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 fsrv_utils
from unittest.mock import patch
class ForeignServerAddTestCase(BaseTestGenerator):
@@ -28,10 +30,8 @@ class ForeignServerAddTestCase(BaseTestGenerator):
This class will add foreign server under database node.
"""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign server node.
('Check FSRV Node', dict(url='/browser/foreign_server/obj/'))
]
scenarios = utils.generate_scenarios('foreign_server_create',
fsrv_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -48,6 +48,19 @@ class ForeignServerAddTestCase(BaseTestGenerator):
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def create_foreign_server(self):
"""
This function create a foreign server and returns the created
foreign server response
:return: created foreign server response
"""
return self.tester.post(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/',
data=json.dumps(self.data),
content_type='html/json')
def runTest(self):
"""This function will fetch foreign data wrapper present under test
database."""
@@ -62,45 +75,34 @@ class ForeignServerAddTestCase(BaseTestGenerator):
if not fdw_response:
raise Exception("Could not find FDW.")
db_user = self.server["username"]
data = {
"fsrvacl": [
{
"grantee": db_user,
"grantor": db_user,
"privileges":
[
{
"privilege_type": "U",
"privilege": "true",
"with_grant": "false"
}
]
}
],
"fsrvoptions": [
{
"fsrvoption": "host",
"fsrvvalue": self.server['host']
},
{
"fsrvoption": "port",
"fsrvvalue": str(self.server['port'])
},
{
"fsrvoption": "dbname",
"fsrvvalue": self.db_name
}
],
"fsrvowner": db_user,
"name": "test_fsrv_add_%s" % (str(uuid.uuid4())[1:8])
}
response = self.tester.post(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/',
data=json.dumps(data),
content_type='html/json')
self.assertEquals(response.status_code, 200)
self.data = fsrv_utils.get_fs_data(db_user, self.server, self.db_name)
if self.is_positive_test:
response = self.create_foreign_server()
else:
if hasattr(self, "error_fdw_id"):
self.fdw_id = 99999
response = self.create_foreign_server()
if hasattr(self, "missing_parameter"):
del self.data['name']
response = self.create_foreign_server()
if hasattr(self, "internal_server_error"):
return_value_object = eval(self.mock_data["return_value"])
with patch(self.mock_data["function_name"],
side_effect=[return_value_object]):
response = self.create_foreign_server()
if hasattr(self, "error_in_db"):
return_value_object = eval(self.mock_data["return_value"])
with patch(self.mock_data["function_name"],
side_effect=[return_value_object]):
response = self.create_foreign_server()
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 the test database and drop added foreign
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete.py
+32 -12
View File
@@ -19,15 +19,14 @@ 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 fsrv_utils
from unittest.mock import patch
class ForeignServerDeleteTestCase(BaseTestGenerator):
"""This class will add foreign server under FDW node."""
"""This class will delete foreign server under FDW node."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign server node.
('Check FSRV Node', dict(url='/browser/foreign_server/obj/'))
]
scenarios = utils.generate_scenarios('foreign_server_delete',
fsrv_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -45,6 +44,18 @@ class ForeignServerDeleteTestCase(BaseTestGenerator):
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
def delete_foreign_server(self):
"""
This function returns the foreign server delete response
:return: foreign server delete response
"""
return self.tester.delete(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + '/' +
str(self.fsrv_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch foreign server present under test
database."""
@@ -62,13 +73,22 @@ class ForeignServerDeleteTestCase(BaseTestGenerator):
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id),
follow_redirects=True)
self.assertEquals(delete_response.status_code, 200)
if self.is_positive_test:
delete_response = self.delete_foreign_server()
else:
if hasattr(self, "internal_server_error"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
delete_response = self.delete_foreign_server()
if hasattr(self, "wrong_foreign_server_id"):
self.fsrv_id = 99999
delete_response = self.delete_foreign_server()
actual_response_code = delete_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 the test database and drop
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_delete_multiple.py
+25 -14
View File
@@ -23,12 +23,10 @@ from . import utils as fsrv_utils
class ForeignServerDeleteMultipleTestCase(BaseTestGenerator):
"""This class will add foreign server under FDW node."""
"""This class will delete foreign server under FDW node."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign server node.
('Check FSRV Node', dict(url='/browser/foreign_server/obj/'))
]
scenarios = utils.generate_scenarios('foreign_server_multiple_delete',
fsrv_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -51,8 +49,22 @@ class ForeignServerDeleteMultipleTestCase(BaseTestGenerator):
self.fsrv_names[1],
self.fdw_name)]
def delete_multiple(self, data):
"""
This function returns multiple foreign server delete response
:param data: foreign server ids to delete
:return: foreign server delete response
"""
return self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + "/",
data=json.dumps(data),
content_type='html/json',
follow_redirects=True)
def runTest(self):
"""This function will fetch foreign server present under test
"""This function will delete foreign server present under test
database."""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
@@ -73,14 +85,13 @@ class ForeignServerDeleteMultipleTestCase(BaseTestGenerator):
if not fsrv_response:
raise Exception("Could not find FSRV.")
data = {'ids': self.fsrv_ids}
delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + "/",
data=json.dumps(data),
content_type='html/json',
follow_redirects=True)
self.assertEquals(delete_response.status_code, 200)
if self.is_positive_test:
delete_response = self.delete_multiple(data)
actual_response_code = delete_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 the test database and drop added
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_dependent_dependency.py
+91
View File
@@ -0,0 +1,91 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
tests import utils as fdw_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as fsrv_utils
class ForeignServerDependentsAndDependencyTestCase(BaseTestGenerator):
""" This class will fetch added foreign server dependency and dependent
under test database. """
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('foreign_server_dependent_dependency',
fsrv_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(ForeignServerDependentsAndDependencyTestCase, self).setUp()
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.schema_name = self.schema_data['schema_name']
self.fdw_name = "test_fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "test_fsrv_%s" % (str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
def dependents_foreign_server(self):
"""
This function returns the foreign server dependents response
:return: foreign server dependents response
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + '/' +
str(self.fsrv_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch foreign server dependents and dependency
present under test database."""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
self.db_id)
if not db_con["info"] == "Database connected.":
raise Exception("Could not connect to database.")
fdw_response = fdw_utils.verify_fdw(self.server, self.db_name,
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
if self.is_positive_test:
response = self.dependents_foreign_server()
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 the test database and drop
added foreign data server and dependant objects."""
fdw_utils.delete_fdw(self.server, self.db_name,
self.fdw_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_get.py
+50 -12
View File
@@ -21,15 +21,14 @@ 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 fsrv_utils
from unittest.mock import patch
class ForeignServerGetTestCase(BaseTestGenerator):
"""This class will add foreign server under FDW node."""
"""This class will get foreign server properties under FDW node."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign server node.
('Check FSRV Node', dict(url='/browser/foreign_server/obj/'))
]
scenarios = utils.generate_scenarios('foreign_server_get',
fsrv_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -49,8 +48,31 @@ class ForeignServerGetTestCase(BaseTestGenerator):
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
def get_foreign_server(self):
"""
This function returns the foreign server get response
:return: foreign server get response
"""
return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + '/' +
str(self.fsrv_id),
content_type='html/json')
def get_foreign_server_list(self):
"""
This functions returns the foreign server list
:return: foreign server list
"""
return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + "/",
content_type='html/json')
def runTest(self):
"""This function will fetch foreign server present under test
"""This function will get foreign server properties present under test
database."""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
@@ -62,12 +84,28 @@ class ForeignServerGetTestCase(BaseTestGenerator):
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
fsrv_response = self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' + str(self.fsrv_id),
content_type='html/json')
self.assertEquals(fsrv_response.status_code, 200)
if self.is_positive_test:
if hasattr(self, "foreign_server_list"):
fsrv_response = self.get_foreign_server_list()
else:
fsrv_response = self.get_foreign_server()
else:
if hasattr(self, "error_fetching_fsrv"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "foreign_server_list"):
fsrv_response = self.get_foreign_server_list()
else:
fsrv_response = self.get_foreign_server()
if hasattr(self, "wrong_fsrv_id"):
self.fsrv_id = 99999
fsrv_response = self.get_foreign_server()
actual_response_code = fsrv_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 the test database and drop added extension
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_msql.py
+91
View File
@@ -0,0 +1,91 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from pgadmin.browser.server_groups.servers.databases.extensions.tests import \
utils as extension_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
tests import utils as fdw_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as fsrv_utils
class ForeignServerGetSQLTestCase(BaseTestGenerator):
"""This class will fetch foreign server msql under FDW node."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('foreign_server_get_msql',
fsrv_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(ForeignServerGetSQLTestCase, self).setUp()
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.schema_name = self.schema_data['schema_name']
self.extension_name = "cube"
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "test_fsrv_add_%s" % (str(uuid.uuid4())[1:8])
self.extension_id = extension_utils.create_extension(
self.server, self.db_name, self.extension_name, self.schema_name)
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
def msql_foreign_server(self):
"""
This function returns the foreign server msql response
:return: foreign server msql response
"""
return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + '/' +
str(self.fsrv_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch foreign server msql present under test
database."""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
self.db_id)
if not db_con["info"] == "Database connected.":
raise Exception("Could not connect to database.")
fdw_response = fdw_utils.verify_fdw(self.server, self.db_name,
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
if self.is_positive_test:
response = self.msql_foreign_server()
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):
# Disconnect the database
extension_utils.drop_extension(self.server, self.db_name,
self.extension_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_node.py
+102
View File
@@ -0,0 +1,102 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
import json
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
tests import utils as fdw_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as fsrv_utils
from unittest.mock import patch
class ForeignServerGetNodeTestCase(BaseTestGenerator):
"""This class will update foreign server node under FDW node."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('foreign_server_get_nodes',
fsrv_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(ForeignServerGetNodeTestCase, self).setUp()
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.schema_name = self.schema_data['schema_name']
self.fdw_name = "test_fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "test_fsrv_%s" % (str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
def get_foreign_server_node(self, fsrv_id):
"""
This function returns the foreign server node response
:return: foreign server node response
"""
return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + '/' +
str(fsrv_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch foreign server node present under test
database."""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
self.db_id)
if not db_con["info"] == "Database connected.":
raise Exception("Could not connect to database.")
fdw_response = fdw_utils.verify_fdw(self.server, self.db_name,
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
if self.is_positive_test:
if hasattr(self, "node"):
response = self.get_foreign_server_node(self.fsrv_id)
else:
response = self.get_foreign_server_node("")
else:
if hasattr(self, "internal_server_error"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "node"):
response = self.get_foreign_server_node(self.fsrv_id)
else:
response = self.get_foreign_server_node("")
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 the test database and drop
added foreign data server and dependant objects."""
fdw_utils.delete_fdw(self.server, self.db_name,
self.fdw_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_put.py
+30 -13
View File
@@ -22,15 +22,14 @@ 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 fsrv_utils
from unittest.mock import patch
class ForeignServerPutTestCase(BaseTestGenerator):
"""This class will add foreign server under FDW node."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign server node.
('Check FSRV Node', dict(url='/browser/foreign_server/obj/'))
]
scenarios = utils.generate_scenarios('foreign_server_update',
fsrv_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -50,6 +49,19 @@ class ForeignServerPutTestCase(BaseTestGenerator):
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
def update_fsrv(self):
"""
This functions update foreign servers
:return: foreign server update request details
"""
return self.tester.put(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + '/' +
str(self.fsrv_id),
data=json.dumps(self.test_data),
follow_redirects=True)
def runTest(self):
"""This function will update foreign server present under test
database."""
@@ -67,15 +79,20 @@ class ForeignServerPutTestCase(BaseTestGenerator):
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
data = {"description": "This is foreign server update comment",
"id": self.fsrv_id}
put_response = self.tester.put(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id), data=json.dumps(data),
follow_redirects=True)
self.assertEquals(put_response.status_code, 200)
self.test_data["id"] = self.fsrv_id
if self.is_positive_test:
put_response = self.update_fsrv()
else:
if hasattr(self, "error_in_db"):
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
put_response = self.update_fsrv()
actual_response_code = put_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 the test database and drop added extension
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/test_foreign_servers_sql.py
+102
View File
@@ -0,0 +1,102 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from pgadmin.browser.server_groups.servers.databases.extensions.tests import \
utils as extension_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
tests import utils as fdw_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as fsrv_utils
from unittest.mock import patch
class ForeignServerGetSQLTestCase(BaseTestGenerator):
"""This class will add foreign server under FDW node."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('foreign_server_get_sql',
fsrv_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(ForeignServerGetSQLTestCase, self).setUp()
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.schema_name = self.schema_data['schema_name']
self.extension_name = "cube"
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "test_fsrv_add_%s" % (str(uuid.uuid4())[1:8])
self.extension_id = extension_utils.create_extension(
self.server, self.db_name, self.extension_name, self.schema_name)
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
def sql_foreign_server(self):
"""
This function returns the foreign server sql response
:return: foreign server sql response
"""
return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + '/' +
str(self.fsrv_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch foreign server present under test
database."""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
self.db_id)
if not db_con["info"] == "Database connected.":
raise Exception("Could not connect to database.")
fdw_response = fdw_utils.verify_fdw(self.server, self.db_name,
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
if self.is_positive_test:
response = self.sql_foreign_server()
else:
if hasattr(self, "internal_server_error"):
return_value_object = eval(self.mock_data["return_value"])
with patch(self.mock_data["function_name"],
side_effect=[return_value_object]):
response = self.sql_foreign_server()
if hasattr(self, "wrong_foreign_server_id"):
self.fsrv_id = 99999
response = self.sql_foreign_server()
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):
# Disconnect the database
extension_utils.drop_extension(self.server, self.db_name,
self.extension_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/tests/utils.py
+41
View File
@@ -11,10 +11,51 @@ from __future__ import print_function
import os
import sys
import uuid
import json
from regression.python_test_utils.test_utils import get_db_connection
file_name = os.path.basename(__file__)
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/foreign_servers_test_data.json") as data_file:
test_cases = json.load(data_file)
def get_fs_data(db_user, server, db_name):
data = {
"fsrvacl": [
{
"grantee": db_user,
"grantor": db_user,
"privileges":
[
{
"privilege_type": "U",
"privilege": "true",
"with_grant": "false"
}
]
}
],
"fsrvoptions": [
{
"fsrvoption": "host",
"fsrvvalue": server['host']
},
{
"fsrvoption": "port",
"fsrvvalue": str(server['port'])
},
{
"fsrvoption": "dbname",
"fsrvvalue": db_name
}
],
"fsrvowner": db_user,
"name": "test_fsrv_add_%s" % (str(uuid.uuid4())[1:8])
}
return data
def create_fsrv(server, db_name, fsrv_name, fdw_name):
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_add.py
+48 -27
View File
@@ -23,16 +23,15 @@ from pgadmin.browser.server_groups.servers.databases.tests import \
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 user_utils
from unittest.mock import patch
class UserMappingAddTestCase(BaseTestGenerator):
"""This class will add user mapping under foreign server node."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for user mapping node.
('Check user mapping Node', dict(url='/browser/user_mapping/obj/'))
]
scenarios = utils.generate_scenarios('user_mapping_create',
user_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -52,9 +51,23 @@ class UserMappingAddTestCase(BaseTestGenerator):
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
def create_user_mapping(self):
"""
This function create a user mapping and returns the created
user mapping response
:return: created user mapping response
"""
return self.tester.post(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(
self.db_id) +
'/' + str(self.fdw_id) + '/' + str(self.fsrv_id) + '/',
data=json.dumps(self.data),
content_type='html/json')
def runTest(self):
"""This function will update foreign server present under test
database. """
"""This function will fetch foreign server and add user mappping
present under test database. """
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
@@ -70,26 +83,34 @@ class UserMappingAddTestCase(BaseTestGenerator):
if not fsrv_response:
raise Exception("Could not find FSRV.")
db_user = self.server["username"]
data = {"name": db_user,
"um_options": [],
"umoptions": [
{
"umoption": "user",
"umvalue": self.server["username"]
},
{
"umoption": "password",
"umvalue": self.server["db_password"]
}
]}
response = self.tester.post(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(
self.db_id) +
'/' + str(self.fdw_id) + '/' + str(self.fsrv_id) + '/',
data=json.dumps(data),
content_type='html/json')
self.assertEquals(response.status_code, 200)
self.data = user_utils.get_um_data(db_user, self.server)
if self.is_positive_test:
response = self.create_user_mapping()
else:
if hasattr(self, "error_fsrv_id"):
self.fsrv_id = 99999
response = self.create_user_mapping()
if hasattr(self, "internal_server_error"):
return_value_object = eval(self.mock_data["return_value"])
with patch(self.mock_data["function_name"],
side_effect=[return_value_object]):
response = self.create_user_mapping()
if hasattr(self, "missing_parameter"):
del self.data['name']
response = self.create_user_mapping()
if hasattr(self, "error_in_db"):
return_value_object = eval(self.mock_data["return_value"])
with patch(self.mock_data["function_name"],
side_effect=[return_value_object]):
response = self.create_user_mapping()
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 the test database and drop added extension
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete.py
+40 -11
View File
@@ -23,15 +23,14 @@ 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 um_utils
from unittest.mock import patch
class UserMappingDeleteTestCase(BaseTestGenerator):
"""This class will delete user mapping under foreign server node."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for user mapping node.
('Check user mapping Node', dict(url='/browser/user_mapping/obj/'))
]
scenarios = utils.generate_scenarios('user_mapping_delete',
um_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -50,6 +49,18 @@ class UserMappingDeleteTestCase(BaseTestGenerator):
self.um_id = um_utils.create_user_mapping(self.server, self.db_name,
self.fsrv_name)
def delete_user_mapping(self):
"""
This function returns the user mapping delete response
:return: user mapping delete response
"""
return self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + '/' + str(self.um_id),
follow_redirects=True)
def runTest(self):
"""This function will delete user mapping present under test
database. """
@@ -71,13 +82,31 @@ class UserMappingDeleteTestCase(BaseTestGenerator):
self.fsrv_name)
if not um_response:
raise Exception("Could not find user mapping.")
delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + '/' + str(self.um_id),
follow_redirects=True)
self.assertEquals(delete_response.status_code, 200)
if self.is_positive_test:
delete_response = self.delete_user_mapping()
else:
if hasattr(self, "internal_server_error"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
delete_response = self.delete_user_mapping()
if hasattr(self, "error_in_db"):
return_value_object = eval(self.mock_data["return_value"])
with patch(self.mock_data["function_name"],
side_effect=[return_value_object]):
delete_response = self.delete_user_mapping()
if hasattr(self, "wrong_foreign_server_id"):
self.fsrv_id = 99999
delete_response = self.delete_user_mapping()
if hasattr(self, "wrong_user_mapping_id"):
self.um_id = 99999
delete_response = self.delete_user_mapping()
actual_response_code = delete_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 the test database and drop
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_delete_multiple.py
+25 -15
View File
@@ -27,12 +27,10 @@ from . import utils as um_utils
class UserMappingDeleteMultipleTestCase(BaseTestGenerator):
"""This class will delete user mapping under foreign server node."""
"""This class will delete user mappings under foreign server node."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for user mapping node.
('Check user mapping Node', dict(url='/browser/user_mapping/obj/'))
]
scenarios = utils.generate_scenarios('user_mapping_multiple_delete',
um_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -51,6 +49,21 @@ class UserMappingDeleteMultipleTestCase(BaseTestGenerator):
self.um_id = um_utils.create_user_mapping(self.server, self.db_name,
self.fsrv_name)
def delete_multiple(self):
"""
This function returns multiple user mapping delete response
:param data: user mapping ids to delete
:return: user mapping delete response
"""
return self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + "/",
follow_redirects=True,
data=json.dumps(self.data),
content_type='html/json')
def runTest(self):
"""This function will delete user mapping present under test
database. """
@@ -72,16 +85,13 @@ class UserMappingDeleteMultipleTestCase(BaseTestGenerator):
self.fsrv_name)
if not um_response:
raise Exception("Could not find user mapping.")
data = {'ids': [self.um_id]}
delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + "/",
follow_redirects=True,
data=json.dumps(data),
content_type='html/json')
self.assertEquals(delete_response.status_code, 200)
self.data = {'ids': [self.um_id]}
if self.is_positive_test:
delete_response = self.delete_multiple()
actual_response_code = delete_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 the test database and drop
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_dependent_dependency.py
+100
View File
@@ -0,0 +1,100 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from pgadmin.browser.server_groups.servers.databases.extensions.tests import \
utils as extension_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
foreign_servers.tests import utils as fsrv_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
tests import utils as fdw_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as um_utils
from unittest.mock import patch
class UserMappingDependentsAndDependencyTestCase(BaseTestGenerator):
""" This class will fetch added user mapping dependency and dependent
under test database. """
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('user_mapping_dependent_dependency',
um_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(UserMappingDependentsAndDependencyTestCase, self).setUp()
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.schema_name = self.schema_data['schema_name']
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
self.um_id = um_utils.create_user_mapping(self.server, self.db_name,
self.fsrv_name)
def dependents_user_mapping(self):
"""
This function returns the user mapping dependents response
:return: user mapping dependents response
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + '/' + str(self.um_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch user mapping present under test
database. """
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
self.db_id)
if not db_con["info"] == "Database connected.":
raise Exception("Could not connect to database.")
fdw_response = fdw_utils.verify_fdw(self.server, self.db_name,
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
um_response = um_utils.verify_user_mapping(self.server, self.db_name,
self.fsrv_name)
if not um_response:
raise Exception("Could not find user mapping.")
if self.is_positive_test:
response = self.dependents_user_mapping()
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 the test database and drop
foreign data wrapper and dependant objects."""
fdw_utils.delete_fdw(self.server, self.db_name,
self.fdw_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_get.py
+51 -10
View File
@@ -23,15 +23,14 @@ 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 um_utils
from unittest.mock import patch
class UserMappingGetTestCase(BaseTestGenerator):
"""This class will add user mapping under foreign server node."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for user mapping node.
('Check user mapping Node', dict(url='/browser/user_mapping/obj/'))
]
scenarios = utils.generate_scenarios('user_mapping_get',
um_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -53,6 +52,31 @@ class UserMappingGetTestCase(BaseTestGenerator):
self.um_id = um_utils.create_user_mapping(self.server, self.db_name,
self.fsrv_name)
def get_user_mapping(self):
"""
This function returns the user mapping get response
:return: user mapping get response
"""
return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + '/' +
str(self.fsrv_id) + '/' +
str(self.um_id),
content_type='html/json')
def get_user_mapping_list(self):
"""
This functions returns the user mapping list
:return: user mapping list
"""
return self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id) + "/" +
str(self.fsrv_id) + '/',
content_type='html/json')
def runTest(self):
"""This function will update foreign server present under test
database."""
@@ -70,12 +94,29 @@ class UserMappingGetTestCase(BaseTestGenerator):
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
response = self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(
self.db_id) + '/' + str(self.fdw_id) + '/' + str(
self.fsrv_id) + '/' + str(
self.um_id), content_type='html/json')
self.assertEquals(response.status_code, 200)
if self.is_positive_test:
if hasattr(self, "um_list"):
response = self.get_user_mapping_list()
else:
response = self.get_user_mapping()
else:
if hasattr(self, "error_fetching_um"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "um_list"):
response = self.get_user_mapping_list()
else:
response = self.get_user_mapping()
if hasattr(self, "wrong_um_id"):
self.um_id = 99999
response = self.get_user_mapping()
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 the test database and drop added extension
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_msql.py
+101
View File
@@ -0,0 +1,101 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from pgadmin.browser.server_groups.servers.databases.extensions.tests import \
utils as extension_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
foreign_servers.tests import utils as fsrv_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
tests import utils as fdw_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as um_utils
class UserMappingGetMSQLTestCase(BaseTestGenerator):
"""This class will add user mapping under foreign server node."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('user_mapping_get_msql',
um_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(UserMappingGetMSQLTestCase, self).setUp()
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.schema_name = self.schema_data['schema_name']
self.extension_name = "cube"
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.extension_id = extension_utils.create_extension(
self.server, self.db_name, self.extension_name, self.schema_name)
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
self.um_id = um_utils.create_user_mapping(self.server, self.db_name,
self.fsrv_name)
def msql_user_mapping(self):
"""
This function returns the user mapping msql response
:return: user mapping msql response
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + '/' + str(self.um_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch foreign server present under test
database"""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
self.db_id)
if not db_con["info"] == "Database connected.":
raise Exception("Could not connect to database.")
fdw_response = fdw_utils.verify_fdw(self.server, self.db_name,
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
fsrv_response = fsrv_utils.verify_fsrv(self.server,
self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
um_response = um_utils.verify_user_mapping(self.server, self.db_name,
self.fsrv_name)
if not um_response:
raise Exception("Could not find user mapping.")
if self.is_positive_test:
response = self.msql_user_mapping()
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 the test database and drop added
extension and dependant objects."""
extension_utils.drop_extension(self.server, self.db_name,
self.extension_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_node.py
+110
View File
@@ -0,0 +1,110 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from pgadmin.browser.server_groups.servers.databases.extensions.tests import \
utils as extension_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
foreign_servers.tests import utils as fsrv_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
tests import utils as fdw_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as um_utils
from unittest.mock import patch
class UserMappingNodesTestCase(BaseTestGenerator):
"""This class will delete user mapping under foreign server node."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('user_mapping_get_nodes',
um_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(UserMappingNodesTestCase, self).setUp()
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.schema_name = self.schema_data['schema_name']
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
self.um_id = um_utils.create_user_mapping(self.server, self.db_name,
self.fsrv_name)
def get_user_mapping_node(self, um_id):
"""
This function returns the user mapping node response
:return: user mapping node response
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + '/' + str(um_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch user mapping present under test
database. """
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
self.db_id)
if not db_con["info"] == "Database connected.":
raise Exception("Could not connect to database.")
fdw_response = fdw_utils.verify_fdw(self.server, self.db_name,
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
um_response = um_utils.verify_user_mapping(self.server, self.db_name,
self.fsrv_name)
if not um_response:
raise Exception("Could not find user mapping.")
if self.is_positive_test:
if hasattr(self, "node"):
response = self.get_user_mapping_node(self.um_id)
else:
response = self.get_user_mapping_node("")
else:
if hasattr(self, "internal_server_error"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "node"):
response = self.get_user_mapping_node(self.um_id)
else:
response = self.get_user_mapping_node("")
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 the test database and drop
foreign data wrapper and dependant objects."""
fdw_utils.delete_fdw(self.server, self.db_name,
self.fdw_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_put.py
+31 -25
View File
@@ -24,15 +24,14 @@ 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 um_utils
from unittest.mock import patch
class UserMappingPutTestCase(BaseTestGenerator):
"""This class will update user mapping under foreign server node."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for user mapping node.
('Check user mapping Node', dict(url='/browser/user_mapping/obj/'))
]
scenarios = utils.generate_scenarios('user_mapping_update',
um_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@@ -54,8 +53,21 @@ class UserMappingPutTestCase(BaseTestGenerator):
self.um_id = um_utils.create_user_mapping(self.server, self.db_name,
self.fsrv_name)
def update_user_mapping(self):
"""
This functions update user mapping
:return: user mapping update request details
"""
return self.tester.put(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + '/' + str(self.um_id),
data=json.dumps(self.test_data),
follow_redirects=True)
def runTest(self):
"""This function will update foreign server present under test
"""This function will update user mapping present under test
database"""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
@@ -76,26 +88,20 @@ class UserMappingPutTestCase(BaseTestGenerator):
self.fsrv_name)
if not um_response:
raise Exception("Could not find user mapping.")
data = {
"id": self.um_id,
"umoptions":
{
"changed":
[
{"umoption": "user",
"umvalue": "public"
}
]
}
}
put_response = self.tester.put(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + '/' + str(self.um_id),
data=json.dumps(data),
follow_redirects=True)
self.assertEquals(put_response.status_code, 200)
self.test_data["id"] = self.um_id
if self.is_positive_test:
put_response = self.update_user_mapping()
else:
if hasattr(self, "error_in_db"):
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
put_response = self.update_user_mapping()
actual_response_code = put_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 the test database and drop added
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/test_user_mapping_sql.py
+112
View File
@@ -0,0 +1,112 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
from pgadmin.browser.server_groups.servers.databases.extensions.tests import \
utils as extension_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
foreign_servers.tests import utils as fsrv_utils
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
tests import utils as fdw_utils
from pgadmin.browser.server_groups.servers.databases.tests import \
utils as database_utils
from pgadmin.utils.route import BaseTestGenerator
from regression import parent_node_dict
from regression.python_test_utils import test_utils as utils
from . import utils as um_utils
from unittest.mock import patch
class UserMappingGetSQLTestCase(BaseTestGenerator):
"""This class will add user mapping under foreign server node."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('user_mapping_get_sql',
um_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(UserMappingGetSQLTestCase, self).setUp()
self.schema_data = parent_node_dict['schema'][-1]
self.server_id = self.schema_data['server_id']
self.db_id = self.schema_data['db_id']
self.db_name = parent_node_dict["database"][-1]["db_name"]
self.schema_name = self.schema_data['schema_name']
self.extension_name = "cube"
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.extension_id = extension_utils.create_extension(
self.server, self.db_name, self.extension_name, self.schema_name)
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
self.um_id = um_utils.create_user_mapping(self.server, self.db_name,
self.fsrv_name)
def sql_user_mapping(self):
"""
This function returns the user mapping sql response
:return: user mapping sql response
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) +
'/' + str(self.fdw_id) + '/' +
str(self.fsrv_id) + '/' + str(self.um_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch user mapping present under test
database"""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
self.db_id)
if not db_con["info"] == "Database connected.":
raise Exception("Could not connect to database.")
fdw_response = fdw_utils.verify_fdw(self.server, self.db_name,
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
fsrv_response = fsrv_utils.verify_fsrv(self.server,
self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find FSRV.")
um_response = um_utils.verify_user_mapping(self.server, self.db_name,
self.fsrv_name)
if not um_response:
raise Exception("Could not find user mapping.")
if self.is_positive_test:
response = self.sql_user_mapping()
else:
if hasattr(self, "internal_server_error"):
return_value_object = eval(self.mock_data["return_value"])
with patch(self.mock_data["function_name"],
side_effect=[return_value_object]):
response = self.sql_user_mapping()
if hasattr(self, "wrong_um_id"):
self.um_id = 99999
response = self.sql_user_mapping()
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 the test database and drop added
extension and dependant objects."""
extension_utils.drop_extension(self.server, self.db_name,
self.extension_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/user_mapping_test_data.json
+388
View File
@@ -0,0 +1,388 @@
{
"user_mapping_create": [
{
"name": "Create user mapping: With valid data.",
"url": "/browser/user_mapping/obj/",
"is_positive_test": true,
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error creating user mapping: Internal server error.",
"url": "/browser/user_mapping/obj/",
"is_positive_test": false,
"internal_server_error": true,
"test_data": {},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while creating a user mapping')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error creating user mapping - wrong fsrv_id",
"url": "/browser/user_mapping/obj/",
"is_positive_test": false,
"error_fsrv_id": true,
"test_data": {},
"mocking_required": false,
"mock_data": {
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error while creating user mapping: With insufficient data.",
"url": "/browser/user_mapping/obj/",
"is_positive_test": false,
"missing_parameter": true,
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error creating user mapping: Internal server error 2.",
"url": "/browser/user_mapping/obj/",
"is_positive_test": false,
"error_in_db": true,
"test_data": {},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_scalar",
"return_value": "(False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
}
],
"user_mapping_delete": [
{
"name": "delete user mapping",
"url": "/browser/user_mapping/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while deleting a user mapping - Internal server error 1",
"url": "/browser/user_mapping/obj/",
"internal_server_error": true,
"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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while deleting a user mapping - wrong foreign server id",
"url": "/browser/user_mapping/obj/",
"wrong_foreign_server_id": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error while deleting a user mapping - wrong user mapping id",
"url": "/browser/user_mapping/obj/",
"wrong_user_mapping_id": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Error creating user mapping: Error in db.",
"url": "/browser/user_mapping/obj/",
"is_positive_test": false,
"error_in_db": true,
"test_data": {},
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while deleting a user mapping')"
},
"expected_data": {
"status_code": 500
}
}
],
"user_mapping_multiple_delete": [
{
"name": "Delete multiple user mapping",
"url": "/browser/user_mapping/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"user_mapping_dependent_dependency": [
{
"name": "Get user mapping dependents",
"url": "/browser/user_mapping/dependent/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get user mapping dependencies",
"url": "/browser/user_mapping/dependency/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"user_mapping_get_nodes": [
{
"name": "Get user mapping nodes",
"url": "/browser/user_mapping/nodes/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error fetching user mapping nodes",
"url": "/browser/user_mapping/nodes/",
"is_positive_test": false,
"internal_server_error": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching user mapping nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get user mapping node",
"url": "/browser/user_mapping/nodes/",
"is_positive_test": true,
"node": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error fetching user mapping node",
"url": "/browser/user_mapping/nodes/",
"is_positive_test": false,
"node": true,
"internal_server_error": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching user mapping nodes')"
},
"expected_data": {
"status_code": 500
}
}
],
"user_mapping_update": [
{
"name": "Update user mapping",
"url": "/browser/user_mapping/obj/",
"is_positive_test": true,
"mocking_required": false,
"test_data": {
"umoptions":
{
"changed":
[
{"umoption": "user",
"umvalue": "public"
}
]
},
"id": "PLACE_HOLDER"
},
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while updating a user mapping - Internal server error",
"url": "/browser/user_mapping/obj/",
"error_in_db": true,
"test_data": {
"umoptions":
{
"changed":
[
{"umoption": "user",
"umvalue": "public"
}
]
},
"id": "PLACE_HOLDER"
},
"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 fetching a user mapping')]"
},
"expected_data": {
"status_code": 500
}
}
],
"user_mapping_get": [
{
"name": "Get user mapping properties",
"url": "/browser/user_mapping/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a user mapping properties",
"url": "/browser/user_mapping/obj/",
"error_fetching_um": 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 fetching a user mapping')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error - Fetching user mapping with wrong ID",
"url": "/browser/user_mapping/obj/",
"wrong_um_id": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get user mapping list",
"url": "/browser/user_mapping/obj/",
"is_positive_test": true,
"um_list": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a user mapping list",
"url": "/browser/user_mapping/obj/",
"error_fetching_um": true,
"is_positive_test": false,
"um_list": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching a user mapping')"
},
"expected_data": {
"status_code": 500
}
}
],
"user_mapping_get_sql": [
{
"name": "Get user mapping SQL",
"url": "/browser/user_mapping/sql/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error - Get user mapping SQL - Internal server error",
"url": "/browser/user_mapping/sql/",
"is_positive_test": false,
"internal_server_error": true,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_dict",
"return_value": "(False, 'Mocked Internal Server Error while fetching a user mapping')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error - Get user mapping SQL using wrong um id",
"url": "/browser/user_mapping/sql/",
"wrong_um_id": true,
"is_positive_test": false,
"mocking_required": false,
"mock_data": {
},
"expected_data": {
"status_code": 410
}
}
],
"user_mapping_get_msql": [
{
"name": "Get user mapping msql",
"is_positive_test": true,
"url": "/browser/user_mapping/msql/",
"test_data": {
"name": "modifying name",
"comment": "Comments to test update"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
]
}
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/foreign_servers/user_mappings/tests/utils.py
+23
View File
@@ -9,11 +9,34 @@
from __future__ import print_function
import os
import sys
import traceback
import json
from regression.python_test_utils.test_utils import get_db_connection
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/user_mapping_test_data.json") as data_file:
test_cases = json.load(data_file)
def get_um_data(db_user, server):
data = {"name": db_user,
"um_options": [],
"umoptions": [
{
"umoption": "user",
"umvalue": server["username"]
},
{
"umoption": "password",
"umvalue": server["db_password"]
}
]}
return data
def create_user_mapping(server, db_name, fsrv_name):
"""