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

Add regression tests for resource groups on EPAS.

Browse Source
This commit is contained in:
Navnath Gadakh
2016-10-18 12:01:01 +01:00
committed by Dave Page
parent 8d2dff1482
commit e31ba5873b
6 changed files with 346 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
web/pgadmin/browser/server_groups/servers/resource_groups/tests/__init__.py Normal file
View File

@@ -0,0 +1,15 @@
# ##########################################################################
#
# #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 ResourceGroupsCreateTestCase(BaseTestGenerator):
def runTest(self):
return

57
web/pgadmin/browser/server_groups/servers/resource_groups/tests/test_resource_groups_add.py Normal file
View File

@@ -0,0 +1,57 @@
# #################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
# ##################################################################
import uuid
import json
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
from regression import parent_node_dict
from pgadmin.browser.server_groups.servers.tests import utils as server_utils
from . import utils as resource_groups_utils
class ResourceGroupsAddTestCase(BaseTestGenerator):
"""This class will test the add resource groups API"""
scenarios = [
('Add resource groups', dict(url='/browser/resource_group/obj/'))
]
def setUp(self):
self.server_id = parent_node_dict["server"][-1]["server_id"]
server_con = server_utils.connect_server(self, self.server_id)
if not server_con["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
if "server_type" in server_con["data"]:
if server_con["data"]["server_type"] == "pg":
message = "Resource groups are not supported by PG."
self.skipTest(message)
def runTest(self):
"""This function will add resource groups under server node"""
self.resource_group = "test_resource_group_add%s" % \
str(uuid.uuid4())[1:6]
data = {"name": self.resource_group,
"cpu_rate_limit": 0,
"dirty_rate_limit": 0}
response = self.tester.post(self.url + str(utils.SERVER_GROUP) +
"/" + str(self.server_id) + "/",
data=json.dumps(data),
content_type='html/json')
self.assertEquals(response.status_code, 200)
def tearDown(self):
"""This function delete the resource group from the database."""
connection = utils.get_db_connection(self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'])
resource_groups_utils.delete_resource_group(connection,
self.resource_group)

59
web/pgadmin/browser/server_groups/servers/resource_groups/tests/test_resource_groups_delete.py Normal file
View File

@@ -0,0 +1,59 @@
# #################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
# ##################################################################
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
from regression import parent_node_dict
from pgadmin.browser.server_groups.servers.tests import utils as server_utils
from . import utils as resource_groups_utils
class ResourceGroupsDeleteTestCase(BaseTestGenerator):
"""This class will delete the resource groups"""
scenarios = [
('Delete resource groups', dict(url='/browser/resource_group/obj/'))
]
def setUp(self):
self.server_id = parent_node_dict["server"][-1]["server_id"]
server_response = server_utils.connect_server(self, self.server_id)
if not server_response["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
if "server_type" in server_response["data"]:
if server_response["data"]["server_type"] == "pg":
message = "Resource groupa are not supported by PG."
self.skipTest(message)
self.resource_group = "test_resource_group_delete%s" % \
str(uuid.uuid4())[1:6]
self.resource_group_id = resource_groups_utils.create_resource_groups(
self.server, self.resource_group)
def runTest(self):
"""This function will delete resource groups."""
resource_grp_response = resource_groups_utils.verify_resource_group(
self.server, self.resource_group)
if not resource_grp_response:
raise Exception("Could not find the resource group to fetch.")
response = self.tester.delete(
"{0}{1}/{2}/{3}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.resource_group_id),
follow_redirects=True)
self.assertEquals(response.status_code, 200)
def tearDown(self):
"""This function delete the resource group from the database."""
connection = utils.get_db_connection(self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'])
resource_groups_utils.delete_resource_group(connection,
self.resource_group)

64
web/pgadmin/browser/server_groups/servers/resource_groups/tests/test_resource_groups_put.py Normal file
View File

@@ -0,0 +1,64 @@
# #################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
# ##################################################################
import uuid
import json
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
from regression import parent_node_dict
from pgadmin.browser.server_groups.servers.tests import utils as server_utils
from . import utils as resource_groups_utils
class ResourceGroupsPutTestCase(BaseTestGenerator):
"""This class will update the resource groups"""
scenarios = [
('Put resource groups', dict(url='/browser/resource_group/obj/'))
]
def setUp(self):
self.server_id = parent_node_dict["server"][-1]["server_id"]
server_response = server_utils.connect_server(self, self.server_id)
if not server_response["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
if "server_type" in server_response["data"]:
if server_response["data"]["server_type"] == "pg":
message = "Resource groups are not supported by PG."
self.skipTest(message)
self.resource_group_name = "test_resource_group_put%s" % \
str(uuid.uuid4())[1:6]
self.resource_group_id = resource_groups_utils.create_resource_groups(
self.server, self.resource_group_name)
def runTest(self):
"""This function will get the resource groups."""
resource_grp_response = resource_groups_utils.verify_resource_group(
self.server, self.resource_group_name)
if not resource_grp_response:
raise Exception("Could not find the resource group to fetch.")
self.resource_group_name = "test_resource_group_put%s" % \
str(uuid.uuid4())[1:6]
data = {"id": self.resource_group_id,
"name": self.resource_group_name}
response = self.tester.put('{0}{1}/{2}/{3}'.format(
self.url, utils.SERVER_GROUP, self.server_id,
self.resource_group_id), data=json.dumps(data),
follow_redirects=True)
self.assertEquals(response.status_code, 200)
def tearDown(self):
"""This function delete the resource group from the database."""
connection = utils.get_db_connection(self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'])
resource_groups_utils.delete_resource_group(connection,
self.resource_group_name)

59
web/pgadmin/browser/server_groups/servers/resource_groups/tests/tests_resource_groups_get.py Normal file
View File

@@ -0,0 +1,59 @@
# #################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2016, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
# ##################################################################
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
from regression import parent_node_dict
from pgadmin.browser.server_groups.servers.tests import utils as server_utils
from . import utils as resource_groups_utils
class ResourceGroupsGetTestCase(BaseTestGenerator):
"""This class will get the resource groups"""
scenarios = [
('Get resource groups', dict(url='/browser/resource_group/obj/'))
]
def setUp(self):
self.server_id = parent_node_dict["server"][-1]["server_id"]
server_response = server_utils.connect_server(self, self.server_id)
if not server_response["info"] == "Server connected.":
raise Exception("Could not connect to server to add resource "
"groups.")
if "server_type" in server_response["data"]:
if server_response["data"]["server_type"] == "pg":
message = "Resource groups are not supported by PG."
self.skipTest(message)
self.resource_group = "test_resource_group_get%s" % \
str(uuid.uuid4())[1:6]
self.resource_group_id = resource_groups_utils.create_resource_groups(
self.server, self.resource_group)
def runTest(self):
"""This function will get the resource groups."""
resource_grp_response = resource_groups_utils.verify_resource_group(
self.server, self.resource_group)
if not resource_grp_response:
raise Exception("Could not find the resource group to fetch.")
response = self.tester.get(
"{0}{1}/{2}/{3}".format(self.url, utils.SERVER_GROUP,
self.server_id, self.resource_group_id),
follow_redirects=True)
self.assertEquals(response.status_code, 200)
def tearDown(self):
"""This function delete the resource group from the database."""
connection = utils.get_db_connection(self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'])
resource_groups_utils.delete_resource_group(connection,
self.resource_group)

92
web/pgadmin/browser/server_groups/servers/resource_groups/tests/utils.py Normal file
View File

@@ -0,0 +1,92 @@
# #################################################################
#
# 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 sys
import traceback
from regression import test_utils as utils
def create_resource_groups(server, resource_group_name):
"""
This function create the resource groups into databases.
:param server: server details
:type server: dict
:param resource_group_name: resource group name
:type resource_group_name: str
:return resource_group_id: resource group id
:rtype: int
"""
try:
connection = utils.get_db_connection(server['db'],
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 RESOURCE GROUP %s" % resource_group_name)
connection.set_isolation_level(old_isolation_level)
connection.commit()
# Get oid of newly created resource group
pg_cursor.execute("SELECT oid FROM edb_resource_group WHERE "
"rgrpname='%s'" % resource_group_name)
resource_group = pg_cursor.fetchone()
resource_group_id = resource_group[0]
connection.close()
return resource_group_id
except Exception:
traceback.print_exc(file=sys.stderr)
def verify_resource_group(server, resource_group_name):
"""
This function verifies the resource group exist in database or not.
:param server: server details
:type server: dict
:param db_name: database name
:type db_name: str
:param resource_group_name: resource group name
:type resource_group_name: str
:return:
"""
try:
connection = utils.get_db_connection(server['db'],
server['username'],
server['db_password'],
server['host'],
server['port'])
pg_cursor = connection.cursor()
pg_cursor.execute("SELECT * FROM edb_resource_group WHERE "
"rgrpname='%s'" % resource_group_name)
resource_group = pg_cursor.fetchone()
connection.close()
return resource_group
except Exception:
traceback.print_exc(file=sys.stderr)
def delete_resource_group(connection, resource_group_name):
try:
pg_cursor = connection.cursor()
pg_cursor.execute("SELECT * FROM edb_resource_group WHERE"
" rgrpname='%s'" % resource_group_name)
resource_group_name_count = len(pg_cursor.fetchall())
if resource_group_name_count:
old_isolation_level = connection.isolation_level
connection.set_isolation_level(0)
pg_cursor.execute("DELETE FROM edb_resource_group where "
"rgrpname='%s'" %
resource_group_name)
connection.set_isolation_level(old_isolation_level)
connection.commit()
connection.close()
except Exception:
traceback.print_exc(file=sys.stderr)