Compare commits
No commits in common. "b92a814577b671a97ef1d20deab926fca7a12e9f" and "8e0e75f30ef786e49d8dc8b24b755264ef26c1d6" have entirely different histories.
b92a814577
...
8e0e75f30e
9 changed files with 210 additions and 265 deletions
|
|
@ -29,11 +29,12 @@ repos:
|
|||
language: system
|
||||
types: [python]
|
||||
require_serial: true
|
||||
- repo: https://github.com/PyCQA/bandit
|
||||
rev: 1.7.5
|
||||
- repo: https://github.com/Lucas-C/pre-commit-hooks-bandit
|
||||
rev: v1.0.5
|
||||
hooks:
|
||||
- id: bandit
|
||||
args: [--skip, "B101", --recursive, "mylib"]
|
||||
- id: python-bandit-vulnerability-check
|
||||
name: bandit
|
||||
args: [--skip, "B101", --recursive, mylib]
|
||||
- repo: local
|
||||
hooks:
|
||||
- id: pytest
|
||||
|
|
|
|||
|
|
@ -24,13 +24,7 @@ log = logging.getLogger(__name__)
|
|||
# Constants
|
||||
DEFAULT_ENCODING = "utf-8"
|
||||
DEFAULT_CONFIG_DIRPATH = os.path.expanduser("./")
|
||||
DEFAULT_LOG_FORMAT = "%(asctime)s - %(module)s:%(lineno)d - %(levelname)s - %(message)s"
|
||||
DEFAULT_CONSOLE_LOG_FORMAT = DEFAULT_LOG_FORMAT
|
||||
DEFAULT_FILELOG_FORMAT = DEFAULT_LOG_FORMAT
|
||||
|
||||
|
||||
class ConfigException(BaseException):
|
||||
"""Configuration exception"""
|
||||
DEFAULT_CONSOLE_LOG_FORMAT = "%(asctime)s - %(module)s:%(lineno)d - %(levelname)s - %(message)s"
|
||||
|
||||
|
||||
class BaseOption: # pylint: disable=too-many-instance-attributes
|
||||
|
|
@ -167,12 +161,12 @@ class BaseOption: # pylint: disable=too-many-instance-attributes
|
|||
args = [self.parser_argument_name]
|
||||
if self.short_arg:
|
||||
args.append(self.short_arg)
|
||||
kwargs = {
|
||||
"action": self.parser_action,
|
||||
"dest": self.parser_dest,
|
||||
"help": self.parser_help,
|
||||
"default": self.default,
|
||||
}
|
||||
kwargs = dict(
|
||||
action=self.parser_action,
|
||||
dest=self.parser_dest,
|
||||
help=self.parser_help,
|
||||
default=self.default,
|
||||
)
|
||||
if self.parser_type: # pylint: disable=using-constant-test
|
||||
kwargs["type"] = self.parser_type
|
||||
|
||||
|
|
@ -826,8 +820,8 @@ class Config: # pylint: disable=too-many-instance-attributes
|
|||
"-v", "--verbose", action="store_true", help="Show verbose messages"
|
||||
)
|
||||
|
||||
console_section = self.add_section("console", comment="Console logging")
|
||||
console_section.add_option(
|
||||
section = self.add_section("console", comment="Console logging")
|
||||
section.add_option(
|
||||
BooleanOption,
|
||||
"enabled",
|
||||
default=False,
|
||||
|
|
@ -835,14 +829,14 @@ class Config: # pylint: disable=too-many-instance-attributes
|
|||
short_arg="-C",
|
||||
comment="Enable/disable console log",
|
||||
)
|
||||
console_section.add_option(
|
||||
section.add_option(
|
||||
BooleanOption,
|
||||
"force_stderr",
|
||||
default=False,
|
||||
arg="--console-stderr",
|
||||
comment="Force console log on stderr",
|
||||
)
|
||||
console_section.add_option(
|
||||
section.add_option(
|
||||
StringOption,
|
||||
"log_format",
|
||||
default=DEFAULT_CONSOLE_LOG_FORMAT,
|
||||
|
|
@ -850,24 +844,6 @@ class Config: # pylint: disable=too-many-instance-attributes
|
|||
comment="Console log format",
|
||||
)
|
||||
|
||||
logfile_section = self.add_section("logfile", comment="Logging file")
|
||||
logfile_section.add_option(StringOption, "path", comment="File log path")
|
||||
logfile_section.add_option(
|
||||
StringOption,
|
||||
"format",
|
||||
default=DEFAULT_FILELOG_FORMAT,
|
||||
comment="File log format",
|
||||
)
|
||||
logfile_section.add_option(
|
||||
StringOption,
|
||||
"level",
|
||||
comment=(
|
||||
"File log level limit : by default, all logged messages (according to main log "
|
||||
"level) will be logged to the log file, but you can set a minimal level if you "
|
||||
f"want. Possible values: {', '.join(logging.getLevelNamesMapping())}."
|
||||
),
|
||||
)
|
||||
|
||||
self.add_options_to_parser(self.options_parser)
|
||||
|
||||
return self.options_parser
|
||||
|
|
@ -965,24 +941,6 @@ class Config: # pylint: disable=too-many-instance-attributes
|
|||
logging.getLogger().addHandler(stdout_console_handler)
|
||||
logging.getLogger().addHandler(stderr_console_handler)
|
||||
|
||||
if self.get("logfile", "path"):
|
||||
logfile_handler = logging.FileHandler(self.get("logfile", "path"))
|
||||
logfile_level = (
|
||||
logging.getLevelNamesMapping().get(self.get("logfile", "level"))
|
||||
if self.get("logfile", "level")
|
||||
else logging.DEBUG
|
||||
)
|
||||
if logfile_level is None:
|
||||
log.fatal("Invalid log file level specified (%s)", self.get("logfile", "level"))
|
||||
sys.exit(1)
|
||||
logfile_handler.setLevel(logfile_level)
|
||||
|
||||
if self.get("logfile", "format"):
|
||||
logfile_formater = logging.Formatter(self.get("logfile", "format"))
|
||||
logfile_handler.setFormatter(logfile_formater)
|
||||
|
||||
logging.getLogger().addHandler(logfile_handler)
|
||||
|
||||
if execute_callback:
|
||||
self._loaded()
|
||||
|
||||
|
|
@ -1174,7 +1132,7 @@ class ConfigurableObject:
|
|||
elif self._config_name:
|
||||
self._options_prefix = self._config_name + "_"
|
||||
else:
|
||||
raise ConfigException(f"No configuration name defined for {__name__}")
|
||||
raise Exception(f"No configuration name defined for {__name__}")
|
||||
|
||||
if config:
|
||||
self._config = config
|
||||
|
|
@ -1183,7 +1141,7 @@ class ConfigurableObject:
|
|||
elif self._config_name:
|
||||
self._config_section = self._config_name
|
||||
else:
|
||||
raise ConfigException(f"No configuration name defined for {__name__}")
|
||||
raise Exception(f"No configuration name defined for {__name__}")
|
||||
|
||||
def _get_option(self, option, default=None, required=False):
|
||||
"""Retreive option value"""
|
||||
|
|
@ -1316,7 +1274,7 @@ class ConfigSectionAsDictWrapper:
|
|||
self.__section.set(key, value)
|
||||
|
||||
def __delitem__(self, key):
|
||||
raise ConfigException("Deleting a configuration option is not supported")
|
||||
raise Exception("Deleting a configuration option is not supported")
|
||||
|
||||
|
||||
# pylint: disable=too-few-public-methods
|
||||
|
|
|
|||
|
|
@ -566,15 +566,15 @@ if __name__ == "__main__":
|
|||
catch_all_addr=options.email_catch_all,
|
||||
just_try=options.just_try,
|
||||
encoding=options.email_encoding,
|
||||
templates={
|
||||
"test": {
|
||||
"subject": "Test email",
|
||||
"text": (
|
||||
templates=dict(
|
||||
test=dict(
|
||||
subject="Test email",
|
||||
text=(
|
||||
"Just a test email sent at {sent_date}."
|
||||
if not options.test_mako
|
||||
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>"
|
||||
if not options.test_mako
|
||||
else MakoTemplate( # nosec
|
||||
|
|
@ -582,8 +582,8 @@ if __name__ == "__main__":
|
|||
"<small>(sent at ${sent_date | h})</small>"
|
||||
)
|
||||
),
|
||||
}
|
||||
},
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
logging.info("Send a test email to %s", options.test_to)
|
||||
|
|
|
|||
|
|
@ -120,7 +120,7 @@ class LdapServer:
|
|||
return ldap.SCOPE_ONELEVEL # pylint: disable=no-member
|
||||
if scope == "sub":
|
||||
return ldap.SCOPE_SUBTREE # pylint: disable=no-member
|
||||
raise LdapServerException(f'Unknown LDAP scope "{scope}"')
|
||||
raise Exception(f'Unknown LDAP scope "{scope}"')
|
||||
|
||||
def search(self, basedn, filterstr=None, attrs=None, sizelimit=None, scope=None):
|
||||
"""Run a search on LDAP server"""
|
||||
|
|
@ -399,16 +399,18 @@ class LdapServer:
|
|||
return default
|
||||
|
||||
|
||||
class LdapException(BaseException):
|
||||
"""Generic LDAP exception"""
|
||||
|
||||
|
||||
class LdapServerException(LdapException):
|
||||
class LdapServerException(BaseException):
|
||||
"""Generic exception raised by LdapServer"""
|
||||
|
||||
def __init__(self, msg):
|
||||
BaseException.__init__(self, msg)
|
||||
|
||||
class LdapClientException(LdapException):
|
||||
"""Generic exception raised by LdapClient"""
|
||||
|
||||
class LdapClientException(LdapServerException):
|
||||
"""Generic exception raised by LdapServer"""
|
||||
|
||||
def __init__(self, msg):
|
||||
LdapServerException.__init__(self, msg)
|
||||
|
||||
|
||||
class LdapClient:
|
||||
|
|
@ -539,7 +541,7 @@ class LdapClient:
|
|||
:param dn: The object DN
|
||||
:param attrs: The object attributes as return by python-ldap search
|
||||
"""
|
||||
obj = {"dn": dn}
|
||||
obj = dict(dn=dn)
|
||||
for attr in attrs:
|
||||
obj[attr] = [self.decode(v) for v in self._conn.get_attr(attrs, attr, all_values=True)]
|
||||
return obj
|
||||
|
|
@ -1016,7 +1018,7 @@ def parse_datetime(value, to_timezone=None, default_timezone=None, naive=None):
|
|||
elif isinstance(default_timezone, datetime.tzinfo):
|
||||
date = date.replace(tzinfo=default_timezone)
|
||||
else:
|
||||
raise LdapException("It's not supposed to happen!")
|
||||
raise Exception("It's not supposed to happen!")
|
||||
elif naive:
|
||||
return date.replace(tzinfo=None)
|
||||
if to_timezone:
|
||||
|
|
@ -1076,7 +1078,7 @@ def format_datetime(value, from_timezone=None, to_timezone=None, naive=None):
|
|||
elif isinstance(from_timezone, datetime.tzinfo):
|
||||
from_value = value.replace(tzinfo=from_timezone)
|
||||
else:
|
||||
raise LdapException("It's not supposed to happen!")
|
||||
raise Exception("It's not supposed to happen!")
|
||||
elif naive:
|
||||
from_value = value.replace(tzinfo=pytz.utc)
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -126,14 +126,14 @@ def test_add_option_custom_args():
|
|||
section = config.add_section("my_section")
|
||||
assert isinstance(section, ConfigSection)
|
||||
name = "my_option"
|
||||
kwargs = {
|
||||
"default": "default value",
|
||||
"comment": "my comment",
|
||||
"no_arg": True,
|
||||
"arg": "--my-option",
|
||||
"short_arg": "-M",
|
||||
"arg_help": "My help",
|
||||
}
|
||||
kwargs = dict(
|
||||
default="default value",
|
||||
comment="my comment",
|
||||
no_arg=True,
|
||||
arg="--my-option",
|
||||
short_arg="-M",
|
||||
arg_help="My help",
|
||||
)
|
||||
option = section.add_option(StringOption, name, **kwargs)
|
||||
assert isinstance(option, StringOption)
|
||||
assert name in section.options and section.options[name] == option
|
||||
|
|
|
|||
|
|
@ -74,14 +74,9 @@ class FakeMySQLdb:
|
|||
just_try = False
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
allowed_kwargs = {
|
||||
"db": str,
|
||||
"user": str,
|
||||
"passwd": (str, None),
|
||||
"host": str,
|
||||
"charset": str,
|
||||
"use_unicode": bool,
|
||||
}
|
||||
allowed_kwargs = dict(
|
||||
db=str, user=str, passwd=(str, None), host=str, charset=str, use_unicode=bool
|
||||
)
|
||||
for arg, value in kwargs.items():
|
||||
assert arg in allowed_kwargs, f'Invalid arg {arg}="{value}"'
|
||||
assert isinstance(
|
||||
|
|
@ -205,23 +200,21 @@ mock_doSelect_just_try = mock_doSQL_just_try
|
|||
|
||||
|
||||
def test_combine_params_with_to_add_parameter():
|
||||
assert MyDB._combine_params({"test1": 1}, {"test2": 2}) == {"test1": 1, "test2": 2}
|
||||
assert MyDB._combine_params(dict(test1=1), dict(test2=2)) == dict(test1=1, test2=2)
|
||||
|
||||
|
||||
def test_combine_params_with_kargs():
|
||||
assert MyDB._combine_params({"test1": 1}, test2=2) == {"test1": 1, "test2": 2}
|
||||
assert MyDB._combine_params(dict(test1=1), test2=2) == dict(test1=1, test2=2)
|
||||
|
||||
|
||||
def test_combine_params_with_kargs_and_to_add_parameter():
|
||||
assert MyDB._combine_params({"test1": 1}, {"test2": 2}, test3=3) == {
|
||||
"test1": 1,
|
||||
"test2": 2,
|
||||
"test3": 3,
|
||||
}
|
||||
assert MyDB._combine_params(dict(test1=1), dict(test2=2), test3=3) == dict(
|
||||
test1=1, test2=2, test3=3
|
||||
)
|
||||
|
||||
|
||||
def test_format_where_clauses_params_are_preserved():
|
||||
args = ("test = test", {"test1": 1})
|
||||
args = ("test = test", dict(test1=1))
|
||||
assert MyDB._format_where_clauses(*args) == args
|
||||
|
||||
|
||||
|
|
@ -230,12 +223,12 @@ def test_format_where_clauses_raw():
|
|||
|
||||
|
||||
def test_format_where_clauses_tuple_clause_with_params():
|
||||
where_clauses = ("test1 = %(test1)s AND test2 = %(test2)s", {"test1": 1, "test2": 2})
|
||||
where_clauses = ("test1 = %(test1)s AND test2 = %(test2)s", dict(test1=1, test2=2))
|
||||
assert MyDB._format_where_clauses(where_clauses) == where_clauses
|
||||
|
||||
|
||||
def test_format_where_clauses_dict():
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
assert MyDB._format_where_clauses(where_clauses) == (
|
||||
"`test1` = %(test1)s AND `test2` = %(test2)s",
|
||||
where_clauses,
|
||||
|
|
@ -243,15 +236,15 @@ def test_format_where_clauses_dict():
|
|||
|
||||
|
||||
def test_format_where_clauses_combined_types():
|
||||
where_clauses = ("test1 = 1", ("test2 LIKE %(test2)s", {"test2": 2}), {"test3": 3, "test4": 4})
|
||||
where_clauses = ("test1 = 1", ("test2 LIKE %(test2)s", dict(test2=2)), dict(test3=3, test4=4))
|
||||
assert MyDB._format_where_clauses(where_clauses) == (
|
||||
"test1 = 1 AND test2 LIKE %(test2)s AND `test3` = %(test3)s AND `test4` = %(test4)s",
|
||||
{"test2": 2, "test3": 3, "test4": 4},
|
||||
dict(test2=2, test3=3, test4=4),
|
||||
)
|
||||
|
||||
|
||||
def test_format_where_clauses_with_where_op():
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
assert MyDB._format_where_clauses(where_clauses, where_op="OR") == (
|
||||
"`test1` = %(test1)s OR `test2` = %(test2)s",
|
||||
where_clauses,
|
||||
|
|
@ -260,7 +253,7 @@ def test_format_where_clauses_with_where_op():
|
|||
|
||||
def test_add_where_clauses():
|
||||
sql = "SELECT * FROM table"
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
assert MyDB._add_where_clauses(sql, None, where_clauses) == (
|
||||
sql + " WHERE `test1` = %(test1)s AND `test2` = %(test2)s",
|
||||
where_clauses,
|
||||
|
|
@ -269,11 +262,11 @@ def test_add_where_clauses():
|
|||
|
||||
def test_add_where_clauses_preserved_params():
|
||||
sql = "SELECT * FROM table"
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
params = {"fake1": 1}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
params = dict(fake1=1)
|
||||
assert MyDB._add_where_clauses(sql, params.copy(), where_clauses) == (
|
||||
sql + " WHERE `test1` = %(test1)s AND `test2` = %(test2)s",
|
||||
{**where_clauses, **params},
|
||||
dict(**where_clauses, **params),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -288,11 +281,11 @@ def test_add_where_clauses_with_op():
|
|||
|
||||
def test_add_where_clauses_with_duplicated_field():
|
||||
sql = "UPDATE table SET test1=%(test1)s"
|
||||
params = {"test1": "new_value"}
|
||||
where_clauses = {"test1": "where_value"}
|
||||
params = dict(test1="new_value")
|
||||
where_clauses = dict(test1="where_value")
|
||||
assert MyDB._add_where_clauses(sql, params, where_clauses) == (
|
||||
sql + " WHERE `test1` = %(test1_1)s",
|
||||
{"test1": "new_value", "test1_1": "where_value"},
|
||||
dict(test1="new_value", test1_1="where_value"),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -302,7 +295,7 @@ def test_quote_table_name():
|
|||
|
||||
|
||||
def test_insert(mocker, test_mydb):
|
||||
values = {"test1": 1, "test2": 2}
|
||||
values = dict(test1=1, test2=2)
|
||||
mocker.patch(
|
||||
"mylib.mysql.MyDB.doSQL",
|
||||
generate_mock_doSQL(
|
||||
|
|
@ -315,18 +308,18 @@ def test_insert(mocker, test_mydb):
|
|||
|
||||
def test_insert_just_try(mocker, test_mydb):
|
||||
mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSQL_just_try)
|
||||
assert test_mydb.insert("mytable", {"test1": 1, "test2": 2}, just_try=True)
|
||||
assert test_mydb.insert("mytable", dict(test1=1, test2=2), just_try=True)
|
||||
|
||||
|
||||
def test_update(mocker, test_mydb):
|
||||
values = {"test1": 1, "test2": 2}
|
||||
where_clauses = {"test3": 3, "test4": 4}
|
||||
values = dict(test1=1, test2=2)
|
||||
where_clauses = dict(test3=3, test4=4)
|
||||
mocker.patch(
|
||||
"mylib.mysql.MyDB.doSQL",
|
||||
generate_mock_doSQL(
|
||||
"UPDATE `mytable` SET `test1` = %(test1)s, `test2` = %(test2)s WHERE `test3` ="
|
||||
" %(test3)s AND `test4` = %(test4)s",
|
||||
{**values, **where_clauses},
|
||||
dict(**values, **where_clauses),
|
||||
),
|
||||
)
|
||||
|
||||
|
|
@ -335,11 +328,11 @@ def test_update(mocker, test_mydb):
|
|||
|
||||
def test_update_just_try(mocker, test_mydb):
|
||||
mocker.patch("mylib.mysql.MyDB.doSQL", mock_doSQL_just_try)
|
||||
assert test_mydb.update("mytable", {"test1": 1, "test2": 2}, None, just_try=True)
|
||||
assert test_mydb.update("mytable", dict(test1=1, test2=2), None, just_try=True)
|
||||
|
||||
|
||||
def test_delete(mocker, test_mydb):
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
mocker.patch(
|
||||
"mylib.mysql.MyDB.doSQL",
|
||||
generate_mock_doSQL(
|
||||
|
|
@ -368,17 +361,17 @@ def test_truncate_just_try(mocker, test_mydb):
|
|||
|
||||
def test_select(mocker, test_mydb):
|
||||
fields = ("field1", "field2")
|
||||
where_clauses = {"test3": 3, "test4": 4}
|
||||
where_clauses = dict(test3=3, test4=4)
|
||||
expected_return = [
|
||||
{"field1": 1, "field2": 2},
|
||||
{"field1": 2, "field2": 3},
|
||||
dict(field1=1, field2=2),
|
||||
dict(field1=2, field2=3),
|
||||
]
|
||||
order_by = "field1, DESC"
|
||||
mocker.patch(
|
||||
"mylib.mysql.MyDB.doSelect",
|
||||
generate_mock_doSQL(
|
||||
"SELECT `field1`, `field2` FROM `mytable` WHERE `test3` = %(test3)s AND `test4` ="
|
||||
" %(test4)s ORDER BY " + order_by, # nosec: B608
|
||||
" %(test4)s ORDER BY " + order_by,
|
||||
where_clauses,
|
||||
expected_return,
|
||||
),
|
||||
|
|
@ -404,14 +397,14 @@ def test_select_just_try(mocker, test_mydb):
|
|||
|
||||
|
||||
def test_connect(mocker, test_mydb):
|
||||
expected_kwargs = {
|
||||
"db": test_mydb._db,
|
||||
"user": test_mydb._user,
|
||||
"host": test_mydb._host,
|
||||
"passwd": test_mydb._pwd,
|
||||
"charset": test_mydb._charset,
|
||||
"use_unicode": True,
|
||||
}
|
||||
expected_kwargs = dict(
|
||||
db=test_mydb._db,
|
||||
user=test_mydb._user,
|
||||
host=test_mydb._host,
|
||||
passwd=test_mydb._pwd,
|
||||
charset=test_mydb._charset,
|
||||
use_unicode=True,
|
||||
)
|
||||
|
||||
mocker.patch("MySQLdb.connect", generate_mock_args(expected_kwargs=expected_kwargs))
|
||||
|
||||
|
|
@ -428,7 +421,7 @@ def test_close_connected(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_params = {"test1": 1}
|
||||
fake_connected_mydb._conn.expected_params = dict(test1=1)
|
||||
fake_connected_mydb.doSQL(
|
||||
fake_connected_mydb._conn.expected_sql, fake_connected_mydb._conn.expected_params
|
||||
)
|
||||
|
|
@ -450,8 +443,8 @@ def test_doSQL_on_exception(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_params = {"test1": 1}
|
||||
fake_connected_mydb._conn.expected_return = [{"test1": 1}]
|
||||
fake_connected_mydb._conn.expected_params = dict(test1=1)
|
||||
fake_connected_mydb._conn.expected_return = [dict(test1=1)]
|
||||
assert (
|
||||
fake_connected_mydb.doSelect(
|
||||
fake_connected_mydb._conn.expected_sql, fake_connected_mydb._conn.expected_params
|
||||
|
|
@ -462,7 +455,7 @@ def test_doSelect(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_return = [{"test1": 1}]
|
||||
fake_connected_mydb._conn.expected_return = [dict(test1=1)]
|
||||
assert (
|
||||
fake_connected_mydb.doSelect(fake_connected_mydb._conn.expected_sql)
|
||||
== fake_connected_mydb._conn.expected_return
|
||||
|
|
@ -476,8 +469,8 @@ def test_doSelect_on_exception(fake_connected_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_params = {"test1": 1}
|
||||
fake_connected_just_try_mydb._conn.expected_return = [{"test1": 1}]
|
||||
fake_connected_just_try_mydb._conn.expected_params = dict(test1=1)
|
||||
fake_connected_just_try_mydb._conn.expected_return = [dict(test1=1)]
|
||||
assert (
|
||||
fake_connected_just_try_mydb.doSelect(
|
||||
fake_connected_just_try_mydb._conn.expected_sql,
|
||||
|
|
|
|||
|
|
@ -73,7 +73,7 @@ class FakeCXOracle:
|
|||
just_try = False
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
allowed_kwargs = {"dsn": str, "user": str, "password": (str, None)}
|
||||
allowed_kwargs = dict(dsn=str, user=str, password=(str, None))
|
||||
for arg, value in kwargs.items():
|
||||
assert arg in allowed_kwargs, f"Invalid arg {arg}='{value}'"
|
||||
assert isinstance(
|
||||
|
|
@ -197,23 +197,21 @@ mock_doSelect_just_try = mock_doSQL_just_try
|
|||
|
||||
|
||||
def test_combine_params_with_to_add_parameter():
|
||||
assert OracleDB._combine_params({"test1": 1}, {"test2": 2}) == {"test1": 1, "test2": 2}
|
||||
assert OracleDB._combine_params(dict(test1=1), dict(test2=2)) == dict(test1=1, test2=2)
|
||||
|
||||
|
||||
def test_combine_params_with_kargs():
|
||||
assert OracleDB._combine_params({"test1": 1}, test2=2) == {"test1": 1, "test2": 2}
|
||||
assert OracleDB._combine_params(dict(test1=1), test2=2) == dict(test1=1, test2=2)
|
||||
|
||||
|
||||
def test_combine_params_with_kargs_and_to_add_parameter():
|
||||
assert OracleDB._combine_params({"test1": 1}, {"test2": 2}, test3=3) == {
|
||||
"test1": 1,
|
||||
"test2": 2,
|
||||
"test3": 3,
|
||||
}
|
||||
assert OracleDB._combine_params(dict(test1=1), dict(test2=2), test3=3) == dict(
|
||||
test1=1, test2=2, test3=3
|
||||
)
|
||||
|
||||
|
||||
def test_format_where_clauses_params_are_preserved():
|
||||
args = ("test = test", {"test1": 1})
|
||||
args = ("test = test", dict(test1=1))
|
||||
assert OracleDB._format_where_clauses(*args) == args
|
||||
|
||||
|
||||
|
|
@ -222,12 +220,12 @@ def test_format_where_clauses_raw():
|
|||
|
||||
|
||||
def test_format_where_clauses_tuple_clause_with_params():
|
||||
where_clauses = ("test1 = :test1 AND test2 = :test2", {"test1": 1, "test2": 2})
|
||||
where_clauses = ("test1 = :test1 AND test2 = :test2", dict(test1=1, test2=2))
|
||||
assert OracleDB._format_where_clauses(where_clauses) == where_clauses
|
||||
|
||||
|
||||
def test_format_where_clauses_dict():
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
assert OracleDB._format_where_clauses(where_clauses) == (
|
||||
'"test1" = :test1 AND "test2" = :test2',
|
||||
where_clauses,
|
||||
|
|
@ -235,15 +233,15 @@ def test_format_where_clauses_dict():
|
|||
|
||||
|
||||
def test_format_where_clauses_combined_types():
|
||||
where_clauses = ("test1 = 1", ("test2 LIKE :test2", {"test2": 2}), {"test3": 3, "test4": 4})
|
||||
where_clauses = ("test1 = 1", ("test2 LIKE :test2", dict(test2=2)), dict(test3=3, test4=4))
|
||||
assert OracleDB._format_where_clauses(where_clauses) == (
|
||||
'test1 = 1 AND test2 LIKE :test2 AND "test3" = :test3 AND "test4" = :test4',
|
||||
{"test2": 2, "test3": 3, "test4": 4},
|
||||
dict(test2=2, test3=3, test4=4),
|
||||
)
|
||||
|
||||
|
||||
def test_format_where_clauses_with_where_op():
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
assert OracleDB._format_where_clauses(where_clauses, where_op="OR") == (
|
||||
'"test1" = :test1 OR "test2" = :test2',
|
||||
where_clauses,
|
||||
|
|
@ -252,7 +250,7 @@ def test_format_where_clauses_with_where_op():
|
|||
|
||||
def test_add_where_clauses():
|
||||
sql = "SELECT * FROM table"
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
assert OracleDB._add_where_clauses(sql, None, where_clauses) == (
|
||||
sql + ' WHERE "test1" = :test1 AND "test2" = :test2',
|
||||
where_clauses,
|
||||
|
|
@ -261,11 +259,11 @@ def test_add_where_clauses():
|
|||
|
||||
def test_add_where_clauses_preserved_params():
|
||||
sql = "SELECT * FROM table"
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
params = {"fake1": 1}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
params = dict(fake1=1)
|
||||
assert OracleDB._add_where_clauses(sql, params.copy(), where_clauses) == (
|
||||
sql + ' WHERE "test1" = :test1 AND "test2" = :test2',
|
||||
{**where_clauses, **params},
|
||||
dict(**where_clauses, **params),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -280,11 +278,11 @@ def test_add_where_clauses_with_op():
|
|||
|
||||
def test_add_where_clauses_with_duplicated_field():
|
||||
sql = "UPDATE table SET test1=:test1"
|
||||
params = {"test1": "new_value"}
|
||||
where_clauses = {"test1": "where_value"}
|
||||
params = dict(test1="new_value")
|
||||
where_clauses = dict(test1="where_value")
|
||||
assert OracleDB._add_where_clauses(sql, params, where_clauses) == (
|
||||
sql + ' WHERE "test1" = :test1_1',
|
||||
{"test1": "new_value", "test1_1": "where_value"},
|
||||
dict(test1="new_value", test1_1="where_value"),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -294,7 +292,7 @@ def test_quote_table_name():
|
|||
|
||||
|
||||
def test_insert(mocker, test_oracledb):
|
||||
values = {"test1": 1, "test2": 2}
|
||||
values = dict(test1=1, test2=2)
|
||||
mocker.patch(
|
||||
"mylib.oracle.OracleDB.doSQL",
|
||||
generate_mock_doSQL(
|
||||
|
|
@ -307,18 +305,18 @@ def test_insert(mocker, test_oracledb):
|
|||
|
||||
def test_insert_just_try(mocker, test_oracledb):
|
||||
mocker.patch("mylib.oracle.OracleDB.doSQL", mock_doSQL_just_try)
|
||||
assert test_oracledb.insert("mytable", {"test1": 1, "test2": 2}, just_try=True)
|
||||
assert test_oracledb.insert("mytable", dict(test1=1, test2=2), just_try=True)
|
||||
|
||||
|
||||
def test_update(mocker, test_oracledb):
|
||||
values = {"test1": 1, "test2": 2}
|
||||
where_clauses = {"test3": 3, "test4": 4}
|
||||
values = dict(test1=1, test2=2)
|
||||
where_clauses = dict(test3=3, test4=4)
|
||||
mocker.patch(
|
||||
"mylib.oracle.OracleDB.doSQL",
|
||||
generate_mock_doSQL(
|
||||
'UPDATE "mytable" SET "test1" = :test1, "test2" = :test2 WHERE "test3" = :test3 AND'
|
||||
' "test4" = :test4',
|
||||
{**values, **where_clauses},
|
||||
dict(**values, **where_clauses),
|
||||
),
|
||||
)
|
||||
|
||||
|
|
@ -327,11 +325,11 @@ def test_update(mocker, test_oracledb):
|
|||
|
||||
def test_update_just_try(mocker, test_oracledb):
|
||||
mocker.patch("mylib.oracle.OracleDB.doSQL", mock_doSQL_just_try)
|
||||
assert test_oracledb.update("mytable", {"test1": 1, "test2": 2}, None, just_try=True)
|
||||
assert test_oracledb.update("mytable", dict(test1=1, test2=2), None, just_try=True)
|
||||
|
||||
|
||||
def test_delete(mocker, test_oracledb):
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
mocker.patch(
|
||||
"mylib.oracle.OracleDB.doSQL",
|
||||
generate_mock_doSQL(
|
||||
|
|
@ -362,17 +360,17 @@ def test_truncate_just_try(mocker, test_oracledb):
|
|||
|
||||
def test_select(mocker, test_oracledb):
|
||||
fields = ("field1", "field2")
|
||||
where_clauses = {"test3": 3, "test4": 4}
|
||||
where_clauses = dict(test3=3, test4=4)
|
||||
expected_return = [
|
||||
{"field1": 1, "field2": 2},
|
||||
{"field1": 2, "field2": 3},
|
||||
dict(field1=1, field2=2),
|
||||
dict(field1=2, field2=3),
|
||||
]
|
||||
order_by = "field1, DESC"
|
||||
mocker.patch(
|
||||
"mylib.oracle.OracleDB.doSelect",
|
||||
generate_mock_doSQL(
|
||||
'SELECT "field1", "field2" FROM "mytable" WHERE "test3" = :test3 AND "test4" = :test4'
|
||||
" ORDER BY " + order_by, # nosec: B608
|
||||
" ORDER BY " + order_by,
|
||||
where_clauses,
|
||||
expected_return,
|
||||
),
|
||||
|
|
@ -400,11 +398,9 @@ def test_select_just_try(mocker, test_oracledb):
|
|||
|
||||
|
||||
def test_connect(mocker, test_oracledb):
|
||||
expected_kwargs = {
|
||||
"dsn": test_oracledb._dsn,
|
||||
"user": test_oracledb._user,
|
||||
"password": test_oracledb._pwd,
|
||||
}
|
||||
expected_kwargs = dict(
|
||||
dsn=test_oracledb._dsn, user=test_oracledb._user, password=test_oracledb._pwd
|
||||
)
|
||||
|
||||
mocker.patch("cx_Oracle.connect", generate_mock_args(expected_kwargs=expected_kwargs))
|
||||
|
||||
|
|
@ -421,7 +417,7 @@ def test_close_connected(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_params = {"test1": 1}
|
||||
fake_connected_oracledb._conn.expected_params = dict(test1=1)
|
||||
fake_connected_oracledb.doSQL(
|
||||
fake_connected_oracledb._conn.expected_sql, fake_connected_oracledb._conn.expected_params
|
||||
)
|
||||
|
|
@ -443,8 +439,8 @@ def test_doSQL_on_exception(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_params = {"test1": 1}
|
||||
fake_connected_oracledb._conn.expected_return = [{"test1": 1}]
|
||||
fake_connected_oracledb._conn.expected_params = dict(test1=1)
|
||||
fake_connected_oracledb._conn.expected_return = [dict(test1=1)]
|
||||
assert (
|
||||
fake_connected_oracledb.doSelect(
|
||||
fake_connected_oracledb._conn.expected_sql,
|
||||
|
|
@ -456,7 +452,7 @@ def test_doSelect(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_return = [{"test1": 1}]
|
||||
fake_connected_oracledb._conn.expected_return = [dict(test1=1)]
|
||||
assert (
|
||||
fake_connected_oracledb.doSelect(fake_connected_oracledb._conn.expected_sql)
|
||||
== fake_connected_oracledb._conn.expected_return
|
||||
|
|
@ -470,8 +466,8 @@ def test_doSelect_on_exception(fake_connected_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_params = {"test1": 1}
|
||||
fake_connected_just_try_oracledb._conn.expected_return = [{"test1": 1}]
|
||||
fake_connected_just_try_oracledb._conn.expected_params = dict(test1=1)
|
||||
fake_connected_just_try_oracledb._conn.expected_return = [dict(test1=1)]
|
||||
assert (
|
||||
fake_connected_just_try_oracledb.doSelect(
|
||||
fake_connected_just_try_oracledb._conn.expected_sql,
|
||||
|
|
|
|||
|
|
@ -63,7 +63,7 @@ class FakePsycopg2:
|
|||
just_try = False
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
allowed_kwargs = {"dbname": str, "user": str, "password": (str, None), "host": str}
|
||||
allowed_kwargs = dict(dbname=str, user=str, password=(str, None), host=str)
|
||||
for arg, value in kwargs.items():
|
||||
assert arg in allowed_kwargs, f'Invalid arg {arg}="{value}"'
|
||||
assert isinstance(
|
||||
|
|
@ -194,23 +194,21 @@ mock_doSelect_just_try = mock_doSQL_just_try
|
|||
|
||||
|
||||
def test_combine_params_with_to_add_parameter():
|
||||
assert PgDB._combine_params({"test1": 1}, {"test2": 2}) == {"test1": 1, "test2": 2}
|
||||
assert PgDB._combine_params(dict(test1=1), dict(test2=2)) == dict(test1=1, test2=2)
|
||||
|
||||
|
||||
def test_combine_params_with_kargs():
|
||||
assert PgDB._combine_params({"test1": 1}, test2=2) == {"test1": 1, "test2": 2}
|
||||
assert PgDB._combine_params(dict(test1=1), test2=2) == dict(test1=1, test2=2)
|
||||
|
||||
|
||||
def test_combine_params_with_kargs_and_to_add_parameter():
|
||||
assert PgDB._combine_params({"test1": 1}, {"test2": 2}, test3=3) == {
|
||||
"test1": 1,
|
||||
"test2": 2,
|
||||
"test3": 3,
|
||||
}
|
||||
assert PgDB._combine_params(dict(test1=1), dict(test2=2), test3=3) == dict(
|
||||
test1=1, test2=2, test3=3
|
||||
)
|
||||
|
||||
|
||||
def test_format_where_clauses_params_are_preserved():
|
||||
args = ("test = test", {"test1": 1})
|
||||
args = ("test = test", dict(test1=1))
|
||||
assert PgDB._format_where_clauses(*args) == args
|
||||
|
||||
|
||||
|
|
@ -219,12 +217,12 @@ def test_format_where_clauses_raw():
|
|||
|
||||
|
||||
def test_format_where_clauses_tuple_clause_with_params():
|
||||
where_clauses = ("test1 = %(test1)s AND test2 = %(test2)s", {"test1": 1, "test2": 2})
|
||||
where_clauses = ("test1 = %(test1)s AND test2 = %(test2)s", dict(test1=1, test2=2))
|
||||
assert PgDB._format_where_clauses(where_clauses) == where_clauses
|
||||
|
||||
|
||||
def test_format_where_clauses_dict():
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
assert PgDB._format_where_clauses(where_clauses) == (
|
||||
'"test1" = %(test1)s AND "test2" = %(test2)s',
|
||||
where_clauses,
|
||||
|
|
@ -232,15 +230,15 @@ def test_format_where_clauses_dict():
|
|||
|
||||
|
||||
def test_format_where_clauses_combined_types():
|
||||
where_clauses = ("test1 = 1", ("test2 LIKE %(test2)s", {"test2": 2}), {"test3": 3, "test4": 4})
|
||||
where_clauses = ("test1 = 1", ("test2 LIKE %(test2)s", dict(test2=2)), dict(test3=3, test4=4))
|
||||
assert PgDB._format_where_clauses(where_clauses) == (
|
||||
'test1 = 1 AND test2 LIKE %(test2)s AND "test3" = %(test3)s AND "test4" = %(test4)s',
|
||||
{"test2": 2, "test3": 3, "test4": 4},
|
||||
dict(test2=2, test3=3, test4=4),
|
||||
)
|
||||
|
||||
|
||||
def test_format_where_clauses_with_where_op():
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
assert PgDB._format_where_clauses(where_clauses, where_op="OR") == (
|
||||
'"test1" = %(test1)s OR "test2" = %(test2)s',
|
||||
where_clauses,
|
||||
|
|
@ -249,7 +247,7 @@ def test_format_where_clauses_with_where_op():
|
|||
|
||||
def test_add_where_clauses():
|
||||
sql = "SELECT * FROM table"
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
assert PgDB._add_where_clauses(sql, None, where_clauses) == (
|
||||
sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s',
|
||||
where_clauses,
|
||||
|
|
@ -258,11 +256,11 @@ def test_add_where_clauses():
|
|||
|
||||
def test_add_where_clauses_preserved_params():
|
||||
sql = "SELECT * FROM table"
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
params = {"fake1": 1}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
params = dict(fake1=1)
|
||||
assert PgDB._add_where_clauses(sql, params.copy(), where_clauses) == (
|
||||
sql + ' WHERE "test1" = %(test1)s AND "test2" = %(test2)s',
|
||||
{**where_clauses, **params},
|
||||
dict(**where_clauses, **params),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -277,11 +275,11 @@ def test_add_where_clauses_with_op():
|
|||
|
||||
def test_add_where_clauses_with_duplicated_field():
|
||||
sql = "UPDATE table SET test1=%(test1)s"
|
||||
params = {"test1": "new_value"}
|
||||
where_clauses = {"test1": "where_value"}
|
||||
params = dict(test1="new_value")
|
||||
where_clauses = dict(test1="where_value")
|
||||
assert PgDB._add_where_clauses(sql, params, where_clauses) == (
|
||||
sql + ' WHERE "test1" = %(test1_1)s',
|
||||
{"test1": "new_value", "test1_1": "where_value"},
|
||||
dict(test1="new_value", test1_1="where_value"),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -291,7 +289,7 @@ def test_quote_table_name():
|
|||
|
||||
|
||||
def test_insert(mocker, test_pgdb):
|
||||
values = {"test1": 1, "test2": 2}
|
||||
values = dict(test1=1, test2=2)
|
||||
mocker.patch(
|
||||
"mylib.pgsql.PgDB.doSQL",
|
||||
generate_mock_doSQL(
|
||||
|
|
@ -304,18 +302,18 @@ def test_insert(mocker, test_pgdb):
|
|||
|
||||
def test_insert_just_try(mocker, test_pgdb):
|
||||
mocker.patch("mylib.pgsql.PgDB.doSQL", mock_doSQL_just_try)
|
||||
assert test_pgdb.insert("mytable", {"test1": 1, "test2": 2}, just_try=True)
|
||||
assert test_pgdb.insert("mytable", dict(test1=1, test2=2), just_try=True)
|
||||
|
||||
|
||||
def test_update(mocker, test_pgdb):
|
||||
values = {"test1": 1, "test2": 2}
|
||||
where_clauses = {"test3": 3, "test4": 4}
|
||||
values = dict(test1=1, test2=2)
|
||||
where_clauses = dict(test3=3, test4=4)
|
||||
mocker.patch(
|
||||
"mylib.pgsql.PgDB.doSQL",
|
||||
generate_mock_doSQL(
|
||||
'UPDATE "mytable" SET "test1" = %(test1)s, "test2" = %(test2)s WHERE "test3" ='
|
||||
' %(test3)s AND "test4" = %(test4)s',
|
||||
{**values, **where_clauses},
|
||||
dict(**values, **where_clauses),
|
||||
),
|
||||
)
|
||||
|
||||
|
|
@ -324,11 +322,11 @@ def test_update(mocker, test_pgdb):
|
|||
|
||||
def test_update_just_try(mocker, test_pgdb):
|
||||
mocker.patch("mylib.pgsql.PgDB.doSQL", mock_doSQL_just_try)
|
||||
assert test_pgdb.update("mytable", {"test1": 1, "test2": 2}, None, just_try=True)
|
||||
assert test_pgdb.update("mytable", dict(test1=1, test2=2), None, just_try=True)
|
||||
|
||||
|
||||
def test_delete(mocker, test_pgdb):
|
||||
where_clauses = {"test1": 1, "test2": 2}
|
||||
where_clauses = dict(test1=1, test2=2)
|
||||
mocker.patch(
|
||||
"mylib.pgsql.PgDB.doSQL",
|
||||
generate_mock_doSQL(
|
||||
|
|
@ -357,17 +355,17 @@ def test_truncate_just_try(mocker, test_pgdb):
|
|||
|
||||
def test_select(mocker, test_pgdb):
|
||||
fields = ("field1", "field2")
|
||||
where_clauses = {"test3": 3, "test4": 4}
|
||||
where_clauses = dict(test3=3, test4=4)
|
||||
expected_return = [
|
||||
{"field1": 1, "field2": 2},
|
||||
{"field1": 2, "field2": 3},
|
||||
dict(field1=1, field2=2),
|
||||
dict(field1=2, field2=3),
|
||||
]
|
||||
order_by = "field1, DESC"
|
||||
mocker.patch(
|
||||
"mylib.pgsql.PgDB.doSelect",
|
||||
generate_mock_doSQL(
|
||||
'SELECT "field1", "field2" FROM "mytable" WHERE "test3" = %(test3)s AND "test4" ='
|
||||
" %(test4)s ORDER BY " + order_by, # nosec: B608
|
||||
" %(test4)s ORDER BY " + order_by,
|
||||
where_clauses,
|
||||
expected_return,
|
||||
),
|
||||
|
|
@ -393,12 +391,9 @@ def test_select_just_try(mocker, test_pgdb):
|
|||
|
||||
|
||||
def test_connect(mocker, test_pgdb):
|
||||
expected_kwargs = {
|
||||
"dbname": test_pgdb._db,
|
||||
"user": test_pgdb._user,
|
||||
"host": test_pgdb._host,
|
||||
"password": test_pgdb._pwd,
|
||||
}
|
||||
expected_kwargs = dict(
|
||||
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))
|
||||
|
||||
|
|
@ -428,7 +423,7 @@ def test_setEncoding_on_exception(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_params = {"test1": 1}
|
||||
fake_connected_pgdb._conn.expected_params = dict(test1=1)
|
||||
fake_connected_pgdb.doSQL(
|
||||
fake_connected_pgdb._conn.expected_sql, fake_connected_pgdb._conn.expected_params
|
||||
)
|
||||
|
|
@ -450,8 +445,8 @@ def test_doSQL_on_exception(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_params = {"test1": 1}
|
||||
fake_connected_pgdb._conn.expected_return = [{"test1": 1}]
|
||||
fake_connected_pgdb._conn.expected_params = dict(test1=1)
|
||||
fake_connected_pgdb._conn.expected_return = [dict(test1=1)]
|
||||
assert (
|
||||
fake_connected_pgdb.doSelect(
|
||||
fake_connected_pgdb._conn.expected_sql, fake_connected_pgdb._conn.expected_params
|
||||
|
|
@ -462,7 +457,7 @@ def test_doSelect(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_return = [{"test1": 1}]
|
||||
fake_connected_pgdb._conn.expected_return = [dict(test1=1)]
|
||||
assert (
|
||||
fake_connected_pgdb.doSelect(fake_connected_pgdb._conn.expected_sql)
|
||||
== fake_connected_pgdb._conn.expected_return
|
||||
|
|
@ -476,8 +471,8 @@ def test_doSelect_on_exception(fake_connected_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_params = {"test1": 1}
|
||||
fake_connected_just_try_pgdb._conn.expected_return = [{"test1": 1}]
|
||||
fake_connected_just_try_pgdb._conn.expected_params = dict(test1=1)
|
||||
fake_connected_just_try_pgdb._conn.expected_return = [dict(test1=1)]
|
||||
assert (
|
||||
fake_connected_just_try_pgdb.doSelect(
|
||||
fake_connected_just_try_pgdb._conn.expected_sql,
|
||||
|
|
|
|||
|
|
@ -25,12 +25,12 @@ def test_create_telltale_file(tmp_path):
|
|||
|
||||
def test_create_telltale_file_with_filepath_and_invalid_dirpath():
|
||||
with pytest.raises(AssertionError):
|
||||
TelltaleFile(filepath="/tmp/test", dirpath="/var/tmp") # nosec: B108
|
||||
TelltaleFile(filepath="/tmp/test", dirpath="/var/tmp")
|
||||
|
||||
|
||||
def test_create_telltale_file_with_filepath_and_invalid_filename():
|
||||
with pytest.raises(AssertionError):
|
||||
TelltaleFile(filepath="/tmp/test", filename="other") # nosec: B108
|
||||
TelltaleFile(filepath="/tmp/test", filename="other")
|
||||
|
||||
|
||||
def test_remove_telltale_file(tmp_path):
|
||||
|
|
|
|||
Loading…
Reference in a new issue