public inbox for [email protected]
help / color / mirror / Atom feedFrom: Navnath Gadakh <[email protected]>
To: Dave Page <[email protected]>
Cc: pgadmin-hackers <[email protected]>
Cc: Kanchan Mohitey <[email protected]>
Subject: pgAdmin IV: Code changes in roles and tablespaces test cases
Date: Mon, 26 Sep 2016 18:14:53 +0530
Message-ID: <CAOAJCYpLwJU+CaP+O5VWSHiMTsriu-O9tzxxk6Pt4y9zHdh_wg@mail.gmail.com> (raw)
List-Unsubscribe: <mailto:[email protected]?body=unsub%20pgadmin-hackers>
Hi Dave,
Please find the attached patch for the role and table spaces(Changes
required due to drop objects functionality).
Also, I have added the code in test result summary to show skipped test
cases for tablespace if path not mentioned.
To run testsuite:
python regression/runtests.py --pkg
browser.server_groups.servers.tablespaces
python regression/runtests.py --pkg browser.server_groups.servers.roles
--
Regards,
Navnath
--
Sent via pgadmin-hackers mailing list ([email protected])
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgadmin-hackers
Attachments:
[application/octet-stream] tablespaces_roles.patch (62.4K, 3-tablespaces_roles.patch)
download | inline diff:
diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_add.py b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_add.py
index 9d8f949..11b9fa7 100644
--- a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_add.py
+++ b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_add.py
@@ -6,9 +6,11 @@
# This software is released under the PostgreSQL Licence
#
# ##################################################################
+import json
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
+from regression import test_server_dict
from pgadmin.browser.server_groups.servers.tests import utils as server_utils
from . import utils as roles_utils
@@ -21,35 +23,33 @@ class LoginRoleAddTestCase(BaseTestGenerator):
('Check Role Node', dict(url='/browser/role/obj/'))
]
- @classmethod
- def setUpClass(cls):
- """
- This function used to add the sever
-
- :return: None
- """
-
- # Add the server
- server_utils.add_server(cls.tester)
-
- # Connect to server
- cls.server_connect_response, cls.server_group, cls.server_ids = \
- server_utils.connect_server(cls.tester)
-
- if len(cls.server_connect_response) == 0:
- raise Exception("No Server(s) connected to add the roles!!!")
+ def setUp(self):
+ pass
def runTest(self):
"""This function test the add role scenario"""
-
- roles_utils.add_role(self.tester, self.server_connect_response,
- self.server_group, self.server_ids)
-
- @classmethod
- def tearDownClass(cls):
- """This function deletes the role,server and parent id file"""
-
- roles_utils.delete_role(cls.tester)
- server_utils.delete_server(cls.tester)
- utils.delete_parent_id_file()
-
+ server_id = test_server_dict["server"][0]["server_id"]
+ server_response = server_utils.connect_server(self, server_id)
+ if not server_response['data']['connected']:
+ raise Exception("Server not found to add the role.")
+
+ data = roles_utils.get_role_data(self.server['db_password'])
+ self.role_name = data['rolname']
+ response = self.tester.post(self.url + str(utils.SERVER_GROUP) + '/'
+ + str(server_id) + '/',
+ data=json.dumps(data),
+ content_type='html/json')
+ self.assertEquals(response.status_code, 200)
+ response_data = json.loads(response.data.decode('utf-8'))
+ role_id = response_data['node']['_id']
+ role_dict = {"server_id": server_id, "role_id": role_id}
+ utils.write_node_info(role_id, "lrid", role_dict)
+
+ def tearDown(self):
+ """This function delete the role from added server"""
+ connection = utils.get_db_connection(self.server['db'],
+ self.server['username'],
+ self.server['db_password'],
+ self.server['host'],
+ self.server['port'])
+ roles_utils.delete_role(connection, self.role_name)
diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_delete.py b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_delete.py
index f3204cd..9a495ff 100644
--- a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_delete.py
+++ b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_delete.py
@@ -6,53 +6,40 @@
# This software is released under the PostgreSQL Licence
#
# ##################################################################
-
+import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
-from pgadmin.browser.server_groups.servers.tests import utils as server_utils
+from regression import test_server_dict
from . import utils as roles_utils
class LoginRoleDeleteTestCase(BaseTestGenerator):
"""This class has delete role scenario"""
-
scenarios = [
# Fetching default URL for roles node.
('Check Role Node', dict(url='/browser/role/obj/'))
]
- @classmethod
- def setUpClass(cls):
- """
- This function used to add the sever and roles
-
- :return: None
- """
-
- # Add the server
- server_utils.add_server(cls.tester)
-
- # Connect to server
- cls.server_connect_response, cls.server_group, cls.server_ids = \
- server_utils.connect_server(cls.tester)
-
- if len(cls.server_connect_response) == 0:
- raise Exception("No Server(s) connected to add the roles!!!")
-
- # Add the role
- roles_utils.add_role(cls.tester, cls.server_connect_response,
- cls.server_group, cls.server_ids)
+ def setUp(self):
+ self.role_name = "role_delete_%s" % str(uuid.uuid4())[1:6]
+ self.role_id = roles_utils.create_role(self.server, self.role_name)
def runTest(self):
- """This function tests the delete role scenario"""
-
- roles_utils.delete_role(self.tester)
-
- @classmethod
- def tearDownClass(self):
- """This function deletes the role,server and parent id file"""
-
- server_utils.delete_server(self.tester)
- utils.delete_parent_id_file()
+ """This function test the delete role scenario"""
+ server_id = test_server_dict["server"][0]["server_id"]
+ response = self.tester.delete(
+ self.url + str(utils.SERVER_GROUP) + '/' +
+ str(server_id) + '/' + str(self.role_id),
+ follow_redirects=True)
+ self.assertEquals(response.status_code, 200)
+
+ def tearDown(self):
+ """This function delete the role from added server"""
+ connection = utils.get_db_connection(self.server['db'],
+ self.server['username'],
+ self.server['db_password'],
+ self.server['host'],
+ self.server['port'])
+ roles_utils.delete_role(connection, self.role_name)
diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_get.py b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_get.py
index 4187fca..736ae3c 100644
--- a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_get.py
+++ b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_get.py
@@ -6,62 +6,39 @@
# This software is released under the PostgreSQL Licence
#
# ##################################################################
+import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
-from pgadmin.browser.server_groups.servers.tests import utils as server_utils
+from regression import test_server_dict
from . import utils as roles_utils
class LoginRoleGetTestCase(BaseTestGenerator):
"""This class tests the get role scenario"""
-
scenarios = [
# Fetching default URL for roles node.
('Check Role Node', dict(url='/browser/role/obj/'))
]
- @classmethod
- def setUpClass(cls):
- """
- This function used to add the sever and roles
-
- :return: None
- """
-
- # Add the server
- server_utils.add_server(cls.tester)
-
- # Connect to server
- cls.server_connect_response, cls.server_group, cls.server_ids = \
- server_utils.connect_server(cls.tester)
-
- if len(cls.server_connect_response) == 0:
- raise Exception("No Server(s) connected to get the roles!!!")
-
- # Add the role
- roles_utils.add_role(cls.tester, cls.server_connect_response,
- cls.server_group, cls.server_ids)
+ def setUp(self):
+ self.role_name = "role_get_%s" % str(uuid.uuid4())[1:6]
+ self.role_id = roles_utils.create_role(self.server, self.role_name)
def runTest(self):
"""This function test the get role scenario"""
-
- all_id = utils.get_ids()
- server_ids = all_id["sid"]
- role_ids_dict = all_id["lrid"][0]
-
- for server_id in server_ids:
- role_id = role_ids_dict[int(server_id)]
- response = self.tester.get(
- self.url + str(utils.SERVER_GROUP) + '/' +
- str(server_id) + '/' + str(role_id),
- follow_redirects=True)
- self.assertEquals(response.status_code, 200)
-
- @classmethod
- def tearDownClass(cls):
- """This function deletes the role,server and parent id file"""
-
- roles_utils.delete_role(cls.tester)
- server_utils.delete_server(cls.tester)
- utils.delete_parent_id_file()
+ server_id = test_server_dict["server"][0]["server_id"]
+ response = self.tester.get(
+ self.url + str(utils.SERVER_GROUP) + '/' +
+ str(server_id) + '/' + str(self.role_id),
+ follow_redirects=True)
+ self.assertEquals(response.status_code, 200)
+
+ def tearDown(self):
+ """This function delete the role from added server"""
+ connection = utils.get_db_connection(self.server['db'],
+ self.server['username'],
+ self.server['db_password'],
+ self.server['host'],
+ self.server['port'])
+ roles_utils.delete_role(connection, self.role_name)
diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_put.py b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_put.py
index f8c353e..8b6a6aa 100644
--- a/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_put.py
+++ b/web/pgadmin/browser/server_groups/servers/roles/tests/test_role_put.py
@@ -8,77 +8,48 @@
# ##################################################################
import json
+import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
-from pgadmin.browser.server_groups.servers.tests import utils as server_utils
-from regression.test_setup import advanced_config_data
+from regression import test_server_dict
from . import utils as roles_utils
class LoginRolePutTestCase(BaseTestGenerator):
"""This class has update role scenario"""
-
scenarios = [
# Fetching default URL for roles node.
('Check Role Node', dict(url='/browser/role/obj/'))
]
- @classmethod
- def setUpClass(cls):
- """
- This function used to add the sever and roles
-
- :return: None
- """
-
- # Add the server
- server_utils.add_server(cls.tester)
-
- # Connect to server
- cls.server_connect_response, cls.server_group, cls.server_ids = \
- server_utils.connect_server(cls.tester)
-
- if len(cls.server_connect_response) == 0:
- raise Exception("No Server(s) connected to get the roles!!!")
-
- # Add the role
- roles_utils.add_role(cls.tester, cls.server_connect_response,
- cls.server_group, cls.server_ids)
+ def setUp(self):
+ self.role_name = "role_put_%s" % str(uuid.uuid4())[1:6]
+ self.role_id = roles_utils.create_role(self.server, self.role_name)
def runTest(self):
"""This function tests the update role data scenario"""
+ server_id = test_server_dict["server"][0]["server_id"]
+ role_response = roles_utils.verify_role(self.server, self.role_name)
+ if len(role_response) == 0:
+ raise Exception("No roles(s) to update!!!")
+ data = {
+ "description": "This is the test description for cast",
+ "lrid": self.role_id
+ }
+ put_response = self.tester.put(
+ self.url + str(utils.SERVER_GROUP) + '/' +
+ str(server_id) + '/' + str(self.role_id),
+ data=json.dumps(data),
+ follow_redirects=True)
+ self.assertEquals(put_response.status_code, 200)
+
+ def tearDown(self):
+ """This function delete the role from added server"""
+ connection = utils.get_db_connection(self.server['db'],
+ self.server['username'],
+ self.server['db_password'],
+ self.server['host'],
+ self.server['port'])
+ roles_utils.delete_role(connection, self.role_name)
- all_id = utils.get_ids()
- server_ids = all_id["sid"]
- role_ids_dict = all_id["lrid"][0]
-
- for server_id in server_ids:
- role_id = role_ids_dict[int(server_id)]
- role_response = roles_utils.verify_role(self.tester,
- utils.SERVER_GROUP,
- server_id,
- role_id)
- if len(role_response) == 0:
- raise Exception("No roles(s) to update!!!")
-
- data = {
- "description": advanced_config_data["lr_update_data"]
- ["comment"],
- "lrid": role_id
- }
- put_response = self.tester.put(
- self.url + str(utils.SERVER_GROUP) + '/' +
- str(server_id) + '/' + str(role_id),
- data=json.dumps(data),
- follow_redirects=True)
-
- self.assertEquals(put_response.status_code, 200)
-
- @classmethod
- def tearDownClass(cls):
- """This function deletes the role,server and parent id file"""
-
- roles_utils.delete_role(cls.tester)
- server_utils.delete_server(cls.tester)
- utils.delete_parent_id_file()
diff --git a/web/pgadmin/browser/server_groups/servers/roles/tests/utils.py b/web/pgadmin/browser/server_groups/servers/roles/tests/utils.py
index 85e0990..491ba2e 100644
--- a/web/pgadmin/browser/server_groups/servers/roles/tests/utils.py
+++ b/web/pgadmin/browser/server_groups/servers/roles/tests/utils.py
@@ -6,45 +6,51 @@
# This software is released under the PostgreSQL Licence
#
# ##################################################################
+from __future__ import print_function
import json
import os
import pickle
import uuid
+import sys
-from regression.test_setup import pickle_path, config_data, advanced_config_data
+from regression.test_setup import config_data
from pgadmin.browser.server_groups.servers.tests import utils as server_utils
from regression import test_utils as utils
ROLE_URL = '/browser/role/obj/'
+file_name = os.path.basename(__file__)
-def verify_role(tester, server_group, server_id, role_id):
+def verify_role(server, role_name):
"""
This function calls the GET API for role to verify
-
- :param tester: test client
- :type tester: flask test client object
- :param server_group: server group id
- :type server_group: int
- :param server_id: server id
- :type server_id: str
- :param role_id: role id
- :type role_id: int
- :return: dict/None
+ :param server: server details
+ :type server: dict
+ :param role_name: role name
+ :type role_name: str
+ :return role: role record from db
+ :rtype role: dict
"""
-
- srv_connect = server_utils.verify_server(tester, server_group, server_id)
- if srv_connect['data']['connected']:
- response = tester.get(
- '{0}{1}/{2}/{3}'.format(ROLE_URL, server_group, server_id,
- role_id),
- content_type='html/json')
- temp_response = json.loads(response.data.decode('utf-8'))
- return temp_response
- else:
- return None
+ 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 pg_catalog.pg_roles pr WHERE pr.rolname='%s'" %
+ role_name)
+ connection.commit()
+ role = pg_cursor.fetchone()
+ connection.close()
+ return role
+ except Exception as exception:
+ exception = "Error while getting role: %s: line:%s %s" % (
+ file_name, sys.exc_traceback.tb_lineno, exception)
+ print(exception, file=sys.stderr)
def test_getrole(tester):
@@ -65,58 +71,55 @@ def test_getrole(tester):
return role_response_data
-def get_role_data():
- """This function returns the role data from config file"""
-
+def get_role_data(lr_pwd):
+ """This function returns the role data"""
data = {
- "rolcanlogin": advanced_config_data['lr_credentials']
- ['can_login'],
- "rolconnlimit": advanced_config_data['lr_credentials']
- ['conn_limit'],
- "rolcreaterole": advanced_config_data['lr_credentials']
- ['create_role'],
- "rolinherit": advanced_config_data['lr_credentials']
- ['role_inherit'],
- "rolmembership": advanced_config_data['lr_credentials']
- ['role_membership'],
- "rolname": str(uuid.uuid4())[1:8],
- "rolpassword": advanced_config_data['lr_credentials']
- ['lr_password'],
- "rolvaliduntil": advanced_config_data['lr_credentials']
- ['lr_validity'],
- "seclabels": advanced_config_data['lr_credentials']
- ['sec_lable'],
- "variables": advanced_config_data['lr_credentials']
- ['variable']
+ "rolcanlogin": "true",
+ "rolconnlimit": -1,
+ "rolcreaterole": "true",
+ "rolinherit": "true",
+ "rolmembership": [],
+ "rolname": "test_role_%s" % str(uuid.uuid4())[1:6],
+ "rolpassword": lr_pwd,
+ "rolvaliduntil": "12/27/2016",
+ "seclabels": [],
+ "variables": [{"name": "work_mem",
+ "database": "postgres",
+ "value": 65}]
}
return data
-def add_role(tester, server_connect_response, server_group, server_ids):
+def create_role(server, role_name):
"""
- This function is used to add the roles to server
-
- :param tester: flask test client
- :type tester: flask test client object
- :param server_connect_response: server connect API response
- :type server_connect_response: dict
- :param server_group: server group
- :type server_group: int
- :param server_ids: list of server id
- :type server_ids: list
- :return: None
+ This function create the role.
+ :param server:
+ :param role_name:
+ :return:
"""
-
- for server_connect, server_id in zip(server_connect_response,
- server_ids):
- if server_connect['data']['connected']:
- data = get_role_data()
- response = tester.post(ROLE_URL + str(server_group) + '/'
- + server_id + '/', data=json.dumps(data),
- content_type='html/json')
- assert response.status_code == 200
- response_data = json.loads(response.data.decode('utf-8'))
- write_role_id(response_data)
+ try:
+ connection = utils.get_db_connection(server['db'],
+ server['username'],
+ server['db_password'],
+ server['host'],
+ server['port'])
+ pg_cursor = connection.cursor()
+ pg_cursor.execute("CREATE ROLE %s LOGIN" % role_name)
+ connection.commit()
+ # Get 'oid' from newly created tablespace
+ pg_cursor.execute(
+ "SELECT pr.oid from pg_catalog.pg_roles pr WHERE pr.rolname='%s'" %
+ role_name)
+ oid = pg_cursor.fetchone()
+ role_id = ''
+ if oid:
+ role_id = oid[0]
+ connection.close()
+ return role_id
+ except Exception as exception:
+ exception = "Error while deleting role: %s: line:%s %s" % (
+ file_name, sys.exc_traceback.tb_lineno, exception)
+ print(exception, file=sys.stderr)
def write_role_id(response_data):
@@ -147,41 +150,30 @@ def write_role_id(response_data):
db_output.close()
-def delete_role(tester):
+def delete_role(connection, role_name):
"""
This function use to delete the existing roles in the servers
- :param tester: flask test client
- :type tester: flask test object
+ :param connection: db connection
+ :type connection: connection object
+ :param role_name: role name
+ :type role_name: str
:return: None
"""
-
- server_ids = None
- role_ids_dict = None
-
- all_id = utils.get_ids()
- if "sid" and "lrid" in all_id.keys():
- server_ids = all_id["sid"]
- if all_id['lrid']:
- role_ids_dict = all_id['lrid'][0]
- else:
- raise Exception("Keys are not found: {}".format(["sid", "lrid"]))
-
- if server_ids and role_ids_dict is not None:
- for server_id in server_ids:
- server_response = server_utils.verify_server(tester,
- utils.SERVER_GROUP,
- server_id)
- if server_response["data"]["connected"]:
- role_id = role_ids_dict[int(server_id)]
- response = tester.delete(
- ROLE_URL + str(utils.SERVER_GROUP) + '/' +
- str(server_id) + '/' + str(role_id),
- follow_redirects=True)
- assert response.status_code == 200
- response_data = json.loads(response.data.decode('utf-8'))
- assert response_data['success'] == 1
- else:
- raise Exception("No servers/roles found.")
-
-
+ try:
+ pg_cursor = connection.cursor()
+ pg_cursor.execute(
+ "SELECT * FROM pg_catalog.pg_roles WHERE rolname='%s'" % role_name)
+ role_count = pg_cursor.fetchone()
+ if role_count:
+ old_isolation_level = connection.isolation_level
+ connection.set_isolation_level(0)
+ pg_cursor = connection.cursor()
+ pg_cursor.execute("DROP ROLE %s" % role_name)
+ connection.set_isolation_level(old_isolation_level)
+ connection.commit()
+ connection.close()
+ except Exception as exception:
+ exception = "Error while deleting role: %s: line:%s %s" % (
+ file_name, sys.exc_traceback.tb_lineno, exception)
+ print(exception, file=sys.stderr)
diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_add.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_add.py
index caa6966..09e9e0e 100644
--- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_add.py
+++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_add.py
@@ -6,50 +6,68 @@
# This software is released under the PostgreSQL Licence
#
# ##################################################################
+from __future__ import print_function
+
+import os
+import sys
+import json
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
+from regression import test_server_dict
from pgadmin.browser.server_groups.servers.tests import utils as server_utils
from . import utils as tablespace_utils
class TableSpaceAddTestCase(BaseTestGenerator):
"""This class will add tablespace node under server"""
-
scenarios = [
# Fetching default URL for tablespace node.
('Check Tablespace Node', dict(url='/browser/tablespace/obj/'))
]
- @classmethod
- def setUpClass(cls):
- """
- This function used to add the sever
-
- :return: None
- """
-
- # Add the server
- server_utils.add_server(cls.tester)
-
- # Connect to server
- cls.server_connect_response, cls.server_group, cls.server_ids = \
- server_utils.connect_server(cls.tester)
-
- if len(cls.server_connect_response) == 0:
- raise Exception("No Server(s) connected to add the roles!!!")
+ def setUp(self):
+ self.tablespace_name = ''
+ if not self.server['tablespace_path']\
+ or self.server['tablespace_path'] is None:
+ message = "Skipped tablespace add test case. Tablespace path" \
+ " not configured for server: %s" % self.server['name']
+ # Skip the test case if tablespace_path not found.
+ self.skipTest(message)
def runTest(self):
- """This function test the add tablespace scenario"""
-
- tablespace_status = tablespace_utils.add_table_space(
- self.tester, self.server_connect_response, self.server_group,
- self.server_ids)
+ """This function test the add tablespace API"""
+ server_id = test_server_dict["server"][0]["server_id"]
+ server_response = server_utils.connect_server(self, server_id)
+ if not server_response['data']['connected']:
+ raise Exception("Unable to connect server to get tablespace.")
- @classmethod
- def tearDownClass(cls):
- """This function deletes the tablespaces,server and parent_id file"""
+ db_owner = server_response['data']['user']['name']
+ table_space_path = self.server['tablespace_path']
+ data = tablespace_utils.get_tablespace_data(
+ table_space_path, db_owner)
+ self.tablespace_name = data['name']
+ response = self.tester.post(
+ self.url + str(utils.SERVER_GROUP) + '/'
+ + str(server_id) + '/',
+ data=json.dumps(data),
+ content_type='html/json')
+ self.assertEquals(response.status_code, 200)
+ response_data = json.loads(response.data.decode('utf-8'))
+ tablespace_id = response_data['node']['_id']
+ tablespace_dict = {"tablespace_id": tablespace_id,
+ "tablespace_name": self.tablespace_name,
+ "server_id": server_id}
+ utils.write_node_info(tablespace_id, "tsid", tablespace_dict)
- tablespace_utils.delete_table_space(cls.tester)
- server_utils.delete_server(cls.tester)
- utils.delete_parent_id_file()
+ def tearDown(self):
+ """
+ This function delete the tablespace from server added in SQLite and
+ clears the node_info_dict
+ """
+ connection = utils.get_db_connection(self.server['db'],
+ self.server['username'],
+ self.server['db_password'],
+ self.server['host'],
+ self.server['port'])
+ tablespace_utils.delete_tablespace(connection, self.tablespace_name)
diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_delete.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_delete.py
index f280ee6..605b6af 100644
--- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_delete.py
+++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_delete.py
@@ -6,11 +6,13 @@
# This software is released under the PostgreSQL Licence
#
# ##################################################################
-
+from __future__ import print_function
+import uuid
+import json
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
-from pgadmin.browser.server_groups.servers.tests import utils as server_utils
+from regression import test_server_dict
from . import utils as tablespace_utils
@@ -22,37 +24,39 @@ class TableSpaceDeleteTestCase(BaseTestGenerator):
('Check Tablespace Node', dict(url='/browser/tablespace/obj/'))
]
- @classmethod
- def setUpClass(cls):
- """
- This function used to add the sever
-
- :return: None
- """
-
- # Add the server
- server_utils.add_server(cls.tester)
-
- # Connect to server
- cls.server_connect_response, cls.server_group, cls.server_ids = \
- server_utils.connect_server(cls.tester)
-
- if len(cls.server_connect_response) == 0:
- raise Exception("No Server(s) connected to add the roles!!!")
-
- # Add tablespace
- tablespace_utils.add_table_space(cls.tester,
- cls.server_connect_response,
- cls.server_group, cls.server_ids)
+ def setUp(self):
+ if not self.server['tablespace_path']\
+ or self.server['tablespace_path'] is None:
+ message = "Skipped tablespace delete test case. Tablespace path" \
+ " not configured for server: %s" % self.server['name']
+ # Skip the test case if tablespace_path not found.
+ self.skipTest(message)
+ self.tablespace_name = "tablespace_delete_%s" % str(uuid.uuid4())[1:6]
+ self.tablespace_id = tablespace_utils.create_tablespace(
+ self.server, self.tablespace_name)
def runTest(self):
- """This function tests the delete table space scenario"""
-
- tablespace_utils.delete_table_space(self.tester)
-
- @classmethod
- def tearDownClass(cls):
- """This function deletes the server and parent id file"""
+ """This function tests the delete table space api"""
+ server_id = test_server_dict["server"][0]["server_id"]
+ tablespace_count = tablespace_utils.verify_table_space(
+ self.server, self.tablespace_name)
+ if tablespace_count == 0:
+ raise Exception("No tablespace(s) to delete!!!")
+
+ response = self.tester.delete(self.url + str(utils.SERVER_GROUP)
+ + '/' + str(server_id) + '/'
+ + str(self.tablespace_id),
+ follow_redirects=True)
+ self.assertEquals(response.status_code, 200)
+ delete_response_data = json.loads(response.data.decode('utf-8'))
+ self.assertEquals(delete_response_data['success'], 1)
+
+ def tearDown(self):
+ """This function deletes the tablespace"""
+ connection = utils.get_db_connection(self.server['db'],
+ self.server['username'],
+ self.server['db_password'],
+ self.server['host'],
+ self.server['port'])
+ tablespace_utils.delete_tablespace(connection, self.tablespace_name)
- server_utils.delete_server(cls.tester)
- utils.delete_parent_id_file()
diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_get.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_get.py
index b26d71d..4f30ed6 100644
--- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_get.py
+++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_get.py
@@ -6,9 +6,11 @@
# 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 test_server_dict
from pgadmin.browser.server_groups.servers.tests import utils as server_utils
from . import utils as tablespace_utils
@@ -21,58 +23,39 @@ class TablespaceGetTestCase(BaseTestGenerator):
('Check Tablespace Node', dict(url='/browser/tablespace/obj/'))
]
- @classmethod
- def setUpClass(cls):
- """
- This function used to add the sever
-
- :return: None
- """
-
- # Add the server
- server_utils.add_server(cls.tester)
-
- # Connect to server
- cls.server_connect_response, cls.server_group, cls.server_ids = \
- server_utils.connect_server(cls.tester)
-
- if len(cls.server_connect_response) == 0:
- raise Exception("No Server(s) connected to add the roles!!!")
-
- # Add tablespace
- tablespace_utils.add_table_space(cls.tester,
- cls.server_connect_response,
- cls.server_group, cls.server_ids)
+ def setUp(self):
+ if not self.server['tablespace_path']\
+ or self.server['tablespace_path'] is None:
+ message = "Skipped tablespace get test case. Tablespace path" \
+ " not configured for server: %s" % self.server['name']
+ # Skip the test case if tablespace_path not found.
+ self.skipTest(message)
+ self.tablespace_name = "tablespace_delete_%s" % str(uuid.uuid4())[1:6]
+ self.tablespace_id = tablespace_utils.create_tablespace(
+ self.server, self.tablespace_name)
def runTest(self):
"""This function test the get table space scenario"""
-
- tablespace_ids_dict = None
- all_id = utils.get_ids()
- server_ids = all_id["sid"]
- if "tsid" in all_id and all_id["tsid"]:
- tablespace_ids_dict = all_id["tsid"][0]
-
- if tablespace_ids_dict:
- for server_id in server_ids:
- tablespace_id = tablespace_ids_dict[int(server_id)]
- server_response = server_utils.verify_server(self.tester,
- utils.SERVER_GROUP,
- server_id)
- if server_response['data']['connected']:
- tablespace_utils.verify_table_space(
- self.tester, utils.SERVER_GROUP, server_id, tablespace_id)
- response = self.tester.get(
- self.url + str(utils.SERVER_GROUP) + '/' +
- str(server_id) + '/' + str(
- tablespace_id),
- follow_redirects=True)
- self.assertEquals(response.status_code, 200)
-
- @classmethod
- def tearDownClass(cls):
- """This function deletes the tablespaces,server and parent id file"""
-
- tablespace_utils.delete_table_space(cls.tester)
- server_utils.delete_server(cls.tester)
- utils.delete_parent_id_file()
+ server_id = test_server_dict["server"][0]["server_id"]
+ server_response = server_utils.connect_server(self, server_id)
+ if not server_response['data']['connected']:
+ raise Exception("Unable to connect server to get tablespace.")
+
+ tablespace_count = tablespace_utils.verify_table_space(
+ self.server, self.tablespace_name)
+ if tablespace_count == 0:
+ raise Exception("No tablespace(s) to update!!!")
+ response = self.tester.get(
+ self.url + str(utils.SERVER_GROUP) + '/' +
+ str(server_id) + '/' + str(self.tablespace_id),
+ follow_redirects=True)
+ self.assertEquals(response.status_code, 200)
+
+ def tearDown(self):
+ """This function delete the tablespace from added server"""
+ connection = utils.get_db_connection(self.server['db'],
+ self.server['username'],
+ self.server['db_password'],
+ self.server['host'],
+ self.server['port'])
+ tablespace_utils.delete_tablespace(connection, self.tablespace_name)
diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_put.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_put.py
index 22257f9..b5e1986 100644
--- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_put.py
+++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/test_tbspc_put.py
@@ -8,11 +8,11 @@
# ##################################################################
import json
+import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression import test_utils as utils
-from pgadmin.browser.server_groups.servers.tests import utils as server_utils
-from test_setup import advanced_config_data
+from regression import test_server_dict
from . import utils as tablespace_utils
@@ -24,67 +24,40 @@ class TableSpaceUpdateTestCase(BaseTestGenerator):
('Check Tablespace Node', dict(url='/browser/tablespace/obj/'))
]
- @classmethod
- def setUpClass(cls):
- """
- This function used to add the sever
-
- :return: None
- """
-
- # Add the server
- server_utils.add_server(cls.tester)
-
- # Connect to server
- cls.server_connect_response, cls.server_group, cls.server_ids = \
- server_utils.connect_server(cls.tester)
-
- if len(cls.server_connect_response) == 0:
- raise Exception("No Server(s) connected to add the roles!!!")
-
- # Add tablespace
- tablespace_utils.add_table_space(cls.tester,
- cls.server_connect_response,
- cls.server_group, cls.server_ids)
+ def setUp(self):
+ if not self.server['tablespace_path']\
+ or self.server['tablespace_path'] is None:
+ message = "Skipped tablespace delete test case. Tablespace path" \
+ " not configured for server: %s" % self.server['name']
+ # Skip the test case if tablespace_path not found.
+ self.skipTest(message)
+ self.tablespace_name = "tablespace_delete_%s" % str(uuid.uuid4())[1:6]
+ self.tablespace_id = tablespace_utils.create_tablespace(
+ self.server, self.tablespace_name)
def runTest(self):
"""This function tests the update tablespace data scenario"""
-
- tablespace_ids_dict = None
- all_id = utils.get_ids()
- server_ids = all_id["sid"]
- if "tsid" in all_id and all_id["tsid"]:
- tablespace_ids_dict = all_id["tsid"][0]
-
- if tablespace_ids_dict:
- for server_id in server_ids:
- tablespace_id = tablespace_ids_dict[int(server_id)]
- tablespace_response = tablespace_utils.verify_table_space(
- self.tester,
- utils.SERVER_GROUP, server_id,
- tablespace_id)
- if len(tablespace_response) == 0:
- raise Exception("No tablespace(s) to update!!!")
-
- data = {
- "description": advanced_config_data["tbspc_update_data"]
- ["comment"],
- "table_space_id": tablespace_id
- }
-
- put_response = self.tester.put(
- self.url + str(utils.SERVER_GROUP) + '/' +
- str(server_id) + '/' + str(
- tablespace_id),
- data=json.dumps(data),
- follow_redirects=True)
-
- self.assertEquals(put_response.status_code, 200)
-
- @classmethod
- def tearDownClass(cls):
- """This function deletes the tablespaces,server and parent id file"""
-
- tablespace_utils.delete_table_space(cls.tester)
- server_utils.delete_server(cls.tester)
- utils.delete_parent_id_file()
+ server_id = test_server_dict["server"][0]["server_id"]
+ tablespace_count = tablespace_utils.verify_table_space(
+ self.server, self.tablespace_name)
+ if tablespace_count == 0:
+ raise Exception("No tablespace(s) to update!!!")
+
+ data = {
+ "description": "This is test description.",
+ "table_space_id": self.tablespace_id
+ }
+ put_response = self.tester.put(
+ self.url + str(utils.SERVER_GROUP) + '/' + str(server_id) + '/'
+ + str(self.tablespace_id), data=json.dumps(data),
+ follow_redirects=True)
+ self.assertEquals(put_response.status_code, 200)
+
+ def tearDown(self):
+ """This function deletes the tablespace"""
+ connection = utils.get_db_connection(self.server['db'],
+ self.server['username'],
+ self.server['db_password'],
+ self.server['host'],
+ self.server['port'])
+ tablespace_utils.delete_tablespace(connection, self.tablespace_name)
diff --git a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/utils.py b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/utils.py
index 4fd21bd..c9b6a8a 100644
--- a/web/pgadmin/browser/server_groups/servers/tablespaces/tests/utils.py
+++ b/web/pgadmin/browser/server_groups/servers/tablespaces/tests/utils.py
@@ -9,187 +9,114 @@
from __future__ import print_function
-import json
import os
-import pickle
import uuid
import sys
-from regression.test_setup import pickle_path, config_data, \
- advanced_config_data
-from pgadmin.browser.server_groups.servers.tests import utils as server_utils
+
from regression import test_utils as utils
TABLE_SPACE_URL = '/browser/tablespace/obj/'
-
-
-# Tablespace utility
-def get_tablespace_data(server_connect):
- """This function returns the tablespace data from config file"""
-
- adv_config_data = None
- data = None
- db_user = server_connect['data']['user']['name']
- server_config_data = config_data['server_credentials']
-
- # Get the config data of appropriate db user
- for config_test_data in advanced_config_data['tablespc_credentials']:
- if db_user == config_test_data['spc_user']:
- # Add the tablespace path from server config
- server_config = (item for item in server_config_data if
- item["db_username"] == db_user).next()
- if "tablespace_path" in server_config:
- config_test_data['spc_location'] = server_config['tablespace_path']
- adv_config_data = config_test_data
- else:
- config_test_data['spc_location'] = None
-
- if adv_config_data is not None:
- data = {
- "name": str(uuid.uuid4())[1:8],
- "seclabels": adv_config_data["spc_seclable"],
- "spcacl": adv_config_data["spc_acl"],
- "spclocation": adv_config_data["spc_location"],
- "spcoptions": adv_config_data["spc_opts"],
- "spcuser": adv_config_data["spc_user"]
- }
+file_name = os.path.basename(__file__)
+
+
+def get_tablespace_data(tablespace_path, db_owner):
+ """This function returns the tablespace data"""
+ data = {
+ "name": "test_%s" % str(uuid.uuid4())[1:8],
+ "seclabels": [],
+ "spcacl": [
+ {
+ "grantee": db_owner,
+ "grantor": db_owner,
+ "privileges": [
+ {
+ "privilege_type": "C",
+ "privilege": True,
+ "with_grant": False
+ }
+ ]
+ }
+ ],
+ "spclocation": tablespace_path,
+ "spcoptions": [],
+ "spcuser": db_owner
+ }
return data
-def write_tablespace_id(response_data):
- """
- This function write the table space id to parent_id.pkl
-
- :param response_data: create table space API response data
- :type response_data: dict
- :return: None
- """
-
- table_space_id = response_data['node']['_id']
- server_id = response_data['node']['_pid']
- pickle_id_dict = utils.get_pickle_id_dict()
- if os.path.isfile(pickle_path):
- existing_server_id = open(pickle_path, 'rb')
- tol_server_id = pickle.load(existing_server_id)
- pickle_id_dict = tol_server_id
- if 'tsid' in pickle_id_dict:
- if pickle_id_dict['tsid']:
- # Add the db_id as value in dict
- pickle_id_dict["tsid"][0].update(
- {server_id: table_space_id})
- else:
- # Create new dict with server_id and db_id
- pickle_id_dict["tsid"].append(
- {server_id: table_space_id})
- db_output = open(pickle_path, 'wb')
- pickle.dump(pickle_id_dict, db_output)
- db_output.close()
-
-
-def add_table_space(tester, server_connect_response, server_group, server_ids):
- """
- This function is used to add the roles to server
-
- :param tester: flask test client
- :type tester: flask test client object
- :param server_connect_response: server connect API response
- :type server_connect_response: dict
- :param server_group: server group
- :type server_group: int
- :param server_ids: list of server id
- :type server_ids: list
- :return: None
- """
-
- total_servers_count = len(server_ids)
- servers_without_tablespace_path = []
-
- for server_connect, server_id in zip(server_connect_response,
- server_ids):
- tablespace_path = server_connect['tablespace_path']
- # Skip the test case if tablespace_path does not exist
- if not tablespace_path or tablespace_path is None:
- file_name = os.path.basename(
- sys._getframe().f_back.f_code.co_filename)
- servers_without_tablespace_path.append(server_id)
- if total_servers_count == len(servers_without_tablespace_path):
- print("Skipping tablespaces test cases for the file <{0}>, "
- "Tablespace path not configured for the servers: "
- "{1}".format(file_name, server_ids), file=sys.stderr)
- return
- else:
- print("Skipping tablespaces test case for the file <{0}>: "
- "Tablespace path not configured for server: {1}".format(
- file_name, server_id), file=sys.stderr)
- continue
-
- if server_connect['data']['connected']:
- data = get_tablespace_data(server_connect)
- response = tester.post(TABLE_SPACE_URL + str(server_group) + '/'
- + server_id + '/',
- data=json.dumps(data),
- content_type='html/json')
- assert response.status_code == 200
- response_data = json.loads(response.data.decode('utf-8'))
- write_tablespace_id(response_data)
-
-
-def verify_table_space(tester, server_group, server_id, tablespace_id):
+def create_tablespace(server, test_tablespace_name):
+ 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 TABLESPACE %s LOCATION '%s'" %
+ (test_tablespace_name, server['tablespace_path']))
+ connection.set_isolation_level(old_isolation_level)
+ connection.commit()
+
+ # Get 'oid' from newly created tablespace
+ pg_cursor.execute(
+ "SELECT ts.oid from pg_tablespace ts WHERE ts.spcname='%s'" %
+ test_tablespace_name)
+ oid = pg_cursor.fetchone()
+ tspc_id = ''
+ if oid:
+ tspc_id = oid[0]
+ connection.close()
+ return tspc_id
+ except Exception as exception:
+ raise Exception("Error while creating tablespace. %s" % exception)
+
+
+def verify_table_space(server, test_tablespace_name):
"""
This function calls the GET API for role to verify
-
- :param tester: test client
- :type tester: flask test client object
- :param server_group: server group id
- :type server_group: int
- :param server_id: server id
- :type server_id: str
- :param tablespace_id: table space id
- :type tablespace_id: int
- :return: dict/None
- """
-
- srv_connect = server_utils.verify_server(tester, server_group, server_id)
- if srv_connect['data']['connected']:
- response = tester.get(
- '{0}{1}/{2}/{3}'.format(TABLE_SPACE_URL, server_group, server_id,
- tablespace_id),
- content_type='html/json')
- assert response.status_code == 200
- temp_response = json.loads(response.data.decode('utf-8'))
- return temp_response
- else:
- return None
-
-
-def delete_table_space(tester):
+ :param server: server info
+ :type server: dict
+ :param test_tablespace_name: tablespace name
+ :type test_tablespace_name: str
+ :return tablespace_count: tablespace count
+ :rtype: int
"""
- This function use to delete the existing tablespace in the servers
-
- :param tester: flask test client
- :type tester: flask test object
- :return: None
- """
-
- all_id = utils.get_ids()
- server_ids = all_id["sid"]
- if "tsid" in all_id and all_id["tsid"]:
- tablespace_ids_dict = all_id["tsid"][0]
- else:
- tablespace_ids_dict = None
-
- if tablespace_ids_dict is not None:
- for server_id in server_ids:
- tablespace_id = tablespace_ids_dict[int(server_id)]
- role_response = verify_table_space(tester, utils.SERVER_GROUP,
- server_id,
- tablespace_id)
- if len(role_response) == 0:
- raise Exception("No tablespace(s) to delete!!!")
- response = tester.delete(
- TABLE_SPACE_URL + str(utils.SERVER_GROUP) + '/' +
- str(server_id) + '/' + str(tablespace_id),
- follow_redirects=True)
- assert response.status_code == 200
- delete_response_data = json.loads(response.data.decode('utf-8'))
- assert delete_response_data['success'] == 1
+ 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 pg_tablespace ts WHERE"
+ " ts.spcname='%s'" % test_tablespace_name)
+ tablespace_count = len(pg_cursor.fetchall())
+ connection.close()
+ return tablespace_count
+ except Exception as exception:
+ exception = "%s: line:%s %s" % (
+ file_name, sys.exc_traceback.tb_lineno, exception)
+ print(exception, file=sys.stderr)
+
+
+def delete_tablespace(connection, test_tablespace_name):
+ try:
+ pg_cursor = connection.cursor()
+ pg_cursor.execute("SELECT * FROM pg_tablespace ts WHERE"
+ " ts.spcname='%s'" % test_tablespace_name)
+ tablespace_count = len(pg_cursor.fetchall())
+ if tablespace_count:
+ old_isolation_level = connection.isolation_level
+ connection.set_isolation_level(0)
+ pg_cursor.execute("DROP TABLESPACE %s" % test_tablespace_name)
+ connection.set_isolation_level(old_isolation_level)
+ connection.commit()
+ connection.close()
+ except Exception as exception:
+ exception = "%s: line:%s %s" % (
+ file_name, sys.exc_traceback.tb_lineno, exception)
+ print(exception, file=sys.stderr)
+ raise Exception(exception)
diff --git a/web/regression/__init__.py b/web/regression/__init__.py
index c001be6..6b7cb2f 100644
--- a/web/regression/__init__.py
+++ b/web/regression/__init__.py
@@ -6,29 +6,32 @@
# #This software is released under the PostgreSQL Licence
#
# ##########################################################################
+import pgadmin.browser.server_groups.servers.roles.tests.utils as roles_utils
+import pgadmin.browser.server_groups.servers.tablespaces.tests.utils as \
+ tablespace_utils
global node_info_dict
node_info_dict = {
- "sid": [], # server
- "did": [], # database
- "lrid": [], # role
- "tsid": [], # tablespace
- "scid": [], # schema
- "tfnid": [], # trigger functions
- "coid": [], # collation
- "cid": [], # casts
- "etid": [], # event_trigger
- "eid": [], # extension
- "fid": [], # FDW
- "fsid": [], # FRS
- "umid": [], # user_mapping
- "seid": [] # sequence
+ "sid": [], # server
+ "did": [], # database
+ "lrid": [], # role
+ "tsid": [], # tablespace
+ "scid": [], # schema
+ "tfnid": [], # trigger functions
+ "coid": [], # collation
+ "cid": [], # casts
+ "etid": [], # event_trigger
+ "eid": [], # extension
+ "fid": [], # FDW
+ "fsid": [], # FRS
+ "umid": [], # user_mapping
+ "seid": [] # sequence
}
global test_server_dict
test_server_dict = {
- "server": [],
- "database": [],
- "tablespace": [],
- "role": []
-}
\ No newline at end of file
+ "server": [],
+ "database": [],
+ "tablespace": [],
+ "role": []
+}
diff --git a/web/regression/runtests.py b/web/regression/runtests.py
index 678adc8..8655ba0 100644
--- a/web/regression/runtests.py
+++ b/web/regression/runtests.py
@@ -182,14 +182,12 @@ def get_tests_result(test_suite):
error_case[0]).split('.')[-1].split()[0].strip(')')
if class_name not in failed_cases_result:
failed_cases_result.append(class_name)
- # TODO: Code remaining to count the skipped test cases
- # if test_suite.skipped:
- # for skip_test in test_suite.skipped:
- # class_name = str(
- # skip_test[0]).split('.')[-1].split()[0].strip(')')
- # if class_name not in failed_cases_result:
- # skipped_cases_result.append(class_name)
- # print(class_name)
+ if test_suite.skipped:
+ for skip_test in test_suite.skipped:
+ class_name = str(
+ skip_test[0]).split('.')[-1].split()[0].strip(')')
+ if class_name not in failed_cases_result:
+ skipped_cases_result.append(class_name)
return total_ran, failed_cases_result, skipped_cases_result
except Exception as exception:
exception = "Exception: %s: line:%s %s" % (
@@ -275,15 +273,22 @@ if __name__ == '__main__':
print("============================", file=sys.stderr)
for server_res in test_result:
failed_cases = "\n\t".join(test_result[server_res][1])
+ skipped_cases = "\n\t".join(test_result[server_res][2])
total_failed = len(test_result[server_res][1])
- # TODO : code remaining to handle '-1' condition
- total_passed = int(test_result[server_res][0]) - total_failed
-
- print("%s: %s test%s passed, %s test%s failed %s%s" %
- (server_res, total_passed, (total_passed != 1 and "s" or ""),
- total_failed, (total_failed != 1 and "s" or ""),
- (total_failed != 0 and ":\n\t" or ""), failed_cases),
- file=sys.stderr)
+ total_skipped = len(test_result[server_res][2])
+ total_passed_cases = int(
+ test_result[server_res][0]) - total_failed - total_skipped
+
+ print(
+ "%s: %s test%s passed, %s test%s failed %s%s,"
+ " %s test%s skipped %s%s" %
+ (server_res, total_passed_cases,
+ (total_passed_cases != 1 and "s" or ""),
+ total_failed, (total_failed != 1 and "s" or ""),
+ (total_failed != 0 and ":\n\t" or ""), failed_cases,
+ total_skipped, (total_skipped != 1 and "s" or ""),
+ (total_skipped != 0 and ":\n\t" or ""), skipped_cases),
+ file=sys.stderr)
print("============================", file=sys.stderr)
diff --git a/web/regression/test_utils.py b/web/regression/test_utils.py
index 73c45a6..731f1b6 100644
--- a/web/regression/test_utils.py
+++ b/web/regression/test_utils.py
@@ -13,11 +13,11 @@ import sys
import uuid
import psycopg2
import sqlite3
+from functools import partial
import config
import test_setup
import regression
-from functools import partial
SERVER_GROUP = test_setup.config_data['server_group']
file_name = os.path.basename(__file__)
@@ -78,7 +78,7 @@ def get_config_data():
"db_password": srv['db_password'],
"role": "",
"sslmode": srv['sslmode'],
- "tablespace_path": srv['tablespace_path']}
+ "tablespace_path": srv.get('tablespace_path', None)}
server_data.append(data)
return server_data
@@ -138,7 +138,9 @@ def create_database(server, db_name):
connection.close()
return db_id
except Exception as exception:
- raise Exception("Error while creating database. %s" % exception)
+ exception = "Error while creating database: %s: line:%s %s" % (
+ file_name, sys.exc_traceback.tb_lineno, exception)
+ print(exception, file=sys.stderr)
def drop_database(connection, database_name):
@@ -217,54 +219,6 @@ def delete_server_from_sqlite(sid):
print(exception, file=sys.stderr)
-def create_tablespace(server, test_tablespace_name):
- try:
- connection = 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 TABLESPACE %s LOCATION '%s'" %
- (test_tablespace_name, server['tablespace_path']))
- connection.set_isolation_level(old_isolation_level)
- connection.commit()
-
- # Get 'oid' from newly created tablespace
- pg_cursor.execute(
- "SELECT ts.oid from pg_tablespace ts WHERE ts.spcname='%s'" %
- test_tablespace_name)
- oid = pg_cursor.fetchone()
- tspc_id = ''
- if oid:
- tspc_id = oid[0]
- connection.close()
- return tspc_id
- except Exception as exception:
- raise Exception("Error while creating tablespace. %s" % exception)
-
-
-def delete_tablespace(connection, test_tablespace_name):
- try:
- pg_cursor = connection.cursor()
- pg_cursor.execute("SELECT * FROM pg_tablespace ts WHERE"
- " ts.spcname='%s'" % test_tablespace_name)
- tablespace_count = len(pg_cursor.fetchall())
- if tablespace_count:
- old_isolation_level = connection.isolation_level
- connection.set_isolation_level(0)
- pg_cursor.execute("DROP TABLESPACE %s" % test_tablespace_name)
- connection.set_isolation_level(old_isolation_level)
- connection.commit()
- connection.close()
- except Exception as exception:
- exception = "%s: line:%s %s" % (
- file_name, sys.exc_traceback.tb_lineno, exception)
- print(exception, file=sys.stderr)
-
-
def create_test_server(server_info):
"""
This function create the test server which will act as parent server,
@@ -280,11 +234,6 @@ def create_test_server(server_info):
test_db_name = "test_db_%s" % str(uuid.uuid4())[1:8]
db_id = create_database(server_info, test_db_name)
- # TODO: Need to decide about test tablespace creation
- # Create tablespace
- # test_tablespace_name = "test_tablespace"
- # tablespace_id = create_tablespace(server, test_tablespace_name)
-
# Add server info to test_server_dict
regression.test_server_dict["server"].append({"server_id": srv_id,
"server": server_info})
@@ -390,7 +339,32 @@ def _drop_objects(tester):
"pg_global"]:
tablespace_name = tablespace[0]
# Delete tablespace
- delete_tablespace(connection, tablespace_name)
+ connection = get_db_connection(server_info[3],
+ server_info[4],
+ db_password,
+ server_info[1],
+ server_info[2])
+ regression.tablespace_utils.delete_tablespace(
+ connection, tablespace_name)
+
+ # Delete role
+ connection = get_db_connection(server_info[3],
+ server_info[4],
+ db_password,
+ server_info[1],
+ server_info[2])
+ pg_cursor = connection.cursor()
+ pg_cursor.execute("SELECT * FROM pg_catalog.pg_roles")
+ roles = pg_cursor.fetchall()
+ if roles:
+ for role_name in roles:
+ # Do not delete default table spaces
+ if role_name[0] not in ["postgres",
+ "enterprisedb"]:
+ role_name = role_name[0]
+ # Delete role
+ regression.roles_utils.delete_role(connection,
+ role_name)
for server_info in all_servers:
server_id = server_info[5]
view thread (2+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected]
Subject: Re: pgAdmin IV: Code changes in roles and tablespaces test cases
In-Reply-To: <CAOAJCYpLwJU+CaP+O5VWSHiMTsriu-O9tzxxk6Pt4y9zHdh_wg@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox