pgadmin-org
/
pgadmin4
mirror of https://github.com/pgadmin-org/pgadmin4.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve code coverage and API test cases for Foreign Tables. Fixes #5328

pull/37/head
Yogesh Mahajan 2020-11-26 11:52:06 +05:30 committed by Akshay Joshi
parent 55e50144f3
commit b54cf2edca
12 changed files with 1345 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/en_US/release_notes_4_29.rst
View File

@ -13,6 +13,7 @@ New features
Housekeeping
************
| `Issue #5328 <https://redmine.postgresql.org/issues/5328>`_ - Improve code coverage and API test cases for Foreign Tables.
| `Issue #5337 <https://redmine.postgresql.org/issues/5337>`_ - Improve code coverage and API test cases for Views and Materialized Views.
| `Issue #6033 <https://redmine.postgresql.org/issues/6033>`_ - Update the cryptography python package for Python 3.5 and above.

602
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/foreign_tables_test_data.json Normal file
View File

@ -0,0 +1,602 @@
{
"foreign_table_create": [
{
"name": "Create Foreign Table: With valid data.",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"acl": [],
"basensp": "schema_name",
"columns": [
{
"attname": "ename",
"datatype": "text",
"coloptions": []
}
],
"constraints": [],
"ftoptions": [],
"inherits": [],
"ftsrvname": "fsrv_name",
"name": "ft_name",
"owner": "server user name",
"relacl": [],
"seclabels": [],
"stracl": []
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
}
},
{
"name": "Create Foreign Table: With insufficient data.",
"is_positive_test": false,
"inventory_data": {},
"test_data": {
"acl": [],
"basensp": "schema_name",
"columns": [
{
"attname": "ename",
"datatype": "text",
"coloptions": []
}
],
"constraints": [],
"ftoptions": [],
"inherits": [],
"name": "ft_name",
"owner": "server user name",
"relacl": [],
"seclabels": [],
"stracl": []
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410,
"error_msg": "Could not find the required parameter (ftsrvname).",
"test_result_data": {}
}
},
{
"name": "Create Foreign Table: With valid data while server down.",
"is_positive_test": false,
"inventory_data": {},
"test_data": {
"acl": [],
"basensp": "schema_name",
"columns": [
{
"attname": "ename",
"datatype": "text",
"coloptions": []
}
],
"constraints": [],
"ftoptions": [],
"inherits": [],
"ftsrvname": "fsrv_name",
"name": "ft_name",
"owner": "server user name",
"relacl": [],
"seclabels": [],
"stracl": []
},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
}
}
],
"foreign_table_get": [
{
"name": "Get Foreign Table: With existing foreign table having precision data type column.",
"is_positive_test": true,
"inventory_data": {
"query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; \" %(schema_name, foreign_table_name, fsrv_name)"
},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table list: With existing foreign table.",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": true
},
{
"name": "Get Foreign Table: With Non-existing foreign table.",
"is_positive_test": false,
"inventory_data": {},
"test_data": {
"ft_id": 999999
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410,
"error_msg": "could not find the specified foreign table.",
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table: With existing foreign table while server is down.",
"is_positive_test": false,
"inventory_data": {},
"test_data": {},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table list: With existing foreign table while server is down.",
"is_positive_test": false,
"inventory_data": {},
"test_data": {},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
},
"is_list": true
}
],
"foreign_table_delete": [
{
"name": "Delete Foreign Table: With existing foreign table.",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Delete Multiple Foreign Tables: With existing foreign tables.",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": true
},
{
"name": "Delete Foreign Table: With existing foreign table while server is down.",
"is_positive_test": false,
"inventory_data": {},
"test_data": {},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
},
"is_list": false
}
],
"foreign_table_put": [
{
"name": "Update Foreign Table: With existing foreign table.",
"is_positive_test": true,
"inventory_data": {
"query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) OPTIONS (columnoptionkey 'columnOptionVal') NULL ) SERVER %s;\" %(schema_name, foreign_table_name, fsrv_name)"
},
"test_data": {
"description": "This is foreign table update comment",
"fsrvacl": {
"added": [
{
"grantee": "PUBLIC",
"grantor": "postgres",
"privileges": [
{
"privilege_type": "U",
"privilege": true,
"with_grant": false
}
]
}
]
},
"columns": {
"added": [
{
"attname": "col2",
"datatype": "character varying[]",
"coloptions": []
}
],
"changed": [
{
"coloptions": [
{
"option": "OptionKey",
"value": "OptionValue"
}
],
"attname": "emp",
"datatype": "\"char\"",
"typdefault": null,
"attnotnull": false,
"collname": "pg_catalog.\"default\"",
"attnum": 1,
"inheritedfrom": null,
"inheritedid": null,
"attstattarget": -1,
"attoptions": null,
"attfdwoptions": [
"ColOptionKey=ColOptionValue"
],
"attndims": 0,
"atttypmod": -1,
"fulltype": "text",
"isdup": false
}
]
}
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Update Foreign Table: With existing foreign table while server is down.",
"is_positive_test": false,
"inventory_data": {},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
},
"is_list": false
}
],
"foreign_table_sql": [
{
"name": "Get Foreign Table Sql: With existing foreign table.",
"is_positive_test": true,
"url": "/browser/foreign_table/sql/",
"inventory_data": {
"query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, foreign_table_name, fsrv_name, schema_name, foreign_table_name)"
},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table Sql: With existing foreign table having inheritted table.",
"is_positive_test": true,
"url": "/browser/foreign_table/sql/",
"inventory_data": {
"query": "\"CREATE TABLE %s.toinherittable (col1 character varying(10) NULL); CREATE FOREIGN TABLE %s.%s () INHERITS (%s.toinherittable) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, schema_name, foreign_table_name, schema_name, fsrv_name, schema_name, foreign_table_name)"
},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table Sql: With existing foreign table while server is down.",
"is_positive_test": false,
"url": "/browser/foreign_table/sql/",
"inventory_data": {},
"test_data": {},
"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,
"error_msg": "Mocked Internal Server Error",
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table Sql: With Non-existing foreign table.",
"is_positive_test": false,
"url": "/browser/foreign_table/sql/",
"inventory_data": {},
"test_data": {
"ft_id": 999999
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 410,
"error_msg": "could not find the specified foreign table.",
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table Insert Sql: With existing foreign table.",
"is_positive_test": true,
"url": "/browser/foreign_table/insert_sql/",
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table Update Sql: With existing foreign table.",
"is_positive_test": true,
"url": "/browser/foreign_table/update_sql/",
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table Delete Sql: With existing foreign table.",
"is_positive_test": true,
"url": "/browser/foreign_table/delete_sql/",
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table Insert Sql: With existing foreign table.",
"is_positive_test": true,
"url": "/browser/foreign_table/select_sql/",
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
}
],
"foreign_table_dependencies_dependents": [
{
"name": "Get Foreign Table Dependents: With existing foreign table.",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_dependent": true
},
{
"name": "Get Foreign Table Dependency: With existing foreign table.",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_dependent": false
}
],
"foreign_table_get_nodes": [
{
"name": "Get Foreign Table Nodes: With existing foreign table.",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table list node: With existing foreign table.",
"is_positive_test": true,
"inventory_data": {},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": true
}
],
"foreign_table_get_msql": [
{
"name": "Get Foreign Table msql: With existing foreign table.",
"is_positive_test": true,
"inventory_data": {},
"test_data": {
"description": "Testing Msql API"
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table msql: Modify existing foreign table by inheriting.",
"is_positive_test": true,
"inventory_data": {
"query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; \" %(schema_name, foreign_table_name, fsrv_name)"
},
"test_data": {
"description": "Testing Msql API",
"inherits": [
"table_id"
]
},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
}
],
"foreign_table_get_creation_parameters": [
{
"name": "Get Foreign Table Pre Creation parameters: Get collation.",
"is_positive_test": true,
"url": "/browser/foreign_table/get_collations/",
"inventory_data": {
"query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, foreign_table_name, fsrv_name, schema_name, foreign_table_name)"
},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table Pre Creation parameters: Get tables.",
"is_positive_test": true,
"url": "/browser/foreign_table/get_tables/",
"inventory_data": {
"query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, foreign_table_name, fsrv_name, schema_name, foreign_table_name)"
},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
},
{
"name": "Get Foreign Table Pre Creation parameters: Get types.",
"is_positive_test": true,
"url": "/browser/foreign_table/get_types/",
"inventory_data": {
"query": "\"CREATE FOREIGN TABLE %s.%s (col1 character varying(10) NULL COLLATE pg_catalog.default) SERVER %s; GRANT SELECT ON TABLE %s.%s TO PUBLIC;\" %(schema_name, foreign_table_name, fsrv_name, schema_name, foreign_table_name)"
},
"test_data": {},
"mocking_required": false,
"mock_data": {},
"expected_data": {
"status_code": 200,
"error_msg": null,
"test_result_data": {}
},
"is_list": false
}
]
}

85
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_add.py
View File

@ -8,8 +8,8 @@
##########################################################################
import json
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
foreign_servers.tests import utils as fsrv_utils
@ -29,22 +29,32 @@ class ForeignTableAddTestCase(BaseTestGenerator):
"""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign server node.
('Check foreign table Node', dict(url='/browser/foreign_table/obj/'))
]
# url
url = '/browser/foreign_table/obj/'
# Generates scenarios
scenarios = utils.generate_scenarios("foreign_table_create",
ft_utils.test_cases)
def setUp(self):
""" This function will create foreign data wrapper and
foreign server. """
super(ForeignTableAddTestCase, self).setUp()
# Load test data
self.data = self.test_data
# Get parent schema info
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"]
# Create schema
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
# Create FDW & server
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,
@ -53,9 +63,14 @@ class ForeignTableAddTestCase(BaseTestGenerator):
self.fsrv_name, self.fdw_name)
self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:8])
self.data['basensp'] = self.schema_name
self.data['ftsrvname'] = self.fsrv_name
self.data['name'] = self.ft_name
self.data['owner'] = self.server["username"]
def runTest(self):
"""This function will add foreign table under test database."""
# Create db connection
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
@ -69,40 +84,40 @@ class ForeignTableAddTestCase(BaseTestGenerator):
if not fsrv_response:
raise Exception("Could not find Foreign Server.")
data = {
"acl": [],
"basensp": self.schema_name,
"columns": [
{
"attname": "ename",
"datatype": "text",
"coloptions": []
}
],
"constraints": [],
"ftoptions": [],
"inherits": [],
"ftsrvname": self.fsrv_name,
"name": self.ft_name,
"owner": self.server["username"],
"relacl": [],
"seclabels": [],
"stracl": []
}
if self.is_positive_test:
response = ft_utils.api_create(self)
response = self.tester.post(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' + str(self.db_id) + '/' +
str(self.schema_id) + '/', data=json.dumps(data),
content_type='html/json')
# Assert response
utils.assert_status_code(self, response)
self.assertEqual(response.status_code, 200)
# Verify in backend
cross_check_res = ft_utils.verify_foreign_table(self.server,
self.db_name,
self.fsrv_name)
self.assertIsNotNone(cross_check_res, "Could not find the newly"
" created foreign table.")
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=eval(self.mock_data["return_value"])):
response = ft_utils.api_create(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
else:
del self.data['ftsrvname']
response = ft_utils.api_create(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
""" This function disconnect the test database and delete test
foreign table object. """
ft_utils.delete_foregin_table(self.server, self.db_name,
self.schema_name, self.ft_name
)
if self.is_positive_test:
ft_utils.delete_foregin_table(self.server, self.db_name,
self.schema_name, self.ft_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)

43
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_delete.py
View File

@ -9,6 +9,7 @@
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
foreign_servers.tests import utils as fsrv_utils
@ -28,22 +29,30 @@ class ForeignTableDeleteTestCase(BaseTestGenerator):
"""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign table node.
('Check foreign table Node', dict(url='/browser/foreign_table/obj/'))
]
# url
url = '/browser/foreign_table/obj/'
# Generates scenarios
scenarios = utils.generate_scenarios("foreign_table_delete",
ft_utils.test_cases)
def setUp(self):
""" This function will create foreign data wrapper, foreign server
and foreign table. """
super(ForeignTableDeleteTestCase, self).setUp()
# Load test data
self.data = self.test_data
# Get parent schema info
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.schema_id = self.schema_data['schema_id']
# Create FDW, server & table
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:8])
@ -56,6 +65,16 @@ class ForeignTableDeleteTestCase(BaseTestGenerator):
self.fsrv_name,
self.ft_name)
# In case of multiple foreign tables
if self.is_list:
self.ft_name_2 = "ft_%s" % (str(uuid.uuid4())[1:8])
self.ft_id_2 = ft_utils.create_foreign_table(self.server,
self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name_2)
self.data['ids'] = [self.ft_id, self.ft_id_2]
def runTest(self):
"""This function will delete foreign table under test database."""
@ -78,15 +97,15 @@ class ForeignTableDeleteTestCase(BaseTestGenerator):
if not ft_response:
raise Exception("Could not find Foreign Table.")
delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(self.ft_id),
follow_redirects=True)
if self.is_positive_test:
self.assertEqual(delete_response.status_code, 200)
if self.is_list:
response = ft_utils.api_delete(self)
else:
response = ft_utils.api_delete(self, self.ft_id)
# Assert response
utils.assert_status_code(self, response)
def tearDown(self):
""" This function disconnect the test database. """

104
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_dependencies_dependents.py Normal file
View File

@ -0,0 +1,104 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.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 ft_utils
class ForeignTableDependenciesDependentsTestCase(BaseTestGenerator):
"""
This class will fetch foreign table under database node.
"""
skip_on_database = ['gpdb']
# url
url = '/browser/foreign_table/'
# Generates scenarios
scenarios = utils.generate_scenarios(
"foreign_table_dependencies_dependents", ft_utils.test_cases)
def setUp(self):
""" This function will create foreign data wrapper, foreign server
and foreign table. """
super(ForeignTableDependenciesDependentsTestCase, self).setUp()
# Load test data
self.data = self.test_data
# Get parent schema info
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"]
# Create schema
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
# Create FDW, server & table
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.ft_name = "ft_%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.ft_id = ft_utils.create_foreign_table(self.server, self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name)
def runTest(self):
"""This function will fetch foreign table under test database."""
# Create db connection
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.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find Foreign Server.")
if self.is_positive_test:
if self.is_dependent:
self.url = self.url + 'dependent/'
response = ft_utils.api_get(self)
else:
self.url = self.url + 'dependency/'
response = ft_utils.api_get(self)
utils.assert_status_code(self, response)
def tearDown(self):
""" This function disconnect the test database and delete test
foreign table object. """
ft_utils.delete_foregin_table(self.server, self.db_name,
self.schema_name, self.ft_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)

83
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get.py
View File

@ -9,6 +9,7 @@
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
foreign_servers.tests import utils as fsrv_utils
@ -28,22 +29,32 @@ class ForeignTableGetTestCase(BaseTestGenerator):
"""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign server node.
('Check foreign table Node', dict(url='/browser/foreign_table/obj/'))
]
# url
url = '/browser/foreign_table/obj/'
# Generates scenarios
scenarios = utils.generate_scenarios("foreign_table_get",
ft_utils.test_cases)
def setUp(self):
""" This function will create foreign data wrapper, foreign server
and foreign table. """
super(ForeignTableGetTestCase, self).setUp()
# Load test data
self.data = self.test_data
# Get parent schema info
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"]
# Create schema
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
# Create FDW, server & table
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:8])
@ -52,14 +63,35 @@ class ForeignTableGetTestCase(BaseTestGenerator):
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
self.ft_id = ft_utils.create_foreign_table(self.server, self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name)
if bool(self.inventory_data):
query = self.inventory_data['query']
self.ft_id = ft_utils.create_foreign_table(self.server,
self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name, query)
else:
self.ft_id = ft_utils.create_foreign_table(self.server,
self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name)
# In case of multiple foreign tables
if self.is_list:
self.ft_name_2 = "ft_%s" % (str(uuid.uuid4())[1:8])
self.ft_id_2 = ft_utils.create_foreign_table(self.server,
self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name_2)
self.data['ids'] = [self.ft_id, self.ft_id_2]
def runTest(self):
"""This function will fetch foreign table under test database."""
# Create db connection
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
@ -74,20 +106,35 @@ class ForeignTableGetTestCase(BaseTestGenerator):
if not fsrv_response:
raise Exception("Could not find Foreign Server.")
response = self.tester.get(self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(self.ft_id),
content_type='html/json')
if self.is_positive_test:
if self.is_list:
response = ft_utils.api_get(self, '')
else:
response = ft_utils.api_get(self)
self.assertEqual(response.status_code, 200)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
if self.is_list:
response = ft_utils.api_get(self, '')
else:
response = ft_utils.api_get(self)
elif 'ft_id' in self.data:
# Non-existing ft_id id
self.ft_id = self.data["ft_id"]
response = ft_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
""" This function disconnect the test database and delete test
foreign table object. """
ft_utils.delete_foregin_table(self.server, self.db_name,
self.schema_name, self.ft_name
)
self.schema_name, self.ft_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)

111
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_msql.py Normal file
View File

@ -0,0 +1,111 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
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 ft_utils
from pgadmin.browser.server_groups.servers.databases.schemas.tables.tests \
import utils as table_utils
class ForeignTableGetMsqlTestCase(BaseTestGenerator):
"""
This class will fetch foreign table msql under database node.
"""
skip_on_database = ['gpdb']
# url
url = '/browser/foreign_table/msql/'
# Generates scenarios
scenarios = utils.generate_scenarios("foreign_table_get_msql",
ft_utils.test_cases)
def setUp(self):
""" This function will create foreign data wrapper, foreign server
and foreign table. """
super(ForeignTableGetMsqlTestCase, self).setUp()
# Load test data
self.data = self.test_data
# Get parent schema info
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"]
# Create schema
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
# Create FDW, server & table
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.ft_name = "ft_%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.ft_id = ft_utils.create_foreign_table(self.server, self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name)
def runTest(self):
"""This function will fetch foreign table msql under test database."""
# Create db connection
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.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find Foreign Server.")
if self.is_positive_test:
self.data['oid'] = self.ft_id
if 'inherits' in self.data:
tid = table_utils.create_table(self.server, self.db_name,
self.schema_name, 'toinherit',
"'CREATE TABLE %s.%s ()' "
"%(schema_name, table_name)")
self.data['inherits'] = [tid]
url_encode_data = self.data
response = ft_utils.api_get_msql(self, url_encode_data)
# Assert response
utils.assert_status_code(self, response)
def tearDown(self):
""" This function disconnect the test database and delete test
foreign table object. """
ft_utils.delete_foregin_table(self.server, self.db_name,
self.schema_name, self.ft_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)

130
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_nodes.py Normal file
View File

@ -0,0 +1,130 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.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 ft_utils
class ForeignTableGetNodesTestCase(BaseTestGenerator):
"""
This class will fetch foreign table nodes under database node.
"""
skip_on_database = ['gpdb']
# url
url = '/browser/foreign_table/nodes/'
# Generates scenarios
scenarios = utils.generate_scenarios("foreign_table_get_nodes",
ft_utils.test_cases)
def setUp(self):
""" This function will create foreign data wrapper, foreign server
and foreign table. """
super(ForeignTableGetNodesTestCase, self).setUp()
# Load test data
self.data = self.test_data
# Get parent schema info
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"]
# Create schema
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
# Create FDW, server & table
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.ft_name = "ft_%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.ft_id = ft_utils.create_foreign_table(self.server, self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name)
# In case of multiple foreign tables
if self.is_list:
self.ft_name_2 = "ft_%s" % (str(uuid.uuid4())[1:8])
self.ft_id_2 = ft_utils.create_foreign_table(self.server,
self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name_2)
self.data['ids'] = [self.ft_id, self.ft_id_2]
def runTest(self):
"""This function will fetch foreign table nodes under test database."""
# Create db connection
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.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find Foreign Server.")
if self.is_positive_test:
if self.is_list:
response = ft_utils.api_get(self, '')
else:
response = ft_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
if self.is_list:
response = ft_utils.api_get(self, '')
else:
response = ft_utils.api_get(self)
elif 'ft_id' in self.data:
# Non-existing ft_id id
self.ft_id = self.data["ft_id"]
response = ft_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
""" This function disconnect the test database and delete test
foreign table object. """
ft_utils.delete_foregin_table(self.server, self.db_name,
self.schema_name, self.ft_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)

60
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_delete_multiple.py → web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_get_precreation_parameters.py
View File

@ -9,7 +9,7 @@
import uuid
import json
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
foreign_servers.tests import utils as fsrv_utils
@ -23,48 +23,52 @@ from regression.python_test_utils import test_utils as utils
from . import utils as ft_utils
class ForeignTableDeleteMultipleTestCase(BaseTestGenerator):
class ForeignTableGetTestCase(BaseTestGenerator):
"""
This class will delete foreign table under database node.
This class will fetch foreign table pre creation parameters under
database node.
"""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign table node.
('Check foreign table Node', dict(url='/browser/foreign_table/obj/'))
]
# url
url = '/browser/foreign_table/obj/'
# Generates scenarios
scenarios = utils.generate_scenarios(
"foreign_table_get_creation_parameters", ft_utils.test_cases)
def setUp(self):
""" This function will create foreign data wrapper, foreign server
and foreign table. """
super(ForeignTableDeleteMultipleTestCase, self).setUp()
super(ForeignTableGetTestCase, self).setUp()
# Load test data
self.data = self.test_data
# Get parent schema info
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"]
# Create schema
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
# Create FDW, server & table
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.ft_name = "ft_%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.ft_ids = [ft_utils.create_foreign_table(
self.server, self.db_name,
self.schema_name, self.fsrv_name,
"ft_%s" % (str(uuid.uuid4())[1:8])),
ft_utils.create_foreign_table(
self.server, self.db_name,
self.schema_name, self.fsrv_name,
"ft_%s" % (str(uuid.uuid4())[1:8]))
]
def runTest(self):
"""This function will delete foreign table under test database."""
"""This function will fetch foreign table pre creation
under test database."""
# Create db connection
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
@ -79,20 +83,12 @@ class ForeignTableDeleteMultipleTestCase(BaseTestGenerator):
if not fsrv_response:
raise Exception("Could not find Foreign Server.")
data = {'ids': self.ft_ids}
if self.is_positive_test:
response = ft_utils.api_get(self, '')
delete_response = self.tester.delete(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/',
content_type='html/json',
follow_redirects=True,
data=json.dumps(data))
self.assertEqual(delete_response.status_code, 200)
# Assert response
utils.assert_status_code(self, response)
def tearDown(self):
""" This function disconnect the test database. """
""" This function disconnect the test database """
database_utils.disconnect_database(self, self.server_id, self.db_id)

67
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_put.py
View File

@ -10,6 +10,7 @@
import json
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
foreign_servers.tests import utils as fsrv_utils
@ -29,22 +30,30 @@ class ForeignTablePutTestCase(BaseTestGenerator):
"""
skip_on_database = ['gpdb']
scenarios = [
# Fetching default URL for foreign server node.
('Check foreign table Node', dict(url='/browser/foreign_table/obj/'))
]
# url
url = '/browser/foreign_table/obj/'
# Generates scenarios
scenarios = utils.generate_scenarios("foreign_table_put",
ft_utils.test_cases)
def setUp(self):
""" This function will create foreign data wrapper, foreign server
and foreign table. """
super(ForeignTablePutTestCase, self).setUp()
# Load test data
self.data = self.test_data
# Get parent schema info
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.schema_id = self.schema_data['schema_id']
# Create FDW, server & table
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:8])
@ -53,14 +62,23 @@ class ForeignTablePutTestCase(BaseTestGenerator):
self.fdw_name)
self.fsrv_id = fsrv_utils.create_fsrv(self.server, self.db_name,
self.fsrv_name, self.fdw_name)
self.ft_id = ft_utils.create_foreign_table(self.server, self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name)
if bool(self.inventory_data):
query = self.inventory_data['query']
self.ft_id = ft_utils.create_foreign_table(self.server,
self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name, query)
else:
self.ft_id = ft_utils.create_foreign_table(self.server,
self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name)
def runTest(self):
"""This function will update foreign table under test database."""
db_con = database_utils.connect_database(self,
utils.SERVER_GROUP,
self.server_id,
@ -80,28 +98,27 @@ class ForeignTablePutTestCase(BaseTestGenerator):
if not ft_response:
raise Exception("Could not find Foreign Table.")
data = \
{
"description": "This is foreign table update comment",
"id": self.ft_id,
}
self.data['id'] = self.ft_id
put_response = self.tester.put(
self.url + str(utils.SERVER_GROUP) + '/' +
str(self.server_id) + '/' +
str(self.db_id) + '/' +
str(self.schema_id) + '/' +
str(self.ft_id),
data=json.dumps(data),
follow_redirects=True)
if self.is_positive_test:
response = ft_utils.api_put(self)
self.assertEqual(put_response.status_code, 200)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
response = ft_utils.api_put(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
""" This function disconnect the test database and delete test
foreign table object. """
ft_utils.delete_foregin_table(self.server, self.db_name,
self.schema_name, self.ft_name
)
self.schema_name, self.ft_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)

123
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/test_foreign_table_sql.py Normal file
View File

@ -0,0 +1,123 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2020, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from unittest.mock import patch
from pgadmin.browser.server_groups.servers.databases.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 ft_utils
class ForeignTableSqlTestCase(BaseTestGenerator):
"""
This class will fetch foreign table under database node.
"""
skip_on_database = ['gpdb']
# Generates scenarios
scenarios = utils.generate_scenarios("foreign_table_sql",
ft_utils.test_cases)
def setUp(self):
""" This function will create foreign data wrapper, foreign server
and foreign table. """
super(ForeignTableSqlTestCase, self).setUp()
# Load test data
self.data = self.test_data
# Get parent schema info
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"]
# Create schema
self.schema_name = self.schema_data['schema_name']
self.schema_id = self.schema_data['schema_id']
# Create FDW, server & table
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:8])
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:8])
self.ft_name = "ft_%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)
if bool(self.inventory_data):
query = self.inventory_data['query']
self.ft_id = ft_utils.create_foreign_table(self.server,
self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name, query)
else:
self.ft_id = ft_utils.create_foreign_table(self.server,
self.db_name,
self.schema_name,
self.fsrv_name,
self.ft_name)
def runTest(self):
"""This function will fetch foreign table under test database."""
# Create db connection
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.")
fsrv_response = fsrv_utils.verify_fsrv(self.server, self.db_name,
self.fsrv_name)
if not fsrv_response:
raise Exception("Could not find Foreign Server.")
if self.is_positive_test:
response = ft_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
else:
if self.mocking_required:
with patch(self.mock_data["function_name"],
side_effect=[eval(self.mock_data["return_value"])]):
if self.is_list:
response = ft_utils.api_get(self, '')
else:
response = ft_utils.api_get(self)
elif 'ft_id' in self.data:
# Non-existing ft_id id
self.ft_id = self.data["ft_id"]
response = ft_utils.api_get(self)
# Assert response
utils.assert_status_code(self, response)
utils.assert_error_message(self, response)
def tearDown(self):
""" This function disconnect the test database and delete test
foreign table object. """
ft_utils.delete_foregin_table(self.server, self.db_name,
self.schema_name, self.ft_name)
database_utils.disconnect_database(self, self.server_id, self.db_id)

66
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/tests/utils.py
View File

@ -6,19 +6,67 @@
# This software is released under the PostgreSQL Licence
#
##########################################################################
import json
import os
import sys
import traceback
from urllib.parse import urlencode
from regression.python_test_utils.test_utils import get_db_connection
from regression.python_test_utils import test_utils as utils
# Load test data from json file.
CURRENT_PATH = os.path.dirname(os.path.realpath(__file__))
with open(CURRENT_PATH + "/foreign_tables_test_data.json") as data_file:
test_cases = json.load(data_file)
file_name = os.path.basename(__file__)
def api_create(self):
return self.tester.post("{0}{1}/{2}/{3}/{4}/".
format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id, self.schema_id),
data=json.dumps(self.data),
content_type='html/json')
def api_get(self, ft_table_id=None):
if ft_table_id is None:
ft_table_id = self.ft_id
return self.tester.get("{0}{1}/{2}/{3}/{4}/{5}".
format(self.url, utils.SERVER_GROUP, self.server_id,
self.db_id, self.schema_id, ft_table_id),
content_type='html/json')
def api_delete(self, ft_tbl_id=''):
return self.tester.delete("{0}{1}/{2}/{3}/{4}/{5}".
format(self.url, utils.SERVER_GROUP,
self.server_id, self.db_id,
self.schema_id, ft_tbl_id),
data=json.dumps(self.data),
follow_redirects=True)
def api_put(self):
return self.tester.put("{0}{1}/{2}/{3}/{4}/{5}".
format(self.url, utils.SERVER_GROUP, self.server_id,
self.db_id, self.schema_id, self.ft_id),
data=json.dumps(self.data),
follow_redirects=True)
def api_get_msql(self, url_encode_data):
return self.tester.get("{0}{1}/{2}/{3}/{4}/{5}?{6}".
format(self.url, utils.SERVER_GROUP, self.server_id,
self.db_id, self.schema_id, self.ft_id,
urlencode(url_encode_data)),
follow_redirects=True)
def create_foreign_table(server, db_name, schema_name, fsrv_name,
foreign_table_name):
foreign_table_name, sql_query=None):
"""
This function will create foreign table under the existing
dummy schema.
@ -28,6 +76,12 @@ def create_foreign_table(server, db_name, schema_name, fsrv_name,
"""
try:
query = "CREATE FOREIGN TABLE " + schema_name + "." + \
foreign_table_name + "(emp_name text NULL) SERVER %s" % \
fsrv_name
if sql_query is not None:
query = eval(sql_query)
connection = get_db_connection(db_name,
server['username'],
server['db_password'],
@ -36,11 +90,7 @@ def create_foreign_table(server, db_name, schema_name, fsrv_name,
old_isolation_level = connection.isolation_level
connection.set_isolation_level(0)
pg_cursor = connection.cursor()
pg_cursor.execute(
"CREATE FOREIGN TABLE " + schema_name + "." + foreign_table_name +
"(emp_name text NULL) SERVER %s" % fsrv_name)
pg_cursor.execute(query)
connection.set_isolation_level(old_isolation_level)
connection.commit()