starlingx/root
Code Issues Proposed changes

stx repomgr: add list_pkgs sub-command

Browse Source 
Extend the repo manager with a new subcommand
to list all the packages in an aptly repository.

Testing
repo_manage.py list_pkgs -r deb-local-build
stx repomgr list_pkgs -r deb-local-build

Story: 2008862
Task: 44695
Signed-off-by: Scott Little <scott.little@windriver.com>
Change-Id: I30cf5f540dcc20306f8d0fce32e1b715df0ff4c2
This commit is contained in:
Scott Little 2022-03-04 13:47:41 -05:00
parent a60422a733
commit 8550b600b0
2 changed files with 68 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
build-tools/stx/aptly_deb_usage.py
View File

@ -205,7 +205,7 @@ class Deb_aptly():
if self.aptly.tasks.show(task.id).state != 'SUCCEEDED':
if backup_name:
self.aptly.tasks.wait_for_task_by_id(self.aptly.snapshots.update(backup_name, name).id)
self.logger.warning('Snapshot for %s creation failed: %s. ' % (name, self.aptly.tasks.show(task.id).state))
self.logger.warning('merge_snapshot: Snapshot for %s creation failed: %s. ' % (name, self.aptly.tasks.show(task.id).state))
return False
# Remove the backup snapshot if it is created above
if backup_name:
@ -250,7 +250,7 @@ class Deb_aptly():
if self.aptly.tasks.show(task.id).state != 'SUCCEEDED':
if backup_name:
self.aptly.tasks.wait_for_task_by_id(self.aptly.snapshots.update(backup_name, name).id)
self.logger.warning('Snapshot for %s creation failed: %s. ' % (name, self.aptly.tasks.show(task.id).state))
self.logger.warning('create_snapshot: Snapshot for %s creation failed: %s. ' % (name, self.aptly.tasks.show(task.id).state))
return False
if backup_name:
task = self.aptly.snapshots.delete(snapshotname=backup_name, force=True)
@ -456,15 +456,15 @@ class Deb_aptly():
return False
repo_list = self.aptly.repos.list()
repo_find = False
repo_found = False
for repo in repo_list:
if repo_name == repo.name:
self.logger.debug('%s find, can be used', repo_name)
repo_find = True
self.logger.debug('repo %s was found and can be used', repo_name)
repo_found = True
break
if not repo_find:
self.logger.warning('%s not exist, please create it firstly.', repo_name)
if not repo_found:
self.logger.warning('repo %s does not exist, please create it first.', repo_name)
return False
# For files with ":" in its filename, tools like 'apt' may replace it
@ -507,6 +507,23 @@ class Deb_aptly():
if self.aptly.tasks.show(task.id).state != 'SUCCEEDED':
self.logger.warning('Delete package failed %s : %s' % (pkg_name, self.aptly.tasks.show(task.id).state))
def pkg_list(self, repo_list):
'''list packages available from any of the listed repos, local or remote.'''
pkg_list=[]
for repo_name in repo_list:
if repo_name.startswith(PREFIX_LOCAL):
query = 'Name'
pkgs_raw = self.aptly.repos.search_packages(repo_name, query=query)
elif repo_name.startswith(PREFIX_REMOTE):
pkgs_raw = self.aptly.mirrors.packages(repo_name)
for pkg in pkgs_raw:
pkg_name = pkg.key.split()[1]
pkg_ver = pkg.key.split()[2]
pkg_arch = pkg.key.split()[0][1:]
pkg_list.append("%s_%s_%s.deb" % (pkg_name, pkg_ver, pkg_arch))
return pkg_list
# Search a package in a set of repos, return True if find, or False
# repolist: a list of repo names, including local repo and mirror
# pkg_name: package name
@ -554,7 +571,7 @@ class Deb_aptly():
ret = self.__publish_snap(name)
return ret
# deploy a loacl repository
# deploy a local repository
# Input: the name of the local repository
# Output: None or DebAptDistributionResponse
def deploy_local(self, name):
@ -571,7 +588,7 @@ class Deb_aptly():
repo_find = True
break
if not repo_find:
self.logger.warning('local repo %s not find.', name)
self.logger.warning('local repo %s not found.', name)
return None
if self.__create_snapshot(name, True):

45
build-tools/stx/repo_manage.py
View File

@ -420,9 +420,37 @@ class RepoMgr():
'''Clear all meta files. Construct a clean environment'''
self.repo.clean_all()
# list a repository
# repo_name: the name of the repo been listed.
# Output: True is all works in order
def list_pkgs(self, repo_name, quiet=False):
'''List a specified repository.'''
local_list = self.repo.list_local(quiet=True)
remote_list = self.repo.list_remotes(quiet=True)
pkg_list = []
for repo in local_list:
if repo == repo_name:
self.logger.info('List a local repo')
pkgs = self.repo.pkg_list([repo])
pkg_list.extend(pkgs)
if not quiet:
self.logger.info("Local repo %s:" % repo_name)
for pkg in pkgs:
self.logger.info(" %s:" % pkg)
for repo in remote_list:
if repo == repo_name:
self.logger.info('List a remote mirror')
pkgs = self.repo.pkg_list([repo])
pkg_list.extend(pkgs)
if not quiet:
self.logger.info("Remote repo %s:" % repo_name)
for pkg in pkgs:
self.logger.info(" %s:" % pkg)
return pkg_list
# delete a repository
# repo_name: the name of the repo been deleted.
# Output: Ture is all works in order
# Output: True is all works in order
def remove_repo(self, repo_name):
'''Remove a specified repository.'''
local_list = self.repo.list_local(quiet=True)
@ -437,7 +465,7 @@ class RepoMgr():
self.logger.info('Remove a remote mirror')
self.repo.remove_remote(repo)
return True
self.logger.warn('Remove repo failed, not find spesified repo')
self.logger.warn("Remove repo failed: repo '%'s not found" % repo_name)
return False
# Before uploading a source package into a local repo, scan all repos,
@ -669,6 +697,11 @@ def _handleList(_args):
repomgr.list()
def _handleListPkgs(args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', applogger)
repomgr.list_pkgs(args.repository)
def _handleClean(_args):
repomgr = RepoMgr('aptly', REPOMGR_URL, '/tmp', applogger)
repomgr.clean()
@ -743,7 +776,7 @@ def subcmd_merge(subparsers):
def main():
# command line arguments
parser = argparse.ArgumentParser(add_help=False,
parser = argparse.ArgumentParser(add_help=True,
description='Repository management Tool',
epilog='''Tips: Use %(prog)s --help to get help for all of '
'parameters\n\n''')
@ -803,6 +836,12 @@ def main():
'remote mirror.\n\n')
list_parser.set_defaults(handle=_handleList)
list_pkgs_parser = subparsers.add_parser('list_pkgs',
help='List contents of a specific repo.\n\n')
list_pkgs_parser.add_argument('--repository', '-r',
help='Name of the repo to be listed')
list_pkgs_parser.set_defaults(handle=_handleListPkgs)
args = parser.parse_args()
if hasattr(args, 'handle'):