Update pre-commit config and fix some pylint & bandit warnings

This commit is contained in:
Benjamin Renard 2023-10-27 13:35:59 +02:00
parent 8e0e75f30e
commit 8a0a65465d
Signed by: bn8
GPG key ID: 3E2E1CE1907115BC
9 changed files with 222 additions and 205 deletions

View file

@ -1,45 +1,44 @@
# Pre-commit hooks to run tests and ensure code is cleaned. # Pre-commit hooks to run tests and ensure code is cleaned.
# See https://pre-commit.com for more information # See https://pre-commit.com for more information
repos: repos:
- repo: https://github.com/asottile/pyupgrade - repo: https://github.com/asottile/pyupgrade
rev: v3.3.1 rev: v3.3.1
hooks: hooks:
- id: pyupgrade - id: pyupgrade
args: ['--keep-percent-format', '--py37-plus'] args: ['--keep-percent-format', '--py37-plus']
- repo: https://github.com/psf/black - repo: https://github.com/psf/black
rev: 22.12.0 rev: 22.12.0
hooks: hooks:
- id: black - id: black
args: ['--target-version', 'py37', '--line-length', '100'] args: ['--target-version', 'py37', '--line-length', '100']
- repo: https://github.com/PyCQA/isort - repo: https://github.com/PyCQA/isort
rev: 5.11.5 rev: 5.11.5
hooks: hooks:
- id: isort - id: isort
args: ['--profile', 'black', '--line-length', '100'] args: ['--profile', 'black', '--line-length', '100']
- repo: https://github.com/PyCQA/flake8 - repo: https://github.com/PyCQA/flake8
rev: 6.0.0 rev: 6.0.0
hooks: hooks:
- id: flake8 - id: flake8
args: ['--max-line-length=100'] args: ['--max-line-length=100']
- repo: local - repo: local
hooks: hooks:
- id: pylint - id: pylint
name: pylint name: pylint
entry: pylint --extension-pkg-whitelist=cx_Oracle entry: pylint --extension-pkg-whitelist=cx_Oracle
language: system language: system
types: [python] types: [python]
require_serial: true require_serial: true
- repo: https://github.com/Lucas-C/pre-commit-hooks-bandit - repo: https://github.com/PyCQA/bandit
rev: v1.0.5 rev: 1.7.5
hooks: hooks:
- id: python-bandit-vulnerability-check - id: bandit
name: bandit args: [--skip, "B101", --recursive, "mylib"]
args: [--skip, "B101", --recursive, mylib] - repo: local
- repo: local hooks:
hooks: - id: pytest
- id: pytest name: pytest
name: pytest entry: python3 -m pytest tests
entry: python3 -m pytest tests language: system
language: system types: [python]
types: [python] pass_filenames: false
pass_filenames: false

View file

@ -27,6 +27,10 @@ DEFAULT_CONFIG_DIRPATH = os.path.expanduser("./")
DEFAULT_CONSOLE_LOG_FORMAT = "%(asctime)s - %(module)s:%(lineno)d - %(levelname)s - %(message)s" DEFAULT_CONSOLE_LOG_FORMAT = "%(asctime)s - %(module)s:%(lineno)d - %(levelname)s - %(message)s"
class ConfigException(BaseException):
"""Configuration exception"""
class BaseOption: # pylint: disable=too-many-instance-attributes class BaseOption: # pylint: disable=too-many-instance-attributes
"""Base configuration option class""" """Base configuration option class"""
@ -161,12 +165,12 @@ class BaseOption: # pylint: disable=too-many-instance-attributes
args = [self.parser_argument_name] args = [self.parser_argument_name]
if self.short_arg: if self.short_arg:
args.append(self.short_arg) args.append(self.short_arg)
kwargs = dict( kwargs = {
action=self.parser_action, "action": self.parser_action,
dest=self.parser_dest, "dest": self.parser_dest,
help=self.parser_help, "help": self.parser_help,
default=self.default, "default": self.default,
) }
if self.parser_type: # pylint: disable=using-constant-test if self.parser_type: # pylint: disable=using-constant-test
kwargs["type"] = self.parser_type kwargs["type"] = self.parser_type
@ -1132,7 +1136,7 @@ class ConfigurableObject:
elif self._config_name: elif self._config_name:
self._options_prefix = self._config_name + "_" self._options_prefix = self._config_name + "_"
else: else:
raise Exception(f"No configuration name defined for {__name__}") raise ConfigException(f"No configuration name defined for {__name__}")
if config: if config:
self._config = config self._config = config
@ -1141,7 +1145,7 @@ class ConfigurableObject:
elif self._config_name: elif self._config_name:
self._config_section = self._config_name self._config_section = self._config_name
else: else:
raise Exception(f"No configuration name defined for {__name__}") raise ConfigException(f"No configuration name defined for {__name__}")
def _get_option(self, option, default=None, required=False): def _get_option(self, option, default=None, required=False):
"""Retreive option value""" """Retreive option value"""
@ -1274,7 +1278,7 @@ class ConfigSectionAsDictWrapper:
self.__section.set(key, value) self.__section.set(key, value)
def __delitem__(self, key): def __delitem__(self, key):
raise Exception("Deleting a configuration option is not supported") raise ConfigException("Deleting a configuration option is not supported")
# pylint: disable=too-few-public-methods # pylint: disable=too-few-public-methods

View file

@ -566,15 +566,15 @@ if __name__ == "__main__":
catch_all_addr=options.email_catch_all, catch_all_addr=options.email_catch_all,
just_try=options.just_try, just_try=options.just_try,
encoding=options.email_encoding, encoding=options.email_encoding,
templates=dict( templates={
test=dict( "test": {
subject="Test email", "subject": "Test email",
text=( "text": (
"Just a test email sent at {sent_date}." "Just a test email sent at {sent_date}."
if not options.test_mako if not options.test_mako
else MakoTemplate("Just a test email sent at ${sent_date | h}.") # nosec else MakoTemplate("Just a test email sent at ${sent_date | h}.") # nosec
), ),
html=( "html": (
"<strong>Just a test email.</strong> <small>(sent at {sent_date | h})</small>" "<strong>Just a test email.</strong> <small>(sent at {sent_date | h})</small>"
if not options.test_mako if not options.test_mako
else MakoTemplate( # nosec else MakoTemplate( # nosec
@ -582,8 +582,8 @@ if __name__ == "__main__":
"<small>(sent at ${sent_date | h})</small>" "<small>(sent at ${sent_date | h})</small>"
) )
), ),
) }
), },
) )
logging.info("Send a test email to %s", options.test_to) logging.info("Send a test email to %s", options.test_to)

View file

@ -120,7 +120,7 @@ class LdapServer:
return ldap.SCOPE_ONELEVEL # pylint: disable=no-member return ldap.SCOPE_ONELEVEL # pylint: disable=no-member
if scope == "sub": if scope == "sub":
return ldap.SCOPE_SUBTREE # pylint: disable=no-member return ldap.SCOPE_SUBTREE # pylint: disable=no-member
raise Exception(f'Unknown LDAP scope "{scope}"') raise LdapServerException(f'Unknown LDAP scope "{scope}"')
def search(self, basedn, filterstr=None, attrs=None, sizelimit=None, scope=None): def search(self, basedn, filterstr=None, attrs=None, sizelimit=None, scope=None):
"""Run a search on LDAP server""" """Run a search on LDAP server"""
@ -399,18 +399,16 @@ class LdapServer:
return default return default
class LdapServerException(BaseException): class LdapException(BaseException):
"""Generic LDAP exception"""
class LdapServerException(LdapException):
"""Generic exception raised by LdapServer""" """Generic exception raised by LdapServer"""
def __init__(self, msg):
BaseException.__init__(self, msg)
class LdapClientException(LdapException):
class LdapClientException(LdapServerException): """Generic exception raised by LdapClient"""
"""Generic exception raised by LdapServer"""
def __init__(self, msg):
LdapServerException.__init__(self, msg)
class LdapClient: class LdapClient:
@ -541,7 +539,7 @@ class LdapClient:
:param dn: The object DN :param dn: The object DN
:param attrs: The object attributes as return by python-ldap search :param attrs: The object attributes as return by python-ldap search
""" """
obj = dict(dn=dn) obj = {"dn": dn}
for attr in attrs: for attr in attrs:
obj[attr] = [self.decode(v) for v in self._conn.get_attr(attrs, attr, all_values=True)] obj[attr] = [self.decode(v) for v in self._conn.get_attr(attrs, attr, all_values=True)]
return obj return obj
@ -1018,7 +1016,7 @@ def parse_datetime(value, to_timezone=None, default_timezone=None, naive=None):
elif isinstance(default_timezone, datetime.tzinfo): elif isinstance(default_timezone, datetime.tzinfo):
date = date.replace(tzinfo=default_timezone) date = date.replace(tzinfo=default_timezone)
else: else:
raise Exception("It's not supposed to happen!") raise LdapException("It's not supposed to happen!")
elif naive: elif naive:
return date.replace(tzinfo=None) return date.replace(tzinfo=None)
if to_timezone: if to_timezone:
@ -1078,7 +1076,7 @@ def format_datetime(value, from_timezone=None, to_timezone=None, naive=None):
elif isinstance(from_timezone, datetime.tzinfo): elif isinstance(from_timezone, datetime.tzinfo):
from_value = value.replace(tzinfo=from_timezone) from_value = value.replace(tzinfo=from_timezone)
else: else:
raise Exception("It's not supposed to happen!") raise LdapException("It's not supposed to happen!")
elif naive: elif naive:
from_value = value.replace(tzinfo=pytz.utc) from_value = value.replace(tzinfo=pytz.utc)
else: else:

View file

@ -126,14 +126,14 @@ def test_add_option_custom_args():
section = config.add_section("my_section") section = config.add_section("my_section")
assert isinstance(section, ConfigSection) assert isinstance(section, ConfigSection)
name = "my_option" name = "my_option"
kwargs = dict( kwargs = {
default="default value", "default": "default value",
comment="my comment", "comment": "my comment",
no_arg=True, "no_arg": True,
arg="--my-option", "arg": "--my-option",
short_arg="-M", "short_arg": "-M",
arg_help="My help", "arg_help": "My help",
) }
option = section.add_option(StringOption, name, **kwargs) option = section.add_option(StringOption, name, **kwargs)
assert isinstance(option, StringOption) assert isinstance(option, StringOption)
assert name in section.options and section.options[name] == option assert name in section.options and section.options[name] == option

View file

@ -74,9 +74,14 @@ class FakeMySQLdb:
just_try = False just_try = False
def __init__(self, **kwargs): def __init__(self, **kwargs):
allowed_kwargs = dict( allowed_kwargs = {
db=str, user=str, passwd=(str, None), host=str, charset=str, use_unicode=bool "db": str,
) "user": str,
"passwd": (str, None),
"host": str,
"charset": str,
"use_unicode": bool,
}
for arg, value in kwargs.items(): for arg, value in kwargs.items():
assert arg in allowed_kwargs, f'Invalid arg {arg}="{value}"' assert arg in allowed_kwargs, f'Invalid arg {arg}="{value}"'
assert isinstance( assert isinstance(
@ -200,21 +205,23 @@ mock_doSelect_just_try = mock_doSQL_just_try
def test_combine_params_with_to_add_parameter(): def test_combine_params_with_to_add_parameter():
assert MyDB._combine_params(dict(test1=1), dict(test2=2)) == dict(test1=1, test2=2) assert MyDB._combine_params({"test1": 1}, {"test2": 2}) == {"test1": 1, "test2": 2}
def test_combine_params_with_kargs(): def test_combine_params_with_kargs():
assert MyDB._combine_params(dict(test1=1), test2=2) == dict(test1=1, test2=2) assert MyDB._combine_params({"test1": 1}, test2=2) == {"test1": 1, "test2": 2}
def test_combine_params_with_kargs_and_to_add_parameter(): def test_combine_params_with_kargs_and_to_add_parameter():
assert MyDB._combine_params(dict(test1=1), dict(test2=2), test3=3) == dict( assert MyDB._combine_params({"test1": 1}, {"test2": 2}, test3=3) == {
test1=1, test2=2, test3=3 "test1": 1,
) "test2": 2,
"test3": 3,
}
def test_format_where_clauses_params_are_preserved(): def test_format_where_clauses_params_are_preserved():
args = ("test = test", dict(test1=1)) args = ("test = test", {"test1": 1})
assert MyDB._format_where_clauses(*args) == args assert MyDB._format_where_clauses(*args) == args
@ -223,12 +230,12 @@ def test_format_where_clauses_raw():
def test_format_where_clauses_tuple_clause_with_params(): def test_format_where_clauses_tuple_clause_with_params():
where_clauses = ("test1 = %(test1)s AND test2 = %(test2)s", dict(test1=1, test2=2)) where_clauses = ("test1 = %(test1)s AND test2 = %(test2)s", {"test1": 1, "test2": 2})
assert MyDB._format_where_clauses(where_clauses) == where_clauses assert MyDB._format_where_clauses(where_clauses) == where_clauses
def test_format_where_clauses_dict(): def test_format_where_clauses_dict():
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
assert MyDB._format_where_clauses(where_clauses) == ( assert MyDB._format_where_clauses(where_clauses) == (
"`test1` = %(test1)s AND `test2` = %(test2)s", "`test1` = %(test1)s AND `test2` = %(test2)s",
where_clauses, where_clauses,
@ -236,15 +243,15 @@ def test_format_where_clauses_dict():
def test_format_where_clauses_combined_types(): def test_format_where_clauses_combined_types():
where_clauses = ("test1 = 1", ("test2 LIKE %(test2)s", dict(test2=2)), dict(test3=3, test4=4)) where_clauses = ("test1 = 1", ("test2 LIKE %(test2)s", {"test2": 2}), {"test3": 3, "test4": 4})
assert MyDB._format_where_clauses(where_clauses) == ( assert MyDB._format_where_clauses(where_clauses) == (
"test1 = 1 AND test2 LIKE %(test2)s AND `test3` = %(test3)s AND `test4` = %(test4)s", "test1 = 1 AND test2 LIKE %(test2)s AND `test3` = %(test3)s AND `test4` = %(test4)s",
dict(test2=2, test3=3, test4=4), {"test2": 2, "test3": 3, "test4": 4},
) )
def test_format_where_clauses_with_where_op(): def test_format_where_clauses_with_where_op():
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
assert MyDB._format_where_clauses(where_clauses, where_op="OR") == ( assert MyDB._format_where_clauses(where_clauses, where_op="OR") == (
"`test1` = %(test1)s OR `test2` = %(test2)s", "`test1` = %(test1)s OR `test2` = %(test2)s",
where_clauses, where_clauses,
@ -253,7 +260,7 @@ def test_format_where_clauses_with_where_op():
def test_add_where_clauses(): def test_add_where_clauses():
sql = "SELECT * FROM table" sql = "SELECT * FROM table"
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
assert MyDB._add_where_clauses(sql, None, where_clauses) == ( assert MyDB._add_where_clauses(sql, None, where_clauses) == (
sql + " WHERE `test1` = %(test1)s AND `test2` = %(test2)s", sql + " WHERE `test1` = %(test1)s AND `test2` = %(test2)s",
where_clauses, where_clauses,
@ -262,11 +269,11 @@ def test_add_where_clauses():
def test_add_where_clauses_preserved_params(): def test_add_where_clauses_preserved_params():
sql = "SELECT * FROM table" sql = "SELECT * FROM table"
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
params = dict(fake1=1) params = {"fake1": 1}
assert MyDB._add_where_clauses(sql, params.copy(), where_clauses) == ( assert MyDB._add_where_clauses(sql, params.copy(), where_clauses) == (
sql + " WHERE `test1` = %(test1)s AND `test2` = %(test2)s", sql + " WHERE `test1` = %(test1)s AND `test2` = %(test2)s",
dict(**where_clauses, **params), {**where_clauses, **params},
) )
@ -281,11 +288,11 @@ def test_add_where_clauses_with_op():
def test_add_where_clauses_with_duplicated_field(): def test_add_where_clauses_with_duplicated_field():
sql = "UPDATE table SET test1=%(test1)s" sql = "UPDATE table SET test1=%(test1)s"
params = dict(test1="new_value") params = {"test1": "new_value"}
where_clauses = dict(test1="where_value") where_clauses = {"test1": "where_value"}
assert MyDB._add_where_clauses(sql, params, where_clauses) == ( assert MyDB._add_where_clauses(sql, params, where_clauses) == (
sql + " WHERE `test1` = %(test1_1)s", sql + " WHERE `test1` = %(test1_1)s",
dict(test1="new_value", test1_1="where_value"), {"test1": "new_value", "test1_1": "where_value"},
) )
@ -295,7 +302,7 @@ def test_quote_table_name():
def test_insert(mocker, test_mydb): def test_insert(mocker, test_mydb):
values = dict(test1=1, test2=2) values = {"test1": 1, "test2": 2}
mocker.patch( mocker.patch(
"mylib.mysql.MyDB.doSQL", "mylib.mysql.MyDB.doSQL",
generate_mock_doSQL( generate_mock_doSQL(
@ -308,18 +315,18 @@ def test_insert(mocker, test_mydb):
def test_insert_just_try(mocker, test_mydb): def test_insert_just_try(mocker, test_mydb):
mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSQL_just_try) mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSQL_just_try)
assert test_mydb.insert("mytable", dict(test1=1, test2=2), just_try=True) assert test_mydb.insert("mytable", {"test1": 1, "test2": 2}, just_try=True)
def test_update(mocker, test_mydb): def test_update(mocker, test_mydb):
values = dict(test1=1, test2=2) values = {"test1": 1, "test2": 2}
where_clauses = dict(test3=3, test4=4) where_clauses = {"test3": 3, "test4": 4}
mocker.patch( mocker.patch(
"mylib.mysql.MyDB.doSQL", "mylib.mysql.MyDB.doSQL",
generate_mock_doSQL( generate_mock_doSQL(
"UPDATE `mytable` SET `test1` = %(test1)s, `test2` = %(test2)s WHERE `test3` =" "UPDATE `mytable` SET `test1` = %(test1)s, `test2` = %(test2)s WHERE `test3` ="
" %(test3)s AND `test4` = %(test4)s", " %(test3)s AND `test4` = %(test4)s",
dict(**values, **where_clauses), {**values, **where_clauses},
), ),
) )
@ -328,11 +335,11 @@ def test_update(mocker, test_mydb):
def test_update_just_try(mocker, test_mydb): def test_update_just_try(mocker, test_mydb):
mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSQL_just_try) mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSQL_just_try)
assert test_mydb.update("mytable", dict(test1=1, test2=2), None, just_try=True) assert test_mydb.update("mytable", {"test1": 1, "test2": 2}, None, just_try=True)
def test_delete(mocker, test_mydb): def test_delete(mocker, test_mydb):
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
mocker.patch( mocker.patch(
"mylib.mysql.MyDB.doSQL", "mylib.mysql.MyDB.doSQL",
generate_mock_doSQL( generate_mock_doSQL(
@ -361,17 +368,17 @@ def test_truncate_just_try(mocker, test_mydb):
def test_select(mocker, test_mydb): def test_select(mocker, test_mydb):
fields = ("field1", "field2") fields = ("field1", "field2")
where_clauses = dict(test3=3, test4=4) where_clauses = {"test3": 3, "test4": 4}
expected_return = [ expected_return = [
dict(field1=1, field2=2), {"field1": 1, "field2": 2},
dict(field1=2, field2=3), {"field1": 2, "field2": 3},
] ]
order_by = "field1, DESC" order_by = "field1, DESC"
mocker.patch( mocker.patch(
"mylib.mysql.MyDB.doSelect", "mylib.mysql.MyDB.doSelect",
generate_mock_doSQL( generate_mock_doSQL(
"SELECT `field1`, `field2` FROM `mytable` WHERE `test3` = %(test3)s AND `test4` =" "SELECT `field1`, `field2` FROM `mytable` WHERE `test3` = %(test3)s AND `test4` ="
" %(test4)s ORDER BY " + order_by, " %(test4)s ORDER BY " + order_by, # nosec: B608
where_clauses, where_clauses,
expected_return, expected_return,
), ),
@ -397,14 +404,14 @@ def test_select_just_try(mocker, test_mydb):
def test_connect(mocker, test_mydb): def test_connect(mocker, test_mydb):
expected_kwargs = dict( expected_kwargs = {
db=test_mydb._db, "db": test_mydb._db,
user=test_mydb._user, "user": test_mydb._user,
host=test_mydb._host, "host": test_mydb._host,
passwd=test_mydb._pwd, "passwd": test_mydb._pwd,
charset=test_mydb._charset, "charset": test_mydb._charset,
use_unicode=True, "use_unicode": True,
) }
mocker.patch("MySQLdb.connect", generate_mock_args(expected_kwargs=expected_kwargs)) mocker.patch("MySQLdb.connect", generate_mock_args(expected_kwargs=expected_kwargs))
@ -421,7 +428,7 @@ def test_close_connected(fake_connected_mydb):
def test_doSQL(fake_connected_mydb): def test_doSQL(fake_connected_mydb):
fake_connected_mydb._conn.expected_sql = "DELETE FROM table WHERE test1 = %(test1)s" fake_connected_mydb._conn.expected_sql = "DELETE FROM table WHERE test1 = %(test1)s"
fake_connected_mydb._conn.expected_params = dict(test1=1) fake_connected_mydb._conn.expected_params = {"test1": 1}
fake_connected_mydb.doSQL( fake_connected_mydb.doSQL(
fake_connected_mydb._conn.expected_sql, fake_connected_mydb._conn.expected_params fake_connected_mydb._conn.expected_sql, fake_connected_mydb._conn.expected_params
) )
@ -443,8 +450,8 @@ def test_doSQL_on_exception(fake_connected_mydb):
def test_doSelect(fake_connected_mydb): def test_doSelect(fake_connected_mydb):
fake_connected_mydb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s" fake_connected_mydb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s"
fake_connected_mydb._conn.expected_params = dict(test1=1) fake_connected_mydb._conn.expected_params = {"test1": 1}
fake_connected_mydb._conn.expected_return = [dict(test1=1)] fake_connected_mydb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_mydb.doSelect( fake_connected_mydb.doSelect(
fake_connected_mydb._conn.expected_sql, fake_connected_mydb._conn.expected_params fake_connected_mydb._conn.expected_sql, fake_connected_mydb._conn.expected_params
@ -455,7 +462,7 @@ def test_doSelect(fake_connected_mydb):
def test_doSelect_without_params(fake_connected_mydb): def test_doSelect_without_params(fake_connected_mydb):
fake_connected_mydb._conn.expected_sql = "SELECT * FROM table" fake_connected_mydb._conn.expected_sql = "SELECT * FROM table"
fake_connected_mydb._conn.expected_return = [dict(test1=1)] fake_connected_mydb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_mydb.doSelect(fake_connected_mydb._conn.expected_sql) fake_connected_mydb.doSelect(fake_connected_mydb._conn.expected_sql)
== fake_connected_mydb._conn.expected_return == fake_connected_mydb._conn.expected_return
@ -469,8 +476,8 @@ def test_doSelect_on_exception(fake_connected_mydb):
def test_doSelect_just_try(fake_connected_just_try_mydb): def test_doSelect_just_try(fake_connected_just_try_mydb):
fake_connected_just_try_mydb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s" fake_connected_just_try_mydb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s"
fake_connected_just_try_mydb._conn.expected_params = dict(test1=1) fake_connected_just_try_mydb._conn.expected_params = {"test1": 1}
fake_connected_just_try_mydb._conn.expected_return = [dict(test1=1)] fake_connected_just_try_mydb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_just_try_mydb.doSelect( fake_connected_just_try_mydb.doSelect(
fake_connected_just_try_mydb._conn.expected_sql, fake_connected_just_try_mydb._conn.expected_sql,

View file

@ -73,7 +73,7 @@ class FakeCXOracle:
just_try = False just_try = False
def __init__(self, **kwargs): def __init__(self, **kwargs):
allowed_kwargs = dict(dsn=str, user=str, password=(str, None)) allowed_kwargs = {"dsn": str, "user": str, "password": (str, None)}
for arg, value in kwargs.items(): for arg, value in kwargs.items():
assert arg in allowed_kwargs, f"Invalid arg {arg}='{value}'" assert arg in allowed_kwargs, f"Invalid arg {arg}='{value}'"
assert isinstance( assert isinstance(
@ -197,21 +197,23 @@ mock_doSelect_just_try = mock_doSQL_just_try
def test_combine_params_with_to_add_parameter(): def test_combine_params_with_to_add_parameter():
assert OracleDB._combine_params(dict(test1=1), dict(test2=2)) == dict(test1=1, test2=2) assert OracleDB._combine_params({"test1": 1}, {"test2": 2}) == {"test1": 1, "test2": 2}
def test_combine_params_with_kargs(): def test_combine_params_with_kargs():
assert OracleDB._combine_params(dict(test1=1), test2=2) == dict(test1=1, test2=2) assert OracleDB._combine_params({"test1": 1}, test2=2) == {"test1": 1, "test2": 2}
def test_combine_params_with_kargs_and_to_add_parameter(): def test_combine_params_with_kargs_and_to_add_parameter():
assert OracleDB._combine_params(dict(test1=1), dict(test2=2), test3=3) == dict( assert OracleDB._combine_params({"test1": 1}, {"test2": 2}, test3=3) == {
test1=1, test2=2, test3=3 "test1": 1,
) "test2": 2,
"test3": 3,
}
def test_format_where_clauses_params_are_preserved(): def test_format_where_clauses_params_are_preserved():
args = ("test = test", dict(test1=1)) args = ("test = test", {"test1": 1})
assert OracleDB._format_where_clauses(*args) == args assert OracleDB._format_where_clauses(*args) == args
@ -220,12 +222,12 @@ def test_format_where_clauses_raw():
def test_format_where_clauses_tuple_clause_with_params(): def test_format_where_clauses_tuple_clause_with_params():
where_clauses = ("test1 = :test1 AND test2 = :test2", dict(test1=1, test2=2)) where_clauses = ("test1 = :test1 AND test2 = :test2", {"test1": 1, "test2": 2})
assert OracleDB._format_where_clauses(where_clauses) == where_clauses assert OracleDB._format_where_clauses(where_clauses) == where_clauses
def test_format_where_clauses_dict(): def test_format_where_clauses_dict():
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
assert OracleDB._format_where_clauses(where_clauses) == ( assert OracleDB._format_where_clauses(where_clauses) == (
'"test1" = :test1 AND "test2" = :test2', '"test1" = :test1 AND "test2" = :test2',
where_clauses, where_clauses,
@ -233,15 +235,15 @@ def test_format_where_clauses_dict():
def test_format_where_clauses_combined_types(): def test_format_where_clauses_combined_types():
where_clauses = ("test1 = 1", ("test2 LIKE :test2", dict(test2=2)), dict(test3=3, test4=4)) where_clauses = ("test1 = 1", ("test2 LIKE :test2", {"test2": 2}), {"test3": 3, "test4": 4})
assert OracleDB._format_where_clauses(where_clauses) == ( assert OracleDB._format_where_clauses(where_clauses) == (
'test1 = 1 AND test2 LIKE :test2 AND "test3" = :test3 AND "test4" = :test4', 'test1 = 1 AND test2 LIKE :test2 AND "test3" = :test3 AND "test4" = :test4',
dict(test2=2, test3=3, test4=4), {"test2": 2, "test3": 3, "test4": 4},
) )
def test_format_where_clauses_with_where_op(): def test_format_where_clauses_with_where_op():
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
assert OracleDB._format_where_clauses(where_clauses, where_op="OR") == ( assert OracleDB._format_where_clauses(where_clauses, where_op="OR") == (
'"test1" = :test1 OR "test2" = :test2', '"test1" = :test1 OR "test2" = :test2',
where_clauses, where_clauses,
@ -250,7 +252,7 @@ def test_format_where_clauses_with_where_op():
def test_add_where_clauses(): def test_add_where_clauses():
sql = "SELECT * FROM table" sql = "SELECT * FROM table"
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
assert OracleDB._add_where_clauses(sql, None, where_clauses) == ( assert OracleDB._add_where_clauses(sql, None, where_clauses) == (
sql + ' WHERE "test1" = :test1 AND "test2" = :test2', sql + ' WHERE "test1" = :test1 AND "test2" = :test2',
where_clauses, where_clauses,
@ -259,11 +261,11 @@ def test_add_where_clauses():
def test_add_where_clauses_preserved_params(): def test_add_where_clauses_preserved_params():
sql = "SELECT * FROM table" sql = "SELECT * FROM table"
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
params = dict(fake1=1) params = {"fake1": 1}
assert OracleDB._add_where_clauses(sql, params.copy(), where_clauses) == ( assert OracleDB._add_where_clauses(sql, params.copy(), where_clauses) == (
sql + ' WHERE "test1" = :test1 AND "test2" = :test2', sql + ' WHERE "test1" = :test1 AND "test2" = :test2',
dict(**where_clauses, **params), {**where_clauses, **params},
) )
@ -278,11 +280,11 @@ def test_add_where_clauses_with_op():
def test_add_where_clauses_with_duplicated_field(): def test_add_where_clauses_with_duplicated_field():
sql = "UPDATE table SET test1=:test1" sql = "UPDATE table SET test1=:test1"
params = dict(test1="new_value") params = {"test1": "new_value"}
where_clauses = dict(test1="where_value") where_clauses = {"test1": "where_value"}
assert OracleDB._add_where_clauses(sql, params, where_clauses) == ( assert OracleDB._add_where_clauses(sql, params, where_clauses) == (
sql + ' WHERE "test1" = :test1_1', sql + ' WHERE "test1" = :test1_1',
dict(test1="new_value", test1_1="where_value"), {"test1": "new_value", "test1_1": "where_value"},
) )
@ -292,7 +294,7 @@ def test_quote_table_name():
def test_insert(mocker, test_oracledb): def test_insert(mocker, test_oracledb):
values = dict(test1=1, test2=2) values = {"test1": 1, "test2": 2}
mocker.patch( mocker.patch(
"mylib.oracle.OracleDB.doSQL", "mylib.oracle.OracleDB.doSQL",
generate_mock_doSQL( generate_mock_doSQL(
@ -305,18 +307,18 @@ def test_insert(mocker, test_oracledb):
def test_insert_just_try(mocker, test_oracledb): def test_insert_just_try(mocker, test_oracledb):
mocker.patch("mylib.oracle.OracleDB.doSQL", mock_doSQL_just_try) mocker.patch("mylib.oracle.OracleDB.doSQL", mock_doSQL_just_try)
assert test_oracledb.insert("mytable", dict(test1=1, test2=2), just_try=True) assert test_oracledb.insert("mytable", {"test1": 1, "test2": 2}, just_try=True)
def test_update(mocker, test_oracledb): def test_update(mocker, test_oracledb):
values = dict(test1=1, test2=2) values = {"test1": 1, "test2": 2}
where_clauses = dict(test3=3, test4=4) where_clauses = {"test3": 3, "test4": 4}
mocker.patch( mocker.patch(
"mylib.oracle.OracleDB.doSQL", "mylib.oracle.OracleDB.doSQL",
generate_mock_doSQL( generate_mock_doSQL(
'UPDATE "mytable" SET "test1" = :test1, "test2" = :test2 WHERE "test3" = :test3 AND' 'UPDATE "mytable" SET "test1" = :test1, "test2" = :test2 WHERE "test3" = :test3 AND'
' "test4" = :test4', ' "test4" = :test4',
dict(**values, **where_clauses), {**values, **where_clauses},
), ),
) )
@ -325,11 +327,11 @@ def test_update(mocker, test_oracledb):
def test_update_just_try(mocker, test_oracledb): def test_update_just_try(mocker, test_oracledb):
mocker.patch("mylib.oracle.OracleDB.doSQL", mock_doSQL_just_try) mocker.patch("mylib.oracle.OracleDB.doSQL", mock_doSQL_just_try)
assert test_oracledb.update("mytable", dict(test1=1, test2=2), None, just_try=True) assert test_oracledb.update("mytable", {"test1": 1, "test2": 2}, None, just_try=True)
def test_delete(mocker, test_oracledb): def test_delete(mocker, test_oracledb):
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
mocker.patch( mocker.patch(
"mylib.oracle.OracleDB.doSQL", "mylib.oracle.OracleDB.doSQL",
generate_mock_doSQL( generate_mock_doSQL(
@ -360,17 +362,17 @@ def test_truncate_just_try(mocker, test_oracledb):
def test_select(mocker, test_oracledb): def test_select(mocker, test_oracledb):
fields = ("field1", "field2") fields = ("field1", "field2")
where_clauses = dict(test3=3, test4=4) where_clauses = {"test3": 3, "test4": 4}
expected_return = [ expected_return = [
dict(field1=1, field2=2), {"field1": 1, "field2": 2},
dict(field1=2, field2=3), {"field1": 2, "field2": 3},
] ]
order_by = "field1, DESC" order_by = "field1, DESC"
mocker.patch( mocker.patch(
"mylib.oracle.OracleDB.doSelect", "mylib.oracle.OracleDB.doSelect",
generate_mock_doSQL( generate_mock_doSQL(
'SELECT "field1", "field2" FROM "mytable" WHERE "test3" = :test3 AND "test4" = :test4' 'SELECT "field1", "field2" FROM "mytable" WHERE "test3" = :test3 AND "test4" = :test4'
" ORDER BY " + order_by, " ORDER BY " + order_by, # nosec: B608
where_clauses, where_clauses,
expected_return, expected_return,
), ),
@ -398,9 +400,11 @@ def test_select_just_try(mocker, test_oracledb):
def test_connect(mocker, test_oracledb): def test_connect(mocker, test_oracledb):
expected_kwargs = dict( expected_kwargs = {
dsn=test_oracledb._dsn, user=test_oracledb._user, password=test_oracledb._pwd "dsn": test_oracledb._dsn,
) "user": test_oracledb._user,
"password": test_oracledb._pwd,
}
mocker.patch("cx_Oracle.connect", generate_mock_args(expected_kwargs=expected_kwargs)) mocker.patch("cx_Oracle.connect", generate_mock_args(expected_kwargs=expected_kwargs))
@ -417,7 +421,7 @@ def test_close_connected(fake_connected_oracledb):
def test_doSQL(fake_connected_oracledb): def test_doSQL(fake_connected_oracledb):
fake_connected_oracledb._conn.expected_sql = "DELETE FROM table WHERE test1 = :test1" fake_connected_oracledb._conn.expected_sql = "DELETE FROM table WHERE test1 = :test1"
fake_connected_oracledb._conn.expected_params = dict(test1=1) fake_connected_oracledb._conn.expected_params = {"test1": 1}
fake_connected_oracledb.doSQL( fake_connected_oracledb.doSQL(
fake_connected_oracledb._conn.expected_sql, fake_connected_oracledb._conn.expected_params fake_connected_oracledb._conn.expected_sql, fake_connected_oracledb._conn.expected_params
) )
@ -439,8 +443,8 @@ def test_doSQL_on_exception(fake_connected_oracledb):
def test_doSelect(fake_connected_oracledb): def test_doSelect(fake_connected_oracledb):
fake_connected_oracledb._conn.expected_sql = "SELECT * FROM table WHERE test1 = :test1" fake_connected_oracledb._conn.expected_sql = "SELECT * FROM table WHERE test1 = :test1"
fake_connected_oracledb._conn.expected_params = dict(test1=1) fake_connected_oracledb._conn.expected_params = {"test1": 1}
fake_connected_oracledb._conn.expected_return = [dict(test1=1)] fake_connected_oracledb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_oracledb.doSelect( fake_connected_oracledb.doSelect(
fake_connected_oracledb._conn.expected_sql, fake_connected_oracledb._conn.expected_sql,
@ -452,7 +456,7 @@ def test_doSelect(fake_connected_oracledb):
def test_doSelect_without_params(fake_connected_oracledb): def test_doSelect_without_params(fake_connected_oracledb):
fake_connected_oracledb._conn.expected_sql = "SELECT * FROM table" fake_connected_oracledb._conn.expected_sql = "SELECT * FROM table"
fake_connected_oracledb._conn.expected_return = [dict(test1=1)] fake_connected_oracledb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_oracledb.doSelect(fake_connected_oracledb._conn.expected_sql) fake_connected_oracledb.doSelect(fake_connected_oracledb._conn.expected_sql)
== fake_connected_oracledb._conn.expected_return == fake_connected_oracledb._conn.expected_return
@ -466,8 +470,8 @@ def test_doSelect_on_exception(fake_connected_oracledb):
def test_doSelect_just_try(fake_connected_just_try_oracledb): def test_doSelect_just_try(fake_connected_just_try_oracledb):
fake_connected_just_try_oracledb._conn.expected_sql = "SELECT * FROM table WHERE test1 = :test1" fake_connected_just_try_oracledb._conn.expected_sql = "SELECT * FROM table WHERE test1 = :test1"
fake_connected_just_try_oracledb._conn.expected_params = dict(test1=1) fake_connected_just_try_oracledb._conn.expected_params = {"test1": 1}
fake_connected_just_try_oracledb._conn.expected_return = [dict(test1=1)] fake_connected_just_try_oracledb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_just_try_oracledb.doSelect( fake_connected_just_try_oracledb.doSelect(
fake_connected_just_try_oracledb._conn.expected_sql, fake_connected_just_try_oracledb._conn.expected_sql,

View file

@ -63,7 +63,7 @@ class FakePsycopg2:
just_try = False just_try = False
def __init__(self, **kwargs): def __init__(self, **kwargs):
allowed_kwargs = dict(dbname=str, user=str, password=(str, None), host=str) allowed_kwargs = {"dbname": str, "user": str, "password": (str, None), "host": str}
for arg, value in kwargs.items(): for arg, value in kwargs.items():
assert arg in allowed_kwargs, f'Invalid arg {arg}="{value}"' assert arg in allowed_kwargs, f'Invalid arg {arg}="{value}"'
assert isinstance( assert isinstance(
@ -194,21 +194,23 @@ mock_doSelect_just_try = mock_doSQL_just_try
def test_combine_params_with_to_add_parameter(): def test_combine_params_with_to_add_parameter():
assert PgDB._combine_params(dict(test1=1), dict(test2=2)) == dict(test1=1, test2=2) assert PgDB._combine_params({"test1": 1}, {"test2": 2}) == {"test1": 1, "test2": 2}
def test_combine_params_with_kargs(): def test_combine_params_with_kargs():
assert PgDB._combine_params(dict(test1=1), test2=2) == dict(test1=1, test2=2) assert PgDB._combine_params({"test1": 1}, test2=2) == {"test1": 1, "test2": 2}
def test_combine_params_with_kargs_and_to_add_parameter(): def test_combine_params_with_kargs_and_to_add_parameter():
assert PgDB._combine_params(dict(test1=1), dict(test2=2), test3=3) == dict( assert PgDB._combine_params({"test1": 1}, {"test2": 2}, test3=3) == {
test1=1, test2=2, test3=3 "test1": 1,
) "test2": 2,
"test3": 3,
}
def test_format_where_clauses_params_are_preserved(): def test_format_where_clauses_params_are_preserved():
args = ("test = test", dict(test1=1)) args = ("test = test", {"test1": 1})
assert PgDB._format_where_clauses(*args) == args assert PgDB._format_where_clauses(*args) == args
@ -217,12 +219,12 @@ def test_format_where_clauses_raw():
def test_format_where_clauses_tuple_clause_with_params(): def test_format_where_clauses_tuple_clause_with_params():
where_clauses = ("test1 = %(test1)s AND test2 = %(test2)s", dict(test1=1, test2=2)) where_clauses = ("test1 = %(test1)s AND test2 = %(test2)s", {"test1": 1, "test2": 2})
assert PgDB._format_where_clauses(where_clauses) == where_clauses assert PgDB._format_where_clauses(where_clauses) == where_clauses
def test_format_where_clauses_dict(): def test_format_where_clauses_dict():
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
assert PgDB._format_where_clauses(where_clauses) == ( assert PgDB._format_where_clauses(where_clauses) == (
'"test1" = %(test1)s AND "test2" = %(test2)s', '"test1" = %(test1)s AND "test2" = %(test2)s',
where_clauses, where_clauses,
@ -230,15 +232,15 @@ def test_format_where_clauses_dict():
def test_format_where_clauses_combined_types(): def test_format_where_clauses_combined_types():
where_clauses = ("test1 = 1", ("test2 LIKE %(test2)s", dict(test2=2)), dict(test3=3, test4=4)) where_clauses = ("test1 = 1", ("test2 LIKE %(test2)s", {"test2": 2}), {"test3": 3, "test4": 4})
assert PgDB._format_where_clauses(where_clauses) == ( assert PgDB._format_where_clauses(where_clauses) == (
'test1 = 1 AND test2 LIKE %(test2)s AND "test3" = %(test3)s AND "test4" = %(test4)s', 'test1 = 1 AND test2 LIKE %(test2)s AND "test3" = %(test3)s AND "test4" = %(test4)s',
dict(test2=2, test3=3, test4=4), {"test2": 2, "test3": 3, "test4": 4},
) )
def test_format_where_clauses_with_where_op(): def test_format_where_clauses_with_where_op():
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
assert PgDB._format_where_clauses(where_clauses, where_op="OR") == ( assert PgDB._format_where_clauses(where_clauses, where_op="OR") == (
'"test1" = %(test1)s OR "test2" = %(test2)s', '"test1" = %(test1)s OR "test2" = %(test2)s',
where_clauses, where_clauses,
@ -247,7 +249,7 @@ def test_format_where_clauses_with_where_op():
def test_add_where_clauses(): def test_add_where_clauses():
sql = "SELECT * FROM table" sql = "SELECT * FROM table"
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
assert PgDB._add_where_clauses(sql, None, where_clauses) == ( assert PgDB._add_where_clauses(sql, None, where_clauses) == (
sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s', sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s',
where_clauses, where_clauses,
@ -256,11 +258,11 @@ def test_add_where_clauses():
def test_add_where_clauses_preserved_params(): def test_add_where_clauses_preserved_params():
sql = "SELECT * FROM table" sql = "SELECT * FROM table"
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
params = dict(fake1=1) params = {"fake1": 1}
assert PgDB._add_where_clauses(sql, params.copy(), where_clauses) == ( assert PgDB._add_where_clauses(sql, params.copy(), where_clauses) == (
sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s', sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s',
dict(**where_clauses, **params), {**where_clauses, **params},
) )
@ -275,11 +277,11 @@ def test_add_where_clauses_with_op():
def test_add_where_clauses_with_duplicated_field(): def test_add_where_clauses_with_duplicated_field():
sql = "UPDATE table SET test1=%(test1)s" sql = "UPDATE table SET test1=%(test1)s"
params = dict(test1="new_value") params = {"test1": "new_value"}
where_clauses = dict(test1="where_value") where_clauses = {"test1": "where_value"}
assert PgDB._add_where_clauses(sql, params, where_clauses) == ( assert PgDB._add_where_clauses(sql, params, where_clauses) == (
sql + ' WHERE "test1" = %(test1_1)s', sql + ' WHERE "test1" = %(test1_1)s',
dict(test1="new_value", test1_1="where_value"), {"test1": "new_value", "test1_1": "where_value"},
) )
@ -289,7 +291,7 @@ def test_quote_table_name():
def test_insert(mocker, test_pgdb): def test_insert(mocker, test_pgdb):
values = dict(test1=1, test2=2) values = {"test1": 1, "test2": 2}
mocker.patch( mocker.patch(
"mylib.pgsql.PgDB.doSQL", "mylib.pgsql.PgDB.doSQL",
generate_mock_doSQL( generate_mock_doSQL(
@ -302,18 +304,18 @@ def test_insert(mocker, test_pgdb):
def test_insert_just_try(mocker, test_pgdb): def test_insert_just_try(mocker, test_pgdb):
mocker.patch("mylib.pgsql.PgDB.doSQL", mock_doSQL_just_try) mocker.patch("mylib.pgsql.PgDB.doSQL", mock_doSQL_just_try)
assert test_pgdb.insert("mytable", dict(test1=1, test2=2), just_try=True) assert test_pgdb.insert("mytable", {"test1": 1, "test2": 2}, just_try=True)
def test_update(mocker, test_pgdb): def test_update(mocker, test_pgdb):
values = dict(test1=1, test2=2) values = {"test1": 1, "test2": 2}
where_clauses = dict(test3=3, test4=4) where_clauses = {"test3": 3, "test4": 4}
mocker.patch( mocker.patch(
"mylib.pgsql.PgDB.doSQL", "mylib.pgsql.PgDB.doSQL",
generate_mock_doSQL( generate_mock_doSQL(
'UPDATE "mytable" SET "test1" = %(test1)s, "test2" = %(test2)s WHERE "test3" =' 'UPDATE "mytable" SET "test1" = %(test1)s, "test2" = %(test2)s WHERE "test3" ='
' %(test3)s AND "test4" = %(test4)s', ' %(test3)s AND "test4" = %(test4)s',
dict(**values, **where_clauses), {**values, **where_clauses},
), ),
) )
@ -322,11 +324,11 @@ def test_update(mocker, test_pgdb):
def test_update_just_try(mocker, test_pgdb): def test_update_just_try(mocker, test_pgdb):
mocker.patch("mylib.pgsql.PgDB.doSQL", mock_doSQL_just_try) mocker.patch("mylib.pgsql.PgDB.doSQL", mock_doSQL_just_try)
assert test_pgdb.update("mytable", dict(test1=1, test2=2), None, just_try=True) assert test_pgdb.update("mytable", {"test1": 1, "test2": 2}, None, just_try=True)
def test_delete(mocker, test_pgdb): def test_delete(mocker, test_pgdb):
where_clauses = dict(test1=1, test2=2) where_clauses = {"test1": 1, "test2": 2}
mocker.patch( mocker.patch(
"mylib.pgsql.PgDB.doSQL", "mylib.pgsql.PgDB.doSQL",
generate_mock_doSQL( generate_mock_doSQL(
@ -355,17 +357,17 @@ def test_truncate_just_try(mocker, test_pgdb):
def test_select(mocker, test_pgdb): def test_select(mocker, test_pgdb):
fields = ("field1", "field2") fields = ("field1", "field2")
where_clauses = dict(test3=3, test4=4) where_clauses = {"test3": 3, "test4": 4}
expected_return = [ expected_return = [
dict(field1=1, field2=2), {"field1": 1, "field2": 2},
dict(field1=2, field2=3), {"field1": 2, "field2": 3},
] ]
order_by = "field1, DESC" order_by = "field1, DESC"
mocker.patch( mocker.patch(
"mylib.pgsql.PgDB.doSelect", "mylib.pgsql.PgDB.doSelect",
generate_mock_doSQL( generate_mock_doSQL(
'SELECT "field1", "field2" FROM "mytable" WHERE "test3" = %(test3)s AND "test4" =' 'SELECT "field1", "field2" FROM "mytable" WHERE "test3" = %(test3)s AND "test4" ='
" %(test4)s ORDER BY " + order_by, " %(test4)s ORDER BY " + order_by, # nosec: B608
where_clauses, where_clauses,
expected_return, expected_return,
), ),
@ -391,9 +393,12 @@ def test_select_just_try(mocker, test_pgdb):
def test_connect(mocker, test_pgdb): def test_connect(mocker, test_pgdb):
expected_kwargs = dict( expected_kwargs = {
dbname=test_pgdb._db, user=test_pgdb._user, host=test_pgdb._host, password=test_pgdb._pwd "dbname": test_pgdb._db,
) "user": test_pgdb._user,
"host": test_pgdb._host,
"password": test_pgdb._pwd,
}
mocker.patch("psycopg2.connect", generate_mock_args(expected_kwargs=expected_kwargs)) mocker.patch("psycopg2.connect", generate_mock_args(expected_kwargs=expected_kwargs))
@ -423,7 +428,7 @@ def test_setEncoding_on_exception(fake_connected_pgdb):
def test_doSQL(fake_connected_pgdb): def test_doSQL(fake_connected_pgdb):
fake_connected_pgdb._conn.expected_sql = "DELETE FROM table WHERE test1 = %(test1)s" fake_connected_pgdb._conn.expected_sql = "DELETE FROM table WHERE test1 = %(test1)s"
fake_connected_pgdb._conn.expected_params = dict(test1=1) fake_connected_pgdb._conn.expected_params = {"test1": 1}
fake_connected_pgdb.doSQL( fake_connected_pgdb.doSQL(
fake_connected_pgdb._conn.expected_sql, fake_connected_pgdb._conn.expected_params fake_connected_pgdb._conn.expected_sql, fake_connected_pgdb._conn.expected_params
) )
@ -445,8 +450,8 @@ def test_doSQL_on_exception(fake_connected_pgdb):
def test_doSelect(fake_connected_pgdb): def test_doSelect(fake_connected_pgdb):
fake_connected_pgdb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s" fake_connected_pgdb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s"
fake_connected_pgdb._conn.expected_params = dict(test1=1) fake_connected_pgdb._conn.expected_params = {"test1": 1}
fake_connected_pgdb._conn.expected_return = [dict(test1=1)] fake_connected_pgdb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_pgdb.doSelect( fake_connected_pgdb.doSelect(
fake_connected_pgdb._conn.expected_sql, fake_connected_pgdb._conn.expected_params fake_connected_pgdb._conn.expected_sql, fake_connected_pgdb._conn.expected_params
@ -457,7 +462,7 @@ def test_doSelect(fake_connected_pgdb):
def test_doSelect_without_params(fake_connected_pgdb): def test_doSelect_without_params(fake_connected_pgdb):
fake_connected_pgdb._conn.expected_sql = "SELECT * FROM table" fake_connected_pgdb._conn.expected_sql = "SELECT * FROM table"
fake_connected_pgdb._conn.expected_return = [dict(test1=1)] fake_connected_pgdb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_pgdb.doSelect(fake_connected_pgdb._conn.expected_sql) fake_connected_pgdb.doSelect(fake_connected_pgdb._conn.expected_sql)
== fake_connected_pgdb._conn.expected_return == fake_connected_pgdb._conn.expected_return
@ -471,8 +476,8 @@ def test_doSelect_on_exception(fake_connected_pgdb):
def test_doSelect_just_try(fake_connected_just_try_pgdb): def test_doSelect_just_try(fake_connected_just_try_pgdb):
fake_connected_just_try_pgdb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s" fake_connected_just_try_pgdb._conn.expected_sql = "SELECT * FROM table WHERE test1 = %(test1)s"
fake_connected_just_try_pgdb._conn.expected_params = dict(test1=1) fake_connected_just_try_pgdb._conn.expected_params = {"test1": 1}
fake_connected_just_try_pgdb._conn.expected_return = [dict(test1=1)] fake_connected_just_try_pgdb._conn.expected_return = [{"test1": 1}]
assert ( assert (
fake_connected_just_try_pgdb.doSelect( fake_connected_just_try_pgdb.doSelect(
fake_connected_just_try_pgdb._conn.expected_sql, fake_connected_just_try_pgdb._conn.expected_sql,

View file

@ -25,12 +25,12 @@ def test_create_telltale_file(tmp_path):
def test_create_telltale_file_with_filepath_and_invalid_dirpath(): def test_create_telltale_file_with_filepath_and_invalid_dirpath():
with pytest.raises(AssertionError): with pytest.raises(AssertionError):
TelltaleFile(filepath="/tmp/test", dirpath="/var/tmp") TelltaleFile(filepath="/tmp/test", dirpath="/var/tmp") # nosec: B108
def test_create_telltale_file_with_filepath_and_invalid_filename(): def test_create_telltale_file_with_filepath_and_invalid_filename():
with pytest.raises(AssertionError): with pytest.raises(AssertionError):
TelltaleFile(filepath="/tmp/test", filename="other") TelltaleFile(filepath="/tmp/test", filename="other") # nosec: B108
def test_remove_telltale_file(tmp_path): def test_remove_telltale_file(tmp_path):