diff --git a/oslo/utils/netutils.py b/oslo/utils/netutils.py index e20106d6..0bfb8346 100644 --- a/oslo/utils/netutils.py +++ b/oslo/utils/netutils.py @@ -92,6 +92,37 @@ class _ModifiedSplitResult(parse.SplitResult): host, port = parse_host_port(netloc) return port + def params(self, collapse=True): + """Extracts the query parameters from the split urls components. + + This method will provide back as a dictionary the query parameter + names and values that were provided in the url. + + :param collapse: Boolean, turn on or off collapsing of query values + with the same name. Since a url can contain the same query parameter + name with different values it may or may not be useful for users to + care that this has happened. This parameter when True uses the + last value that was given for a given name, while if False it will + retain all values provided by associating the query parameter name with + a list of values instead of a single (non-list) value. + """ + if self.query: + if collapse: + return dict(parse.parse_qsl(self.query)) + else: + params = {} + for (key, value) in parse.parse_qsl(self.query): + if key in params: + if isinstance(params[key], list): + params[key].append(value) + else: + params[key] = [params[key], value] + else: + params[key] = value + return params + else: + return {} + def urlsplit(url, scheme='', allow_fragments=True): """Parse a URL using urlparse.urlsplit(), splitting query and fragments. diff --git a/tests/test_netutils.py b/tests/test_netutils.py index 97966282..e1886c4c 100644 --- a/tests/test_netutils.py +++ b/tests/test_netutils.py @@ -103,6 +103,25 @@ class NetworkUtilsTest(test_base.BaseTestCase): self.assertEqual(result.query, 'ab') self.assertEqual(result.fragment, '12') + def test_urlsplit_params(self): + test_url = "http://localhost/?a=b&c=d" + result = netutils.urlsplit(test_url) + self.assertEqual({'a': 'b', 'c': 'd'}, result.params()) + self.assertEqual({'a': 'b', 'c': 'd'}, result.params(collapse=False)) + + test_url = "http://localhost/?a=b&a=c&a=d" + result = netutils.urlsplit(test_url) + self.assertEqual({'a': 'd'}, result.params()) + self.assertEqual({'a': ['b', 'c', 'd']}, result.params(collapse=False)) + + test_url = "http://localhost" + result = netutils.urlsplit(test_url) + self.assertEqual({}, result.params()) + + test_url = "http://localhost?" + result = netutils.urlsplit(test_url) + self.assertEqual({}, result.params()) + def test_set_tcp_keepalive(self): mock_sock = mock.Mock() netutils.set_tcp_keepalive(mock_sock, True, 100, 10, 5)