mirrors/github.com_openSUSE_osc
1
0
Fork 0
mirror of https://github.com/openSUSE/osc.git synced 2024-12-25 17:36:13 +01:00
Code

Merge pull request #1205 from dmach/fix-buildhistory-cli-and-output

Browse Source 
Fix buildhistory cli and output
This commit is contained in:
Daniel Mach 2022-12-19 09:36:06 +01:00 committed by GitHub
commit f8417181a6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 275 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
osc/_private/__init__.py
View File

@ -3,6 +3,7 @@
#
# The cherry-picked imports will be the supported API.
from .api_build import BuildHistory
from .api_source import add_channels
from .api_source import add_containers
from .api_source import enable_channels

12
osc/_private/api.py
View File

@ -4,10 +4,6 @@ and work with related XML data.
"""
from .. import connection as osc_connection
from .. import core as osc_core
def get(apiurl, path, query=None):
"""
Send a GET request to OBS.
@ -21,6 +17,9 @@ def get(apiurl, path, query=None):
:returns: Parsed XML root.
:rtype: xml.etree.ElementTree.Element
"""
from .. import connection as osc_connection
from .. import core as osc_core
assert apiurl
assert path
@ -46,6 +45,9 @@ def post(apiurl, path, query=None):
:returns: Parsed XML root.
:rtype: xml.etree.ElementTree.Element
"""
from .. import connection as osc_connection
from .. import core as osc_core
assert apiurl
assert path
@ -110,6 +112,8 @@ def write_xml_node_to_file(node, path, indent=True):
:param indent: Whether to indent (pretty-print) the written XML.
:type indent: bool
"""
from .. import core as osc_core
if indent:
osc_core.xmlindent(node)
osc_core.ET.ElementTree(node).write(path)

79
osc/_private/api_build.py Normal file
View File

@ -0,0 +1,79 @@
import csv
import io
import time
from . import api
class BuildHistory:
def __init__(
self,
apiurl: str,
project: str,
package: str,
repository: str,
arch: str,
limit: int = 0,
):
self.apiurl = apiurl
self.project = project
self.package = package
self.repository = repository
self.arch = arch
self._limit = limit
self.entries = self._get_entries()
def _get_entries(self):
url_path = [
"build",
self.project,
self.repository,
self.arch,
self.package,
"_history",
]
url_query = {}
if self._limit and self._limit > 0:
query["limit"] = self._limit
root = api.get(self.apiurl, url_path, url_query)
result = []
nodes = api.find_nodes(root, "buildhistory", "entry")
for node in nodes:
item = {
"rev": node.get("rev"),
"srcmd5": node.get("srcmd5"),
"ver_rel": node.get("versrel"),
"build_count": int(node.get("bcnt")),
"time": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(int(node.get("time")))),
"duration": int(node.get("duration")),
}
result.append(item)
return result
def to_csv(self):
out = io.StringIO()
header = ["time", "srcmd5", "rev", "ver_rel", "build_count", "duration"]
writer = csv.DictWriter(out, fieldnames=header, quoting=csv.QUOTE_ALL)
writer.writeheader()
for i in self.entries:
writer.writerow(i)
return out.getvalue()
def to_text_table(self):
from ..core import build_table
header = ("TIME", "SRCMD5", "VER-REL.BUILD#", "REV", "DURATION")
data = []
for i in self.entries:
item = (
i["time"],
i["srcmd5"],
f"{i['ver_rel']}.{i['build_count']}",
i["rev"],
i["duration"],
)
data.extend(item)
return "\n".join(build_table(len(header), data, header))

6
osc/_private/package.py
View File

@ -1,7 +1,5 @@
import functools
from .. import core as osc_core
from .. import store as osc_store
from . import api
@ -40,6 +38,8 @@ class PackageBase:
raise NotImplementedError
def _load_from_directory_node(self, directory_node):
from .. import core as osc_core
# attributes
self.rev = directory_node.get("rev")
self.vrev = directory_node.get("vrev")
@ -78,6 +78,8 @@ class ApiPackage(PackageBase):
class LocalPackage(PackageBase):
def __init__(self, path):
from .. import store as osc_store
self.dir = path
self.store = osc_store.Store(self.dir)
super().__init__(self.store.apiurl, self.store.project, self.store.package)

3
osc/_private/request.py
View File

@ -1,4 +1,3 @@
from .. import core as osc_core
from . import package as osc_package
@ -6,6 +5,8 @@ def forward_request(apiurl, request, interactive=True):
"""
Forward the specified `request` to the projects the packages were branched from.
"""
from .. import core as osc_core
for action in request.get_actions("submit"):
package = osc_package.ApiPackage(apiurl, action.tgt_project, action.tgt_package)

112
osc/commandline.py
View File

@ -129,6 +129,88 @@ def pop_project_package_from_args(args, default_project=None, default_package=No
return project, package
def pop_repository_arch_from_args(args):
"""
Get repository and arch from given `args`.
:param args: List of command-line arguments.
WARNING: `args` gets modified in this function call!
:type args: list(str)
:returns: Repository name and arch name.
:rtype: tuple(str)
"""
assert isinstance(args, list)
try:
repository = args.pop(0)
except IndexError:
raise oscerr.OscValueError("Please specify a repository")
if not isinstance(repository, str):
raise TypeError(f"Repository should be 'str', found: {type(repository).__name__}")
arch = None
if "/" in repository:
# repository/arch
if repository.count("/") != 1:
raise oscerr.OscValueError(f"Argument doesn't match the '<repository>/<arch>' pattern: {repository}")
repository, arch = repository.split("/")
if arch is None:
try:
arch = args.pop(0)
except IndexError:
raise oscerr.OscValueError("Please specify an arch")
if not isinstance(arch, str):
raise TypeError(f"Arch should be 'str', found: {type(arch).__name__}")
return repository, arch
def pop_project_package_repository_arch_from_args(args):
"""
Get project, package, repository and arch from given `args`.
:param args: List of command-line arguments.
WARNING: `args` gets modified in this function call!
:type args: list(str)
:returns: Project name, package name, repository name and arch name.
:rtype: tuple(str)
"""
args_backup = args.copy()
try_working_copy = True
try:
# try this sequence first: project package repository arch
project, package = pop_project_package_from_args(args, package_is_optional=False)
if args:
# we got more than 2 arguments -> we shouldn't try to retrieve project and package from a working copy
try_working_copy = False
repository, arch = pop_repository_arch_from_args(args)
except oscerr.OscValueError as ex:
if not try_working_copy:
raise ex from None
# then read project and package from working copy and try repository arch
args[:] = args_backup.copy()
project, package = pop_project_package_from_args(
[], default_project=".", default_package=".", package_is_optional=False
)
repository, arch = pop_repository_arch_from_args(args)
return project, package, repository, arch
def ensure_no_remaining_args(args):
if not args:
return
args_str = " ".join(args)
raise oscerr.WrongArgs(f"Unexpected args: {args_str}")
class Osc(cmdln.Cmdln):
"""
openSUSE commander is a command-line interface to the Open Build Service.
@ -6732,36 +6814,20 @@ Please submit there instead, or use --nodevelproject to force direct submission.
osc buildhist REPOSITORY ARCHITECTURE
osc buildhist PROJECT PACKAGE[:FLAVOR] REPOSITORY ARCHITECTURE
"""
args = slash_split(args)
if len(args) < 2 and is_package_dir('.'):
self.print_repos()
apiurl = self.get_api_url()
if len(args) == 4:
project = self._process_project_name(args[0])
package = args[1]
repository = args[2]
arch = args[3]
elif len(args) == 2:
wd = Path.cwd()
package = store_read_package(wd)
project = store_read_project(wd)
repository = args[0]
arch = args[1]
else:
raise oscerr.WrongArgs('Wrong number of arguments')
args = list(args)
project, package, repository, arch = pop_project_package_repository_arch_from_args(args)
ensure_no_remaining_args(args)
if opts.multibuild_package:
package = package + ":" + opts.multibuild_package
format = 'text'
history = _private.BuildHistory(apiurl, project, package, repository, arch, limit=opts.limit)
if opts.csv:
format = 'csv'
print('\n'.join(get_buildhistory(apiurl, project, package, repository, arch, format, opts.limit)))
print(history.to_csv(), end="")
else:
print(history.to_text_table())
@cmdln.option('', '--csv', action='store_true',
help='generate output in CSV (separated by |)')

35
osc/core.py
View File

@ -6810,38 +6810,6 @@ def get_source_rev(apiurl: str, project: str, package: str, revision=None):
return e
def get_buildhistory(apiurl: str, prj: str, package: str, repository: str, arch: str, format="text", limit=None):
query = {}
if limit is not None and int(limit) > 0:
query['limit'] = int(limit)
u = makeurl(apiurl, ['build', prj, repository, arch, package, '_history'], query)
f = http_GET(u)
root = ET.parse(f).getroot()
r = []
for node in root.findall('entry'):
rev = node.get('rev')
srcmd5 = node.get('srcmd5')
versrel = node.get('versrel')
bcnt = int(node.get('bcnt'))
duration = node.get('duration')
t = time.gmtime(int(node.get('time')))
t = time.strftime('%Y-%m-%d %H:%M:%S', t)
if duration is None:
duration = ""
if format == 'csv':
r.append('%s|%s|%s|%s.%d|%s' % (t, srcmd5, rev, versrel, bcnt, duration))
else:
bversrel = '%s.%d' % (versrel, bcnt)
r.append('%s %s %s %s %s' % (t, srcmd5, bversrel.ljust(16)[:16], rev, duration.rjust(10)))
if format == 'text':
r.insert(0, 'time srcmd5 vers-rel.bcnt rev duration')
return r
def print_jobhistory(apiurl: str, prj: str, current_package: str, repository: str, arch: str, format="text", limit=20):
query = {}
if current_package:
@ -7266,6 +7234,9 @@ def build_table(col_num, data=None, headline=None, width=1, csv=False):
longest_col.append(0)
if headline and not csv:
data[0:0] = headline
data = [str(i) for i in data]
# find longest entry in each column
i = 0
for itm in data:

89
tests/test_commandline.py
View File

@ -4,6 +4,8 @@ import tempfile
import unittest
from osc.commandline import pop_project_package_from_args
from osc.commandline import pop_project_package_repository_arch_from_args
from osc.commandline import pop_repository_arch_from_args
from osc.oscerr import NoWorkingCopy, OscValueError
from osc.store import Store
@ -140,5 +142,92 @@ class TestPopProjectPackageFromArgs(unittest.TestCase):
self.assertEqual(args, [])
class TestPopRepositoryArchFromArgs(unittest.TestCase):
def test_individial_args(self):
args = ["repo", "arch", "another-arg"]
repo, arch = pop_repository_arch_from_args(args)
self.assertEqual(repo, "repo")
self.assertEqual(arch, "arch")
self.assertEqual(args, ["another-arg"])
def test_slash_separator(self):
args = ["repo/arch", "another-arg"]
repo, arch = pop_repository_arch_from_args(args)
self.assertEqual(repo, "repo")
self.assertEqual(arch, "arch")
self.assertEqual(args, ["another-arg"])
def test_missing_repository(self):
args = []
self.assertRaises(OscValueError, pop_repository_arch_from_args, args)
def test_missing_arch(self):
args = ["repo"]
self.assertRaises(OscValueError, pop_repository_arch_from_args, args)
class TestPopProjectPackageRepositoryArchFromArgs(unittest.TestCase):
def _write_store(self, project=None, package=None):
store = Store(self.tmpdir, check=False)
if project:
store.project = project
store.is_project = True
if package:
store.package = package
store.is_project = False
store.is_package = True
def setUp(self):
self.tmpdir = tempfile.mkdtemp(prefix="osc_test")
os.chdir(self.tmpdir)
def tearDown(self):
try:
shutil.rmtree(self.tmpdir)
except OSError:
pass
def test_individual_args(self):
args = ["project", "package", "repo", "arch", "another-arg"]
project, package, repo, arch = pop_project_package_repository_arch_from_args(args)
self.assertEqual(project, "project")
self.assertEqual(package, "package")
self.assertEqual(repo, "repo")
self.assertEqual(arch, "arch")
self.assertEqual(args, ["another-arg"])
def test_slash_separator(self):
args = ["project/package", "repo/arch", "another-arg"]
project, package, repo, arch = pop_project_package_repository_arch_from_args(args)
self.assertEqual(project, "project")
self.assertEqual(package, "package")
self.assertEqual(repo, "repo")
self.assertEqual(arch, "arch")
self.assertEqual(args, ["another-arg"])
def test_missing_arch(self):
args = ["project", "package", "repo"]
self.assertRaises(OscValueError, pop_project_package_repository_arch_from_args, args)
def test_no_working_copy(self):
args = ["repo", "arch"]
self.assertRaises(NoWorkingCopy, pop_project_package_repository_arch_from_args, args)
def test_working_copy(self):
self._write_store("store_project", "store_package")
args = ["repo", "arch"]
project, package, repo, arch = pop_project_package_repository_arch_from_args(args)
self.assertEqual(project, "store_project")
self.assertEqual(package, "store_package")
self.assertEqual(repo, "repo")
self.assertEqual(arch, "arch")
def test_working_copy_extra_arg(self):
self._write_store("store_project", "store_package")
args = ["repo", "arch", "another-arg"]
# example of invalid usage, working copy is not used when there's 3+ args; [project, package, ...] are expected
self.assertRaises(OscValueError, pop_project_package_repository_arch_from_args, args)
if __name__ == "__main__":
unittest.main()