Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if ruid in locals:
ldata = locals[ruid]
if ldata["mtime"] < rdata["mtime"]:
if ldata["name"] != rdata["name"]:
self.logger.info("removing %s %s" % (obj_type, ldata["name"]))
self.api.remove_item(obj_type, ldata["name"], recursive=True, logger=self.logger)
creator = getattr(self.api, "new_%s" % obj_type)
newobj = creator()
newobj.from_dict(rdata)
try:
self.logger.info("updating %s %s" % (obj_type, rdata["name"]))
if not self.api.add_item(obj_type, newobj):
self.logger.error("failed to update %s %s" % (obj_type, rdata["name"]))
except Exception:
utils.log_exc(self.logger)
def remove_objects_not_on_master(self, obj_type):
locals = utils.lod_to_dod(self.local_data[obj_type], "uid")
remotes = utils.lod_to_dod(self.remote_data[obj_type], "uid")
for (luid, ldata) in list(locals.items()):
if luid not in remotes:
try:
self.logger.info("removing %s %s" % (obj_type, ldata["name"]))
self.api.remove_item(obj_type, ldata["name"], recursive=True, logger=self.logger)
except Exception:
utils.log_exc(self.logger)
self.found_repos[comps_path] = 1
elif comps_path not in self.found_repos:
utils.remove_yum_olddata(comps_path)
cmd = "createrepo %s --groupfile %s %s" % (self.settings.createrepo_flags, os.path.join(comps_path, masterdir, comps_file), comps_path)
utils.subprocess_call(self.logger, cmd, shell=True)
self.found_repos[comps_path] = 1
# for older distros, if we have a "base" dir parallel with "repodata", we need to copy comps.xml up one...
p1 = os.path.join(comps_path, "repodata", "comps.xml")
p2 = os.path.join(comps_path, "base", "comps.xml")
if os.path.exists(p1) and os.path.exists(p2):
shutil.copyfile(p1, p2)
except:
self.logger.error("error launching createrepo (not installed?), ignoring")
utils.log_exc(self.logger)
tmpfile.flush()
logger.debug("Successfully got file from %s" % self.settings().signature_url)
# test the import without caching it
try:
utils.load_signatures(tmpfile.name, cache=False)
except:
logger.error("Downloaded signatures failed test load (tempfile = %s)" % tmpfile.name)
# rewrite the real signature file and import it for real
f = open(self.settings().signature_path, "w")
f.write(sigjson.text)
f.close()
utils.load_signatures(self.settings().signature_path)
except:
utils.log_exc(logger)
def generate_autoinstall(self, profile=None, system=None, REMOTE_ADDR=None, REMOTE_MAC=None, **rest):
self._log("generate_autoinstall")
try:
return self.autoinstall_mgr.generate_autoinstall(profile, system)
except Exception:
utils.log_exc(self.logger)
return "# This automatic OS installation file had errors that prevented it from being rendered correctly.\n# The cobbler.log should have information relating to this failure."
try:
blip = __import__("cobbler.modules.%s" % (modname), globals(), locals(), [modname])
if not hasattr(blip, "register"):
if not modname.startswith("__init__"):
errmsg = _("%(module_path)s/%(modname)s is not a proper module")
print(errmsg % {'module_path': module_path, 'modname': modname})
continue
category = blip.register()
if category:
MODULE_CACHE[modname] = blip
if category not in MODULES_BY_CATEGORY:
MODULES_BY_CATEGORY[category] = {}
MODULES_BY_CATEGORY[category][modname] = blip
except Exception:
logger.info('Exception raised when loading module %s' % modname)
log_exc(logger)
return (MODULE_CACHE, MODULES_BY_CATEGORY)
if os.getenv(k):
old_env[k] = os.getenv(k)
else:
os.environ[k] = env[k]
# which may actually NOT reposync if the repo is set to not mirror locally
# but that's a technicality
for x in range(self.tries + 1, 1, -1):
success = False
try:
self.sync(repo)
success = True
break
except:
utils.log_exc(self.logger)
self.logger.warning("reposync failed, tries left: %s" % (x - 2))
# cleanup/restore any environment variables that were
# added or changed above
for k in list(env.keys()):
if env[k] is not None:
if k in old_env:
self.logger.debug("resetting repo environment: %s=%s" % (k, old_env[k]))
os.environ[k] = old_env[k]
else:
self.logger.debug("removing repo environment: %s=%s" % (k, env[k]))
del os.environ[k]
if not success:
report_failure = True