openstack
/
trove
Code Issues Proposed changes

Don't ignore E125 check 

E125: https://www.flake8rules.com/rules/E125.html
Change-Id: I8156bb87ce3e4d1a19d2df9e2c3f67f789bffa46
This commit is contained in:
wu.chunyang 2023-06-14 14:35:07 +08:00
parent 09fd606163
commit 89514c0d5b
3 changed files with 16 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tox.ini
View File

@ -58,7 +58,7 @@ ignore-path = .venv,.tox,.git,dist,doc,*egg-info,tools,etc,build,*.po,*.pot,inte
[flake8]
show-source = True
ignore = E125,E129,E402,E731,F601,F821,H301,H404,H405,H501,W503,W504,W605
ignore = E129,E402,E731,F601,F821,H301,H404,H405,H501,W503,W504,W605
enable-extensions = H203,H106
import-order-style = pep8
builtins = _

28
trove/guestagent/datastore/mysql_common/service.py
View File

@ -107,7 +107,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
LOG.debug("Associating dbs to user %(name)s at %(host)s.",
{'name': user.name, 'host': user.host})
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
q = sql_query.Query()
q.columns = ["grantee", "table_schema"]
q.tables = ["information_schema.SCHEMA_PRIVILEGES"]
@ -124,7 +124,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
"""Change the passwords of one or more existing users."""
LOG.debug("Changing the password of some users.")
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
for item in users:
LOG.debug("Changing password for user %s.", item)
user_dict = {'_name': item['name'],
@ -149,7 +149,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
if new_name or new_host or new_password:
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
if new_password is not None:
uu = sql_query.SetPassword(
user.name, host=user.host,
@ -169,7 +169,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
def create_databases(self, databases):
"""Create the list of specified databases."""
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
for item in databases:
mydb = models.MySQLSchema.deserialize(item)
mydb.check_create()
@ -185,7 +185,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
specified databases.
"""
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
for item in users:
user = models.MySQLUser.deserialize(item)
user.check_create()
@ -206,7 +206,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
def delete_database(self, database):
"""Delete the specified database."""
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
mydb = models.MySQLSchema.deserialize(database)
mydb.check_delete()
dd = sql_query.DropDatabase(mydb.name)
@ -221,7 +221,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
def delete_user_by_name(self, name, host='%'):
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
du = sql_query.DropUser(name, host=host)
t = text(str(du))
LOG.debug("delete_user_by_name: %s", t)
@ -248,7 +248,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
{'user': username, 'reason': err_msg}
)
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
q = sql_query.Query()
q.columns = ['User', 'Host']
q.tables = ['mysql.user']
@ -271,7 +271,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
user = self._get_user(username, hostname)
mydb = None # cache the model as we just want name validation
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
for database in databases:
try:
if mydb:
@ -375,7 +375,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
"be omitted from the listing: %s", ignored_user_names)
users = []
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
iq = sql_query.Query() # Inner query.
iq.columns = ['User', 'Host', "CONCAT(User, '@', Host) as Marker"]
iq.tables = ['mysql.user']
@ -418,7 +418,7 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
"""Revoke a user's permission to use a given database."""
user = self._get_user(username, hostname)
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
r = sql_query.Revoke(database=database,
user=user.name,
host=user.host)
@ -477,7 +477,7 @@ class BaseMySqlApp(service.BaseDbApp):
def execute_sql(self, sql_statement, use_flush=False):
LOG.debug("Executing SQL: %s", sql_statement)
with mysql_util.SqlClient(
self.get_engine(), use_flush=use_flush) as client:
self.get_engine(), use_flush=use_flush) as client:
return client.execute(sql_statement)
def get_data_dir(self):
@ -863,7 +863,7 @@ class BaseMySqlRootAccess(object):
"""
user = models.MySQLUser.root(password=root_password)
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
try:
cu = sql_query.CreateUser(user.name, host=user.host)
t = text(str(cu))
@ -874,7 +874,7 @@ class BaseMySqlRootAccess(object):
LOG.debug(err)
with mysql_util.SqlClient(
self.mysql_app.get_engine(), use_flush=True) as client:
self.mysql_app.get_engine(), use_flush=True) as client:
uu = sql_query.SetPassword(
user.name, host=user.host, new_password=user.password,
ds=CONF.datastore_manager, ds_version=CONF.datastore_version

2
trove/tests/unittests/taskmanager/test_models.py
View File

@ -357,7 +357,7 @@ class FreshInstanceTasksTest(BaseFreshInstanceTasksTest):
new_callable=PropertyMock,
return_value='fake-name')
def test_servers_create_block_device_mapping_v2_since_yoga(
self, mock_hostname, mock_name):
self, mock_hostname, mock_name):
# This testcase is for testing with config_drive=False.
self.freshinstancetasks.prepare_userdata = Mock(