openstack/nova
Code Issues Proposed changes

Rename minixpath_select to get_from_path

Browse Source
This commit is contained in:
Justin Santa Barbara
2011-02-23 14:07:08 -08:00
parent 89ade95d2e
commit 1183c9e11b
4 changed files with 188 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
nova/api/openstack/servers.py
View File

@@ -30,7 +30,6 @@ from nova.auth import manager as auth_manager
from nova.compute import instance_types
from nova.compute import power_state
import nova.api.openstack
import types
LOG = logging.getLogger('server')
LOG.setLevel(logging.DEBUG)
@@ -63,10 +62,10 @@ def _translate_detail_keys(inst):
inst_dict['status'] = power_mapping[inst_dict['status']]
inst_dict['addresses'] = dict(public=[], private=[])
private_ips = utils.minixpath_select(inst, 'fixed_ip/address')
private_ips = utils.get_from_path(inst, 'fixed_ip/address')
inst_dict['addresses']['private'] = private_ips
public_ips = utils.minixpath_select(inst, 'fixed_ip/floating_ips/address')
public_ips = utils.get_from_path(inst, 'fixed_ip/floating_ips/address')
inst_dict['addresses']['public'] = public_ips
inst_dict['metadata'] = {}

177
nova/tests/test_minixpath.py
View File

@@ -1,177 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 Justin Santa Barbara
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from nova import test
from nova import utils
from nova import exception
class MiniXPathTestCase(test.TestCase):
def test_tolerates_nones(self):
xp = utils.minixpath_select
input = []
self.assertEquals([], xp(input, "a"))
self.assertEquals([], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [None]
self.assertEquals([], xp(input, "a"))
self.assertEquals([], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': None}]
self.assertEquals([], xp(input, "a"))
self.assertEquals([], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': {'b': None}}]
self.assertEquals([{'b': None}], xp(input, "a"))
self.assertEquals([], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': {'b': {'c': None}}}]
self.assertEquals([{'b': {'c': None}}], xp(input, "a"))
self.assertEquals([{'c': None}], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': {'b': {'c': None}}}, {'a': None}]
self.assertEquals([{'b': {'c': None}}], xp(input, "a"))
self.assertEquals([{'c': None}], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': {'b': {'c': None}}}, {'a': {'b': None}}]
self.assertEquals([{'b': {'c': None}}, {'b': None}], xp(input, "a"))
self.assertEquals([{'c': None}], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
def test_does_select(self):
xp = utils.minixpath_select
input = [{'a': 'a_1'}]
self.assertEquals(['a_1'], xp(input, "a"))
self.assertEquals([], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': {'b': 'b_1'}}]
self.assertEquals([{'b': 'b_1'}], xp(input, "a"))
self.assertEquals(['b_1'], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': {'b': {'c': 'c_1'}}}]
self.assertEquals([{'b': {'c': 'c_1'}}], xp(input, "a"))
self.assertEquals([{'c': 'c_1'}], xp(input, "a/b"))
self.assertEquals(['c_1'], xp(input, "a/b/c"))
input = [{'a': {'b': {'c': 'c_1'}}}, {'a': None}]
self.assertEquals([{'b': {'c': 'c_1'}}],
xp(input, "a"))
self.assertEquals([{'c': 'c_1'}], xp(input, "a/b"))
self.assertEquals(['c_1'], xp(input, "a/b/c"))
input = [{'a': {'b': {'c': 'c_1'}}},
{'a': {'b': None}}]
self.assertEquals([{'b': {'c': 'c_1'}}, {'b': None}],
xp(input, "a"))
self.assertEquals([{'c': 'c_1'}], xp(input, "a/b"))
self.assertEquals(['c_1'], xp(input, "a/b/c"))
input = [{'a': {'b': {'c': 'c_1'}}},
{'a': {'b': {'c': 'c_2'}}}]
self.assertEquals([{'b': {'c': 'c_1'}}, {'b': {'c': 'c_2'}}],
xp(input, "a"))
self.assertEquals([{'c': 'c_1'}, {'c': 'c_2'}],
xp(input, "a/b"))
self.assertEquals(['c_1', 'c_2'], xp(input, "a/b/c"))
self.assertEquals([], xp(input, "a/b/c/d"))
self.assertEquals([], xp(input, "c/a/b/d"))
self.assertEquals([], xp(input, "i/r/t"))
def test_flattens_lists(self):
xp = utils.minixpath_select
input = [{'a': [1, 2, 3]}]
self.assertEquals([1, 2, 3], xp(input, "a"))
self.assertEquals([], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': {'b': [1, 2, 3]}}]
self.assertEquals([{'b': [1, 2, 3]}], xp(input, "a"))
self.assertEquals([1, 2, 3], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': {'b': [1, 2, 3]}}, {'a': {'b': [4, 5, 6]}}]
self.assertEquals([1, 2, 3, 4, 5, 6], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': [{'b': [1, 2, 3]}, {'b': [4, 5, 6]}]}]
self.assertEquals([1, 2, 3, 4, 5, 6], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = [{'a': [1, 2, {'b': 'b_1'}]}]
self.assertEquals([1, 2, {'b': 'b_1'}], xp(input, "a"))
self.assertEquals(['b_1'], xp(input, "a/b"))
def test_bad_xpath(self):
xp = utils.minixpath_select
self.assertRaises(exception.Error, xp, [], None)
self.assertRaises(exception.Error, xp, [], "")
self.assertRaises(exception.Error, xp, [], "/")
self.assertRaises(exception.Error, xp, [], "/a")
self.assertRaises(exception.Error, xp, [], "/a/")
self.assertRaises(exception.Error, xp, [], "//")
self.assertRaises(exception.Error, xp, [], "//a")
self.assertRaises(exception.Error, xp, [], "a//a")
self.assertRaises(exception.Error, xp, [], "a//a/")
self.assertRaises(exception.Error, xp, [], "a/a/")
def test_real_failure1(self):
# Real world failure case...
# We weren't coping when the input was a Dictionary instead of a List
# This led to test_accepts_dictionaries
xp = utils.minixpath_select
inst = {'fixed_ip': {'floating_ips': [{'address': '1.2.3.4'}],
'address': '192.168.0.3'},
'hostname': ''}
private_ips = xp(inst, 'fixed_ip/address')
public_ips = xp(inst, 'fixed_ip/floating_ips/address')
self.assertEquals(['192.168.0.3'], private_ips)
self.assertEquals(['1.2.3.4'], public_ips)
def test_accepts_dictionaries(self):
xp = utils.minixpath_select
input = {'a': [1, 2, 3]}
self.assertEquals([1, 2, 3], xp(input, "a"))
self.assertEquals([], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = {'a': {'b': [1, 2, 3]}}
self.assertEquals([{'b': [1, 2, 3]}], xp(input, "a"))
self.assertEquals([1, 2, 3], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = {'a': [{'b': [1, 2, 3]}, {'b': [4, 5, 6]}]}
self.assertEquals([1, 2, 3, 4, 5, 6], xp(input, "a/b"))
self.assertEquals([], xp(input, "a/b/c"))
input = {'a': [1, 2, {'b': 'b_1'}]}
self.assertEquals([1, 2, {'b': 'b_1'}], xp(input, "a"))
self.assertEquals(['b_1'], xp(input, "a/b"))

174
nova/tests/test_utils.py Normal file
View File

@@ -0,0 +1,174 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 Justin Santa Barbara
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from nova import test
from nova import utils
from nova import exception
class GetFromPathTestCase(test.TestCase):
def test_tolerates_nones(self):
f = utils.get_from_path
input = []
self.assertEquals([], f(input, "a"))
self.assertEquals([], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [None]
self.assertEquals([], f(input, "a"))
self.assertEquals([], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': None}]
self.assertEquals([], f(input, "a"))
self.assertEquals([], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': {'b': None}}]
self.assertEquals([{'b': None}], f(input, "a"))
self.assertEquals([], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': {'b': {'c': None}}}]
self.assertEquals([{'b': {'c': None}}], f(input, "a"))
self.assertEquals([{'c': None}], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': {'b': {'c': None}}}, {'a': None}]
self.assertEquals([{'b': {'c': None}}], f(input, "a"))
self.assertEquals([{'c': None}], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': {'b': {'c': None}}}, {'a': {'b': None}}]
self.assertEquals([{'b': {'c': None}}, {'b': None}], f(input, "a"))
self.assertEquals([{'c': None}], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
def test_does_select(self):
f = utils.get_from_path
input = [{'a': 'a_1'}]
self.assertEquals(['a_1'], f(input, "a"))
self.assertEquals([], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': {'b': 'b_1'}}]
self.assertEquals([{'b': 'b_1'}], f(input, "a"))
self.assertEquals(['b_1'], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': {'b': {'c': 'c_1'}}}]
self.assertEquals([{'b': {'c': 'c_1'}}], f(input, "a"))
self.assertEquals([{'c': 'c_1'}], f(input, "a/b"))
self.assertEquals(['c_1'], f(input, "a/b/c"))
input = [{'a': {'b': {'c': 'c_1'}}}, {'a': None}]
self.assertEquals([{'b': {'c': 'c_1'}}], f(input, "a"))
self.assertEquals([{'c': 'c_1'}], f(input, "a/b"))
self.assertEquals(['c_1'], f(input, "a/b/c"))
input = [{'a': {'b': {'c': 'c_1'}}},
{'a': {'b': None}}]
self.assertEquals([{'b': {'c': 'c_1'}}, {'b': None}], f(input, "a"))
self.assertEquals([{'c': 'c_1'}], f(input, "a/b"))
self.assertEquals(['c_1'], f(input, "a/b/c"))
input = [{'a': {'b': {'c': 'c_1'}}},
{'a': {'b': {'c': 'c_2'}}}]
self.assertEquals([{'b': {'c': 'c_1'}}, {'b': {'c': 'c_2'}}],
f(input, "a"))
self.assertEquals([{'c': 'c_1'}, {'c': 'c_2'}], f(input, "a/b"))
self.assertEquals(['c_1', 'c_2'], f(input, "a/b/c"))
self.assertEquals([], f(input, "a/b/c/d"))
self.assertEquals([], f(input, "c/a/b/d"))
self.assertEquals([], f(input, "i/r/t"))
def test_flattens_lists(self):
f = utils.get_from_path
input = [{'a': [1, 2, 3]}]
self.assertEquals([1, 2, 3], f(input, "a"))
self.assertEquals([], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': {'b': [1, 2, 3]}}]
self.assertEquals([{'b': [1, 2, 3]}], f(input, "a"))
self.assertEquals([1, 2, 3], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': {'b': [1, 2, 3]}}, {'a': {'b': [4, 5, 6]}}]
self.assertEquals([1, 2, 3, 4, 5, 6], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': [{'b': [1, 2, 3]}, {'b': [4, 5, 6]}]}]
self.assertEquals([1, 2, 3, 4, 5, 6], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = [{'a': [1, 2, {'b': 'b_1'}]}]
self.assertEquals([1, 2, {'b': 'b_1'}], f(input, "a"))
self.assertEquals(['b_1'], f(input, "a/b"))
def test_bad_xpath(self):
f = utils.get_from_path
self.assertRaises(exception.Error, f, [], None)
self.assertRaises(exception.Error, f, [], "")
self.assertRaises(exception.Error, f, [], "/")
self.assertRaises(exception.Error, f, [], "/a")
self.assertRaises(exception.Error, f, [], "/a/")
self.assertRaises(exception.Error, f, [], "//")
self.assertRaises(exception.Error, f, [], "//a")
self.assertRaises(exception.Error, f, [], "a//a")
self.assertRaises(exception.Error, f, [], "a//a/")
self.assertRaises(exception.Error, f, [], "a/a/")
def test_real_failure1(self):
# Real world failure case...
# We weren't coping when the input was a Dictionary instead of a List
# This led to test_accepts_dictionaries
f = utils.get_from_path
inst = {'fixed_ip': {'floating_ips': [{'address': '1.2.3.4'}],
'address': '192.168.0.3'},
'hostname': ''}
private_ips = f(inst, 'fixed_ip/address')
public_ips = f(inst, 'fixed_ip/floating_ips/address')
self.assertEquals(['192.168.0.3'], private_ips)
self.assertEquals(['1.2.3.4'], public_ips)
def test_accepts_dictionaries(self):
f = utils.get_from_path
input = {'a': [1, 2, 3]}
self.assertEquals([1, 2, 3], f(input, "a"))
self.assertEquals([], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = {'a': {'b': [1, 2, 3]}}
self.assertEquals([{'b': [1, 2, 3]}], f(input, "a"))
self.assertEquals([1, 2, 3], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = {'a': [{'b': [1, 2, 3]}, {'b': [4, 5, 6]}]}
self.assertEquals([1, 2, 3, 4, 5, 6], f(input, "a/b"))
self.assertEquals([], f(input, "a/b/c"))
input = {'a': [1, 2, {'b': 'b_1'}]}
self.assertEquals([1, 2, {'b': 'b_1'}], f(input, "a"))
self.assertEquals(['b_1'], f(input, "a/b"))

24
nova/utils.py
View File

@@ -32,10 +32,10 @@ import string
import struct
import sys
import time
import types
from xml.sax import saxutils
import re
import netaddr
import types
from eventlet import event
from eventlet import greenthread
@@ -503,18 +503,19 @@ def ensure_b64_encoding(val):
return base64.b64encode(val)
def minixpath_select(items, minixpath):
""" Takes an xpath-like expression e.g. prop1/prop2/prop3, and for each
item in items, looks up items[prop1][prop2][prop3]. Like XPath, if any of
the intermediate results are lists it will treat each list item
individually. A 'None' in items or any child expressions will be ignored,
this function will not throw because of None (anywhere) in items. The
returned list will contain no None values."""
def get_from_path(items, path):
""" Returns a list of items matching the specified path. Takes an
XPath-like expression e.g. prop1/prop2/prop3, and for each item in items,
looks up items[prop1][prop2][prop3]. Like XPath, if any of the
intermediate results are lists it will treat each list item individually.
A 'None' in items or any child expressions will be ignored, this function
will not throw because of None (anywhere) in items. The returned list
will contain no None values."""
if minixpath is None:
if path is None:
raise exception.Error("Invalid mini_xpath")
(first_token, sep, remainder) = minixpath.partition("/")
(first_token, sep, remainder) = path.partition("/")
if first_token == "":
raise exception.Error("Invalid mini_xpath")
@@ -537,7 +538,6 @@ def minixpath_select(items, minixpath):
child = get_method(first_token)
if child is None:
continue
#print "%s => %s" % (first_token, child)
if isinstance(child, types.ListType):
# Flatten intermediate lists
for x in child:
@@ -549,4 +549,4 @@ def minixpath_select(items, minixpath):
# No more tokens
return results
else:
return minixpath_select(results, remainder)
return get_from_path(results, remainder)