mirror of
https://github.com/openSUSE/osc.git
synced 2024-12-29 11:16:14 +01:00
Refactor makeurl(), deprecate query taking string or list arguments, drop osc_urlencode()
This commit is contained in:
parent
9070d03cb6
commit
3f14cef53a
93
osc/core.py
93
osc/core.py
@ -30,12 +30,13 @@ import sys
|
|||||||
import tempfile
|
import tempfile
|
||||||
import textwrap
|
import textwrap
|
||||||
import time
|
import time
|
||||||
|
import warnings
|
||||||
from functools import cmp_to_key, total_ordering
|
from functools import cmp_to_key, total_ordering
|
||||||
from http.client import IncompleteRead
|
from http.client import IncompleteRead
|
||||||
from io import StringIO
|
from io import StringIO
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Optional, Dict, Union, List, Iterable
|
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, quote_plus, urlencode, unquote
|
||||||
from urllib.error import HTTPError
|
from urllib.error import HTTPError
|
||||||
from urllib.request import pathname2url
|
from urllib.request import pathname2url
|
||||||
from xml.etree import ElementTree as ET
|
from xml.etree import ElementTree as ET
|
||||||
@ -3606,39 +3607,71 @@ def pathjoin(a, *p):
|
|||||||
return path
|
return path
|
||||||
|
|
||||||
|
|
||||||
def osc_urlencode(data):
|
class UrlQueryArray(list):
|
||||||
"""
|
"""
|
||||||
An urlencode wrapper that encodes dictionaries in OBS compatible way:
|
Passing values wrapped in this object causes ``makeurl()`` to encode the list
|
||||||
{"file": ["foo", "bar"]} -> &file[]=foo&file[]=bar
|
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)
|
pass
|
||||||
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)
|
|
||||||
|
|
||||||
|
|
||||||
def makeurl(baseurl: str, l, query=None):
|
def makeurl(apiurl: str, path: List[str], query: Optional[dict] = 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.
|
|
||||||
"""
|
"""
|
||||||
query = query or []
|
Construct an URL based on the given arguments.
|
||||||
_private.print_msg("makeurl:", baseurl, l, query, print_to="debug")
|
|
||||||
|
|
||||||
if isinstance(query, list):
|
:param apiurl: URL to the API server.
|
||||||
query = '&'.join(query)
|
:param path: List of URL path components.
|
||||||
elif isinstance(query, dict):
|
:param query: Optional dictionary with URL query data.
|
||||||
query = osc_urlencode(query)
|
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]
|
path = apiurl_path.split("/") + [i.strip("/") for i in path]
|
||||||
return urlunsplit((scheme, netloc, '/'.join([path] + list(l)), query, ''))
|
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):
|
def check_store_version(dir):
|
||||||
@ -5426,7 +5459,7 @@ def server_diff(
|
|||||||
query['view'] = 'xml'
|
query['view'] = 'xml'
|
||||||
query['unified'] = 0
|
query['unified'] = 0
|
||||||
if files:
|
if files:
|
||||||
query["file"] = files
|
query["file"] = UrlQueryArray(files)
|
||||||
|
|
||||||
u = makeurl(apiurl, ['source', new_project, new_package], query=query)
|
u = makeurl(apiurl, ['source', new_project, new_package], query=query)
|
||||||
f = http_POST(u, retry_on_400=False)
|
f = http_POST(u, retry_on_400=False)
|
||||||
@ -6495,8 +6528,8 @@ def show_results_meta(
|
|||||||
repository: Optional[List[str]] = None,
|
repository: Optional[List[str]] = None,
|
||||||
arch: Optional[List[str]] = None,
|
arch: Optional[List[str]] = None,
|
||||||
oldstate: Optional[str] = None,
|
oldstate: Optional[str] = None,
|
||||||
multibuild=False,
|
multibuild: Optional[bool] = None,
|
||||||
locallink=False,
|
locallink: Optional[bool] = None,
|
||||||
code: Optional[str] = None,
|
code: Optional[str] = None,
|
||||||
):
|
):
|
||||||
repository = repository or []
|
repository = repository or []
|
||||||
|
@ -1,5 +1,7 @@
|
|||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
|
from osc.core import makeurl
|
||||||
|
from osc.core import UrlQueryArray
|
||||||
from osc.core import parseRevisionOption
|
from osc.core import parseRevisionOption
|
||||||
from osc.oscerr import OscInvalidRevision
|
from osc.oscerr import OscInvalidRevision
|
||||||
|
|
||||||
@ -47,5 +49,94 @@ class TestParseRevisionOption(unittest.TestCase):
|
|||||||
self.assertRaises(OscInvalidRevision, parseRevisionOption, rev)
|
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__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
Loading…
Reference in New Issue
Block a user