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

Improve code coverage and API test cases for Foreign Data Wrapper. Fixes #5323.

Browse Source
This commit is contained in:
Satish V 2020-07-14 15:58:22 +05:30 committed by Akshay Joshi
parent 161c9b0fc7
commit a91e4d8a7f
14 changed files with 1088 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/en_US/release_notes_4_24.rst
View File

@ -17,6 +17,7 @@ New features
Housekeeping
************
| `Issue #5323 <https://redmine.postgresql.org/issues/5323>`_ - Improve code coverage and API test cases for Foreign Data Wrapper.
| `Issue #5326 <https://redmine.postgresql.org/issues/5326>`_ - Improve code coverage and API test cases for Domain and Domain Constraints.
| `Issue #5329 <https://redmine.postgresql.org/issues/5329>`_ - Improve code coverage and API test cases for FTS Configuration, FTS Parser, FTS Dictionaries, and FTS Template.

409
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/fdw_test_data.json Normal file
View File

@ -0,0 +1,409 @@
{
"fdw_create": [
{
"name": "Create fdw: With valid data.",
"url": "/browser/foreign_data_wrapper/obj/",
"is_positive_test": true,
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while creating fdw while server down",
"url": "/browser/foreign_data_wrapper/obj/",
"is_positive_test": false,
"error_creating_fdw": 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 data wrapper')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while creating fdw - internal server error",
"url": "/browser/foreign_data_wrapper/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": "(True, True), (False, 'Mocked Internal Server Error')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while creating fdw - invalid data",
"url": "/browser/foreign_data_wrapper/obj/",
"is_positive_test": false,
"invalid_data": true,
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
}
],
"fdw_dependency_dependent": [
{
"name": "Get fdw dependency",
"url": "/browser/foreign_data_wrapper/dependency/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Get fdw dependent",
"url": "/browser/foreign_data_wrapper/dependent/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"fdw_delete": [
{
"name": "delete fdw",
"url": "/browser/foreign_data_wrapper/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while deleting a fdw",
"url": "/browser/foreign_data_wrapper/obj/",
"error_deleting_fdw": 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 while deleting a fdw')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while deleting a fdw - internal server error",
"url": "/browser/foreign_data_wrapper/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": "(True, True), (False, 'Mocked Internal Server Error while deleting a fdw')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "delete fdw using wrong fdw id",
"url": "/browser/foreign_data_wrapper/obj/",
"wrong_fdw_id": true,
"is_positive_test": false,
"mocking_required": false,
"mock_data": {
},
"expected_data": {
"status_code": 410
}
}
],
"fdw_delete_multiple": [
{
"name": "Delete multiple fdw",
"url": "/browser/foreign_data_wrapper/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
}
],
"fdw_get_nodes_and_node": [
{
"name": "Get fdw nodes",
"url": "/browser/foreign_data_wrapper/nodes/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching fdw nodes",
"url": "/browser/foreign_data_wrapper/nodes/",
"error_fetching_fdw": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching fdw nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Get fdw node",
"url": "/browser/foreign_data_wrapper/nodes/",
"is_positive_test": true,
"node": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching fdw node",
"url": "/browser/foreign_data_wrapper/nodes/",
"error_fetching_fdw": true,
"is_positive_test": false,
"node": 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 fdw nodes')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Fetching fdw node with wrong id",
"url": "/browser/foreign_data_wrapper/nodes/",
"wrong_id": true,
"is_positive_test": false,
"node": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
}
],
"fdw_get_handlers": [
{
"name": "Get get_handlers",
"url": "/browser/foreign_data_wrapper/get_handlers/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while getting handlers",
"url": "/browser/foreign_data_wrapper/get_handlers/",
"internal_server_error": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching handlers')"
},
"expected_data": {
"status_code": 500
}
}
],
"fdw_get_validators": [
{
"name": "Get get_validators",
"url": "/browser/foreign_data_wrapper/get_validators/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while getting validtors",
"url": "/browser/foreign_data_wrapper/get_validators/",
"internal_server_error": true,
"is_positive_test": false,
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.utils.driver.psycopg2.connection.Connection.execute_2darray",
"return_value": "(False, 'Mocked Internal Server Error while fetching handlers')"
},
"expected_data": {
"status_code": 500
}
}
],
"fdw_update": [
{
"name": "Update fdw",
"url": "/browser/foreign_data_wrapper/obj/",
"is_positive_test": true,
"mocking_required": false,
"test_data": {
"description": "This is fdw update comment",
"id": "PLACE_HOLDER"
},
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while updating a fdw",
"url": "/browser/foreign_data_wrapper/obj/",
"internal_server_error": true,
"test_data": {
"description": "This is fdw 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 updating fdw')"
},
"expected_data": {
"status_code": 500
}
}
],
"fdw_get": [
{
"name": "Get fdw",
"url": "/browser/foreign_data_wrapper/obj/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching a fdw",
"url": "/browser/foreign_data_wrapper/obj/",
"error_fetching_fdw": 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 fdw')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Fetch fdw using wrong fdw id",
"url": "/browser/foreign_data_wrapper/obj/",
"wrong_fdw_id": true,
"is_positive_test": false,
"mocking_required": false,
"mock_data": {
},
"expected_data": {
"status_code": 410
}
},
{
"name": "Get fdw list",
"url": "/browser/foreign_data_wrapper/obj/",
"is_positive_test": true,
"fdw_list": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching fdw list",
"url": "/browser/foreign_data_wrapper/obj/",
"error_fetching_fdw": true,
"is_positive_test": false,
"fdw_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 fdw')"
},
"expected_data": {
"status_code": 500
}
}
],
"fdw_get_sql": [
{
"name": "Get fdw SQL",
"url": "/browser/foreign_data_wrapper/sql/",
"is_positive_test": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200
}
},
{
"name": "Error while fetching SQL generated - Internal server error",
"url": "/browser/foreign_data_wrapper/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')"
},
"expected_data": {
"status_code": 500
}
},
{
"name": "Error while fetching SQL generated - Internal server error",
"url": "/browser/foreign_data_wrapper/sql/",
"is_positive_test": false,
"wrong_fdw_id": true,
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410
}
}
],
"fdw_get_msql": [
{
"name": "Get fdw msql",
"is_positive_test": true,
"url": "/browser/foreign_data_wrapper/msql/",
"mocking_required": true,
"mock_data": {
"function_name": "pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.ForeignDataWrapperView.get_sql",
"return_value": "('', 'Mocked get_sql function')"
},
"expected_data": {
"status_code": 200
}
}
]
}

47
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_add.py
View File

@ -17,17 +17,15 @@ 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 fdw_utils
from unittest.mock import patch
class FDWDAddTestCase(BaseTestGenerator):
""" This class will add foreign data wrappers under database node. """
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign_data_wrapper node.
('Check FDW Node',
dict(url='/browser/foreign_data_wrapper/obj/'))
]
scenarios = utils.generate_scenarios('fdw_create',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension."""
@ -39,6 +37,18 @@ class FDWDAddTestCase(BaseTestGenerator):
self.schema_name = self.schema_data['schema_name']
self.db_name = parent_node_dict["database"][-1]["db_name"]
def create_foreign_data_wrapper(self):
"""
This function create a foreign data wrapper and returns the created
foreign data wrapper response
:return: created foreign data wrapper response
"""
return self.tester.post(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) + '/',
data=json.dumps(self.data),
content_type='html/json')
def runTest(self):
"""This function will add foreign data wrapper under test database."""
db_con = database_utils.connect_database(self,
@ -49,15 +59,28 @@ class FDWDAddTestCase(BaseTestGenerator):
raise Exception("Could not connect to database.")
self.data = fdw_utils.get_fdw_data(self.schema_name,
self.server['username'])
response = self.tester.post(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) + '/',
data=json.dumps(self.data),
content_type='html/json')
self.assertEquals(response.status_code, 200)
if self.is_positive_test:
response = self.create_foreign_data_wrapper()
else:
if hasattr(self, "error_creating_fdw"):
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_data_wrapper()
if hasattr(self, "internal_server_error"):
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = self.create_foreign_data_wrapper()
if hasattr(self, "invalid_data"):
del self.data['name']
response = self.create_foreign_data_wrapper()
self.assertEquals(response.status_code,
self.expected_data['status_code'])
def tearDown(self):
"""This function delete the FDW and disconnect the test database """
fdw_utils.delete_fdw(self.server, self.db_name, self.data["name"])
database_utils.disconnect_database(self, self.server_id,
self.db_id)

51
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_delete.py
View File

@ -17,18 +17,18 @@ 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 fdw_utils
from unittest.mock import patch
class FDWDDeleteTestCase(BaseTestGenerator):
class FDWDeleteTestCase(BaseTestGenerator):
"""This class will delete foreign data wrappers under test database."""
skip_on_database = ['gpdb']
scenarios = [ # Fetching default URL for foreign_data_wrapper node.
('Check FDW Node',
dict(url='/browser/foreign_data_wrapper/obj/'))]
scenarios = utils.generate_scenarios('fdw_delete',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(FDWDDeleteTestCase, self).setUp()
super(FDWDeleteTestCase, 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']
@ -38,6 +38,18 @@ class FDWDDeleteTestCase(BaseTestGenerator):
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def delete_fdw(self):
"""
This function deletes fdw
:return: fdw delete
"""
return self.tester.delete(self.url +
str(utils.SERVER_GROUP) +
'/' + str(self.server_id) + '/' +
str(self.db_id) +
'/' + str(self.fdw_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch foreign data wrapper present under test
database."""
@ -51,13 +63,28 @@ class FDWDDeleteTestCase(BaseTestGenerator):
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
delete_response = self.tester.delete(self.url +
str(utils.SERVER_GROUP) +
'/' + str(self.server_id) + '/' +
str(self.db_id) +
'/' + str(self.fdw_id),
follow_redirects=True)
self.assertEquals(delete_response.status_code, 200)
if self.is_positive_test:
response = self.delete_fdw()
else:
if hasattr(self, "error_deleting_fdw"):
return_value_object = eval(self.mock_data["return_value"])
with patch(self.mock_data["function_name"],
side_effect=[return_value_object]):
response = self.delete_fdw()
if hasattr(self, "internal_server_error"):
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = self.delete_fdw()
if hasattr(self, "wrong_fdw_id"):
self.fdw_id = 99999
response = self.delete_fdw()
self.assertEquals(response.status_code,
self.expected_data['status_code'])
def tearDown(self):
"""This function disconnect the test database and drop added extension

30
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_delete_multiple.py
View File

@ -23,9 +23,8 @@ from . import utils as fdw_utils
class FDWDDeleteMultipleTestCase(BaseTestGenerator):
"""This class will delete foreign data wrappers under test database."""
skip_on_database = ['gpdb']
scenarios = [ # Fetching default URL for foreign_data_wrapper node.
('Check FDW Node',
dict(url='/browser/foreign_data_wrapper/obj/'))]
scenarios = utils.generate_scenarios('fdw_delete_multiple',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@ -42,6 +41,19 @@ class FDWDDeleteMultipleTestCase(BaseTestGenerator):
fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_names[1])]
def delete_multiple(self, data):
"""
This function returns multiple fdw delete response
:param data: fdw ids to delete
:return: fdw delete response
"""
return self.tester.delete(self.url + str(utils.SERVER_GROUP) +
'/' + str(self.server_id) +
'/' + str(self.db_id) + '/',
follow_redirects=True,
data=json.dumps(data),
content_type='html/json')
def runTest(self):
"""This function will fetch foreign data wrapper present under test
database."""
@ -60,14 +72,10 @@ class FDWDDeleteMultipleTestCase(BaseTestGenerator):
if not fdw_response:
raise Exception("Could not find FDW.")
data = {'ids': self.fdw_ids}
delete_response = self.tester.delete(self.url +
str(utils.SERVER_GROUP) +
'/' + str(self.server_id) + '/' +
str(self.db_id) + '/',
follow_redirects=True,
data=json.dumps(data),
content_type='html/json')
self.assertEquals(delete_response.status_code, 200)
delete_response = self.delete_multiple(data)
self.assertEquals(delete_response.status_code,
self.expected_data['status_code'])
def tearDown(self):
"""This function disconnect the test database and drop added extension

77
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_dependency_dependent.py Normal file
View File

@ -0,0 +1,77 @@
##########################################################################
#
# 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.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 fdw_utils
class FDWDependencyAndDependentTestCase(BaseTestGenerator):
"""This class will test fdw dependents under test database."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('fdw_dependency_dependent',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(FDWDependencyAndDependentTestCase, 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_{0}".format(str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def get_dependency_dependent(self):
"""
This function returns the fdw dependency and dependent
:return: fdw dependency and dependent
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.fdw_id),
follow_redirects=True)
def runTest(self):
"""This function will fetch foreign data wrapper 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.")
if self.is_positive_test:
response = self.get_dependency_dependent()
expected_response_code = self.expected_data['status_code']
self.assertEquals(response.status_code, expected_response_code)
def tearDown(self):
"""This function disconnect the test database and drop added extension
and dependant objects."""
database_utils.disconnect_database(self, self.server_id,
self.db_id)

83
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get-handlers.py Normal file
View File

@ -0,0 +1,83 @@
##########################################################################
#
# 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.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 fdw_utils
from unittest.mock import patch
class FDWHandlersTestCase(BaseTestGenerator):
"""This class will test fdw handler under test database."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('fdw_get_handlers',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(FDWHandlersTestCase, 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_{0}".format(str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def get_handlers_fdw(self):
"""
This function gets fdw handlers
:return: fdw handlers
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/',
content_type='html/json')
def runTest(self):
"""This function will fetch foreign data wrapper 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.")
if self.is_positive_test:
response = self.get_handlers_fdw()
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.get_handlers_fdw()
self.assertEquals(response.status_code,
self.expected_data['status_code'])
def tearDown(self):
"""This function disconnect the test database and drop added extension
and dependant objects."""
database_utils.disconnect_database(self, self.server_id,
self.db_id)

61
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get.py
View File

@ -17,16 +17,15 @@ 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 fdw_utils
from unittest.mock import patch
class FDWDGetTestCase(BaseTestGenerator):
""" This class will add foreign data wrappers under test database. """
""" This class will test fdw properties
and list API under test database. """
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign_data_wrapper node.
('Check FDW Node',
dict(url='/browser/foreign_data_wrapper/obj/'))
]
scenarios = utils.generate_scenarios('fdw_get',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@ -40,6 +39,28 @@ class FDWDGetTestCase(BaseTestGenerator):
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def get_fdw(self):
"""
This functions returns the fdw properties
:return: fdw properties
"""
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 get_fdw_list(self):
"""
This functions returns the fdw list
:return: fdw list
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' + str(
self.server_id) + '/' +
str(self.db_id) + '/',
content_type='html/json')
def runTest(self):
"""This function will fetch foreign data wrapper present under test
database."""
@ -49,12 +70,28 @@ class FDWDGetTestCase(BaseTestGenerator):
self.db_id)
if not db_con["info"] == "Database connected.":
raise Exception("Could not connect to database.")
response = 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')
self.assertEquals(response.status_code, 200)
if self.is_positive_test:
if hasattr(self, "fdw_list"):
response = self.get_fdw_list()
else:
response = self.get_fdw()
else:
if hasattr(self, "error_fetching_fdw"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "fdw_list"):
response = self.get_fdw_list()
else:
response = self.get_fdw()
if hasattr(self, "wrong_fdw_id"):
self.fdw_id = 99999
response = self.get_fdw()
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 delete the FDW and disconnect the test database """

105
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get_nodes.py Normal file
View File

@ -0,0 +1,105 @@
##########################################################################
#
# 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.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 fdw_utils
from unittest.mock import patch
class FDWDNodesTestCase(BaseTestGenerator):
"""This class will delete foreign data wrappers under test database."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('fdw_get_nodes_and_node',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(FDWDNodesTestCase, 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_{0}".format(str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def get_fdw_nodes(self):
"""
This function returns fdw nodes
:return: fdw nodes
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/',
content_type='html/json')
def get_fdw_node(self):
"""
This functions returns the fdw node
:return: fdw node
"""
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 data wrapper 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.")
if self.is_positive_test:
if hasattr(self, "node"):
response = self.get_fdw_node()
else:
response = self.get_fdw_nodes()
else:
if hasattr(self, "error_fetching_fdw"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
if hasattr(self, "node"):
response = self.get_fdw_node()
else:
response = self.get_fdw_nodes()
if hasattr(self, "wrong_id"):
self.fdw_id = 99999
response = self.get_fdw_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 extension
and dependant objects."""
database_utils.disconnect_database(self, self.server_id,
self.db_id)

81
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_get_validators.py Normal file
View File

@ -0,0 +1,81 @@
##########################################################################
#
# 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.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 fdw_utils
from unittest.mock import patch
class FDWValidatorsTestCase(BaseTestGenerator):
"""This class will test fdw validators under test database."""
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('fdw_get_validators',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(FDWValidatorsTestCase, 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_{0}".format(str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def get_validators_fdw(self):
"""
This function gets fdw validators
:return: fdw validators
"""
return self.tester.get(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/',
content_type='html/json')
def runTest(self):
"""This function will fetch foreign data wrapper 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.")
if self.is_positive_test:
response = self.get_validators_fdw()
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.get_validators_fdw()
self.assertEquals(response.status_code,
self.expected_data['status_code'])
def tearDown(self):
"""This function delete the FDW and disconnect the test database """
database_utils.disconnect_database(self, self.server_id,
self.db_id)

72
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_msql.py Normal file
View File

@ -0,0 +1,72 @@
##########################################################################
#
# 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.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 fdw_utils
from unittest.mock import patch
class FDWDGetMSQLTestCase(BaseTestGenerator):
""" This class will test fdw msql under test database. """
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('fdw_get_msql',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(FDWDGetMSQLTestCase, 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_{0}".format(str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def get_fdw_msql(self):
"""
This functions returns the fdw msql
:return: fdw sql
"""
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 data wrapper 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.")
if self.is_positive_test:
response = self.get_fdw_msql()
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 database to delete it
database_utils.disconnect_database(self, self.server_id, self.db_id)

46
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_put.py
View File

@ -18,17 +18,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 fdw_utils
from unittest.mock import patch
class FDWDPutTestCase(BaseTestGenerator):
"""This class will update foreign data wrappers under test database."""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign_data_wrapper node.
('Check FDW Node',
dict(url='/browser/foreign_data_wrapper/obj/'))
]
scenarios = utils.generate_scenarios('fdw_update',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
@ -42,6 +39,18 @@ class FDWDPutTestCase(BaseTestGenerator):
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def update_fdw(self):
"""
This function returns the fdw update response
:return: fdw update response
"""
return self.tester.put(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.fdw_id),
data=json.dumps(self.test_data),
follow_redirects=True)
def runTest(self):
""" This function will fetch foreign data wrapper present under
test database. """
@ -55,20 +64,21 @@ class FDWDPutTestCase(BaseTestGenerator):
self.fdw_name)
if not fdw_response:
raise Exception("Could not find FDW.")
data = {
"description": "This is FDW update comment",
"id": self.fdw_id
}
put_response = self.tester.put(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' + str(self.fdw_id),
data=json.dumps(data),
follow_redirects=True)
self.assertEquals(put_response.status_code, 200)
self.test_data['id'] = self.fdw_id
if self.is_positive_test:
put_response = self.update_fdw()
else:
if hasattr(self, "internal_server_error"):
with patch(self.mock_data["function_name"],
return_value=eval(self.mock_data["return_value"])):
put_response = self.update_fdw()
self.assertEquals(put_response.status_code,
self.expected_data['status_code'])
def tearDown(self):
"""This function delete the FDW and disconnect the test database """
fdw_utils.delete_fdw(self.server, self.db_name, self.fdw_name)
database_utils.disconnect_database(self, self.server_id,
self.db_id)

82
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/test_fdw_sql.py Normal file
View File

@ -0,0 +1,82 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from __future__ import print_function
import uuid
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 fdw_utils
from unittest.mock import patch
class FDWDGetSQLTestCase(BaseTestGenerator):
""" This class will add foreign data wrappers under test database. """
skip_on_database = ['gpdb']
scenarios = utils.generate_scenarios('fdw_get_sql',
fdw_utils.test_cases)
def setUp(self):
""" This function will create extension and foreign data wrapper."""
super(FDWDGetSQLTestCase, 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_{0}".format(str(uuid.uuid4())[1:8])
self.fdw_id = fdw_utils.create_fdw(self.server, self.db_name,
self.fdw_name)
def get_fdw_sql(self):
"""
This functions returns the collation sql
:return: collation sql
"""
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 data wrapper 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.")
if self.is_positive_test:
response = self.get_fdw_sql()
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.get_fdw_sql()
if hasattr(self, "wrong_fdw_id"):
self.fdw_id = 99999
response = self.get_fdw_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):
# Disconnect database to delete it
database_utils.disconnect_database(self, self.server_id, self.db_id)

9
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/tests/utils.py
View File

@ -12,9 +12,15 @@ from __future__ import print_function
import sys
import traceback
import uuid
import json
import os
from regression.python_test_utils.test_utils import get_db_connection
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/fdw_test_data.json") as data_file:
test_cases = json.load(data_file)
def get_fdw_data(schema_name, db_user):
data = {
@ -63,7 +69,8 @@ def create_fdw(server, db_name, fdw_name):
old_isolation_level = connection.isolation_level
connection.set_isolation_level(0)
pg_cursor = connection.cursor()
pg_cursor.execute('''CREATE FOREIGN DATA WRAPPER "%s"''' % fdw_name)
pg_cursor.execute('''CREATE FOREIGN DATA WRAPPER "%s"
OPTIONS (op1 '5')''' % fdw_name)
connection.set_isolation_level(old_isolation_level)
connection.commit()
# Get 'oid' from newly created foreign data wrapper