Class for uncritical errors, which could and should be caught.
"""
+ pass
+
+
+# =============================================================================
+class DataDirOptionAction(argparse.Action, UncriticalHookError):
+
+ # -------------------------------------------------------------------------
+ def __call__(self, parser, namespace, data_dir, option_string=None):
+
+ if not data_dir.exists():
+ msg = "Data directory {!r} does not exists.".format(str(data_dir))
+ raise argparse.ArgumentError(self, msg)
+
+ if not data_dir.is_dir():
+ msg = "Path to data directory {!r} is not a directory.".format(str(data_dir))
+ raise argparse.ArgumentError(self, msg)
+
+ setattr(namespace, self.dest, data_dir)
+
# =============================================================================
class BaseHookApp(BaseApplication):
"""
main_branches = ('development', 'test', 'production')
# -------------------------------------------------------------------------
- def __init__(self, appname=None, base_dir=None, verbose=0, version=__version__, no_cc=False):
+ def __init__(
+ self, appname=None, base_dir=None, verbose=0, version=__version__,
+ initialized=False, usage=None, description=None, argparse_epilog=None,
+ argparse_prefix_chars='-', env_prefix=None, output_type='json', no_cc=False):
"""Constructor."""
if not base_dir:
base_dir = pathlib.Path(self.cgi_bin_dir)
- if not getattr(self, 'description', None):
- self.description = "Base gitlab webhook application."
+ if not description:
+ description = "Base gitlab webhook application."
self.data_dir = self.default_data_dir
self._read_stdin = True
self._mail_subject = None
self._mail_headline = None
self._sort_by_name = False
-
- if not hasattr(self, '_output_type'):
- self._output_type = self.default_output_type
+ self._output_type = self.default_output_type
self._mime_type = self.default_mime_type
- super(BaseHookApp, self).__init__(
- appname=appname, verbose=verbose, version=version,
- base_dir=base_dir, initialized=False,
- )
-
self._mime_type = self.valid_output_types[self.output_type]
- self._start_verbose = self.verbose
- self._simulate = False
self.data = None
self.json_data = None
self.tz = None
self.query = {}
- self.cmdline_args = None
-
self.curl_bin = None
self.handler = None
self.mail_cc_addresses = [self.default_email, ]
self.sender_address = DEFAULT_FROM_SENDER
+ self._log_directory = pathlib.Path('/var/log/webhooks')
+
try:
pwd_info = pwd.getpwuid(os.geteuid())
self.sender_address = "{u} at {h} <{a}>".format(
u=pwd_info.pw_name, h=socket.gethostname(), a=DEFAULT_FROM_EMAIL)
except Exception as e:
- sys.stderr.write("\n{cn}: {e}\n\n".format(cn=e.__class__.__name__, e=e))
+ self.handle_error(e)
+ sys.stderr.write('\n')
- self._log_directory = os.sep + os.path.join('var', 'log', 'webhooks')
-
- self.do_arg_parser()
+ super(BaseHookApp, self).__init__(
+ appname=appname, verbose=verbose, version=version, base_dir=base_dir,
+ description=description, initialized=False,
+ )
- self.read_config()
+ self.output_type = output_type
- if self.cmdline_args.data_dir:
- self._log_directory = self.data_dir
+ # -------------------------------------------------------------------------
+ def post_init(self):
+ self.initialized = False
+ self.perform_arg_parser_logging()
self.init_logging()
- if 'REQUEST_METHOD' in os.environ:
+ self.perform_arg_parser()
+
+ if self.is_cgi:
LOG.debug("We are in a CGI: REQUEST_METHOD {!r}".format(os.environ['REQUEST_METHOD']))
else:
LOG.debug("We are NOT in CGI.")
simulate=self.simulate)
self.tz = pytz.timezone(self.tz_name)
-
- self.post_init()
+ self._start_verbose = self.verbose
return
+ # -------------------------------------------------------------------------
+ def init_arg_parser(self):
+ """
+ Public available method to initiate the argument parser.
+ """
+
+ super(BaseHookApp, self).init_arg_parser()
+
+ hook_group = self.arg_parser.add_argument_group("General hook options")
+
+ hook_group.add_argument(
+ "-N", "--no-error-mail", action='store_true', dest='no_error_mail',
+ help="Don't send error messages in case of some exceptions.",
+ )
+
+ hook_group.add_argument(
+ "-D", '--data', '--data-dir', metavar='DIR', dest='data_dir',
+ action=DataDirOptionAction, type=pathlib.Path,
+ help="Data directory, default: {!r}.".format(self.data_dir),
+ )
+
+ sort_group = hook_group.add_mutually_exclusive_group()
+
+ sort_group.add_argument(
+ '-o', '--sort-by-name', action="store_true", dest='sort_by_name',
+ help="Sorting all lists of modules by name and vendor, in this order."
+ )
+
+ sort_group.add_argument(
+ '-O', '--sort-by-fullname', action="store_false", dest='sort_by_name',
+ help="Sorting all lists of modules by the full name of the module (default)."
+ )
+
+ hook_group.add_argument(
+ "-C", '--cgi', action='store_true', dest='cgi',
+ help='Enforces behaviour as called as a CGI script.',
+ )
+
+ self.arg_parser.add_argument(
+ 'query', nargs='?',
+ help="An optional query string like on HTTP GET requests."
+ )
+
+ # -------------------------------------------------------------------------
+ def perform_arg_parser_logging(self):
+
+ if self.args.data_dir:
+ self._log_directory = self.data_dir
+
+ # -------------------------------------------------------------------------
+ def perform_arg_parser(self):
+
+ if self.args.no_error_mail:
+ self.no_error_mail = True
+
+ if self.args.data_dir:
+ path = self.args.data_dir
+ if not path.is_absolute():
+ path = path.resolve()
+ self.data_dir = path
+
+ if self.args.cgi:
+ if not os.environ.get('REQUEST_METHOD', None):
+ os.environ['REQUEST_METHOD'] = 'GET'
+
+ if self.args.query:
+ if not os.environ.get('QUERY_STRING', None):
+ os.environ['QUERY_STRING'] = self.args.query
+
+ if self.args.sort_by_name:
+ self._sort_by_name = True
+
+ # -------------------------------------------------------------------------
+ @property
+ def is_cgi(self):
+ """Shows, that the current script is executing as a CGI script."""
+ if 'REQUEST_METHOD' in os.environ:
+ return True
+ return False
+
# -----------------------------------------------------------
@property
def simulate(self):
# -----------------------------------------------------------
@property
def read_stdin(self):
- """Flag, tzhat STDIN shoud be read."""
+ """Flag, that STDIN shoud be read."""
return getattr(self, '_read_stdin', True)
@read_stdin.setter
@property
def logfile(self):
"""The logfile of this application."""
- return os.path.join(self.log_directory, self.appname + '.log')
+ return self.log_directory.joinpath(self.appname + '.log')
# -----------------------------------------------------------
@property
def error_logfile(self):
"""The logfile for STDERR of this application."""
- return os.path.join(self.log_directory, self.appname + '.error.log')
+ return self.log_directory.joinpath(self.appname + '.error.log')
# -----------------------------------------------------------
@property
"""
res = super(BaseHookApp, self).as_dict(short=short)
- res['simulate'] = self.simulate
+ res['is_cgi'] = self.is_cgi
res['cgi_bin_dir'] = self.cgi_bin_dir
res['log_directory'] = self.log_directory
res['error_logfile'] = self.error_logfile
self._perform_args(arg_parser)
- # -------------------------------------------------------------------------
- def init_arg_parser(self, arg_parser):
-
- pass
-
- # -------------------------------------------------------------------------
- def _init_arg_parser(self, arg_parser):
-
- self.init_arg_parser(arg_parser)
-
- general_group = arg_parser.add_argument_group('General options')
-
- general_group.add_argument(
- "-N", "--no-error-mail", action='store_true', dest='no_error_mail',
- help="Don't send error messages in case of some exceptions.",
- )
-
- general_group.add_argument(
- "-D", '--data', '--data-dir', metavar='DIR', dest='data_dir',
- help="Data directory, default: {!r}.".format(self.data_dir),
- )
-
- sort_group = general_group.add_mutually_exclusive_group()
-
- sort_group.add_argument(
- '-o', '--sort-by-name', action="store_true", dest='sort_by_name',
- help="Sorting all lists of modules by name and vendor, in this order."
- )
-
- sort_group.add_argument(
- '-O', '--sort-by-fullname', action="store_false", dest='sort_by_name',
- help="Sorting all lists of modules by the full name of the module (default)."
- )
-
- general_group.add_argument(
- "-v", "--verbose", action="count", dest='verbose',
- help='Increase the verbosity level',
- )
-
- general_group.add_argument(
- '-s', '--simulate', '--test', action='store_true', dest='simulate',
- help="Simulation mode, nothing is really done.",
- )
-
- general_group.add_argument(
- "-C", '--cgi', action='store_true', dest='cgi',
- help='Enforces behaviour as called as a CGI script.',
- )
-
- general_group.add_argument(
- "-h", "--help", action='help', dest='help',
- help='Show this help message and exit'
- )
-
- general_group.add_argument(
- "--usage", action='store_true', dest='usage',
- help="Display brief usage message and exit"
- )
-
- general_group.add_argument(
- "-V", '--version', action='version',
- version='Version of %(prog)s: {}'.format(self.version),
- help="Show program's version number and exit"
- )
-
- arg_parser.add_argument(
- 'query', nargs='?',
- help="An optional query string like on HTTP GET requests."
- )
-
- # -------------------------------------------------------------------------
- def _perform_args(self, arg_parser):
-
- self.cmdline_args = arg_parser.parse_args()
-
- if self.cmdline_args.usage:
- arg_parser.print_usage(sys.stdout)
- sys.exit(0)
-
- if self.cmdline_args.no_error_mail:
- self.no_error_mail = True
-
- if self.cmdline_args.data_dir:
- path = self.cmdline_args.data_dir
- if not os.path.isabs(path):
- path = os.path.normpath(os.path.join(os.getcwd(), path))
- self.data_dir = path
-
- if self.cmdline_args.verbose is not None:
- if self.cmdline_args.verbose > self.verbose:
- self.verbose = self.cmdline_args.verbose
- if self.cmdline_args.verbose > self._start_verbose:
- self._start_verbose = self.cmdline_args.verbose
-
- if self.cmdline_args.cgi:
- if not os.environ.get('REQUEST_METHOD', None):
- os.environ['REQUEST_METHOD'] = 'GET'
-
- if self.cmdline_args.query:
- if not os.environ.get('QUERY_STRING', None):
- os.environ['QUERY_STRING'] = self.cmdline_args.query
-
- if self.cmdline_args.simulate:
- sys.stderr.write("\nSimulation mode - nothing is really done.\n\n")
- self.simulate = True
-
- if self.cmdline_args.sort_by_name:
- self._sort_by_name = True
-
- self._get_query()
- if 'output_type' in self.query:
- try:
- self.output_type = self.query['output_type']
- except ValueError as e:
- LOG.error(str(e))
-
- self.perform_args(arg_parser)
-
- # -------------------------------------------------------------------------
- def perform_args(self, arg_parser):
-
- pass
-
+# # -------------------------------------------------------------------------
+# def __init_arg_parser(self):
+#
+# self.init_arg_parser(arg_parser)
+#
+# general_group = arg_parser.add_argument_group('General options')
+#
+# general_group.add_argument(
+# "-N", "--no-error-mail", action='store_true', dest='no_error_mail',
+# help="Don't send error messages in case of some exceptions.",
+# )
+#
+# general_group.add_argument(
+# "-D", '--data', '--data-dir', metavar='DIR', dest='data_dir',
+# help="Data directory, default: {!r}.".format(self.data_dir),
+# )
+#
+# sort_group = general_group.add_mutually_exclusive_group()
+#
+# sort_group.add_argument(
+# '-o', '--sort-by-name', action="store_true", dest='sort_by_name',
+# help="Sorting all lists of modules by name and vendor, in this order."
+# )
+#
+# sort_group.add_argument(
+# '-O', '--sort-by-fullname', action="store_false", dest='sort_by_name',
+# help="Sorting all lists of modules by the full name of the module (default)."
+# )
+#
+# general_group.add_argument(
+# "-v", "--verbose", action="count", dest='verbose',
+# help='Increase the verbosity level',
+# )
+#
+# general_group.add_argument(
+# '-s', '--simulate', '--test', action='store_true', dest='simulate',
+# help="Simulation mode, nothing is really done.",
+# )
+#
+# general_group.add_argument(
+# "-C", '--cgi', action='store_true', dest='cgi',
+# help='Enforces behaviour as called as a CGI script.',
+# )
+#
+# general_group.add_argument(
+# "-h", "--help", action='help', dest='help',
+# help='Show this help message and exit'
+# )
+#
+# general_group.add_argument(
+# "--usage", action='store_true', dest='usage',
+# help="Display brief usage message and exit"
+# )
+#
+# general_group.add_argument(
+# "-V", '--version', action='version',
+# version='Version of %(prog)s: {}'.format(self.version),
+# help="Show program's version number and exit"
+# )
+#
+# arg_parser.add_argument(
+# 'query', nargs='?',
+# help="An optional query string like on HTTP GET requests."
+# )
+#
+# # -------------------------------------------------------------------------
+# def __perform_args(self, arg_parser):
+#
+# self.cmdline_args = arg_parser.parse_args()
+#
+# if self.cmdline_args.usage:
+# arg_parser.print_usage(sys.stdout)
+# sys.exit(0)
+#
+# if self.cmdline_args.no_error_mail:
+# self.no_error_mail = True
+#
+# if self.cmdline_args.data_dir:
+# path = self.cmdline_args.data_dir
+# if not os.path.isabs(path):
+# path = os.path.normpath(os.path.join(os.getcwd(), path))
+# self.data_dir = path
+#
+# if self.cmdline_args.verbose is not None:
+# if self.cmdline_args.verbose > self.verbose:
+# self.verbose = self.cmdline_args.verbose
+# if self.cmdline_args.verbose > self._start_verbose:
+# self._start_verbose = self.cmdline_args.verbose
+#
+# if self.cmdline_args.cgi:
+# if not os.environ.get('REQUEST_METHOD', None):
+# os.environ['REQUEST_METHOD'] = 'GET'
+#
+# if self.cmdline_args.query:
+# if not os.environ.get('QUERY_STRING', None):
+# os.environ['QUERY_STRING'] = self.cmdline_args.query
+#
+# if self.cmdline_args.simulate:
+# sys.stderr.write("\nSimulation mode - nothing is really done.\n\n")
+# self.simulate = True
+#
+# if self.cmdline_args.sort_by_name:
+# self._sort_by_name = True
+#
+# self._get_query()
+# if 'output_type' in self.query:
+# try:
+# self.output_type = self.query['output_type']
+# except ValueError as e:
+# LOG.error(str(e))
+#
+# self.perform_args(arg_parser)
+#
+# # -------------------------------------------------------------------------
+# def perform_args(self, arg_parser):
+#
+# pass
+#
# -------------------------------------------------------------------------
def _get_query(self):
file.flush()
# -------------------------------------------------------------------------
- def post_init(self):
- """ Dummy function, which is called after initialization of the application object.
- May be overridden in child classes.
- """
-
- pass
-
- # -------------------------------------------------------------------------
- def __call__(self):
- """Helper method to make the resulting object callable."""
+ def pre_run(self):
+ """Method to run before the main routine."""
self.print_out("Content-Type: {};charset=utf-8\n".format(self.mime_type))
if self.output_type == 'txt':
self.data = input_stream.read()
else:
self.data = '{}'
+
+ # -------------------------------------------------------------------------
+ def _run(self):
+ '''Main routine.'''
+
try:
self.json_data = json.loads(self.data)
if self.verbose > 1:
LOG.debug("Got JSON data:\n{}".format(pp(self.json_data)))
try:
- if self.pre_run():
- self.run()
+ if self.pre_run_hook():
+ self.run_hook()
except BaseHookError as e:
cn = e.__class__.__name__
n = ''
msg += "\nInput data: {!r}".format(self.data)
LOG.error(msg)
self.error_data.append(msg)
- finally:
- if self.full_name:
- self.send_error_msgs(self.full_name)
- else:
- self.send_error_msgs()
- if self.output_type == 'html':
- self.print_htlm_end()
- LOG.debug("Finished.")
- sys.exit(0)
+
+ # -------------------------------------------------------------------------
+ def run_hook(self):
+ """Dummy method, must be overridden."""
+
+ raise FunctionNotImplementedError('_run()', self.__class__.__name__)
+
+ # -------------------------------------------------------------------------
+ def post_run(self):
+
+ if self.verbose > 1:
+ LOG.info("Executing {} ...".format('post_run()'))
+
+ if self.full_name:
+ self.send_error_msgs(self.full_name)
+ else:
+ self.send_error_msgs()
+ if self.output_type == 'html':
+ self.print_htlm_end()
+ LOG.debug("Finished.")
+ self.exit_value = 0
# -------------------------------------------------------------------------
def print_htlm_start(self):
self.print_out('</body>\n</html>\n')
# -------------------------------------------------------------------------
- def pre_run(self):
+ def pre_run_hook(self):
if not self.json_data and not self.read_stdin:
return True
return
- # -------------------------------------------------------------------------
- def run(self):
- """Main routine, must be overridden in descendant classes."""
-
- msg = "Method run() must be overridden in descendant classes of {!r}.".format(
- self.__class__.__name__)
- raise NotImplementedError(msg)
+# # -------------------------------------------------------------------------
+# def run(self):
+# """Main routine, must be overridden in descendant classes."""
+#
+# msg = "Method run() must be overridden in descendant classes of {!r}.".format(
+# self.__class__.__name__)
+# raise NotImplementedError(msg)
# -------------------------------------------------------------------------
def read_cache_file(self, only_main_branches=True):
projects / pixelpark / puppetmaster-webhooks.git / commitdiff