Improve image.v2.tag

While looking into create_by_id usage, I came across image.v2.tag as
something that needed to be adjusted. It was a bit cumbersome to use and
is doesn't benefit from a lot of the usual resource things since it
it just needs to join a string to the base_path and doesn't send or
receive anything else.

    t = tag.Tag({"image": image_resource_or_id})
    t.create(conn.session, "new tag")
    t.delete(conn.session, "old tag")

Building a proxy will be trivial for this. Viewing tags isn't offered by
this resource, but you can see them via image.Image.tags

Change-Id: I2bd549c55a8c21ac292752509122896a6fd51fe8
This commit is contained in:
Brian Curtin
2015-03-11 21:13:54 -05:00
parent 6a7e7cce4d
commit cba0caab10
2 changed files with 62 additions and 29 deletions

View File

@@ -11,25 +11,45 @@
# under the License.
from openstack.image import image_service
from openstack.image.v2 import image
from openstack import resource
from openstack import utils
class Tag(resource.Resource):
id_attribute = 'tag'
resources_key = 'tags'
base_path = '/images/%(image_id)s/tags'
id_attribute = "image"
base_path = "/images/%(image)s/tags"
service = image_service.ImageService()
# capabilities
allow_create = True
allow_delete = True
# Properties
image_id = resource.prop('image_id')
#: The image to manipulate
image = resource.prop("image", type=image.Image)
def create(self, session):
"""Create a remote resource from this instance."""
# Service expects a naked PUT. Omit properties.
self.create_by_id(session, None, self.id, path_args=self)
self._reset_dirty()
return self
def create(self, session, tag):
"""Set a tag on the image
:param session: The session to use for making this request.
:type session: :class:`~openstack.session.Session`
:param string tag: A tag to set on the image
:return: ``None``
"""
url = utils.urljoin(self.base_path %
{"image": self.image.id}, tag)
session.put(url, service=self.service, accept=None)
def delete(self, session, tag):
"""Delete a tag on the image
:param session: The session to use for making this request.
:type session: :class:`~openstack.session.Session`
:param string tag: The tag to delete on the image
:return: ``None``
"""
url = utils.urljoin(self.base_path %
{"image": self.image.id}, tag)
session.delete(url, service=self.service, accept=None)

View File

@@ -13,23 +13,27 @@
import mock
import testtools
from openstack.image.v2 import image
from openstack.image.v2 import tag
IDENTIFIER = 'IDENTIFIER'
EXAMPLE = {
'image_id': 'IMAGE_ID',
'tag': IDENTIFIER,
}
class TestTag(testtools.TestCase):
def setUp(self):
super(TestTag, self).setUp()
self.session = mock.Mock()
self.session.put = mock.Mock()
self.session.delete = mock.Mock()
self.img = image.Image({"id": "123"})
def test_basic(self):
sot = tag.Tag()
self.assertIsNone(sot.resource_key)
self.assertEqual('tags', sot.resources_key)
self.assertEqual('/images/%(image_id)s/tags', sot.base_path)
self.assertEqual('/images/%(image)s/tags', sot.base_path)
self.assertEqual('image', sot.service.service_type)
self.assertEqual('tag', sot.id_attribute)
self.assertEqual('image', sot.id_attribute)
self.assertTrue(sot.allow_create)
self.assertFalse(sot.allow_retrieve)
self.assertFalse(sot.allow_update)
@@ -37,14 +41,23 @@ class TestTag(testtools.TestCase):
self.assertFalse(sot.allow_list)
def test_make_it(self):
sot = tag.Tag(EXAMPLE)
self.assertEqual(IDENTIFIER, sot.id)
self.assertEqual(EXAMPLE['image_id'], sot.image_id)
sot = tag.Tag({"image": self.img})
self.assertEqual(self.img, sot.image)
def _test_action(self, sot_method, session_method):
test_tag = "testing"
sot = tag.Tag({"image": self.img})
rv = getattr(sot, sot_method)(self.session, test_tag)
url = 'images/%(image)s/tags/%(tag)s' % {
"image": self.img.get_id(self.img), "tag": test_tag}
self.assertIsNone(rv)
session_method.assert_called_with(url, service=sot.service,
accept=None)
def test_create(self):
sess = mock.Mock()
resp = mock.Mock()
sess.put = mock.Mock(return_value=resp)
url = 'images/{image_id}/tags/{tag}'.format(**EXAMPLE)
tag.Tag(EXAMPLE).create(sess)
sess.put.assert_called_with(url, service=tag.Tag.service, json=None)
self._test_action("create", self.session.put)
def test_delete(self):
self._test_action("delete", self.session.delete)