PEP8 fixes. Fixes #3175

This commit is contained in:
Murtuza Zabuawala 2018-03-07 11:47:01 +00:00 committed by Dave Page
parent 427314cfdf
commit 06ff05762e
9 changed files with 275 additions and 179 deletions

View File

@ -995,10 +995,13 @@ if hasattr(config, 'SECURITY_CHANGEABLE') and config.SECURITY_CHANGEABLE:
except Exception as e:
# Handle other exceptions.
logging.exception(str(e), exc_info=True)
flash(gettext(u'Error: {}\n'
u'Your password has not been changed.'
).format(e),
'danger')
flash(
gettext(
u'Error: {}\n'
u'Your password has not been changed.'
).format(e),
'danger'
)
has_error = True
if request.json is None and not has_error:
@ -1098,10 +1101,13 @@ if hasattr(config, 'SECURITY_RECOVERABLE') and config.SECURITY_RECOVERABLE:
# url_for('browser.forgot_password')
# So hard code the url '/browser/reset_password' while passing as
# parameter to slash_url_suffix function.
@blueprint.route('/reset_password' + slash_url_suffix(
'/browser/reset_password', '<token>'),
methods=['GET', 'POST'],
endpoint='reset_password')
@blueprint.route(
'/reset_password' + slash_url_suffix(
'/browser/reset_password', '<token>'
),
methods=['GET', 'POST'],
endpoint='reset_password'
)
@anonymous_user_required
def reset_password(token):
"""View function that handles a reset password request."""

View File

@ -86,6 +86,7 @@ class DatabaseModule(CollectionNodeModule):
"""
return False
blueprint = DatabaseModule(__name__)
@ -105,20 +106,52 @@ class DatabaseView(PGChildNodeView):
{'get': 'properties', 'delete': 'delete', 'put': 'update'},
{'get': 'list', 'post': 'create'}
],
'nodes': [{'get': 'node'}, {'get': 'nodes'}],
'get_databases': [{'get': 'get_databases'}, {'get': 'get_databases'}],
'sql': [{'get': 'sql'}],
'msql': [{'get': 'msql'}, {'get': 'msql'}],
'stats': [{'get': 'statistics'}, {'get': 'statistics'}],
'dependency': [{'get': 'dependencies'}],
'dependent': [{'get': 'dependents'}],
'children': [{'get': 'children'}],
'connect': [{
'get': 'connect_status', 'post': 'connect', 'delete': 'disconnect'
}],
'get_encodings': [{'get': 'get_encodings'}, {'get': 'get_encodings'}],
'get_ctypes': [{'get': 'get_ctypes'}, {'get': 'get_ctypes'}],
'vopts': [{}, {'get': 'variable_options'}]
'nodes': [
{'get': 'node'},
{'get': 'nodes'}
],
'get_databases': [
{'get': 'get_databases'},
{'get': 'get_databases'}
],
'sql': [
{'get': 'sql'}
],
'msql': [
{'get': 'msql'},
{'get': 'msql'}
],
'stats': [
{'get': 'statistics'},
{'get': 'statistics'}
],
'dependency': [
{'get': 'dependencies'}
],
'dependent': [
{'get': 'dependents'}
],
'children': [
{'get': 'children'}
],
'connect': [
{
'get': 'connect_status',
'post': 'connect',
'delete': 'disconnect'
}
],
'get_encodings': [
{'get': 'get_encodings'},
{'get': 'get_encodings'}
],
'get_ctypes': [
{'get': 'get_ctypes'},
{'get': 'get_ctypes'}
],
'vopts': [
{}, {'get': 'variable_options'}
]
})
def check_precondition(action=None):
@ -132,7 +165,11 @@ class DatabaseView(PGChildNodeView):
@wraps(f)
def wrapped(self, *args, **kwargs):
self.manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(kwargs['sid'])
self.manager = get_driver(
PG_DEFAULT_DRIVER
).connection_manager(
kwargs['sid']
)
if self.manager is None:
return gone(errormsg="Could not find the server.")
@ -152,7 +189,9 @@ class DatabaseView(PGChildNodeView):
self.conn = self.manager.connection()
# set template path for sql scripts
self.template_path = 'databases/sql/#{0}#'.format(self.manager.version)
self.template_path = 'databases/sql/#{0}#'.format(
self.manager.version
)
return f(self, *args, **kwargs)
@ -170,12 +209,16 @@ class DatabaseView(PGChildNodeView):
db_disp_res = None
params = None
if self.manager and self.manager.db_res:
db_disp_res = ", ".join(['%s'] * len(self.manager.db_res.split(',')))
db_disp_res = ", ".join(
['%s'] * len(self.manager.db_res.split(','))
)
params = tuple(self.manager.db_res.split(','))
SQL = render_template(
"/".join([self.template_path, 'properties.sql']),
conn=self.conn, last_system_oid=last_system_oid, db_restrictions=db_disp_res
conn=self.conn,
last_system_oid=last_system_oid,
db_restrictions=db_disp_res
)
status, res = self.conn.execute_dict(SQL, params)
@ -191,8 +234,8 @@ class DatabaseView(PGChildNodeView):
res = []
last_system_oid = 0 if self.blueprint.show_system_objects or \
show_system_templates else (
(self.manager.db_info[self.manager.did])['datlastsysoid'] \
if self.manager.db_info is not None and \
(self.manager.db_info[self.manager.did])['datlastsysoid']
if self.manager.db_info is not None and
self.manager.did in self.manager.db_info else 0
)
server_node_res = self.manager
@ -200,7 +243,9 @@ class DatabaseView(PGChildNodeView):
db_disp_res = None
params = None
if server_node_res and server_node_res.db_res:
db_disp_res = ", ".join(['%s']*len(server_node_res.db_res.split(',')))
db_disp_res = ", ".join(
['%s'] * len(server_node_res.db_res.split(','))
)
params = tuple(server_node_res.db_res.split(','))
SQL = render_template(
"/".join([self.template_path, 'nodes.sql']),
@ -218,7 +263,7 @@ class DatabaseView(PGChildNodeView):
connected = True
canDrop = canDisConn = False
else:
conn = self.manager.connection(dbname,did=row['did'])
conn = self.manager.connection(dbname, did=row['did'])
connected = conn.connected()
canDrop = canDisConn = True
@ -281,8 +326,8 @@ class DatabaseView(PGChildNodeView):
row['did'],
sid,
row['name'],
icon="icon-database-not-connected" if not connected \
else "pg-icon-database",
icon="icon-database-not-connected" if not connected
else "pg-icon-database",
connected=connected,
spcname=row['spcname'],
allowConn=row['datallowconn'],
@ -495,7 +540,8 @@ class DatabaseView(PGChildNodeView):
if 'datacl' in data:
data['datacl'] = parse_priv_to_db(data['datacl'], 'DATABASE')
# The below SQL will execute rest DMLs because we cannot execute CREATE with any other
# The below SQL will execute rest DMLs because we cannot execute
# CREATE with any other
SQL = render_template(
"/".join([self.template_path, 'grant.sql']),
data=data, conn=self.conn
@ -547,9 +593,8 @@ class DatabaseView(PGChildNodeView):
status, errmsg = conn.connect()
if not status:
current_app.logger.error(
"Could not create database connection for offline updates\nErr: {0}".format(
errmsg
)
"Could not create database connection for offline updates\n"
"Err: {0}".format(errmsg)
)
return internal_server_error(errmsg)
@ -594,9 +639,8 @@ class DatabaseView(PGChildNodeView):
if not status:
current_app.logger.error(
"Could not connected to database(#{0}).\nError: {1}".format(
did, errmsg
)
"Could not connected to database(#{0}).\n"
"Error: {1}".format(did, errmsg)
)
return internal_server_error(errmsg)
@ -783,9 +827,10 @@ class DatabaseView(PGChildNodeView):
)
SQL_acl = render_template(
"/".join([self.template_path, 'grant.sql']),
data=data, conn=self.conn
)
"/".join([self.template_path, 'grant.sql']),
data=data,
conn=self.conn
)
SQL = render_template(
"/".join([self.template_path, 'create.sql']),
@ -867,14 +912,21 @@ class DatabaseView(PGChildNodeView):
db_disp_res = None
params = None
if self.manager and self.manager.db_res:
db_disp_res = ", ".join(['%s'] * len(self.manager.db_res.split(',')))
db_disp_res = ", ".join(
['%s'] * len(self.manager.db_res.split(','))
)
params = tuple(self.manager.db_res.split(','))
conn = self.manager.connection()
status, res = conn.execute_dict(render_template(
"/".join([self.template_path, 'stats.sql']),
did=did, conn=conn, last_system_oid=last_system_oid, db_restrictions=db_disp_res
),params
status, res = conn.execute_dict(
render_template(
"/".join([self.template_path, 'stats.sql']),
did=did,
conn=conn,
last_system_oid=last_system_oid,
db_restrictions=db_disp_res
),
params
)
if not status:

View File

@ -60,4 +60,3 @@ class DatabaseAddTestCase(BaseTestGenerator):
self.server['host'],
self.server['port'])
utils.drop_database(connection, self.db_name)

View File

@ -17,81 +17,60 @@ DATABASE_CONNECT_URL = '/browser/database/connect/'
def get_db_data(db_owner):
"""
This function returns the database details in dict format
"""
"""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
}
]
}
],
"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:8],
"name": "db_add_%s" % str(uuid.uuid4())[1: 8],
"privileges": [],
"securities": [],
"variables": []
@ -105,7 +84,9 @@ def create_database(connection, db_name):
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)
pg_cursor.execute(
'''CREATE DATABASE "%s" TEMPLATE template0''' % db_name
)
connection.set_isolation_level(old_isolation_level)
connection.commit()
return pg_cursor
@ -134,9 +115,15 @@ def connect_database(self, server_group, server_id, db_id):
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)
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
@ -144,7 +131,13 @@ def connect_database(self, server_group, server_id, db_id):
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)
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

View File

@ -76,7 +76,10 @@ def parse_variables_from_db(db_variables):
if var_value == 'false' or var_value == 'off':
var_value = False
var_dict = {'name': var_name, 'value': var_value}
var_dict = {
'name': var_name,
'value': var_value
}
if 'user_name' in row:
var_dict['role'] = row['user_name']
if 'db_name' in row:

View File

@ -84,16 +84,28 @@ class ChangePasswordTestCase(BaseTestGenerator):
# Check for 'valid_password' exists in self to test 'valid password'
# test case
if 'valid_password' in dir(self):
response = self.tester.post('/user_management/user/', data=dict(
email=self.username, newPassword=self.password,
confirmPassword=self.password, active=1, role="2"),
follow_redirects=True)
response = self.tester.post(
'/user_management/user/',
data=dict(
email=self.username,
newPassword=self.password,
confirmPassword=self.password,
active=1,
role="2"
),
follow_redirects=True
)
user_id = json.loads(response.data.decode('utf-8'))['id']
# Logout the Administrator before login normal user
test_utils.logout_tester_account(self.tester)
response = self.tester.post('/login', data=dict(
email=self.username, password=self.password),
follow_redirects=True)
response = self.tester.post(
'/login',
data=dict(
email=self.username,
password=self.password
),
follow_redirects=True
)
self.assertEquals(response.status_code, 200)
# test the 'change password' test case
utils.change_password(self)
@ -103,7 +115,8 @@ class ChangePasswordTestCase(BaseTestGenerator):
test_utils.login_tester_account(self.tester)
response = self.tester.delete(
'/user_management/user/' + str(user_id),
follow_redirects=True)
follow_redirects=True
)
self.assertEquals(response.status_code, 200)
else:
utils.change_password(self)

View File

@ -82,10 +82,14 @@ class LoginTestCase(BaseTestGenerator):
def runTest(self):
"""This function checks login functionality."""
response = self.tester.post('/login', data=dict(
email=self.email, password=self.password),
follow_redirects=True)
response = self.tester.post(
'/login',
data=dict(
email=self.email,
password=self.password
),
follow_redirects=True
)
self.assertTrue(self.respdata in response.data.decode('utf8'))
@classmethod

View File

@ -7,12 +7,7 @@
#
##########################################################################
import json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils
from regression.test_setup import config_data
from pgadmin.browser.utils import is_version_in_range
@ -24,30 +19,45 @@ class VersionInRangeTestCase(BaseTestGenerator):
"""
scenarios = [
('TestCase for Validating pgversion 8.23 and min_version is 91000, should not show', dict(
sversion=82300,
min_version=90100,
max_version=1000000000,
scenario=2
)),
('TestCase for Validating pgversion 9.2 and should show by default', dict(
sversion=90200,
min_version=0,
max_version=1000000000,
scenario=1
)),
('TestCase for Validating pgversion 9.2 and min/max are None, should show by default', dict(
sversion=90200,
min_version=None,
max_version=None,
scenario=1
)),
('TestCase for Validating pgversion 9.6 and max is lower, should not show', dict(
sversion=90600,
min_version=None,
max_version=90400,
scenario=2
))
(
'TestCase for Validating pgversion 8.23 and min_version is 91000, '
'should not show',
dict(
sversion=82300,
min_version=90100,
max_version=1000000000,
scenario=2
)
),
(
'TestCase for Validating pgversion 9.2 and should show by default',
dict(
sversion=90200,
min_version=0,
max_version=1000000000,
scenario=1
)
),
(
'TestCase for Validating pgversion 9.2 and min/max are None, '
'should show by default',
dict(
sversion=90200,
min_version=None,
max_version=None,
scenario=1
)
),
(
'TestCase for Validating pgversion 9.6 and max is lower, should '
'not show',
dict(
sversion=90600,
min_version=None,
max_version=90400,
scenario=2
)
)
]
@classmethod
@ -62,7 +72,15 @@ class VersionInRangeTestCase(BaseTestGenerator):
self.test_result_is_false()
def test_result_is_true(self):
self.assertTrue(is_version_in_range(self.sversion, self.min_version, self.max_version))
self.assertTrue(
is_version_in_range(
self.sversion, self.min_version, self.max_version
)
)
def test_result_is_false(self):
self.assertFalse(is_version_in_range(self.sversion, self.min_version, self.max_version))
self.assertFalse(
is_version_in_range(
self.sversion, self.min_version, self.max_version
)
)

View File

@ -7,14 +7,22 @@
#
##########################################################################
def change_password(self):
response = self.tester.get('/browser/change_password', follow_redirects=True)
self.assertTrue('pgAdmin 4 Password Change' in response.data.decode(
'utf-8'))
response = self.tester.post('/browser/change_password', data=dict(
password=self.password,
new_password=self.new_password,
new_password_confirm=self.new_password_confirm),
follow_redirects=True)
def change_password(self):
response = self.tester.get(
'/browser/change_password', follow_redirects=True
)
self.assertTrue(
'pgAdmin 4 Password Change' in response.data.decode('utf-8')
)
response = self.tester.post(
'/browser/change_password',
data=dict(
password=self.password,
new_password=self.new_password,
new_password_confirm=self.new_password_confirm
),
follow_redirects=True
)
self.assertTrue(self.respdata in response.data.decode('utf-8'))