| Server IP : 188.40.95.74 / Your IP : 216.73.216.142 Web Server : Apache System : Linux cp01.striminghost.net 3.10.0-1160.119.1.el7.tuxcare.els13.x86_64 #1 SMP Fri Nov 22 06:29:45 UTC 2024 x86_64 User : vlasotin ( 1054) PHP Version : 5.6.40 Disable Function : NONE MySQL : ON | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /usr/share/leapp-repository/repositories/system_upgrade/common/libraries/ |
Upload File : |
import json
import os
from leapp.libraries.common import mounting, utils
from leapp.libraries.stdlib import api
from leapp.models import fields, RepositoryData, RepositoryFile
try:
import dnf
except ImportError:
api.current_logger().warning('repofileutils.py: failed to import dnf')
class InvalidRepoDefinition(Exception):
"""Raised when a repository definition is invalid."""
def __init__(self, msg, repofile, repoid):
message = 'Invalid repository definition: {repoid} in: {repofile}: {msg}'.format(
repoid=repoid, repofile=repofile, msg=msg)
super(InvalidRepoDefinition, self).__init__(message)
self.repofile = repofile
self.repoid = repoid
def _parse_repository(repoid, repo_data):
def asbool(x):
return x == '1'
prepared = {'repoid': repoid, 'additional_fields': {}}
for key in repo_data.keys():
if key in RepositoryData.fields:
if isinstance(RepositoryData.fields[key], fields.Boolean):
repo_data[key] = asbool(repo_data[key])
prepared[key] = repo_data[key]
else:
prepared['additional_fields'][key] = repo_data[key]
prepared['additional_fields'] = json.dumps(prepared['additional_fields'])
return RepositoryData(**prepared)
def parse_repofile(repofile):
"""
Parse the given repo file.
:param repofile: Path to the repo file
:type repofile: str
:rtype: RepositoryFile
:raises InvalidRepoDefinition: If the repository definition is invalid,
this can for example occur if 'name' field in repository is missing or it is invalid.
"""
data = []
with open(repofile, mode='r') as fp:
cp = utils.parse_config(fp, strict=False)
for repoid in cp.sections():
try:
data.append(_parse_repository(repoid, dict(cp.items(repoid))))
except fields.ModelViolationError as e:
raise InvalidRepoDefinition(e, repofile=repofile, repoid=repoid)
return RepositoryFile(file=repofile, data=data)
def get_repodirs():
"""
Return all directories yum scans for repository files, if they exist.
By default, the possible paths on RHEL should be:
['/etc/yum.repos.d', '/etc/yum/repos.d', '/etc/distro.repos.d']
ATTENTION: Requires the dnf module to be present.
TODO: Get repodirs inside given context.
"""
with dnf.base.Base() as base:
base.conf.read(priority=dnf.conf.PRIO_MAINCONFIG)
return list({os.path.realpath(d) for d in base.conf.reposdir if os.path.isdir(d)})
def get_parsed_repofiles(context=mounting.NotIsolatedActions(base_dir='/')):
"""
Scan all repositories on the system.
Repositories are scanned under repository directories (as reported by dnf)
of the given context. By default the context is the host system.
ATTENTION: Do not forget to ensure the redhat.repo file is regenerated
by RHSM when used.
:param context: An instance of a mounting.IsolatedActions class
:type context: mounting.IsolatedActions class
:rtype: List(RepositoryFile)
"""
repofiles = []
cmd = ['find', '-L'] + get_repodirs() + ['-maxdepth', '1', '-type', 'f', '-name', '*.repo']
repofiles_paths = context.call(cmd, split=True)['stdout']
for repofile_path in repofiles_paths:
repofile = parse_repofile(context.full_path(repofile_path))
# we want full path in cotext, not the real full path
repofile.file = repofile_path
repofiles.append(repofile)
return repofiles
def _invert_dict(data):
"""{a: [b]} -> {b: [a]}"""
inv_dict = {}
for key in data.keys():
for value in data[key]:
inv_dict[value] = inv_dict.get(value, []) + [key]
return inv_dict
def get_duplicate_repositories(repofiles):
"""
Return dict of duplicate repositories {repoid: [repofile_path]}
A repository is defined multiple times if it exists in multiple repofiles.
Redefinition inside one repository file is ignored (same in DNF).
:param repofiles:
:type repofiles: List(RepositoryFile)
:rtype: dict {repoid: repofilepath}
"""
rf_repos = {repofile.file: [repo.repoid for repo in repofile.data] for repofile in repofiles}
repos = _invert_dict(rf_repos)
return {repo: set(rfiles) for repo, rfiles in repos.items() if len(set(rfiles)) > 1}