mirror of
https://github.com/pgadmin-org/pgadmin4.git
synced 2025-02-25 18:55:31 -06:00
151 lines
3.9 KiB
Python
151 lines
3.9 KiB
Python
##########################################################################
|
|
#
|
|
# pgAdmin 4 - PostgreSQL Tools
|
|
#
|
|
# Copyright (C) 2013 - 2017, The pgAdmin Development Team
|
|
# This software is released under the PostgreSQL Licence
|
|
#
|
|
##########################################################################
|
|
|
|
import json
|
|
import uuid
|
|
|
|
from pgadmin.browser.server_groups.servers.tests import utils as server_utils
|
|
from regression.python_test_utils import test_utils as utils
|
|
|
|
DATABASE_CONNECT_URL = '/browser/database/connect/'
|
|
|
|
|
|
def get_db_data(db_owner):
|
|
"""
|
|
This function returns the database details in dict format
|
|
"""
|
|
data = {
|
|
"datconnlimit": -1,
|
|
"datowner": db_owner,
|
|
"deffuncacl": [
|
|
{
|
|
"grantee": db_owner,
|
|
"grantor": db_owner,
|
|
"privileges": [
|
|
{
|
|
"privilege_type": "X",
|
|
"privilege": True,
|
|
"with_grant": False
|
|
}
|
|
]
|
|
}
|
|
],
|
|
"defseqacl": [
|
|
{
|
|
"grantee": db_owner,
|
|
"grantor": db_owner,
|
|
"privileges": [
|
|
{
|
|
"privilege_type": "r",
|
|
"privilege": True,
|
|
"with_grant": False
|
|
},
|
|
{
|
|
"privilege_type": "w",
|
|
"privilege": True,
|
|
"with_grant": False
|
|
},
|
|
{
|
|
"privilege_type": "U",
|
|
"privilege": True,
|
|
"with_grant": False
|
|
}
|
|
]
|
|
}
|
|
],
|
|
"deftblacl": [
|
|
{
|
|
"grantee": db_owner,
|
|
"grantor": db_owner,
|
|
"privileges": [
|
|
{
|
|
"privilege_type": "a",
|
|
"privilege": True,
|
|
"with_grant": True
|
|
},
|
|
{
|
|
"privilege_type": "r",
|
|
"privilege": True,
|
|
"with_grant": False
|
|
}
|
|
]
|
|
}
|
|
],
|
|
"deftypeacl": [
|
|
{
|
|
"grantee": db_owner,
|
|
"grantor": db_owner,
|
|
"privileges": [
|
|
{
|
|
"privilege_type": "U",
|
|
"privilege": True,
|
|
"with_grant": False
|
|
}
|
|
]
|
|
}
|
|
],
|
|
"encoding": "UTF8",
|
|
"name": "db_add_%s" % str(uuid.uuid4())[1:6],
|
|
"privileges": [],
|
|
"securities": [],
|
|
"variables": []
|
|
}
|
|
return data
|
|
|
|
|
|
def create_database(connection, db_name):
|
|
"""This function used to create database"""
|
|
try:
|
|
old_isolation_level = connection.isolation_level
|
|
connection.set_isolation_level(0)
|
|
pg_cursor = connection.cursor()
|
|
pg_cursor.execute('''CREATE DATABASE "%s" TEMPLATE template0''' % db_name)
|
|
connection.set_isolation_level(old_isolation_level)
|
|
connection.commit()
|
|
return pg_cursor
|
|
except Exception as exception:
|
|
raise Exception("Error while creating database. %s" % exception)
|
|
|
|
|
|
def connect_database(self, server_group, server_id, db_id):
|
|
"""
|
|
This function verifies that database is exists and whether it connect
|
|
successfully or not
|
|
|
|
:param self: class object of test case class
|
|
:type self: class
|
|
:param server_group: server group id
|
|
:type server_group: int
|
|
:param server_id: server id
|
|
:type server_id: str
|
|
:param db_id: database id
|
|
:type db_id: str
|
|
:return: temp_db_con
|
|
:rtype: list
|
|
"""
|
|
|
|
# Verify servers
|
|
server_utils.connect_server(self, server_id)
|
|
|
|
# Connect to database
|
|
db_con = self.tester.post('{0}{1}/{2}/{3}'.format(
|
|
DATABASE_CONNECT_URL, server_group, server_id, db_id),
|
|
follow_redirects=True)
|
|
assert db_con.status_code == 200
|
|
db_con = json.loads(db_con.data.decode('utf-8'))
|
|
return db_con
|
|
|
|
|
|
def disconnect_database(self, server_id, db_id):
|
|
"""This function disconnect the db"""
|
|
db_con = self.tester.delete('{0}{1}/{2}/{3}'.format(
|
|
'browser/database/connect/', utils.SERVER_GROUP, server_id, db_id),
|
|
follow_redirects=True)
|
|
assert db_con.status_code == 200
|