mirror of
https://github.com/pgadmin-org/pgadmin4.git
synced 2025-02-25 18:55:31 -06:00
Add regression tests for Foreignb Tables.
This commit is contained in:
Priyanka Shendge
Dave Page
parent
8b91babdc2
commit
540c389d92
@ -0,0 +1,16 @@
|
||||
# #################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
# ##################################################################
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
|
||||
|
||||
class ForeignTableGeneratorTestCase(BaseTestGenerator):
|
||||
|
||||
def runTest(self):
|
||||
return []
|
||||
@ -0,0 +1,100 @@
|
||||
# #################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
# ##################################################################
|
||||
from __future__ import print_function
|
||||
import uuid
|
||||
import json
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from pgadmin.browser.server_groups.servers.databases.tests import \
|
||||
utils as database_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
|
||||
tests import utils as fdw_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
|
||||
foreign_servers.tests import utils as fsrv_utils
|
||||
from regression import parent_node_dict
|
||||
from regression import test_utils as utils
|
||||
|
||||
|
||||
class ForeignTableAddTestCase(BaseTestGenerator):
|
||||
"""
|
||||
This class will add foreign table under database node.
|
||||
"""
|
||||
|
||||
scenarios = [
|
||||
# Fetching default URL for foreign server node.
|
||||
('Check foreign table Node', dict(url='/browser/foreign-table/obj/'))
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
""" This function will create foreign data wrapper and
|
||||
foreign server. """
|
||||
|
||||
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']
|
||||
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:4])
|
||||
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:4])
|
||||
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 runTest(self):
|
||||
"""This function will add foreign table 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.")
|
||||
|
||||
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.")
|
||||
|
||||
data = {
|
||||
"acl": [],
|
||||
"basensp": self.schema_name,
|
||||
"columns":
|
||||
[
|
||||
{
|
||||
"attname": "ename",
|
||||
"datatype": "text",
|
||||
"coloptions": []
|
||||
}
|
||||
],
|
||||
"constraints": [],
|
||||
"ftoptions": [],
|
||||
"inherits": [],
|
||||
"ftsrvname": self.fsrv_name,
|
||||
"name": "ft_%s" % (str(uuid.uuid4())[1:4]),
|
||||
"owner": self.server["username"],
|
||||
"relacl": [],
|
||||
"seclabels": [],
|
||||
"stracl": []
|
||||
}
|
||||
|
||||
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')
|
||||
|
||||
self.assertEquals(response.status_code, 200)
|
||||
|
||||
def tearDown(self):
|
||||
""" This function disconnect the test database. """
|
||||
|
||||
database_utils.disconnect_database(self, self.server_id, self.db_id)
|
||||
@ -0,0 +1,90 @@
|
||||
# #################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
# ##################################################################
|
||||
from __future__ import print_function
|
||||
import uuid
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from pgadmin.browser.server_groups.servers.databases.tests import \
|
||||
utils as database_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
|
||||
tests import utils as fdw_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
|
||||
foreign_servers.tests import utils as fsrv_utils
|
||||
from regression import parent_node_dict
|
||||
from regression import test_utils as utils
|
||||
from . import utils as ft_utils
|
||||
|
||||
|
||||
class ForeignTableDeleteTestCase(BaseTestGenerator):
|
||||
"""
|
||||
This class will delete foreign table under database node.
|
||||
"""
|
||||
|
||||
scenarios = [
|
||||
# Fetching default URL for foreign table node.
|
||||
('Check foreign table Node', dict(url='/browser/foreign-table/obj/'))
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
""" This function will create foreign data wrapper, foreign server
|
||||
and foreign table. """
|
||||
|
||||
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']
|
||||
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:4])
|
||||
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:4])
|
||||
self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:4])
|
||||
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 delete foreign table 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.")
|
||||
|
||||
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.")
|
||||
|
||||
ft_response = ft_utils.verify_foreign_table(self.server, self.db_name,
|
||||
self.fsrv_name)
|
||||
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)
|
||||
|
||||
self.assertEquals(delete_response.status_code, 200)
|
||||
|
||||
def tearDown(self):
|
||||
""" This function disconnect the test database. """
|
||||
|
||||
database_utils.disconnect_database(self, self.server_id, self.db_id)
|
||||
@ -0,0 +1,84 @@
|
||||
# #################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
# ##################################################################
|
||||
from __future__ import print_function
|
||||
import uuid
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from pgadmin.browser.server_groups.servers.databases.tests import \
|
||||
utils as database_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
|
||||
tests import utils as fdw_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers.\
|
||||
foreign_servers.tests import utils as fsrv_utils
|
||||
from regression import parent_node_dict
|
||||
from regression import test_utils as utils
|
||||
from . import utils as ft_utils
|
||||
|
||||
|
||||
class ForeignTableGetTestCase(BaseTestGenerator):
|
||||
"""
|
||||
This class will fetch foreign table under database node.
|
||||
"""
|
||||
scenarios = [
|
||||
# Fetching default URL for foreign server node.
|
||||
('Check foreign table Node', dict(url='/browser/foreign-table/obj/'))
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
""" This function will create foreign data wrapper, foreign server
|
||||
and foreign table. """
|
||||
|
||||
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']
|
||||
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:4])
|
||||
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:4])
|
||||
self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:4])
|
||||
|
||||
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."""
|
||||
|
||||
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.")
|
||||
|
||||
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')
|
||||
|
||||
self.assertEquals(response.status_code, 200)
|
||||
|
||||
def tearDown(self):
|
||||
""" This function disconnect the test database. """
|
||||
|
||||
database_utils.disconnect_database(self, self.server_id, self.db_id)
|
||||
@ -0,0 +1,98 @@
|
||||
# #################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
# ##################################################################
|
||||
from __future__ import print_function
|
||||
import uuid
|
||||
import json
|
||||
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from pgadmin.browser.server_groups.servers.databases.tests import \
|
||||
utils as database_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
|
||||
tests import utils as fdw_utils
|
||||
from pgadmin.browser.server_groups.servers.databases.foreign_data_wrappers. \
|
||||
foreign_servers.tests import utils as fsrv_utils
|
||||
from regression import parent_node_dict
|
||||
from regression import test_utils as utils
|
||||
from . import utils as ft_utils
|
||||
|
||||
|
||||
class ForeignTablePutTestCase(BaseTestGenerator):
|
||||
"""
|
||||
This class will fetch foreign table under database node.
|
||||
"""
|
||||
scenarios = [
|
||||
# Fetching default URL for foreign server node.
|
||||
('Check foreign table Node', dict(url='/browser/foreign-table/obj/'))
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
""" This function will create foreign data wrapper, foreign server
|
||||
and foreign table. """
|
||||
|
||||
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']
|
||||
self.fdw_name = "fdw_%s" % (str(uuid.uuid4())[1:4])
|
||||
self.fsrv_name = "fsrv_%s" % (str(uuid.uuid4())[1:4])
|
||||
self.ft_name = "ft_%s" % (str(uuid.uuid4())[1:4])
|
||||
|
||||
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 update foreign table 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.")
|
||||
|
||||
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.")
|
||||
|
||||
ft_response = ft_utils.verify_foreign_table(self.server, self.db_name,
|
||||
self.fsrv_name)
|
||||
if not ft_response:
|
||||
raise Exception("Could not find Foreign Table.")
|
||||
|
||||
data = \
|
||||
{
|
||||
"description": "This is foreign table update comment",
|
||||
"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)
|
||||
|
||||
self.assertEquals(put_response.status_code, 200)
|
||||
|
||||
def tearDown(self):
|
||||
""" This function disconnect the test database. """
|
||||
|
||||
database_utils.disconnect_database(self, self.server_id, self.db_id)
|
||||
@ -0,0 +1,81 @@
|
||||
# #################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
# ##################################################################
|
||||
|
||||
from __future__ import print_function
|
||||
import os
|
||||
import sys
|
||||
import traceback
|
||||
from regression.test_utils import get_db_connection
|
||||
|
||||
file_name = os.path.basename(__file__)
|
||||
|
||||
|
||||
def create_foreign_table(server, db_name, schema_name, fsrv_name,
|
||||
foreign_table_name):
|
||||
"""
|
||||
This function will create foreign table under the existing
|
||||
dummy schema.
|
||||
|
||||
:param server: test_server, test_db, fsrv_name, foreign_table_name
|
||||
:return: ft_id
|
||||
"""
|
||||
|
||||
try:
|
||||
connection = get_db_connection(db_name,
|
||||
server['username'],
|
||||
server['db_password'],
|
||||
server['host'],
|
||||
server['port'])
|
||||
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)
|
||||
|
||||
connection.set_isolation_level(old_isolation_level)
|
||||
connection.commit()
|
||||
|
||||
# Get 'oid' from newly created foreign table
|
||||
pg_cursor.execute(
|
||||
"SELECT ftrelid FROM pg_foreign_table WHERE ftserver = "
|
||||
"(SELECT oid FROM pg_foreign_server WHERE srvname = '%s') ORDER BY "
|
||||
"ftrelid ASC limit 1" % fsrv_name)
|
||||
|
||||
oid = pg_cursor.fetchone()
|
||||
ft_id = ''
|
||||
if oid:
|
||||
ft_id = oid[0]
|
||||
connection.close()
|
||||
return ft_id
|
||||
except Exception:
|
||||
traceback.print_exc(file=sys.stderr)
|
||||
|
||||
|
||||
def verify_foreign_table(server, db_name, fsrv_name):
|
||||
""" This function will verify current foreign table."""
|
||||
|
||||
try:
|
||||
connection = get_db_connection(db_name,
|
||||
server['username'],
|
||||
server['db_password'],
|
||||
server['host'],
|
||||
server['port'])
|
||||
pg_cursor = connection.cursor()
|
||||
|
||||
pg_cursor.execute(
|
||||
"SELECT ftrelid FROM pg_foreign_table WHERE ftserver = "
|
||||
"(SELECT oid FROM pg_foreign_server WHERE srvname = '%s') ORDER BY "
|
||||
"ftrelid ASC limit 1" % fsrv_name)
|
||||
fts = pg_cursor.fetchone()
|
||||
connection.close()
|
||||
return fts
|
||||
except Exception:
|
||||
traceback.print_exc(file=sys.stderr)
|
||||
Loading…
Reference in New Issue
Block a user