mirrors/github.com_openSUSE_osc
1
0
Fork 0
mirror of https://github.com/openSUSE/osc.git synced 2024-11-14 16:26:13 +01:00
Code

Merge pull request #1476 from dmach/makeurl

Browse Source 
Refactor makeurl(), deprecate query taking string or list arguments, drop osc_urlencode()
This commit is contained in:
Daniel Mach 2024-02-09 14:28:06 +01:00 committed by GitHub
commit dbecb16404
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 234 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
osc/build.py
View File

@ -21,7 +21,7 @@ from . import conf
from . import connection
from . import core
from . import oscerr
from .core import get_buildinfo, meta_exists, quote_plus, get_buildconfig, dgst
from .core import get_buildinfo, meta_exists, get_buildconfig, dgst
from .core import get_binarylist, get_binary_file, run_external, return_external, raw_input
from .fetch import Fetcher, OscFileGrabber, verify_pacs
from .meter import create_text_meter
@ -1021,13 +1021,13 @@ def main(apiurl, store, opts, argv):
except HTTPError as e:
if e.code == 404:
# check what caused the 404
if meta_exists(metatype='prj', path_args=(quote_plus(prj), ),
if meta_exists(metatype='prj', path_args=(prj, ),
template_args=None, create_new=False, apiurl=apiurl):
pkg_meta_e = None
try:
# take care, not to run into double trouble.
pkg_meta_e = meta_exists(metatype='pkg', path_args=(quote_plus(prj),
quote_plus(pac)), template_args=None, create_new=False,
pkg_meta_e = meta_exists(metatype='pkg', path_args=(prj, pac),
template_args=None, create_new=False,
apiurl=apiurl)
except:
pass

38
osc/commandline.py
View File

@ -1950,7 +1950,7 @@ class Osc(cmdln.Cmdln):
edit=True,
force=opts.force,
remove_linking_repositories=opts.remove_linking_repositories,
path_args=quote_plus(project),
path_args=(project, ),
apiurl=apiurl,
msg=opts.message,
template_args=({
@ -1959,7 +1959,7 @@ class Osc(cmdln.Cmdln):
elif cmd == 'pkg':
edit_meta(metatype='pkg',
edit=True,
path_args=(quote_plus(project), quote_plus(package)),
path_args=(project, package),
apiurl=apiurl,
template_args=({
'name': package,
@ -1967,20 +1967,20 @@ class Osc(cmdln.Cmdln):
elif cmd == 'prjconf':
edit_meta(metatype='prjconf',
edit=True,
path_args=quote_plus(project),
path_args=(project, ),
apiurl=apiurl,
msg=opts.message,
template_args=None)
elif cmd == 'user':
edit_meta(metatype='user',
edit=True,
path_args=(quote_plus(user)),
path_args=(user, ),
apiurl=apiurl,
template_args=({'user': user}))
elif cmd == 'group':
edit_meta(metatype='group',
edit=True,
path_args=(quote_plus(group)),
path_args=(group, ),
apiurl=apiurl,
template_args=({'group': group}))
elif cmd == 'pattern':
@ -1993,7 +1993,7 @@ class Osc(cmdln.Cmdln):
edit_meta(
metatype='attribute',
edit=True,
path_args=(quote_plus(project), quote_plus(opts.attribute)),
path_args=(project, opts.attribute),
apiurl=apiurl,
# PUT is not supported
method="POST",
@ -2062,32 +2062,32 @@ class Osc(cmdln.Cmdln):
remove_linking_repositories=opts.remove_linking_repositories,
apiurl=apiurl,
msg=opts.message,
path_args=quote_plus(project))
path_args=(project, ))
elif cmd == 'pkg':
edit_meta(metatype='pkg',
data=f,
edit=opts.edit,
apiurl=apiurl,
path_args=(quote_plus(project), quote_plus(package)))
path_args=(project, package))
elif cmd == 'prjconf':
edit_meta(metatype='prjconf',
data=f,
edit=opts.edit,
apiurl=apiurl,
msg=opts.message,
path_args=quote_plus(project))
path_args=(project, ))
elif cmd == 'user':
edit_meta(metatype='user',
data=f,
edit=opts.edit,
apiurl=apiurl,
path_args=(quote_plus(user)))
path_args=(user, ))
elif cmd == 'group':
edit_meta(metatype='group',
data=f,
edit=opts.edit,
apiurl=apiurl,
path_args=(quote_plus(group)))
path_args=(group, ))
elif cmd == 'pattern':
edit_meta(metatype='pattern',
data=f,
@ -6026,7 +6026,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
raise e
@cmdln.alias('r')
@cmdln.option('-l', '--last-build', action='store_true',
@cmdln.option('-l', '--last-build', action='store_true', default=None,
help='show last build results (succeeded/failed/unknown)')
@cmdln.option('-r', '--repo', action='append', default=[],
help='Show results only for specified repo(s)')
@ -6277,7 +6277,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
query['last'] = 1
if opts.lastsucceeded:
query['lastsucceeded'] = 1
u = makeurl(self.get_api_url(), ['build', quote_plus(project), quote_plus(repository), quote_plus(arch), quote_plus(package), '_log'], query=query)
u = makeurl(self.get_api_url(), ['build', project, repository, arch, package, '_log'], query=query)
f = http_GET(u)
root = ET.parse(f).getroot()
offset = int(root.find('entry').get('size'))
@ -6290,7 +6290,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
elif opts.offset:
offset = int(opts.offset)
strip_time = opts.strip_time or conf.config['buildlog_strip_time']
print_buildlog(apiurl, quote_plus(project), quote_plus(package), quote_plus(repository), quote_plus(arch), offset, strip_time, opts.last, opts.lastsucceeded)
print_buildlog(apiurl, project, package, repository, arch, offset, strip_time, opts.last, opts.lastsucceeded)
def print_repos(self, repos_only=False, exc_class=oscerr.WrongArgs, exc_msg='Missing arguments', project=None):
wd = Path.cwd()
@ -6371,7 +6371,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
query['last'] = 1
if opts.lastsucceeded:
query['lastsucceeded'] = 1
u = makeurl(self.get_api_url(), ['build', quote_plus(project), quote_plus(repository), quote_plus(arch), quote_plus(package), '_log'], query=query)
u = makeurl(self.get_api_url(), ['build', project, repository, arch, package, '_log'], query=query)
f = http_GET(u)
root = ET.parse(f).getroot()
offset = int(root.find('entry').get('size'))
@ -6384,7 +6384,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
elif opts.offset:
offset = int(opts.offset)
strip_time = opts.strip_time or conf.config['buildlog_strip_time']
print_buildlog(apiurl, quote_plus(project), quote_plus(package), quote_plus(repository), quote_plus(arch), offset, strip_time, opts.last, opts.lastsucceeded)
print_buildlog(apiurl, project, package, repository, arch, offset, strip_time, opts.last, opts.lastsucceeded)
def _find_last_repo_arch(self, repo=None, fatal=True):
files = glob.glob(os.path.join(Path.cwd(), store, "_buildinfo-*"))
@ -8712,7 +8712,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
apiurl = osc_store.Store(project_dir).apiurl
user = conf.get_apiurl_usr(apiurl)
data = meta_exists(metatype='pkg',
path_args=(quote_plus(project), quote_plus(pac)),
path_args=(project, pac),
template_args=({
'name': pac,
'user': user}), apiurl=apiurl)
@ -8726,7 +8726,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
print('error - cannot get meta data', file=sys.stderr)
sys.exit(1)
edit_meta(metatype='pkg',
path_args=(quote_plus(project), quote_plus(pac)),
path_args=(project, pac),
data=data, apiurl=apiurl)
Package.init_package(apiurl, project, pac, os.path.join(project_dir, pac))
else:
@ -9257,7 +9257,7 @@ Please submit there instead, or use --nodevelproject to force direct submission.
o = open(destfile, 'wb')
if md5 != '':
query = {'rev': dir['srcmd5']}
u = makeurl(dir['apiurl'], ['source', dir['project'], dir['package'], pathname2url(name)], query=query)
u = makeurl(dir['apiurl'], ['source', dir['project'], dir['package'], name], query=query)
for buf in streamfile(u, http_GET, BUFSIZE):
o.write(buf)
o.close()

215
osc/core.py
View File

@ -30,14 +30,14 @@ import sys
import tempfile
import textwrap
import time
import warnings
from functools import cmp_to_key, total_ordering
from http.client import IncompleteRead
from io import StringIO
from pathlib import Path
from typing import Optional, Dict, Union, List, Iterable
from urllib.parse import urlsplit, urlunsplit, urlparse, quote_plus, urlencode, unquote
from urllib.parse import urlsplit, urlunsplit, urlparse, quote, urlencode, unquote
from urllib.error import HTTPError
from urllib.request import pathname2url
from xml.etree import ElementTree as ET
try:
@ -364,7 +364,7 @@ class Serviceinfo:
def getProjectGlobalServices(self, apiurl: str, project: str, package: str):
self.apiurl = apiurl
# get all project wide services in one file, we don't store it yet
u = makeurl(apiurl, ['source', project, package], query='cmd=getprojectservices')
u = makeurl(apiurl, ["source", project, package], query={"cmd": "getprojectservices"})
try:
f = http_POST(u)
root = ET.parse(f).getroot()
@ -1115,7 +1115,7 @@ class Project:
else:
user = conf.get_apiurl_usr(self.apiurl)
edit_meta(metatype='pkg',
path_args=(quote_plus(self.name), quote_plus(pac)),
path_args=(self.name, pac),
template_args=({
'name': pac,
'user': user}),
@ -1170,11 +1170,11 @@ class Project:
package = store_read_package(pac_path)
apiurl = store.apiurl
if not meta_exists(metatype='pkg',
path_args=(quote_plus(project), quote_plus(package)),
path_args=(project, package),
template_args=None, create_new=False, apiurl=apiurl):
user = conf.get_apiurl_usr(self.apiurl)
edit_meta(metatype='pkg',
path_args=(quote_plus(project), quote_plus(package)),
path_args=(project, package),
template_args=({'name': pac, 'user': user}), apiurl=apiurl)
p = Package(pac_path)
p.todo = files
@ -1541,7 +1541,7 @@ class Package:
def delete_remote_source_file(self, n):
"""delete a remote source file (e.g. from the server)"""
query = 'rev=upload'
u = makeurl(self.apiurl, ['source', self.prjname, self.name, pathname2url(n)], query=query)
u = makeurl(self.apiurl, ['source', self.prjname, self.name, n], query=query)
http_DELETE(u)
def put_source_file(self, n, tdir, copy_only=False):
@ -1551,7 +1551,7 @@ class Package:
# escaping '+' in the URL path (note: not in the URL query string) is
# only a workaround for ruby on rails, which swallows it otherwise
if not copy_only:
u = makeurl(self.apiurl, ['source', self.prjname, self.name, pathname2url(n)], query=query)
u = makeurl(self.apiurl, ['source', self.prjname, self.name, n], query=query)
http_PUT(u, file=tfilename)
if n in self.to_be_added:
self.to_be_added.remove(n)
@ -3606,39 +3606,71 @@ def pathjoin(a, *p):
return path
def osc_urlencode(data):
class UrlQueryArray(list):
"""
An urlencode wrapper that encodes dictionaries in OBS compatible way:
{"file": ["foo", "bar"]} -> &file[]=foo&file[]=bar
Passing values wrapped in this object causes ``makeurl()`` to encode the list
in Ruby on Rails compatible way (adding square brackets to the parameter names):
{"file": UrlQueryArray(["foo", "bar"])} -> &file[]=foo&file[]=bar
"""
data = copy.deepcopy(data)
if isinstance(data, dict):
for key, value in list(data.items()):
if isinstance(value, list):
del data[key]
data[f"{key}[]"] = value
return urlencode(data, doseq=True)
pass
def makeurl(baseurl: str, l, query=None):
"""Given a list of path compoments, construct a complete URL.
Optional parameters for a query string can be given as a list, as a
dictionary, or as an already assembled string.
In case of a dictionary, the parameters will be urlencoded by this
function. In case of a list not -- this is to be backwards compatible.
def makeurl(apiurl: str, path: List[str], query: Optional[dict] = None):
"""
query = query or []
_private.print_msg("makeurl:", baseurl, l, query, print_to="debug")
Construct an URL based on the given arguments.
if isinstance(query, list):
query = '&'.join(query)
elif isinstance(query, dict):
query = osc_urlencode(query)
:param apiurl: URL to the API server.
:param path: List of URL path components.
:param query: Optional dictionary with URL query data.
Values can be: ``str``, ``int``, ``bool``, ``[str]``, ``[int]``.
Items with value equal to ``None`` will be skipped.
"""
apiurl_scheme, apiurl_netloc, apiurl_path = urlsplit(apiurl)[0:3]
scheme, netloc, path = urlsplit(baseurl)[0:3]
return urlunsplit((scheme, netloc, '/'.join([path] + list(l)), query, ''))
path = apiurl_path.split("/") + [i.strip("/") for i in path]
path = [quote(i, safe="/:") for i in path]
path_str = "/".join(path)
# DEPRECATED
if isinstance(query, (list, tuple)):
warnings.warn(
"makeurl() query taking a list or a tuple is deprecated. Use dict instead.",
DeprecationWarning
)
query_str = "&".join(query)
return urlunsplit((apiurl_scheme, apiurl_netloc, path_str, query_str, ""))
# DEPRECATED
if isinstance(query, str):
warnings.warn(
"makeurl() query taking a string is deprecated. Use dict instead.",
DeprecationWarning
)
query_str = query
return urlunsplit((apiurl_scheme, apiurl_netloc, path_str, query_str, ""))
if query is None:
query = {}
query = copy.deepcopy(query)
for key in list(query):
value = query[key]
if value in (None, [], ()):
# remove items with value equal to None or [] or ()
del query[key]
elif isinstance(value, bool):
# convert boolean values to "0" or "1"
query[key] = str(int(value))
elif isinstance(value, UrlQueryArray):
# encode lists in Ruby on Rails compatible way:
# {"file": ["foo", "bar"]} -> &file[]=foo&file[]=bar
del query[key]
query[f"{key}[]"] = value
query_str = urlencode(query, doseq=True)
return urlunsplit((apiurl_scheme, apiurl_netloc, path_str, query_str, ""))
def check_store_version(dir):
@ -3818,11 +3850,9 @@ def show_attribute_meta(apiurl: str, prj: str, pac, subpac, attribute, with_defa
path.append('_attribute')
if attribute:
path.append(attribute)
query = []
if with_defaults:
query.append("with_default=1")
if with_project:
query.append("with_project=1")
query = {}
query["with_default"] = with_defaults
query["with_project"] = with_project
url = makeurl(apiurl, path, query)
try:
f = http_GET(url)
@ -4269,11 +4299,10 @@ def show_upstream_xsrcmd5(
def show_project_sourceinfo(apiurl: str, project: str, nofilename: bool, *packages):
query = ['view=info']
if packages:
query.extend([f'package={quote_plus(p)}' for p in packages])
if nofilename:
query.append('nofilename=1')
query = {}
query["view"] = "info"
query["package"] = packages
query["nofilename"] = nofilename
f = http_GET(makeurl(apiurl, ['source', project], query=query))
return f.read()
@ -4670,7 +4699,7 @@ def create_submit_request(
options_block,
_html_escape(message))
u = makeurl(apiurl, ['request'], query='cmd=create')
u = makeurl(apiurl, ["request"], query={"cmd": "create"})
r = None
try:
f = http_POST(u, data=xml)
@ -5149,7 +5178,7 @@ def get_group(apiurl: str, group: str):
def get_group_meta(apiurl: str, group: str):
u = makeurl(apiurl, ['group', quote_plus(group)])
u = makeurl(apiurl, ['group', group])
try:
f = http_GET(u)
return b''.join(f.readlines())
@ -5159,7 +5188,7 @@ def get_group_meta(apiurl: str, group: str):
def get_user_meta(apiurl: str, user: str):
u = makeurl(apiurl, ['person', quote_plus(user)])
u = makeurl(apiurl, ['person', user])
try:
f = http_GET(u)
return b''.join(f.readlines())
@ -5240,7 +5269,7 @@ def get_source_file(
query['rev'] = revision
u = makeurl(
apiurl,
["source", prj, package, pathname2url(filename.encode(locale.getpreferredencoding(), "replace"))],
["source", prj, package, filename],
query=query,
)
download(u, targetfilename, progress_obj, mtime)
@ -5426,7 +5455,7 @@ def server_diff(
query['view'] = 'xml'
query['unified'] = 0
if files:
query["file"] = files
query["file"] = UrlQueryArray(files)
u = makeurl(apiurl, ['source', new_project, new_package], query=query)
f = http_POST(u, retry_on_400=False)
@ -5671,7 +5700,7 @@ def checkout_package(
# before we create directories and stuff, check if the package actually
# exists
meta_data = b''.join(show_package_meta(apiurl, quote_plus(project), quote_plus(package)))
meta_data = b''.join(show_package_meta(apiurl, project, package))
root = ET.fromstring(meta_data)
scmsync_element = root.find("scmsync")
if scmsync_element is not None and scmsync_element.text is not None:
@ -5756,7 +5785,7 @@ def link_to_branch(apiurl: str, project: str, package: str):
"""
if '_link' in meta_get_filelist(apiurl, project, package):
u = makeurl(apiurl, ['source', project, package], 'cmd=linktobranch')
u = makeurl(apiurl, ["source", project, package], {"cmd": "linktobranch"})
http_POST(u)
else:
raise oscerr.OscIOError(None, f'no _link file inside project \'{project}\' package \'{package}\'')
@ -5791,7 +5820,7 @@ def link_pac(
apiurl = conf.config['apiurl']
try:
dst_meta = meta_exists(metatype='pkg',
path_args=(quote_plus(dst_project), quote_plus(dst_package)),
path_args=(dst_project, dst_package),
template_args=None,
create_new=False, apiurl=apiurl)
root = ET.fromstring(parse_meta_to_string(dst_meta))
@ -5921,7 +5950,7 @@ def aggregate_pac(
try:
dst_meta = meta_exists(metatype='pkg',
path_args=(quote_plus(dst_project), quote_plus(dst_package_meta)),
path_args=(dst_project, dst_package_meta),
template_args=None,
create_new=False, apiurl=apiurl)
root = ET.fromstring(parse_meta_to_string(dst_meta))
@ -6215,7 +6244,7 @@ def copy_pac(
src_meta = replace_pkg_meta(src_meta, dst_package, dst_project, keep_maintainers,
dst_userid, keep_develproject)
url = make_meta_url('pkg', (quote_plus(dst_project),) + (quote_plus(dst_package),), dst_apiurl)
url = make_meta_url('pkg', (dst_project, dst_package), dst_apiurl)
found = None
try:
found = http_GET(url).readlines()
@ -6264,7 +6293,7 @@ def copy_pac(
with tempfile.NamedTemporaryFile(prefix='osc-copypac') as f:
get_source_file(src_apiurl, src_project, src_package, filename,
targetfilename=f.name, revision=revision)
path = ['source', dst_project, dst_package, pathname2url(filename)]
path = ['source', dst_project, dst_package, filename]
u = makeurl(dst_apiurl, path, query={'rev': 'repository'})
http_PUT(u, file=f.name)
tfilelist = Package.commit_filelist(dst_apiurl, dst_project, dst_package,
@ -6495,29 +6524,21 @@ def show_results_meta(
repository: Optional[List[str]] = None,
arch: Optional[List[str]] = None,
oldstate: Optional[str] = None,
multibuild=False,
locallink=False,
multibuild: Optional[bool] = None,
locallink: Optional[bool] = None,
code: Optional[str] = None,
):
repository = repository or []
arch = arch or []
query = []
if package:
query.append(f'package={quote_plus(package)}')
if oldstate:
query.append(f'oldstate={quote_plus(oldstate)}')
if lastbuild:
query.append('lastbuild=1')
if multibuild:
query.append('multibuild=1')
if locallink:
query.append('locallink=1')
if code:
query.append(f'code={quote_plus(code)}')
for repo in repository:
query.append(f'repository={quote_plus(repo)}')
for a in arch:
query.append(f'arch={quote_plus(a)}')
query = {}
query["package"] = package
query["oldstate"] = oldstate
query["lastbuild"] = lastbuild
query["multibuild"] = multibuild
query["locallink"] = locallink
query["code"] = code
query["repository"] = repository
query["arch"] = arch
u = makeurl(apiurl, ['build', prj, '_result'], query=query)
f = http_GET(u)
return f.readlines()
@ -7017,15 +7038,13 @@ def print_buildlog(
def get_dependson(apiurl: str, project: str, repository: str, arch: str, packages=None, reverse=None):
query = []
if packages:
for i in packages:
query.append(f'package={quote_plus(i)}')
query = {}
query["package"] = packages
if reverse:
query.append('view=revpkgnames')
query["view"] = "revpkgnames"
else:
query.append('view=pkgnames')
query["view"] = "pkgnames"
u = makeurl(apiurl, ['build', project, repository, arch, '_builddepinfo'], query=query)
f = http_GET(u)
@ -7035,12 +7054,9 @@ def get_dependson(apiurl: str, project: str, repository: str, arch: str, package
def get_buildinfo(
apiurl: str, prj: str, package: str, repository: str, arch: str, specfile=None, addlist=None, debug=None
):
query = []
if addlist:
for i in addlist:
query.append(f'add={quote_plus(i)}')
if debug:
query.append('debug=1')
query = {}
query["add"] = addlist
query["debug"] = debug
u = makeurl(apiurl, ['build', prj, repository, arch, package, '_buildinfo'], query=query)
@ -7052,10 +7068,8 @@ def get_buildinfo(
def get_buildconfig(apiurl: str, prj: str, repository: str, path=None):
query = []
if path:
for prp in path:
query.append(f'path={quote_plus(prp)}')
query = {}
query["path"] = path
u = makeurl(apiurl, ['build', prj, repository, '_buildconfig'], query=query)
f = http_GET(u)
return f.read()
@ -7859,10 +7873,10 @@ def addMaintainer(apiurl: str, prj: str, pac: str, user: str):
def addPerson(apiurl: str, prj: str, pac: str, user: str, role="maintainer"):
""" add a new person to a package or project """
path = quote_plus(prj),
path = (prj, )
kind = 'prj'
if pac:
path = path + (quote_plus(pac),)
path = path + (pac ,)
kind = 'pkg'
data = meta_exists(metatype=kind,
path_args=path,
@ -7895,10 +7909,10 @@ def delMaintainer(apiurl: str, prj: str, pac: str, user: str):
def delPerson(apiurl: str, prj: str, pac: str, user: str, role="maintainer"):
""" delete a person from a package or project """
path = quote_plus(prj),
path = (prj, )
kind = 'prj'
if pac:
path = path + (quote_plus(pac), )
path = path + (pac, )
kind = 'pkg'
data = meta_exists(metatype=kind,
path_args=path,
@ -7924,10 +7938,10 @@ def delPerson(apiurl: str, prj: str, pac: str, user: str, role="maintainer"):
def setBugowner(apiurl: str, prj: str, pac: str, user=None, group=None):
""" delete all bugowners (user and group entries) and set one new one in a package or project """
path = quote_plus(prj),
path = (prj, )
kind = 'prj'
if pac:
path = path + (quote_plus(pac), )
path = path + (pac, )
kind = 'pkg'
data = meta_exists(metatype=kind,
path_args=path,
@ -7957,7 +7971,7 @@ def setBugowner(apiurl: str, prj: str, pac: str, user=None, group=None):
def setDevelProject(apiurl, prj, pac, dprj, dpkg=None):
""" set the <devel project="..."> element to package metadata"""
path = (quote_plus(prj),) + (quote_plus(pac),)
path = (prj, pac)
data = meta_exists(metatype='pkg',
path_args=path,
template_args=None,
@ -8821,7 +8835,7 @@ def which(name: str):
def get_comments(apiurl: str, kind, *args):
url = makeurl(apiurl, ('comments', kind) + args)
url = makeurl(apiurl, ["comments", kind] + list(args))
f = http_GET(url)
return ET.parse(f).getroot()
@ -8844,9 +8858,8 @@ def print_comments(apiurl: str, kind, *args):
def create_comment(apiurl: str, kind, comment, *args, **kwargs) -> Optional[str]:
query = {}
if kwargs.get('parent') is not None:
query = {'parent_id': kwargs['parent']}
u = makeurl(apiurl, ('comments', kind) + args, query=query)
query["parent_id"] = kwargs.get("parent", None)
u = makeurl(apiurl, ["comments", kind] + list(args), query=query)
f = http_POST(u, data=comment)
ret = ET.fromstring(f.read()).find('summary')
if ret is None:

9
osc/fetch.py
View File

@ -10,7 +10,6 @@ import shutil
import subprocess
import sys
import tempfile
from urllib.parse import quote_plus
from urllib.request import HTTPError
from . import checker as osc_checker
@ -51,10 +50,10 @@ class Fetcher:
def __download_cpio_archive(self, apiurl, project, repo, arch, package, **pkgs):
if not pkgs:
return
query = [f'binary={quote_plus(i)}' for i in pkgs]
query.append('view=cpio')
for module in self.modules:
query.append(f"module={module}")
query = {}
query["binary"] = pkgs
query["view"] = "cpio"
query["module"] = self.modules
try:
url = makeurl(apiurl, ['build', project, repo, arch, package], query=query)
sys.stdout.write("preparing download ...\r")

91
tests/test_core.py
View File

@ -1,5 +1,7 @@
import unittest
from osc.core import makeurl
from osc.core import UrlQueryArray
from osc.core import parseRevisionOption
from osc.oscerr import OscInvalidRevision
@ -47,5 +49,94 @@ class TestParseRevisionOption(unittest.TestCase):
self.assertRaises(OscInvalidRevision, parseRevisionOption, rev)
class TestMakeurl(unittest.TestCase):
def test_basic(self):
url = makeurl("https://example.com/api/v1", ["path", "to", "resource"], {"k1": "v1", "k2": ["v2", "v3"]})
self.assertEqual(url, "https://example.com/api/v1/path/to/resource?k1=v1&k2=v2&k2=v3")
def test_array(self):
url = makeurl("https://example.com/api/v1", ["path", "to", "resource"], {"k1": "v1", "k2": UrlQueryArray(["v2", "v3"])})
self.assertEqual(url, "https://example.com/api/v1/path/to/resource?k1=v1&k2%5B%5D=v2&k2%5B%5D=v3")
def test_query_none(self):
url = makeurl("https://example.com/api/v1", [], {"none": None})
self.assertEqual(url, "https://example.com/api/v1")
def test_query_empty_list(self):
url = makeurl("https://example.com/api/v1", [], {"empty_list": []})
self.assertEqual(url, "https://example.com/api/v1")
def test_query_int(self):
url = makeurl("https://example.com/api/v1", [], {"int": 1})
self.assertEqual(url, "https://example.com/api/v1?int=1")
def test_query_bool(self):
url = makeurl("https://example.com/api/v1", [], {"bool": True})
self.assertEqual(url, "https://example.com/api/v1?bool=1")
url = makeurl("https://example.com/api/v1", [], {"bool": False})
self.assertEqual(url, "https://example.com/api/v1?bool=0")
def test_quote_path(self):
mapping = (
# (character, expected encoded character)
(" ", "%20"),
("!", "%21"),
('"', "%22"),
("#", "%23"),
("$", "%24"),
("%", "%25"),
("&", "%26"),
("'", "%27"),
("(", "%28"),
(")", "%29"),
("*", "%2A"),
("+", "%2B"),
(",", "%2C"),
("/", "/"),
(":", ":"), # %3A
(";", "%3B"),
("=", "%3D"),
("?", "%3F"),
("@", "%40"),
("[", "%5B"),
("]", "%5D"),
)
for char, encoded_char in mapping:
url = makeurl("https://example.com/api/v1", [f"PREFIX_{char}_SUFFIX"])
self.assertEqual(url, f"https://example.com/api/v1/PREFIX_{encoded_char}_SUFFIX")
def test_quote_query(self):
mapping = (
# (character, expected encoded character)
(" ", "+"),
("!", "%21"),
('"', "%22"),
("#", "%23"),
("$", "%24"),
("%", "%25"),
("&", "%26"),
("'", "%27"),
("(", "%28"),
(")", "%29"),
("*", "%2A"),
("+", "%2B"),
(",", "%2C"),
("/", "%2F"),
(":", "%3A"),
(";", "%3B"),
("=", "%3D"),
("?", "%3F"),
("@", "%40"),
("[", "%5B"),
("]", "%5D"),
)
for char, encoded_char in mapping:
url = makeurl("https://example.com/api/v1", [], {char: char})
self.assertEqual(url, f"https://example.com/api/v1?{encoded_char}={encoded_char}")
if __name__ == "__main__":
unittest.main()

4
tests/test_update.py
View File

@ -193,8 +193,8 @@ class TestUpdate(OscTestCase):
@GET('http://localhost/source/osctest/services?rev=latest', file='testUpdateServiceFilesAddDelete_filesremote')
@GET('http://localhost/source/osctest/services/bigfile?rev=2', file='testUpdateServiceFilesAddDelete_bigfile')
@GET('http://localhost/source/osctest/services/_service%3Abar?rev=2', file='testUpdateServiceFilesAddDelete__service:bar')
@GET('http://localhost/source/osctest/services/_service%3Afoo?rev=2', file='testUpdateServiceFilesAddDelete__service:foo')
@GET('http://localhost/source/osctest/services/_service:bar?rev=2', file='testUpdateServiceFilesAddDelete__service:bar')
@GET('http://localhost/source/osctest/services/_service:foo?rev=2', file='testUpdateServiceFilesAddDelete__service:foo')
@GET('http://localhost/source/osctest/services/_meta', file='meta.xml')
def testUpdateAddDeleteServiceFiles(self):
"""update package with _service:* files"""