bn8/python-mylib
1
0
Fork 0
Code Issues Pull requests Projects Releases 12 Wiki Activity Actions

Config: add stuff to handle just-try mode in ConfigurableObject class

Browse source
This commit is contained in:
Benjamin Renard 2023-06-19 17:07:59 +02:00
parent 72877dd13e
commit f597164305
Signed by: bn8
GPG key ID: 3E2E1CE1907115BC
8 changed files with 135 additions and 76 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

78
mylib/config.py
View file

@ -1107,7 +1107,9 @@ class ConfigurableObject:
# Default options value
# Important: all supported options MUST HAVE a default value defined
_defaults = {}
_defaults = {
"just_try": None,
}
# Store options passed throuht __init__ method
_kwargs = {}
@ -1166,25 +1168,93 @@ class ConfigurableObject:
for option, default_value in default_values.items():
self.set_default(option, default_value)
def configure(self, comment=None, **kwargs):
"""Configure options on registered mylib.Config object"""
def configure(
self,
comment=None,
just_try=False,
just_try_default=False,
just_try_help="Just-try mode",
**kwargs,
):
"""
Configure options on registered mylib.Config object
:param comment: Configuration section comment (default: self._config_comment)
:param just_try: Add just-try mode option (default: False)
:param just_try_default: Default just-try mode option value (default: False)
:param just_try_help: Default just-try mode option help message (default: "Just-try mode")
:param kwargs: Other provided parameters are directly passed to Config.add_section() method
"""
assert self._config, (
"mylib.Config object not registered. Must be passed to __init__ as config keyword"
" argument."
)
return self._config.add_section(
section = self._config.add_section(
self._config_section,
comment=comment if comment else self._config_comment,
loaded_callback=self.initialize,
**kwargs,
)
if just_try:
self._defaults["just_try"] = just_try_default
section.add_option(
BooleanOption,
"just_try",
default=self._defaults["just_try"],
comment=just_try_help if just_try_help else "Just-try mode",
)
return section
def initialize(self, loaded_config=None):
"""Configuration initialized hook"""
if loaded_config:
self.config = loaded_config # pylint: disable=attribute-defined-outside-init
@property
def _just_try(self):
"""Check if just-try mode is enabled"""
# If "just_try" provided to constructor, use it value
if "just_try" in self._kwargs:
log.debug(
"Just-try mode is %s by value passed to constructor",
"enabled" if self._kwargs["just_try"] else "disabled",
)
return self._kwargs["just_try"]
# If options provided and just-try option exist and is enabled, just-try mode enabled
if (
self._options
and hasattr(self._options, f"{self._options_prefix}just_try")
and getattr(self._options, f"{self._options_prefix}just_try")
):
log.debug("Just-try mode for %s is enabled", __class__.__name__)
return True
# If options provided and a just_try option exist and is enabled, just-try mode enabled
if (
self._options
and hasattr(self._options, "just_try")
and getattr(self._options, "just_try")
):
log.debug("Just-try mode is globally enabled")
return True
# If Config provided, config section defined and just-try enabled in config, just-try mode
# enabled
if (
self._config
and self._config.defined(self._config_section, "just_try")
and self._config.get(self._config_section, "just_try")
):
log.debug("Just-try mode for %s is enabled in configuration", self._config_section)
return True
# If Config provided, use it's get_option() method to obtain a global just_try parameter
# value with a defaut to False, otherwise always false
return self._config.get_option("just_try", default=False) if self._config else False
class ConfigSectionAsDictWrapper:
"""

19
mylib/email.py
View file

@ -60,9 +60,12 @@ class EmailClient(
self.initialize()
# pylint: disable=arguments-differ,arguments-renamed
def configure(self, use_smtp=True, just_try=True, **kwargs):
def configure(self, use_smtp=True, **kwargs):
"""Configure options on registered mylib.Config object"""
section = super().configure(**kwargs)
section = super().configure(
just_try_help=kwargs.pop("just_try_help", "Just-try mode: do not really send emails"),
**kwargs,
)
if use_smtp:
section.add_option(
@ -132,14 +135,6 @@ class EmailClient(
comment="Catch all sent emails to this specified email address",
)
if just_try:
section.add_option(
BooleanOption,
"just_try",
default=self._defaults["just_try"],
comment="Just-try mode: do not really send emails",
)
section.add_option(
StringOption,
"templates_path",
@ -302,7 +297,7 @@ class EmailClient(
return msg
def send(
self, recipients, msg=None, subject=None, just_try=False, cc=None, bcc=None, **forge_args
self, recipients, msg=None, subject=None, just_try=None, cc=None, bcc=None, **forge_args
):
"""
Send an email
@ -350,7 +345,7 @@ class EmailClient(
]
)
if just_try or self._get_option("just_try"):
if just_try if just_try is not None else self._just_try:
log.debug(
'Just-try mode: do not really send this email to %s (subject="%s")',
", ".join(recipients),

12
mylib/report.py
View file

@ -19,6 +19,7 @@ class Report(ConfigurableObject): # pylint: disable=useless-object-inheritance
"subject": "Report",
"loglevel": "WARNING",
"logformat": "%(asctime)s - %(levelname)s - %(message)s",
"just_try": False,
}
content = []
@ -37,7 +38,10 @@ class Report(ConfigurableObject): # pylint: disable=useless-object-inheritance
def configure(self, **kwargs): # pylint: disable=arguments-differ
"""Configure options on registered mylib.Config object"""
section = super().configure(**kwargs)
section = super().configure(
just_try_help=kwargs.pop("just_try_help", "Just-try mode: do not really send report"),
**kwargs,
)
section.add_option(StringOption, "recipient", comment="Report recipient email address")
section.add_option(
@ -97,7 +101,7 @@ class Report(ConfigurableObject): # pylint: disable=useless-object-inheritance
"""Add attachment payload"""
self._attachment_payloads.append(payload)
def send(self, subject=None, rcpt_to=None, email_client=None, just_try=False):
def send(self, subject=None, rcpt_to=None, email_client=None, just_try=None):
"""Send report using an EmailClient"""
if rcpt_to is None:
rcpt_to = self._get_option("recipient")
@ -124,7 +128,9 @@ class Report(ConfigurableObject): # pylint: disable=useless-object-inheritance
attachment_files=self._attachment_files,
attachment_payloads=self._attachment_payloads,
)
if email_client.send(rcpt_to, msg=msg, just_try=just_try):
if email_client.send(
rcpt_to, msg=msg, just_try=just_try if just_try is not None else self._just_try
):
log.debug("Report sent to %s", rcpt_to)
return True
log.error("Fail to send report to %s", rcpt_to)

2
mylib/scripts/email_test_with_config.py
View file

@ -22,7 +22,7 @@ def main(argv=None): # pylint: disable=too-many-locals,too-many-statements
"templates_path",
os.path.join(os.path.dirname(os.path.realpath(__file__)), "email_templates"),
)
email_client.configure()
email_client.configure(just_try=True)
# Options parser
parser = config.get_arguments_parser(description=__doc__)

18
mylib/scripts/helpers.py
View file

@ -240,23 +240,7 @@ def init_email_client(options, **kwargs):
from mylib.email import EmailClient # pylint: disable=import-outside-toplevel
log.info("Initialize Email client")
return EmailClient(
smtp_host=options.email_smtp_host,
smtp_port=options.email_smtp_port,
smtp_ssl=options.email_smtp_ssl,
smtp_tls=options.email_smtp_tls,
smtp_user=options.email_smtp_user,
smtp_password=options.email_smtp_password,
smtp_debug=options.email_smtp_debug,
sender_name=options.email_sender_name,
sender_email=options.email_sender_email,
catch_all_addr=options.email_catch_all,
just_try=options.just_try if hasattr(options, "just_try") else False,
encoding=options.email_encoding,
templates_path=options.email_templates_path,
initialize=True,
**kwargs,
)
return EmailClient(options=options, initialize=True, **kwargs)
def add_sftp_opts(parser):

11
mylib/scripts/report_test.py
View file

@ -20,16 +20,21 @@ def main(argv=None): # pylint: disable=too-many-locals,too-many-statements
report_opts = parser.add_argument_group("Report options")
report_opts.add_argument(
"-t", "--to", action="store", type=str, dest="report_rcpt", help="Send report to this email"
"-t",
"--to",
action="store",
type=str,
dest="report_recipient",
help="Send report to this email",
)
options = parser.parse_args()
if not options.report_rcpt:
if not options.report_recipient:
parser.error("You must specify a report recipient using -t/--to parameter")
# Initialize logs
report = Report(rcpt_to=options.report_rcpt, subject="Test report")
report = Report(options=options, subject="Test report")
init_logging(options, "Test Report", report=report)
email_client = init_email_client(options)

49
mylib/scripts/sftp_test.py
View file

@ -43,7 +43,7 @@ def main(argv=None): # pylint: disable=too-many-locals,too-many-statements
options.sftp_password = getpass.getpass("Please enter SFTP password: ")
log.info("Initialize Email client")
sftp = SFTPClient(options=options, just_try=options.just_try)
sftp = SFTPClient(options=options)
sftp.connect()
atexit.register(sftp.close)
@ -74,30 +74,31 @@ def main(argv=None): # pylint: disable=too-many-locals,too-many-statements
else os.path.basename(tmp_file)
)
with tempfile.NamedTemporaryFile() as tmp_file2:
log.info("Retrieve test file to %s", tmp_file2.name)
if not sftp.get_file(remote_filepath, tmp_file2.name):
log.error("Fail to retrieve test file")
else:
with open(tmp_file2.name, "rb") as file_desc:
content = file_desc.read()
log.debug("Read content: %s", content)
if test_content == content:
log.info("Content file retrieved match with uploaded one")
else:
log.error("Content file retrieved doest not match with uploaded one")
if not sftp._just_try: # pylint: disable=protected-access
with tempfile.NamedTemporaryFile() as tmp_file2:
log.info("Retrieve test file to %s", tmp_file2.name)
if not sftp.get_file(remote_filepath, tmp_file2.name):
log.error("Fail to retrieve test file")
else:
with open(tmp_file2.name, "rb") as file_desc:
content = file_desc.read()
log.debug("Read content: %s", content)
if test_content == content:
log.info("Content file retrieved match with uploaded one")
else:
log.error("Content file retrieved doest not match with uploaded one")
try:
log.info("Remotly open test file %s", remote_filepath)
file_desc = sftp.open_file(remote_filepath)
content = file_desc.read()
log.debug("Read content: %s", content)
if test_content == content:
log.info("Content of remote file match with uploaded one")
else:
log.error("Content of remote file doest not match with uploaded one")
except Exception: # pylint: disable=broad-except
log.exception("An exception occurred remotly opening test file %s", remote_filepath)
try:
log.info("Remotly open test file %s", remote_filepath)
file_desc = sftp.open_file(remote_filepath)
content = file_desc.read()
log.debug("Read content: %s", content)
if test_content == content:
log.info("Content of remote file match with uploaded one")
else:
log.error("Content of remote file doest not match with uploaded one")
except Exception: # pylint: disable=broad-except
log.exception("An exception occurred remotly opening test file %s", remote_filepath)
if sftp.remove_file(remote_filepath):
log.info("Test file removed on SFTP server")

22
mylib/sftp.py
View file

@ -40,9 +40,15 @@ class SFTPClient(ConfigurableObject):
initial_directory = None
# pylint: disable=arguments-differ,arguments-renamed
def configure(self, just_try=True, **kwargs):
def configure(self, **kwargs):
"""Configure options on registered mylib.Config object"""
section = super().configure(**kwargs)
section = super().configure(
just_try=kwargs.pop("just_try", True),
just_try_help=kwargs.pop(
"just_try_help", "Just-try mode: do not really make change on remote SFTP host"
),
**kwargs,
)
section.add_option(
StringOption,
@ -80,14 +86,6 @@ class SFTPClient(ConfigurableObject):
comment="Auto add unknown host key",
)
if just_try:
section.add_option(
BooleanOption,
"just_try",
default=self._defaults["just_try"],
comment="Just-try mode: do not really make change on remote SFTP host",
)
return section
def initialize(self, loaded_config=None):
@ -141,7 +139,7 @@ class SFTPClient(ConfigurableObject):
os.path.basename(filepath),
)
log.debug("Upload file '%s' to '%s'", filepath, remote_filepath)
if self._get_option("just_try"):
if self._just_try:
log.debug(
"Just-try mode: do not really upload file '%s' to '%s'", filepath, remote_filepath
)
@ -153,7 +151,7 @@ class SFTPClient(ConfigurableObject):
"""Remove a file on SFTP server"""
self.connect()
log.debug("Remove file '%s'", filepath)
if self._get_option("just_try"):
if self._just_try:
log.debug("Just - try mode: do not really remove file '%s'", filepath)
return True
return self.sftp_client.remove(filepath) is None