Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.verbose = True
self.api = collection_mgr.api
self.collection_mgr = collection_mgr
self.distros = collection_mgr.distros()
self.profiles = collection_mgr.profiles()
self.systems = collection_mgr.systems()
self.settings = collection_mgr.settings()
self.repos = collection_mgr.repos()
self.rflags = self.settings.reposync_flags
self.tries = tries
self.nofail = nofail
self.logger = logger
self.dlmgr = download_manager.DownloadManager(self.collection_mgr, self.logger)
if logger is None:
self.logger = clogger.Logger()
self.logger.info("hello, reposync")
Starts a new background task.
token -- token from login() call, all tasks require tokens
role_name -- used to check token against authn/authz layers
thr_obj_fn -- function handle to run in a background thread
name -- display name to show in logs/events
args -- usually this is a single dict, containing options
on_done -- an optional second function handle to run after success (and only success)
Returns a task id.
"""
self.check_access(token, role_name)
event_id = self.__generate_event_id(role_name) # use short form for logfile suffix
event_id = str(event_id)
self.events[event_id] = [float(time.time()), str(name), EVENT_RUNNING, []]
self._log("start_task(%s); event_id(%s)" % (name, event_id))
logatron = clogger.Logger("/var/log/cobbler/tasks/%s.log" % event_id)
thr_obj = CobblerThread(event_id, self, logatron, args, role_name, self.api)
thr_obj._run = thr_obj_fn
if on_done is not None:
thr_obj.on_done = on_done.__get__(thr_obj, CobblerThread)
thr_obj.start()
return event_id
self.__dict__ = CobblerAPI.__shared_state
self.perms_ok = False
if not CobblerAPI.__has_loaded:
# NOTE: we do not log all API actions, because
# a simple CLI invocation may call adds and such
# to load the config, which would just fill up
# the logs, so we'll do that logging at CLI
# level (and remote.py web service level) instead.
random.seed()
self.is_cobblerd = is_cobblerd
try:
self.logger = clogger.Logger()
except CX:
# return to CLI/other but perms are not valid
# perms_ok is False
return
# FIXME: conslidate into 1 server instance
self.selinux_enabled = utils.is_selinux_enabled()
self.dist, self.os_version = utils.os_release()
CobblerAPI.__has_loaded = True
# load the modules first, or nothing else works...
module_loader.load_modules()
self._collection_mgr = manager.CollectionManager(self)
def load_modules(module_path=mod_path, blacklist=None):
logger = clogger.Logger()
filenames = glob.glob("%s/*.py" % module_path)
filenames += glob.glob("%s/*.pyc" % module_path)
filenames += glob.glob("%s/*.pyo" % module_path)
# Allow recursive modules
filenames += glob.glob("%s/**/*.py" % module_path)
filenames += glob.glob("%s/**/*.pyc" % module_path)
filenames += glob.glob("%s/**/*.pyo" % module_path)
mods = set()
for fn in filenames:
# TODO: Add nested module-names to allow recursivly stored modules to be loaded.
basename = os.path.basename(fn)
if basename == "__init__.py":
continue
def __init__(self, collection_mgr, verbose=False, logger=None):
"""
Constructor
"""
self.verbose = verbose
self.collection_mgr = collection_mgr
self.distros = collection_mgr.distros()
self.profiles = collection_mgr.profiles()
self.systems = collection_mgr.systems()
self.images = collection_mgr.images()
self.settings = collection_mgr.settings()
self.repos = collection_mgr.repos()
if logger is None:
logger = clogger.Logger()
self.logger = logger
self.tftpd = module_loader.get_module_from_file("tftpd", "module", "in_tftpd").get_manager(collection_mgr, logger)
self.sync = collection_mgr.api.get_sync(verbose, logger=self.logger)
self.sync.make_tftpboot()
def __init__(self, collection_mgr, logger=None):
"""
Constructor
@param CollectionManager collection_mgr collection manager
@param Logger logger logger
"""
self.collection_mgr = collection_mgr
self.snippets_base_dir = self.collection_mgr.settings().autoinstall_snippets_dir
self.templates_base_dir = self.collection_mgr.settings().autoinstall_templates_dir
self.autoinstallgen = autoinstallgen.AutoInstallationGen(self.collection_mgr)
if logger is None:
logger = clogger.Logger()
self.logger = logger
def die(logger, msg):
global main_logger
if main_logger is None:
main_logger = clogger.Logger()
# log the exception once in the per-task log or the main
# log if this is not a background op.
try:
raise CX(msg)
except:
if logger is not None:
log_exc(logger)
else:
log_exc(main_logger)
# now re-raise it so the error can fail the operation
raise CX(msg)