openstack
/
glance
Code Issues Proposed changes

Merge "Remove v1 tests"

This commit is contained in:
Zuul 2018-04-18 12:03:29 +00:00 committed by Gerrit Code Review
parent 7c3a9c83da 55318560a7
commit b7ff0782fd
10 changed files with 0 additions and 9793 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
glance/tests/functional/v1/__init__.py
View File

961
glance/tests/functional/v1/test_api.py
View File

@ -1,961 +0,0 @@
# Copyright 2011 OpenStack Foundation
# All Rights Reserved.
#
# 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.
"""Functional test case that utilizes httplib2 against the API server"""
import hashlib
import httplib2
import sys
from oslo_serialization import jsonutils
from oslo_utils import units
from six.moves import http_client
# NOTE(jokke): simplified transition to py3, behaves like py2 xrange
from six.moves import range
from glance.tests import functional
from glance.tests.utils import minimal_headers
from glance.tests.utils import skip_if_disabled
FIVE_KB = 5 * units.Ki
FIVE_GB = 5 * units.Gi
class TestApi(functional.FunctionalTest):
"""Functional tests using httplib2 against the API server"""
def _check_image_create(self, headers, status=http_client.CREATED,
image_data="*" * FIVE_KB):
# performs image_create request, checks the response and returns
# content
http = httplib2.Http()
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
response, content = http.request(
path, 'POST', headers=headers, body=image_data)
self.assertEqual(status, response.status)
return content.decode()
def test_checksum_32_chars_at_image_create(self):
self.cleanup()
self.start_servers(**self.__dict__.copy())
headers = minimal_headers('Image1')
image_data = b"*" * FIVE_KB
# checksum can be no longer that 32 characters (String(32))
headers['X-Image-Meta-Checksum'] = 'x' * 42
content = self._check_image_create(headers, http_client.BAD_REQUEST)
self.assertIn("Invalid checksum", content)
# test positive case as well
headers['X-Image-Meta-Checksum'] = hashlib.md5(image_data).hexdigest()
self._check_image_create(headers)
def test_param_int_too_large_at_create(self):
# currently 2 params min_disk/min_ram can cause DBError on save
self.cleanup()
self.start_servers(**self.__dict__.copy())
# Integer field can't be greater than max 8-byte signed integer
for param in ['min_disk', 'min_ram']:
headers = minimal_headers('Image1')
# check that long numbers result in 400
headers['X-Image-Meta-%s' % param] = str(sys.maxsize + 1)
content = self._check_image_create(headers,
http_client.BAD_REQUEST)
self.assertIn("'%s' value out of range" % param, content)
# check that integers over 4 byte result in 400
headers['X-Image-Meta-%s' % param] = str(2 ** 31)
content = self._check_image_create(headers,
http_client.BAD_REQUEST)
self.assertIn("'%s' value out of range" % param, content)
# verify positive case as well
headers['X-Image-Meta-%s' % param] = str((2 ** 31) - 1)
self._check_image_create(headers)
def test_updating_is_public(self):
"""Verify that we can update the is_public attribute."""
self.cleanup()
self.start_servers(**self.__dict__.copy())
# Verify no public images
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual('{"images": []}', content.decode())
# Verify no public images
path = "http://%s:%d/v1/images/detail" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual('{"images": []}', content.decode())
# POST /images with private image named Image1
# attribute and no custom properties. Verify a 200 OK is returned
image_data = b"*" * FIVE_KB
headers = minimal_headers('Image1', public=False)
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
image_id = data['image']['id']
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image1", data['image']['name'])
self.assertFalse(data['image']['is_public'])
# Retrieve image again to verify it was created appropriately
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
expected_image_headers = {
'x-image-meta-id': image_id,
'x-image-meta-name': 'Image1',
'x-image-meta-is_public': 'False',
'x-image-meta-status': 'active',
'x-image-meta-disk_format': 'raw',
'x-image-meta-container_format': 'ovf',
'x-image-meta-size': str(FIVE_KB)}
expected_std_headers = {
'content-length': str(FIVE_KB),
'content-type': 'application/octet-stream'}
for expected_key, expected_value in expected_image_headers.items():
self.assertEqual(expected_value, response[expected_key],
"For key '%s' expected header value '%s'. "
"Got '%s'" % (expected_key,
expected_value,
response[expected_key]))
for expected_key, expected_value in expected_std_headers.items():
self.assertEqual(expected_value, response[expected_key],
"For key '%s' expected header value '%s'. "
"Got '%s'" % (expected_key,
expected_value,
response[expected_key]))
self.assertEqual(image_data, content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
hashlib.md5(content).hexdigest())
# PUT image with custom properties to make public and then
# Verify 200 returned
headers = {'X-Image-Meta-is_public': 'True'}
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.OK, response.status)
image = jsonutils.loads(content)
is_public = image['image']['is_public']
self.assertTrue(
is_public,
"Expected image to be public but received %s" % is_public)
# PUT image with custom properties to make private and then
# Verify 200 returned
headers = {'X-Image-Meta-is_public': 'False'}
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.OK, response.status)
image = jsonutils.loads(content)
is_public = image['image']['is_public']
self.assertFalse(
is_public,
"Expected image to be private but received %s" % is_public)
@skip_if_disabled
def test_get_head_simple_post(self):
"""
We test the following sequential series of actions:
0. GET /images
- Verify no public images
1. GET /images/detail
- Verify no public images
2. POST /images with public image named Image1
and no custom properties
- Verify 201 returned
3. HEAD image
- Verify HTTP headers have correct information we just added
4. GET image
- Verify all information on image we just added is correct
5. GET /images
- Verify the image we just added is returned
6. GET /images/detail
- Verify the image we just added is returned
7. PUT image with custom properties of "distro" and "arch"
- Verify 200 returned
8. PUT image with too many custom properties
- Verify 413 returned
9. GET image
- Verify updated information about image was stored
10. PUT image
- Remove a previously existing property.
11. PUT image
- Add a previously deleted property.
12. PUT image/members/member1
- Add member1 to image
13. PUT image/members/member2
- Add member2 to image
14. GET image/members
- List image members
15. DELETE image/members/member1
- Delete image member1
16. PUT image/members
- Attempt to replace members with an overlimit amount
17. PUT image/members/member11
- Attempt to add a member while at limit
18. POST /images with another public image named Image2
- attribute and three custom properties, "distro", "arch" & "foo"
- Verify a 200 OK is returned
19. HEAD image2
- Verify image2 found now
20. GET /images
- Verify 2 public images
21. GET /images with filter on user-defined property "distro".
- Verify both images are returned
22. GET /images with filter on user-defined property 'distro' but
- with non-existent value. Verify no images are returned
23. GET /images with filter on non-existent user-defined property
- "boo". Verify no images are returned
24. GET /images with filter 'arch=i386'
- Verify only image2 is returned
25. GET /images with filter 'arch=x86_64'
- Verify only image1 is returned
26. GET /images with filter 'foo=bar'
- Verify only image2 is returned
27. DELETE image1
- Delete image
28. GET image/members
- List deleted image members
29. PUT image/members/member2
- Update existing member2 of deleted image
30. PUT image/members/member3
- Add member3 to deleted image
31. DELETE image/members/member2
- Delete member2 from deleted image
32. DELETE image2
- Delete image
33. GET /images
- Verify no images are listed
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
# 0. GET /images
# Verify no public images
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual('{"images": []}', content.decode())
# 1. GET /images/detail
# Verify no public images
path = "http://%s:%d/v1/images/detail" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual('{"images": []}', content.decode())
# 2. POST /images with public image named Image1
# attribute and no custom properties. Verify a 200 OK is returned
image_data = b"*" * FIVE_KB
headers = minimal_headers('Image1')
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
image_id = data['image']['id']
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image1", data['image']['name'])
self.assertTrue(data['image']['is_public'])
# 3. HEAD image
# Verify image found now
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
self.assertEqual("Image1", response['x-image-meta-name'])
# 4. GET image
# Verify all information on image we just added is correct
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
expected_image_headers = {
'x-image-meta-id': image_id,
'x-image-meta-name': 'Image1',
'x-image-meta-is_public': 'True',
'x-image-meta-status': 'active',
'x-image-meta-disk_format': 'raw',
'x-image-meta-container_format': 'ovf',
'x-image-meta-size': str(FIVE_KB)}
expected_std_headers = {
'content-length': str(FIVE_KB),
'content-type': 'application/octet-stream'}
for expected_key, expected_value in expected_image_headers.items():
self.assertEqual(expected_value, response[expected_key],
"For key '%s' expected header value '%s'. "
"Got '%s'" % (expected_key,
expected_value,
response[expected_key]))
for expected_key, expected_value in expected_std_headers.items():
self.assertEqual(expected_value, response[expected_key],
"For key '%s' expected header value '%s'. "
"Got '%s'" % (expected_key,
expected_value,
response[expected_key]))
self.assertEqual(image_data, content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
hashlib.md5(content).hexdigest())
# 5. GET /images
# Verify one public image
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
expected_result = {"images": [
{"container_format": "ovf",
"disk_format": "raw",
"id": image_id,
"name": "Image1",
"checksum": "c2e5db72bd7fd153f53ede5da5a06de3",
"size": 5120}]}
self.assertEqual(expected_result, jsonutils.loads(content))
# 6. GET /images/detail
# Verify image and all its metadata
path = "http://%s:%d/v1/images/detail" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
expected_image = {
"status": "active",
"name": "Image1",
"deleted": False,
"container_format": "ovf",
"disk_format": "raw",
"id": image_id,
"is_public": True,
"deleted_at": None,
"properties": {},
"size": 5120}
image = jsonutils.loads(content)
for expected_key, expected_value in expected_image.items():
self.assertEqual(expected_value, image['images'][0][expected_key],
"For key '%s' expected header value '%s'. "
"Got '%s'" % (expected_key,
expected_value,
image['images'][0][expected_key]))
# 7. PUT image with custom properties of "distro" and "arch"
# Verify 200 returned
headers = {'X-Image-Meta-Property-Distro': 'Ubuntu',
'X-Image-Meta-Property-Arch': 'x86_64'}
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertEqual("x86_64", data['image']['properties']['arch'])
self.assertEqual("Ubuntu", data['image']['properties']['distro'])
# 8. PUT image with too many custom properties
# Verify 413 returned
headers = {}
for i in range(11): # configured limit is 10
headers['X-Image-Meta-Property-foo%d' % i] = 'bar'
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.REQUEST_ENTITY_TOO_LARGE,
response.status)
# 9. GET /images/detail
# Verify image and all its metadata
path = "http://%s:%d/v1/images/detail" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
expected_image = {
"status": "active",
"name": "Image1",
"deleted": False,
"container_format": "ovf",
"disk_format": "raw",
"id": image_id,
"is_public": True,
"deleted_at": None,
"properties": {'distro': 'Ubuntu', 'arch': 'x86_64'},
"size": 5120}
image = jsonutils.loads(content)
for expected_key, expected_value in expected_image.items():
self.assertEqual(expected_value, image['images'][0][expected_key],
"For key '%s' expected header value '%s'. "
"Got '%s'" % (expected_key,
expected_value,
image['images'][0][expected_key]))
# 10. PUT image and remove a previously existing property.
headers = {'X-Image-Meta-Property-Arch': 'x86_64'}
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.OK, response.status)
path = "http://%s:%d/v1/images/detail" % ("127.0.0.1", self.api_port)
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)['images'][0]
self.assertEqual(1, len(data['properties']))
self.assertEqual("x86_64", data['properties']['arch'])
# 11. PUT image and add a previously deleted property.
headers = {'X-Image-Meta-Property-Distro': 'Ubuntu',
'X-Image-Meta-Property-Arch': 'x86_64'}
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
path = "http://%s:%d/v1/images/detail" % ("127.0.0.1", self.api_port)
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)['images'][0]
self.assertEqual(2, len(data['properties']))
self.assertEqual("x86_64", data['properties']['arch'])
self.assertEqual("Ubuntu", data['properties']['distro'])
self.assertNotEqual(data['created_at'], data['updated_at'])
# 12. Add member to image
path = ("http://%s:%d/v1/images/%s/members/pattieblack" %
("127.0.0.1", self.api_port, image_id))
http = httplib2.Http()
response, content = http.request(path, 'PUT')
self.assertEqual(http_client.NO_CONTENT, response.status)
# 13. Add member to image
path = ("http://%s:%d/v1/images/%s/members/pattiewhite" %
("127.0.0.1", self.api_port, image_id))
http = httplib2.Http()
response, content = http.request(path, 'PUT')
self.assertEqual(http_client.NO_CONTENT, response.status)
# 14. List image members
path = ("http://%s:%d/v1/images/%s/members" %
("127.0.0.1", self.api_port, image_id))
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
data = jsonutils.loads(content)
self.assertEqual(2, len(data['members']))
self.assertEqual('pattieblack', data['members'][0]['member_id'])
self.assertEqual('pattiewhite', data['members'][1]['member_id'])
# 15. Delete image member
path = ("http://%s:%d/v1/images/%s/members/pattieblack" %
("127.0.0.1", self.api_port, image_id))
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.NO_CONTENT, response.status)
# 16. Attempt to replace members with an overlimit amount
# Adding 11 image members should fail since configured limit is 10
path = ("http://%s:%d/v1/images/%s/members" %
("127.0.0.1", self.api_port, image_id))
memberships = []
for i in range(11):
member_id = "foo%d" % i
memberships.append(dict(member_id=member_id))
http = httplib2.Http()
body = jsonutils.dumps(dict(memberships=memberships))
response, content = http.request(path, 'PUT', body=body)
self.assertEqual(http_client.REQUEST_ENTITY_TOO_LARGE,
response.status)
# 17. Attempt to add a member while at limit
# Adding an 11th member should fail since configured limit is 10
path = ("http://%s:%d/v1/images/%s/members" %
("127.0.0.1", self.api_port, image_id))
memberships = []
for i in range(10):
member_id = "foo%d" % i
memberships.append(dict(member_id=member_id))
http = httplib2.Http()
body = jsonutils.dumps(dict(memberships=memberships))
response, content = http.request(path, 'PUT', body=body)
self.assertEqual(http_client.NO_CONTENT, response.status)
path = ("http://%s:%d/v1/images/%s/members/fail_me" %
("127.0.0.1", self.api_port, image_id))
http = httplib2.Http()
response, content = http.request(path, 'PUT')
self.assertEqual(http_client.REQUEST_ENTITY_TOO_LARGE,
response.status)
# 18. POST /images with another public image named Image2
# attribute and three custom properties, "distro", "arch" & "foo".
# Verify a 200 OK is returned
image_data = b"*" * FIVE_KB
headers = minimal_headers('Image2')
headers['X-Image-Meta-Property-Distro'] = 'Ubuntu'
headers['X-Image-Meta-Property-Arch'] = 'i386'
headers['X-Image-Meta-Property-foo'] = 'bar'
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
image2_id = data['image']['id']
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image2", data['image']['name'])
self.assertTrue(data['image']['is_public'])
self.assertEqual('Ubuntu', data['image']['properties']['distro'])
self.assertEqual('i386', data['image']['properties']['arch'])
self.assertEqual('bar', data['image']['properties']['foo'])
# 19. HEAD image2
# Verify image2 found now
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image2_id)
http = httplib2.Http()
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
self.assertEqual("Image2", response['x-image-meta-name'])
# 20. GET /images
# Verify 2 public images
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
images = jsonutils.loads(content)['images']
self.assertEqual(2, len(images))
self.assertEqual(image2_id, images[0]['id'])
self.assertEqual(image_id, images[1]['id'])
# 21. GET /images with filter on user-defined property 'distro'.
# Verify both images are returned
path = "http://%s:%d/v1/images?property-distro=Ubuntu" % (
"127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
images = jsonutils.loads(content)['images']
self.assertEqual(2, len(images))
self.assertEqual(image2_id, images[0]['id'])
self.assertEqual(image_id, images[1]['id'])
# 22. GET /images with filter on user-defined property 'distro' but
# with non-existent value. Verify no images are returned
path = "http://%s:%d/v1/images?property-distro=fedora" % (
"127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
images = jsonutils.loads(content)['images']
self.assertEqual(0, len(images))
# 23. GET /images with filter on non-existent user-defined property
# 'boo'. Verify no images are returned
path = "http://%s:%d/v1/images?property-boo=bar" % ("127.0.0.1",
self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
images = jsonutils.loads(content)['images']
self.assertEqual(0, len(images))
# 24. GET /images with filter 'arch=i386'
# Verify only image2 is returned
path = "http://%s:%d/v1/images?property-arch=i386" % ("127.0.0.1",
self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
images = jsonutils.loads(content)['images']
self.assertEqual(1, len(images))
self.assertEqual(image2_id, images[0]['id'])
# 25. GET /images with filter 'arch=x86_64'
# Verify only image1 is returned
path = "http://%s:%d/v1/images?property-arch=x86_64" % ("127.0.0.1",
self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
images = jsonutils.loads(content)['images']
self.assertEqual(1, len(images))
self.assertEqual(image_id, images[0]['id'])
# 26. GET /images with filter 'foo=bar'
# Verify only image2 is returned
path = "http://%s:%d/v1/images?property-foo=bar" % ("127.0.0.1",
self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
images = jsonutils.loads(content)['images']
self.assertEqual(1, len(images))
self.assertEqual(image2_id, images[0]['id'])
# 27. DELETE image1
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
# 28. Try to list members of deleted image
path = ("http://%s:%d/v1/images/%s/members" %
("127.0.0.1", self.api_port, image_id))
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.NOT_FOUND, response.status)
# 29. Try to update member of deleted image
path = ("http://%s:%d/v1/images/%s/members" %
("127.0.0.1", self.api_port, image_id))
http = httplib2.Http()
fixture = [{'member_id': 'pattieblack', 'can_share': 'false'}]
body = jsonutils.dumps(dict(memberships=fixture))
response, content = http.request(path, 'PUT', body=body)
self.assertEqual(http_client.NOT_FOUND, response.status)
# 30. Try to add member to deleted image
path = ("http://%s:%d/v1/images/%s/members/chickenpattie" %
("127.0.0.1", self.api_port, image_id))
http = httplib2.Http()
response, content = http.request(path, 'PUT')
self.assertEqual(http_client.NOT_FOUND, response.status)
# 31. Try to delete member of deleted image
path = ("http://%s:%d/v1/images/%s/members/pattieblack" %
("127.0.0.1", self.api_port, image_id))
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.NOT_FOUND, response.status)
# 32. DELETE image2
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image2_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
# 33. GET /images
# Verify no images are listed
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
images = jsonutils.loads(content)['images']
self.assertEqual(0, len(images))
# 34. HEAD /images/detail
path = "http://%s:%d/v1/images/detail" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.METHOD_NOT_ALLOWED, response.status)
self.assertEqual('GET', response.get('allow'))
self.stop_servers()
def test_download_non_exists_image_raises_http_forbidden(self):
"""
We test the following sequential series of actions::
0. POST /images with public image named Image1
and no custom properties
- Verify 201 returned
1. HEAD image
- Verify HTTP headers have correct information we just added
2. GET image
- Verify all information on image we just added is correct
3. DELETE image1
- Delete the newly added image
4. GET image
- Verify that 403 HTTPForbidden exception is raised prior to
404 HTTPNotFound
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
image_data = b"*" * FIVE_KB
headers = minimal_headers('Image1')
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
image_id = data['image']['id']
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image1", data['image']['name'])
self.assertTrue(data['image']['is_public'])
# 1. HEAD image
# Verify image found now
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
self.assertEqual("Image1", response['x-image-meta-name'])
# 2. GET /images
# Verify one public image
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
expected_result = {"images": [
{"container_format": "ovf",
"disk_format": "raw",
"id": image_id,
"name": "Image1",
"checksum": "c2e5db72bd7fd153f53ede5da5a06de3",
"size": 5120}]}
self.assertEqual(expected_result, jsonutils.loads(content))
# 3. DELETE image1
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
# 4. GET image
# Verify that 403 HTTPForbidden exception is raised prior to
# 404 HTTPNotFound
rules = {"download_image": '!'}
self.set_policy_rules(rules)
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.FORBIDDEN, response.status)
self.stop_servers()
def test_download_non_exists_image_raises_http_not_found(self):
"""
We test the following sequential series of actions:
0. POST /images with public image named Image1
and no custom properties
- Verify 201 returned
1. HEAD image
- Verify HTTP headers have correct information we just added
2. GET image
- Verify all information on image we just added is correct
3. DELETE image1
- Delete the newly added image
4. GET image
- Verify that 404 HTTPNotFound exception is raised
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
image_data = b"*" * FIVE_KB
headers = minimal_headers('Image1')
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
image_id = data['image']['id']
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image1", data['image']['name'])
self.assertTrue(data['image']['is_public'])
# 1. HEAD image
# Verify image found now
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
self.assertEqual("Image1", response['x-image-meta-name'])
# 2. GET /images
# Verify one public image
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
expected_result = {"images": [
{"container_format": "ovf",
"disk_format": "raw",
"id": image_id,
"name": "Image1",
"checksum": "c2e5db72bd7fd153f53ede5da5a06de3",
"size": 5120}]}
self.assertEqual(expected_result, jsonutils.loads(content))
# 3. DELETE image1
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
# 4. GET image
# Verify that 404 HTTPNotFound exception is raised
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.NOT_FOUND, response.status)
self.stop_servers()
def test_status_cannot_be_manipulated_directly(self):
self.cleanup()
self.start_servers(**self.__dict__.copy())
headers = minimal_headers('Image1')
# Create a 'queued' image
http = httplib2.Http()
headers = {'Content-Type': 'application/octet-stream',
'X-Image-Meta-Disk-Format': 'raw',
'X-Image-Meta-Container-Format': 'bare'}
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
response, content = http.request(path, 'POST', headers=headers,
body=None)
self.assertEqual(http_client.CREATED, response.status)
image = jsonutils.loads(content)['image']
self.assertEqual('queued', image['status'])
# Ensure status of 'queued' image can't be changed
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
image['id'])
http = httplib2.Http()
headers = {'X-Image-Meta-Status': 'active'}
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.FORBIDDEN, response.status)
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
self.assertEqual('queued', response['x-image-meta-status'])
# We allow 'setting' to the same status
http = httplib2.Http()
headers = {'X-Image-Meta-Status': 'queued'}
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.OK, response.status)
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
self.assertEqual('queued', response['x-image-meta-status'])
# Make image active
http = httplib2.Http()
headers = {'Content-Type': 'application/octet-stream'}
response, content = http.request(path, 'PUT', headers=headers,
body='data')
self.assertEqual(http_client.OK, response.status)
image = jsonutils.loads(content)['image']
self.assertEqual('active', image['status'])
# Ensure status of 'active' image can't be changed
http = httplib2.Http()
headers = {'X-Image-Meta-Status': 'queued'}
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.FORBIDDEN, response.status)
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
self.assertEqual('active', response['x-image-meta-status'])
# We allow 'setting' to the same status
http = httplib2.Http()
headers = {'X-Image-Meta-Status': 'active'}
response, content = http.request(path, 'PUT', headers=headers)
self.assertEqual(http_client.OK, response.status)
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
self.assertEqual('active', response['x-image-meta-status'])
# Create a 'queued' image, ensure 'status' header is ignored
http = httplib2.Http()
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
headers = {'Content-Type': 'application/octet-stream',
'X-Image-Meta-Status': 'active'}
response, content = http.request(path, 'POST', headers=headers,
body=None)
self.assertEqual(http_client.CREATED, response.status)
image = jsonutils.loads(content)['image']
self.assertEqual('queued', image['status'])
# Create an 'active' image, ensure 'status' header is ignored
http = httplib2.Http()
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
headers = {'Content-Type': 'application/octet-stream',
'X-Image-Meta-Disk-Format': 'raw',
'X-Image-Meta-Status': 'queued',
'X-Image-Meta-Container-Format': 'bare'}
response, content = http.request(path, 'POST', headers=headers,
body='data')
self.assertEqual(http_client.CREATED, response.status)
image = jsonutils.loads(content)['image']
self.assertEqual('active', image['status'])
self.stop_servers()

300
glance/tests/functional/v1/test_copy_to_file.py
View File

@ -1,300 +0,0 @@
# Copyright 2011 OpenStack Foundation
# Copyright 2012 Red Hat, Inc
# All Rights Reserved.
#
# 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.
"""
Tests copying images to a Glance API server which uses a filesystem-
based storage backend.
"""
import hashlib
import tempfile
import time
import httplib2
from oslo_serialization import jsonutils
from oslo_utils import units
from six.moves import http_client
# NOTE(jokke): simplified transition to py3, behaves like py2 xrange
from six.moves import range
from glance.tests import functional
from glance.tests.functional.store_utils import get_http_uri
from glance.tests.functional.store_utils import setup_http
from glance.tests.utils import skip_if_disabled
FIVE_KB = 5 * units.Ki
class TestCopyToFile(functional.FunctionalTest):
"""
Functional tests for copying images from the HTTP storage
backend to file
"""
def _do_test_copy_from(self, from_store, get_uri):
"""
Ensure we can copy from an external image in from_store.
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
setup_http(self)
# POST /images with public image to be stored in from_store,
# to stand in for the 'external' image
image_data = b"*" * FIVE_KB
headers = {'Content-Type': 'application/octet-stream',
'X-Image-Meta-Name': 'external',
'X-Image-Meta-Store': from_store,
'X-Image-Meta-disk_format': 'raw',
'X-Image-Meta-container_format': 'ovf',
'X-Image-Meta-Is-Public': 'True'}
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status, content)
data = jsonutils.loads(content)
original_image_id = data['image']['id']
copy_from = get_uri(self, original_image_id)
# POST /images with public image copied from_store (to file)
headers = {'X-Image-Meta-Name': 'copied',
'X-Image-Meta-disk_format': 'raw',
'X-Image-Meta-container_format': 'ovf',
'X-Image-Meta-Is-Public': 'True',
'X-Glance-API-Copy-From': copy_from}
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers)
self.assertEqual(http_client.CREATED, response.status, content)
data = jsonutils.loads(content)
copy_image_id = data['image']['id']
self.assertNotEqual(copy_image_id, original_image_id)
# GET image and make sure image content is as expected
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
copy_image_id)
def _await_status(expected_status):
for i in range(100):
time.sleep(0.01)
http = httplib2.Http()
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
if response['x-image-meta-status'] == expected_status:
return
self.fail('unexpected image status %s' %
response['x-image-meta-status'])
_await_status('active')
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual(str(FIVE_KB), response['content-length'])
self.assertEqual(image_data, content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
hashlib.md5(content).hexdigest())
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("copied", data['image']['name'])
# DELETE original image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
original_image_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
# GET image again to make sure the existence of the original
# image in from_store is not depended on
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
copy_image_id)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual(str(FIVE_KB), response['content-length'])
self.assertEqual(image_data, content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
hashlib.md5(content).hexdigest())
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("copied", data['image']['name'])
# DELETE copied image
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
copy_image_id)
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
self.stop_servers()
@skip_if_disabled
def test_copy_from_http_store(self):
"""
Ensure we can copy from an external image in HTTP store.
"""
self._do_test_copy_from('file', get_http_uri)
@skip_if_disabled
def test_copy_from_http_exists(self):
"""Ensure we can copy from an external image in HTTP."""
self.cleanup()
self.start_servers(**self.__dict__.copy())
setup_http(self)
copy_from = get_http_uri(self, 'foobar')
# POST /images with public image copied from HTTP (to file)
headers = {'X-Image-Meta-Name': 'copied',
'X-Image-Meta-disk_format': 'raw',
'X-Image-Meta-container_format': 'ovf',
'X-Image-Meta-Is-Public': 'True',
'X-Glance-API-Copy-From': copy_from}
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers)
self.assertEqual(http_client.CREATED, response.status, content)
data = jsonutils.loads(content)
copy_image_id = data['image']['id']
self.assertEqual('queued', data['image']['status'], content)
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
copy_image_id)
def _await_status(expected_status):
for i in range(100):
time.sleep(0.01)
http = httplib2.Http()
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
if response['x-image-meta-status'] == expected_status:
return
self.fail('unexpected image status %s' %
response['x-image-meta-status'])
_await_status('active')
# GET image and make sure image content is as expected
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual(str(FIVE_KB), response['content-length'])
self.assertEqual(b"*" * FIVE_KB, content)
self.assertEqual(hashlib.md5(b"*" * FIVE_KB).hexdigest(),
hashlib.md5(content).hexdigest())
# DELETE copied image
http = httplib2.Http()
response, content = http.request(path, 'DELETE')
self.assertEqual(http_client.OK, response.status)
self.stop_servers()
@skip_if_disabled
def test_copy_from_http_nonexistent_location_url(self):
# Ensure HTTP 404 response returned when try to create
# image with non-existent http location URL.
self.cleanup()
self.start_servers(**self.__dict__.copy())
setup_http(self)
uri = get_http_uri(self, 'foobar')
copy_from = uri.replace('images', 'snafu')
# POST /images with public image copied from HTTP (to file)
headers = {'X-Image-Meta-Name': 'copied',
'X-Image-Meta-disk_format': 'raw',
'X-Image-Meta-container_format': 'ovf',
'X-Image-Meta-Is-Public': 'True',
'X-Glance-API-Copy-From': copy_from}
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers)
self.assertEqual(http_client.NOT_FOUND, response.status, content)
expected = 'HTTP datastore could not find image at URI.'
self.assertIn(expected, content.decode())
self.stop_servers()
@skip_if_disabled
def test_copy_from_file(self):
"""
Ensure we can't copy from file
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
with tempfile.NamedTemporaryFile() as image_file:
image_file.write(b"XXX")
image_file.flush()
copy_from = 'file://' + image_file.name
# POST /images with public image copied from file (to file)
headers = {'X-Image-Meta-Name': 'copied',
'X-Image-Meta-disk_format': 'raw',
'X-Image-Meta-container_format': 'ovf',
'X-Image-Meta-Is-Public': 'True',
'X-Glance-API-Copy-From': copy_from}
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers)
self.assertEqual(http_client.BAD_REQUEST, response.status, content)
expected = 'External sources are not supported: \'%s\'' % copy_from
msg = 'expected "%s" in "%s"' % (expected, content)
self.assertIn(expected, content.decode(), msg)
self.stop_servers()
@skip_if_disabled
def test_copy_from_swift_config(self):
"""
Ensure we can't copy from swift+config
"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
# POST /images with public image copied from file (to file)
headers = {'X-Image-Meta-Name': 'copied',
'X-Image-Meta-disk_format': 'raw',
'X-Image-Meta-container_format': 'ovf',
'X-Image-Meta-Is-Public': 'True',
'X-Glance-API-Copy-From': 'swift+config://xxx'}
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers)
self.assertEqual(http_client.BAD_REQUEST, response.status, content)
expected = 'External sources are not supported: \'swift+config://xxx\''
msg = 'expected "%s" in "%s"' % (expected, content)
self.assertIn(expected, content.decode(), msg)
self.stop_servers()

122
glance/tests/functional/v1/test_misc.py
View File

@ -1,122 +0,0 @@
# Copyright 2011 OpenStack Foundation
# All Rights Reserved.
#
# 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.
import hashlib
import os
import httplib2
from oslo_serialization import jsonutils
from oslo_utils import units
from six.moves import http_client
from glance.tests import functional
from glance.tests.utils import minimal_headers
FIVE_KB = 5 * units.Ki
FIVE_GB = 5 * units.Gi
class TestMiscellaneous(functional.FunctionalTest):
"""Some random tests for various bugs and stuff"""
def setUp(self):
super(TestMiscellaneous, self).setUp()
# NOTE(sirp): This is needed in case we are running the tests under an
# environment in which OS_AUTH_STRATEGY=keystone. The test server we
# spin up won't have keystone support, so we need to switch to the
# NoAuth strategy.
os.environ['OS_AUTH_STRATEGY'] = 'noauth'
os.environ['OS_AUTH_URL'] = ''
def test_api_response_when_image_deleted_from_filesystem(self):
"""
A test for LP bug #781410 -- glance should fail more gracefully
on requests for images that have been removed from the fs
"""
self.cleanup()
self.start_servers()
# 1. POST /images with public image named Image1
# attribute and no custom properties. Verify a 200 OK is returned
image_data = b"*" * FIVE_KB
headers = minimal_headers('Image1')
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'POST', headers=headers,
body=image_data)
self.assertEqual(http_client.CREATED, response.status)
data = jsonutils.loads(content)
self.assertEqual(hashlib.md5(image_data).hexdigest(),
data['image']['checksum'])
self.assertEqual(FIVE_KB, data['image']['size'])
self.assertEqual("Image1", data['image']['name'])
self.assertTrue(data['image']['is_public'])
# 2. REMOVE the image from the filesystem
image_path = "%s/images/%s" % (self.test_dir, data['image']['id'])
os.remove(image_path)
# 3. HEAD /images/1
# Verify image found now
path = "http://%s:%d/v1/images/%s" % ("127.0.0.1", self.api_port,
data['image']['id'])
http = httplib2.Http()
response, content = http.request(path, 'HEAD')
self.assertEqual(http_client.OK, response.status)
self.assertEqual("Image1", response['x-image-meta-name'])
# 4. GET /images/1
# Verify the api throws the appropriate 404 error
path = "http://%s:%d/v1/images/1" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.NOT_FOUND, response.status)
self.stop_servers()
def test_exception_not_eaten_from_registry_to_api(self):
"""
A test for LP bug #704854 -- Exception thrown by registry
server is consumed by API server.
We start both servers daemonized.
We then use Glance API to try adding an image that does not
meet validation requirements on the registry server and test
that the error returned from the API server is appropriate
"""
self.cleanup()
self.start_servers()
api_port = self.api_port
path = 'http://127.0.0.1:%d/v1/images' % api_port
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual(b'{"images": []}', content)
headers = {'Content-Type': 'application/octet-stream',
'X-Image-Meta-Name': 'ImageName',
'X-Image-Meta-Disk-Format': 'Invalid', }
ignored, content = http.request(path, 'POST', headers=headers)
self.assertIn(b'Invalid disk format', content,
"Could not find 'Invalid disk format' "
"in output: %s" % content)
self.stop_servers()

80
glance/tests/functional/v1/test_multiprocessing.py
View File

@ -1,80 +0,0 @@
# Copyright 2012 OpenStack Foundation
# All Rights Reserved.
#
# 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.
import time
import httplib2
import psutil
from six.moves import http_client
# NOTE(jokke): simplified transition to py3, behaves like py2 xrange
from six.moves import range
from glance.tests import functional
from glance.tests.utils import execute
class TestMultiprocessing(functional.FunctionalTest):
"""Functional tests for the bin/glance CLI tool"""
def setUp(self):
self.workers = 2
super(TestMultiprocessing, self).setUp()
def test_multiprocessing(self):
"""Spin up the api servers with multiprocessing on"""
self.cleanup()
self.start_servers(**self.__dict__.copy())
path = "http://%s:%d/v1/images" % ("127.0.0.1", self.api_port)
http = httplib2.Http()
response, content = http.request(path, 'GET')
self.assertEqual(http_client.OK, response.status)
self.assertEqual(b'{"images": []}', content)
self.stop_servers()
def _get_children(self):
api_pid = self.api_server.process_pid
process = psutil.Process(api_pid)
try:
# psutils version >= 2
children = process.children()
except AttributeError:
# psutils version < 2
children = process.get_children()
pids = [str(child.pid) for child in children]
return pids
def test_interrupt_avoids_respawn_storm(self):
"""
Ensure an interrupt signal does not cause a respawn storm.
See bug #978130
"""
self.start_servers(**self.__dict__.copy())
children = self._get_children()
cmd = "kill -INT %s" % ' '.join(children)
execute(cmd, raise_error=True)
for _ in range(9):
# Yeah. This totally isn't a race condition. Randomly fails
# set at 0.05. Works most of the time at 0.10
time.sleep(0.10)
# ensure number of children hasn't grown
self.assertGreaterEqual(len(children), len(self._get_children()))
for child in self._get_children():
# ensure no new children spawned
self.assertIn(child, children, child)
self.stop_servers()

0
glance/tests/unit/v1/__init__.py
View File

4841
glance/tests/unit/v1/test_api.py
View File

File diff suppressed because it is too large Load Diff

2162
glance/tests/unit/v1/test_registry_api.py
View File

File diff suppressed because it is too large Load Diff

982
glance/tests/unit/v1/test_registry_client.py
View File

@ -1,982 +0,0 @@
# Copyright 2010-2011 OpenStack Foundation
# All Rights Reserved.
#
# 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.
import copy
import datetime
import os
import uuid
from mock import patch
from six.moves import http_client as http
from six.moves import reload_module
import testtools
from glance.api.v1.images import Controller as acontroller
from glance.common import client as test_client
from glance.common import config
from glance.common import exception
from glance.common import timeutils
from glance import context
from glance.db.sqlalchemy import api as db_api
from glance.registry.api.v1.images import Controller as rcontroller
import glance.registry.client.v1.api as rapi
from glance.registry.client.v1.api import client as rclient
from glance.tests.unit import base
from glance.tests import utils as test_utils
import webob
_gen_uuid = lambda: str(uuid.uuid4())
UUID1 = _gen_uuid()
UUID2 = _gen_uuid()
# NOTE(bcwaldon): needed to init config_dir cli opt
config.parse_args(args=[])
class TestRegistryV1Client(base.IsolatedUnitTest, test_utils.RegistryAPIMixIn):
"""
Test proper actions made for both valid and invalid requests
against a Registry service
"""
def setUp(self):
"""Establish a clean test environment"""
super(TestRegistryV1Client, self).setUp()
db_api.get_engine()
self.context = context.RequestContext(is_admin=True)
self.FIXTURES = [
self.get_fixture(
id=UUID1, name='fake image #1', is_public=False,
disk_format='ami', container_format='ami', size=13,
location="swift://user:passwd@acct/container/obj.tar.0",
properties={'type': 'kernel'}),
self.get_fixture(id=UUID2, name='fake image #2', properties={},
size=19, location="file:///tmp/glance-tests/2")]
self.destroy_fixtures()
self.create_fixtures()
self.client = rclient.RegistryClient("0.0.0.0")
def tearDown(self):
"""Clear the test environment"""
super(TestRegistryV1Client, self).tearDown()
self.destroy_fixtures()
def test_get_image_index(self):
"""Test correct set of public image returned"""
fixture = {
'id': UUID2,
'name': 'fake image #2'
}
images = self.client.get_images()
self.assertEqualImages(images, (UUID2,), unjsonify=False)
for k, v in fixture.items():
self.assertEqual(v, images[0][k])
def test_create_image_with_null_min_disk_min_ram(self):
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, min_disk=None, min_ram=None)
db_api.image_create(self.context, extra_fixture)
image = self.client.get_image(UUID3)
self.assertEqual(0, image["min_ram"])
self.assertEqual(0, image["min_disk"])
def test_get_index_sort_name_asc(self):
"""
Tests that the /images registry API returns list of
public images sorted alphabetically by name in
ascending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='asdf')
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='xyz')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(sort_key='name', sort_dir='asc')
self.assertEqualImages(images, (UUID3, UUID2, UUID4), unjsonify=False)
def test_get_index_sort_status_desc(self):
"""
Tests that the /images registry API returns list of
public images sorted alphabetically by status in
descending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='asdf',
status='queued')
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='xyz')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(sort_key='status', sort_dir='desc')
self.assertEqualImages(images, (UUID3, UUID4, UUID2), unjsonify=False)
def test_get_index_sort_disk_format_asc(self):
"""
Tests that the /images registry API returns list of
public images sorted alphabetically by disk_format in
ascending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='asdf',
disk_format='ami',
container_format='ami')
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='xyz',
disk_format='vdi')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(sort_key='disk_format',
sort_dir='asc')
self.assertEqualImages(images, (UUID3, UUID4, UUID2), unjsonify=False)
def test_get_index_sort_container_format_desc(self):
"""
Tests that the /images registry API returns list of
public images sorted alphabetically by container_format in
descending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='asdf',
disk_format='ami',
container_format='ami')
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='xyz',
disk_format='iso',
container_format='bare')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(sort_key='container_format',
sort_dir='desc')
self.assertEqualImages(images, (UUID2, UUID4, UUID3), unjsonify=False)
def test_get_index_sort_size_asc(self):
"""
Tests that the /images registry API returns list of
public images sorted by size in ascending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='asdf',
disk_format='ami',
container_format='ami', size=100)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='asdf',
disk_format='iso',
container_format='bare', size=2)
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(sort_key='size', sort_dir='asc')
self.assertEqualImages(images, (UUID4, UUID2, UUID3), unjsonify=False)
def test_get_index_sort_created_at_asc(self):
"""
Tests that the /images registry API returns list of
public images sorted by created_at in ascending order.
"""
now = timeutils.utcnow()
time1 = now + datetime.timedelta(seconds=5)
time2 = now
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, created_at=time1)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, created_at=time2)
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(sort_key='created_at', sort_dir='asc')
self.assertEqualImages(images, (UUID2, UUID4, UUID3), unjsonify=False)
def test_get_index_sort_updated_at_desc(self):
"""
Tests that the /images registry API returns list of
public images sorted by updated_at in descending order.
"""
now = timeutils.utcnow()
time1 = now + datetime.timedelta(seconds=5)
time2 = now
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, created_at=None,
updated_at=time1)
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, created_at=None,
updated_at=time2)
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(sort_key='updated_at', sort_dir='desc')
self.assertEqualImages(images, (UUID3, UUID4, UUID2), unjsonify=False)
def test_get_image_index_marker(self):
"""Test correct set of images returned with marker param."""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='new name! #123',
status='saving')
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='new name! #125',
status='saving')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(marker=UUID4)
self.assertEqualImages(images, (UUID3, UUID2), unjsonify=False)
def test_get_image_index_invalid_marker(self):
"""Test exception is raised when marker is invalid"""
self.assertRaises(exception.Invalid,
self.client.get_images,
marker=_gen_uuid())
def test_get_image_index_forbidden_marker(self):
"""Test exception is raised when marker is forbidden"""
UUID5 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID5, owner='0123',
status='saving', is_public=False)
db_api.image_create(self.context, extra_fixture)
def non_admin_get_images(self, context, *args, **kwargs):
"""Convert to non-admin context"""
context.is_admin = False
rcontroller.__get_images(self, context, *args, **kwargs)
rcontroller.__get_images = rcontroller._get_images
self.stubs.Set(rcontroller, '_get_images', non_admin_get_images)
self.assertRaises(exception.Invalid,
self.client.get_images,
marker=UUID5)
def test_get_image_index_private_marker(self):
"""Test exception is not raised if private non-owned marker is used"""
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, owner='1234',
status='saving', is_public=False)
db_api.image_create(self.context, extra_fixture)
try:
self.client.get_images(marker=UUID4)
except Exception as e:
self.fail("Unexpected exception '%s'" % e)
def test_get_image_index_limit(self):
"""Test correct number of images returned with limit param."""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(limit=2)
self.assertEqual(2, len(images))
def test_get_image_index_marker_limit(self):
"""Test correct set of images returned with marker/limit params."""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='new name! #123',
status='saving')
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='new name! #125',
status='saving')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(marker=UUID3, limit=1)
self.assertEqualImages(images, (UUID2,), unjsonify=False)
def test_get_image_index_limit_None(self):
"""Test correct set of images returned with limit param == None."""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(limit=None)
self.assertEqual(3, len(images))
def test_get_image_index_by_name(self):
"""
Test correct set of public, name-filtered image returned. This
is just a sanity check, we test the details call more in-depth.
"""
extra_fixture = self.get_fixture(id=_gen_uuid(), name='new name! #123')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images(filters={'name': 'new name! #123'})
self.assertEqual(1, len(images))
for image in images:
self.assertEqual('new name! #123', image['name'])
def test_get_image_details(self):
"""Tests that the detailed info about public images returned"""
fixture = self.get_fixture(id=UUID2, name='fake image #2',
properties={}, size=19, is_public=True)
images = self.client.get_images_detailed()
self.assertEqual(1, len(images))
for k, v in fixture.items():
self.assertEqual(v, images[0][k])
def test_get_image_details_marker_limit(self):
"""Test correct set of images returned with marker/limit params."""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, status='saving')
db_api.image_create(self.context, extra_fixture)
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images_detailed(marker=UUID3, limit=1)
self.assertEqualImages(images, (UUID2,), unjsonify=False)
def test_get_image_details_invalid_marker(self):
"""Test exception is raised when marker is invalid"""
self.assertRaises(exception.Invalid,
self.client.get_images_detailed,
marker=_gen_uuid())
def test_get_image_details_forbidden_marker(self):
"""Test exception is raised when marker is forbidden"""
UUID5 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID5, is_public=False,
owner='0123', status='saving')
db_api.image_create(self.context, extra_fixture)
def non_admin_get_images(self, context, *args, **kwargs):
"""Convert to non-admin context"""
context.is_admin = False
rcontroller.__get_images(self, context, *args, **kwargs)
rcontroller.__get_images = rcontroller._get_images
self.stubs.Set(rcontroller, '_get_images', non_admin_get_images)
self.assertRaises(exception.Invalid,
self.client.get_images_detailed,
marker=UUID5)
def test_get_image_details_private_marker(self):
"""Test exception is not raised if private non-owned marker is used"""
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, is_public=False,
owner='1234', status='saving')
db_api.image_create(self.context, extra_fixture)
try:
self.client.get_images_detailed(marker=UUID4)
except Exception as e:
self.fail("Unexpected exception '%s'" % e)
def test_get_image_details_by_name(self):
"""Tests that a detailed call can be filtered by name"""
extra_fixture = self.get_fixture(id=_gen_uuid(), name='new name! #123')
db_api.image_create(self.context, extra_fixture)
filters = {'name': 'new name! #123'}
images = self.client.get_images_detailed(filters=filters)
self.assertEqual(1, len(images))
for image in images:
self.assertEqual('new name! #123', image['name'])
def test_get_image_details_by_status(self):
"""Tests that a detailed call can be filtered by status"""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images_detailed(filters={'status': 'saving'})
self.assertEqual(1, len(images))
for image in images:
self.assertEqual('saving', image['status'])
def test_get_image_details_by_container_format(self):
"""Tests that a detailed call can be filtered by container_format"""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
filters = {'container_format': 'ovf'}
images = self.client.get_images_detailed(filters=filters)
self.assertEqual(2, len(images))
for image in images:
self.assertEqual('ovf', image['container_format'])
def test_get_image_details_by_disk_format(self):
"""Tests that a detailed call can be filtered by disk_format"""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
filters = {'disk_format': 'vhd'}
images = self.client.get_images_detailed(filters=filters)
self.assertEqual(2, len(images))
for image in images:
self.assertEqual('vhd', image['disk_format'])
def test_get_image_details_with_maximum_size(self):
"""Tests that a detailed call can be filtered by size_max"""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving',
size=21)
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images_detailed(filters={'size_max': 20})
self.assertEqual(1, len(images))
for image in images:
self.assertLessEqual(image['size'], 20)
def test_get_image_details_with_minimum_size(self):
"""Tests that a detailed call can be filtered by size_min"""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images_detailed(filters={'size_min': 20})
self.assertEqual(1, len(images))
for image in images:
self.assertGreaterEqual(image['size'], 20)
def test_get_image_details_with_changes_since(self):
"""Tests that a detailed call can be filtered by changes-since"""
dt1 = timeutils.utcnow() - datetime.timedelta(1)
iso1 = timeutils.isotime(dt1)
dt2 = timeutils.utcnow() + datetime.timedelta(1)
iso2 = timeutils.isotime(dt2)
dt3 = timeutils.utcnow() + datetime.timedelta(2)
dt4 = timeutils.utcnow() + datetime.timedelta(3)
iso4 = timeutils.isotime(dt4)
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='fake image #3')
db_api.image_create(self.context, extra_fixture)
db_api.image_destroy(self.context, UUID3)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='fake image #4',
created_at=dt3, updated_at=dt3)
db_api.image_create(self.context, extra_fixture)
# Check a standard list, 4 images in db (2 deleted)
images = self.client.get_images_detailed(filters={})
self.assertEqualImages(images, (UUID4, UUID2), unjsonify=False)
# Expect 3 images (1 deleted)
filters = {'changes-since': iso1}
images = self.client.get_images(filters=filters)
self.assertEqualImages(images, (UUID4, UUID3, UUID2), unjsonify=False)
# Expect 1 images (0 deleted)
filters = {'changes-since': iso2}
images = self.client.get_images_detailed(filters=filters)
self.assertEqualImages(images, (UUID4,), unjsonify=False)
# Expect 0 images (0 deleted)
filters = {'changes-since': iso4}
images = self.client.get_images(filters=filters)
self.assertEqualImages(images, (), unjsonify=False)
def test_get_image_details_with_size_min(self):
"""Tests that a detailed call can be filtered by size_min"""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images_detailed(filters={'size_min': 20})
self.assertEqual(1, len(images))
for image in images:
self.assertGreaterEqual(image['size'], 20)
def test_get_image_details_by_property(self):
"""Tests that a detailed call can be filtered by a property"""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving',
properties={'p a': 'v a'})
db_api.image_create(self.context, extra_fixture)
filters = {'property-p a': 'v a'}
images = self.client.get_images_detailed(filters=filters)
self.assertEqual(1, len(images))
for image in images:
self.assertEqual('v a', image['properties']['p a'])
def test_get_image_is_public_v1(self):
"""Tests that a detailed call can be filtered by a property"""
extra_fixture = self.get_fixture(id=_gen_uuid(), status='saving',
properties={'is_public': 'avalue'})
context = copy.copy(self.context)
db_api.image_create(context, extra_fixture)
filters = {'property-is_public': 'avalue'}
images = self.client.get_images_detailed(filters=filters)
self.assertEqual(1, len(images))
for image in images:
self.assertEqual('avalue', image['properties']['is_public'])
def test_get_image_details_sort_disk_format_asc(self):
"""
Tests that a detailed call returns list of
public images sorted alphabetically by disk_format in
ascending order.
"""
UUID3 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID3, name='asdf',
disk_format='ami',
container_format='ami')
db_api.image_create(self.context, extra_fixture)
UUID4 = _gen_uuid()
extra_fixture = self.get_fixture(id=UUID4, name='xyz',
disk_format='vdi')
db_api.image_create(self.context, extra_fixture)
images = self.client.get_images_detailed(sort_key='disk_format',
sort_dir='asc')
self.assertEqualImages(images, (UUID3, UUID4, UUID2), unjsonify=False)
def test_get_image(self):
"""Tests that the detailed info about an image returned"""
fixture = self.get_fixture(id=UUID1, name='fake image #1',
disk_format='ami', container_format='ami',
is_public=False, size=13,
properties={'type': 'kernel'})
data = self.client.get_image(UUID1)
for k, v in fixture.items():
el = data[k]
self.assertEqual(v, data[k],
"Failed v != data[k] where v = %(v)s and "
"k = %(k)s and data[k] = %(el)s" % {'v': v,
'k': k,
'el': el})
def test_get_image_non_existing(self):
"""Tests that NotFound is raised when getting a non-existing image"""
self.assertRaises(exception.NotFound,
self.client.get_image,
_gen_uuid())
def test_add_image_basic(self):
"""Tests that we can add image metadata and returns the new id"""
fixture = self.get_fixture(is_public=True)
new_image = self.client.add_image(fixture)
# Test all other attributes set
data = self.client.get_image(new_image['id'])
for k, v in fixture.items():
self.assertEqual(v, data[k])
# Test status was updated properly
self.assertIn('status', data.keys())
self.assertEqual('active', data['status'])
def test_add_image_with_properties(self):
"""Tests that we can add image metadata with properties"""
fixture = self.get_fixture(location="file:///tmp/glance-tests/2",
properties={'distro': 'Ubuntu 10.04 LTS'},
is_public=True)
new_image = self.client.add_image(fixture)
del fixture['location']
for k, v in fixture.items():
self.assertEqual(v, new_image[k])
# Test status was updated properly
self.assertIn('status', new_image.keys())
self.assertEqual('active', new_image['status'])
def test_add_image_with_location_data(self):
"""Tests that we can add image metadata with properties"""
location = "file:///tmp/glance-tests/2"
loc_meta = {'key': 'value'}
fixture = self.get_fixture(location_data=[{'url': location,
'metadata': loc_meta,
'status': 'active'}],
properties={'distro': 'Ubuntu 10.04 LTS'})
new_image = self.client.add_image(fixture)
self.assertEqual(location, new_image['location'])
self.assertEqual(location, new_image['location_data'][0]['url'])
self.assertEqual(loc_meta, new_image['location_data'][0]['metadata'])
def test_add_image_with_location_data_with_encryption(self):
"""Tests that we can add image metadata with properties and
enable encryption.
"""
self.client.metadata_encryption_key = '1234567890123456'
location = "file:///tmp/glance-tests/%d"
loc_meta = {'key': 'value'}
fixture = {'name': 'fake public image',
'is_public': True,
'disk_format': 'vmdk',
'container_format': 'ovf',
'size': 19,
'location_data': [{'url': location % 1,
'metadata': loc_meta,
'status': 'active'},
{'url': location % 2,
'metadata': {},
'status': 'active'}],
'properties': {'distro': 'Ubuntu 10.04 LTS'}}
new_image = self.client.add_image(fixture)
self.assertEqual(location % 1, new_image['location'])
self.assertEqual(2, len(new_image['location_data']))
self.assertEqual(location % 1, new_image['location_data'][0]['url'])
self.assertEqual(loc_meta, new_image['location_data'][0]['metadata'])
self.assertEqual(location % 2, new_image['location_data'][1]['url'])
self.assertEqual({}, new_image['location_data'][1]['metadata'])
self.client.metadata_encryption_key = None
def test_add_image_already_exists(self):
"""Tests proper exception is raised if image with ID already exists"""
fixture = self.get_fixture(id=UUID2,
location="file:///tmp/glance-tests/2")
self.assertRaises(exception.Duplicate,
self.client.add_image,
fixture)
def test_add_image_with_bad_status(self):
"""Tests proper exception is raised if a bad status is set"""
fixture = self.get_fixture(status='bad status',
location="file:///tmp/glance-tests/2")
self.assertRaises(exception.Invalid,
self.client.add_image,
fixture)
def test_update_image(self):
"""Tests that the /images PUT registry API updates the image"""
fixture = {'name': 'fake public image #2',
'disk_format': 'vmdk'}
self.assertTrue(self.client.update_image(UUID2, fixture))
# Test all other attributes set
data = self.client.get_image(UUID2)
for k, v in fixture.items():
self.assertEqual(v, data[k])
def test_update_image_public(self):
"""Tests that the /images PUT registry API updates the image"""
fixture = {'name': 'fake public image #2',
'is_public': True,
'disk_format': 'vmdk'}
self.assertTrue(self.client.update_image(UUID2, fixture))
# Test all other attributes set
data = self.client.get_image(UUID2)
for k, v in fixture.items():
self.assertEqual(v, data[k])
def test_update_image_private(self):
"""Tests that the /images PUT registry API updates the image"""
fixture = {'name': 'fake public image #2',
'is_public': False,
'disk_format': 'vmdk'}
self.assertTrue(self.client.update_image(UUID2, fixture))
# Test all other attributes set
data = self.client.get_image(UUID2)
for k, v in fixture.items():
self.assertEqual(v, data[k])
def test_update_image_not_existing(self):
"""Tests non existing image update doesn't work"""
fixture = self.get_fixture(status='bad status')
self.assertRaises(exception.NotFound,
self.client.update_image,
_gen_uuid(),
fixture)
def test_delete_image(self):
"""Tests that image metadata is deleted properly"""
# Grab the original number of images
orig_num_images = len(self.client.get_images())
# Delete image #2
image = self.FIXTURES[1]
deleted_image = self.client.delete_image(image['id'])
self.assertTrue(deleted_image)
self.assertEqual(image['id'], deleted_image['id'])
self.assertTrue(deleted_image['deleted'])
self.assertTrue(deleted_image['deleted_at'])
# Verify one less image
new_num_images = len(self.client.get_images())
self.assertEqual(orig_num_images - 1, new_num_images)
def test_delete_image_not_existing(self):
"""Check that one cannot delete non-existing image."""
self.assertRaises(exception.NotFound,
self.client.delete_image,
_gen_uuid())
def test_get_image_members(self):
"""Test getting image members."""
memb_list = self.client.get_image_members(UUID2)
num_members = len(memb_list)
self.assertEqual(0, num_members)
def test_get_image_members_not_existing(self):
"""Test getting non-existent image members."""
self.assertRaises(exception.NotFound,
self.client.get_image_members,
_gen_uuid())
def test_get_member_images(self):
"""Test getting member images."""
memb_list = self.client.get_member_images('pattieblack')
num_members = len(memb_list)
self.assertEqual(0, num_members)
def test_add_replace_members(self):
"""Test replacing image members."""
self.assertTrue(self.client.add_member(UUID2, 'pattieblack'))
self.assertTrue(self.client.replace_members(UUID2,
dict(member_id='pattie'
'black2')))
def test_add_delete_member(self):
"""Tests deleting image members"""
self.client.add_member(UUID2, 'pattieblack')
self.assertTrue(self.client.delete_member(UUID2, 'pattieblack'))
class TestBaseClient(testtools.TestCase):
"""
Test proper actions made for both valid and invalid requests
against a Registry service
"""
def test_connect_kwargs_default_values(self):
actual = test_client.BaseClient('127.0.0.1').get_connect_kwargs()
self.assertEqual({'timeout': None}, actual)
def test_connect_kwargs(self):
base_client = test_client.BaseClient(
host='127.0.0.1', port=80, timeout=1, use_ssl=True)
actual = base_client.get_connect_kwargs()
expected = {'insecure': False,
'key_file': None,
'cert_file': None,
'timeout': 1}
for k in expected.keys():
self.assertEqual(expected[k], actual[k])
class TestRegistryV1ClientApi(base.IsolatedUnitTest):
def setUp(self):
"""Establish a clean test environment."""
super(TestRegistryV1ClientApi, self).setUp()
self.context = context.RequestContext()
reload_module(rapi)
def test_get_registry_client(self):
actual_client = rapi.get_registry_client(self.context)
self.assertIsNone(actual_client.identity_headers)
def test_get_registry_client_with_identity_headers(self):
self.config(send_identity_headers=True)
expected_identity_headers = {
'X-User-Id': '',
'X-Tenant-Id': '',
'X-Roles': ','.join(self.context.roles),
'X-Identity-Status': 'Confirmed',
'X-Service-Catalog': 'null',
}
actual_client = rapi.get_registry_client(self.context)
self.assertEqual(expected_identity_headers,
actual_client.identity_headers)
def test_configure_registry_client_not_using_use_user_token(self):
self.config(use_user_token=False)
with patch.object(rapi, 'configure_registry_admin_creds') as mock_rapi:
rapi.configure_registry_client()
mock_rapi.assert_called_once_with()
def _get_fake_config_creds(self, auth_url='auth_url', strategy='keystone'):
return {
'user': 'user',
'password': 'password',
'username': 'user',
'tenant': 'tenant',
'auth_url': auth_url,
'strategy': strategy,
'region': 'region'
}
def test_configure_registry_admin_creds(self):
expected = self._get_fake_config_creds(auth_url=None,
strategy='configured_strategy')
self.config(admin_user=expected['user'])
self.config(admin_password=expected['password'])
self.config(admin_tenant_name=expected['tenant'])
self.config(auth_strategy=expected['strategy'])
self.config(auth_region=expected['region'])
self.stubs.Set(os, 'getenv', lambda x: None)
self.assertIsNone(rapi._CLIENT_CREDS)
rapi.configure_registry_admin_creds()
self.assertEqual(expected, rapi._CLIENT_CREDS)
def test_configure_registry_admin_creds_with_auth_url(self):
expected = self._get_fake_config_creds()
self.config(admin_user=expected['user'])
self.config(admin_password=expected['password'])
self.config(admin_tenant_name=expected['tenant'])
self.config(auth_url=expected['auth_url'])
self.config(auth_strategy='test_strategy')
self.config(auth_region=expected['region'])
self.assertIsNone(rapi._CLIENT_CREDS)
rapi.configure_registry_admin_creds()
self.assertEqual(expected, rapi._CLIENT_CREDS)
class FakeResponse(object):
status = http.ACCEPTED
def getheader(*args, **kwargs):
return None
class TestRegistryV1ClientRequests(base.IsolatedUnitTest):
def setUp(self):
super(TestRegistryV1ClientRequests, self).setUp()
def test_do_request_with_identity_headers(self):
identity_headers = {'foo': 'bar'}
self.client = rclient.RegistryClient("0.0.0.0",
identity_headers=identity_headers)
with patch.object(test_client.BaseClient, 'do_request',
return_value=FakeResponse()) as mock_do_request:
self.client.do_request("GET", "/images")
mock_do_request.assert_called_once_with("GET", "/images",
headers=identity_headers)
def test_do_request(self):
self.client = rclient.RegistryClient("0.0.0.0")
with patch.object(test_client.BaseClient, 'do_request',
return_value=FakeResponse()) as mock_do_request:
self.client.do_request("GET", "/images")
mock_do_request.assert_called_once_with("GET", "/images",
headers={})
def test_registry_invalid_token_exception_handling(self):
self.image_controller = acontroller()
request = webob.Request.blank('/images')
request.method = 'GET'
request.context = context.RequestContext()
with patch.object(rapi, 'get_images_detail') as mock_detail:
mock_detail.side_effect = exception.NotAuthenticated()
self.assertRaises(webob.exc.HTTPUnauthorized,
self.image_controller.detail, request)

345
glance/tests/unit/v1/test_upload_utils.py
View File

@ -1,345 +0,0 @@
# Copyright 2013 OpenStack Foundation
# All Rights Reserved.
#
# 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 contextlib import contextmanager
import glance_store
import mock
from mock import patch
import webob.exc
from glance.api.v1 import upload_utils
from glance.common import exception
from glance.common import store_utils
from glance.common import utils
import glance.registry.client.v1.api as registry
from glance.tests.unit import base
import glance.tests.unit.utils as unit_test_utils
class TestUploadUtils(base.StoreClearingUnitTest):
def setUp(self):
super(TestUploadUtils, self).setUp()
self.config(debug=True)
def test_initiate_delete(self):
req = unit_test_utils.get_fake_request()
location = {"url": "file://foo/bar",
"metadata": {},
"status": "active"}
id = unit_test_utils.UUID1
with patch.object(store_utils,
"safe_delete_from_backend") as mock_store_utils:
upload_utils.initiate_deletion(req, location, id)
mock_store_utils.assert_called_once_with(req.context,
id,
location)
def test_initiate_delete_with_delayed_delete(self):
self.config(delayed_delete=True)
req = unit_test_utils.get_fake_request()
location = {"url": "file://foo/bar",
"metadata": {},
"status": "active"}
id = unit_test_utils.UUID1
with patch.object(store_utils, "schedule_delayed_delete_from_backend",
return_value=True) as mock_store_utils:
upload_utils.initiate_deletion(req, location, id)
mock_store_utils.assert_called_once_with(req.context,
id,
location)
def test_safe_kill(self):
req = unit_test_utils.get_fake_request()
id = unit_test_utils.UUID1
with patch.object(registry, "update_image_metadata") as mock_registry:
upload_utils.safe_kill(req, id, 'saving')
mock_registry.assert_called_once_with(req.context, id,
{'status': 'killed'},
from_state='saving')
def test_safe_kill_with_error(self):
req = unit_test_utils.get_fake_request()
id = unit_test_utils.UUID1
with patch.object(registry, "update_image_metadata",
side_effect=Exception()) as mock_registry:
upload_utils.safe_kill(req, id, 'saving')
mock_registry.assert_called_once_with(req.context, id,
{'status': 'killed'},
from_state='saving')
@contextmanager
def _get_store_and_notifier(self, image_size=10, ext_update_data=None,
ret_checksum="checksum", exc_class=None):
location = "file://foo/bar"
checksum = "checksum"
size = 10
update_data = {'checksum': checksum}
if ext_update_data is not None:
update_data.update(ext_update_data)
image_meta = {'id': unit_test_utils.UUID1,
'size': image_size}
image_data = "blah"
store = mock.MagicMock()
notifier = mock.MagicMock()
if exc_class is not None:
store.add.side_effect = exc_class
else:
store.add.return_value = (location, size, ret_checksum, {})
yield (location, checksum, image_meta, image_data, store, notifier,
update_data)
def test_upload_data_to_store(self):
# 'user_storage_quota' is not set
def store_add(image_id, data, size, **kwargs):
# Check if 'data' is instance of 'CooperativeReader' when
# 'user_storage_quota' is disabled.
self.assertIsInstance(data, utils.CooperativeReader)
return location, 10, "checksum", {}
req = unit_test_utils.get_fake_request()
with self._get_store_and_notifier(
ext_update_data={'size': 10},
exc_class=store_add) as (location, checksum, image_meta,
image_data, store, notifier,
update_data):
ret = image_meta.update(update_data)
with patch.object(registry, 'update_image_metadata',
return_value=ret) as mock_update_image_metadata:
actual_meta, location_data = upload_utils.upload_data_to_store(
req, image_meta, image_data, store, notifier)
self.assertEqual(location, location_data['url'])
self.assertEqual(image_meta.update(update_data), actual_meta)
mock_update_image_metadata.assert_called_once_with(
req.context, image_meta['id'], update_data,
from_state='saving')
def test_upload_data_to_store_user_storage_quota_enabled(self):
# Enable user_storage_quota
self.config(user_storage_quota='100B')
def store_add(image_id, data, size, **kwargs):
# Check if 'data' is instance of 'LimitingReader' when
# 'user_storage_quota' is enabled.
self.assertIsInstance(data, utils.LimitingReader)
return location, 10, "checksum", {}
req = unit_test_utils.get_fake_request()
with self._get_store_and_notifier(
ext_update_data={'size': 10},
exc_class=store_add) as (location, checksum, image_meta,
image_data, store, notifier,
update_data):
ret = image_meta.update(update_data)
# mock 'check_quota'
mock_check_quota = patch('glance.api.common.check_quota',
return_value=100)
mock_check_quota.start()
self.addCleanup(mock_check_quota.stop)
with patch.object(registry, 'update_image_metadata',
return_value=ret) as mock_update_image_metadata:
actual_meta, location_data = upload_utils.upload_data_to_store(
req, image_meta, image_data, store, notifier)
self.assertEqual(location, location_data['url'])
self.assertEqual(image_meta.update(update_data), actual_meta)
mock_update_image_metadata.assert_called_once_with(
req.context, image_meta['id'], update_data,
from_state='saving')
# 'check_quota' is called two times
check_quota_call_count = (
mock_check_quota.target.check_quota.call_count)
self.assertEqual(2, check_quota_call_count)
def test_upload_data_to_store_mismatch_size(self):
req = unit_test_utils.get_fake_request()
with self._get_store_and_notifier(
image_size=11) as (location, checksum, image_meta, image_data,
store, notifier, update_data):
ret = image_meta.update(update_data)
with patch.object(registry, 'update_image_metadata',
return_value=ret) as mock_update_image_metadata:
self.assertRaises(webob.exc.HTTPBadRequest,
upload_utils.upload_data_to_store,
req, image_meta, image_data, store,
notifier)
mock_update_image_metadata.assert_called_with(
req.context, image_meta['id'], {'status': 'killed'},
from_state='saving')
def test_upload_data_to_store_mismatch_checksum(self):
req = unit_test_utils.get_fake_request()
with self._get_store_and_notifier(
ret_checksum='fake') as (location, checksum, image_meta,
image_data, store, notifier, update_data):
ret = image_meta.update(update_data)
with patch.object(registry, "update_image_metadata",
return_value=ret) as mock_update_image_metadata:
self.assertRaises(webob.exc.HTTPBadRequest,
upload_utils.upload_data_to_store,
req, image_meta, image_data, store,
notifier)
mock_update_image_metadata.assert_called_with(
req.context, image_meta['id'], {'status': 'killed'},
from_state='saving')
def _test_upload_data_to_store_exception(self, exc_class, expected_class):
req = unit_test_utils.get_fake_request()
with self._get_store_and_notifier(
exc_class=exc_class) as (location, checksum, image_meta,
image_data, store, notifier, update_data):
with patch.object(upload_utils, 'safe_kill') as mock_safe_kill:
self.assertRaises(expected_class,
upload_utils.upload_data_to_store,
req, image_meta, image_data, store, notifier)
mock_safe_kill.assert_called_once_with(
req, image_meta['id'], 'saving')
def _test_upload_data_to_store_exception_with_notify(self,
exc_class,
expected_class,
image_killed=True):
req = unit_test_utils.get_fake_request()
with self._get_store_and_notifier(
exc_class=exc_class) as (location, checksum, image_meta,
image_data, store, notifier, update_data):
with patch.object(upload_utils, 'safe_kill') as mock_safe_kill:
self.assertRaises(expected_class,
upload_utils.upload_data_to_store,
req, image_meta, image_data, store,
notifier)
if image_killed:
mock_safe_kill.assert_called_with(req, image_meta['id'],
'saving')
def test_upload_data_to_store_raises_store_disabled(self):
"""Test StoreDisabled exception is raised while uploading data"""
self._test_upload_data_to_store_exception_with_notify(
glance_store.StoreAddDisabled,
webob.exc.HTTPGone,
image_killed=True)
def test_upload_data_to_store_duplicate(self):
"""See note in glance.api.v1.upload_utils on why we don't want image to
be deleted in this case.
"""
self._test_upload_data_to_store_exception_with_notify(
exception.Duplicate,
webob.exc.HTTPConflict,
image_killed=False)
def test_upload_data_to_store_forbidden(self):
self._test_upload_data_to_store_exception_with_notify(
exception.Forbidden,
webob.exc.HTTPForbidden)
def test_upload_data_to_store_storage_full(self):
self._test_upload_data_to_store_exception_with_notify(
glance_store.StorageFull,
webob.exc.HTTPRequestEntityTooLarge)
def test_upload_data_to_store_storage_write_denied(self):
self._test_upload_data_to_store_exception_with_notify(
glance_store.StorageWriteDenied,
webob.exc.HTTPServiceUnavailable)
def test_upload_data_to_store_size_limit_exceeded(self):
self._test_upload_data_to_store_exception_with_notify(
exception.ImageSizeLimitExceeded,
webob.exc.HTTPRequestEntityTooLarge)
def test_upload_data_to_store_http_error(self):
self._test_upload_data_to_store_exception_with_notify(
webob.exc.HTTPError,
webob.exc.HTTPError)
def test_upload_data_to_store_client_disconnect(self):
self._test_upload_data_to_store_exception(
ValueError,
webob.exc.HTTPBadRequest)
def test_upload_data_to_store_client_disconnect_ioerror(self):
self._test_upload_data_to_store_exception(
IOError,
webob.exc.HTTPBadRequest)
def test_upload_data_to_store_exception(self):
self._test_upload_data_to_store_exception_with_notify(
Exception,
webob.exc.HTTPInternalServerError)
def test_upload_data_to_store_not_found_after_upload(self):
req = unit_test_utils.get_fake_request()
with self._get_store_and_notifier(
ext_update_data={'size': 10}) as (location, checksum, image_meta,
image_data, store, notifier,
update_data):
exc = exception.ImageNotFound
with patch.object(registry, 'update_image_metadata',
side_effect=exc) as mock_update_image_metadata:
with patch.object(upload_utils,
"initiate_deletion") as mock_initiate_del:
with patch.object(upload_utils,
"safe_kill") as mock_safe_kill:
self.assertRaises(webob.exc.HTTPPreconditionFailed,
upload_utils.upload_data_to_store,
req, image_meta, image_data, store,
notifier)
mock_update_image_metadata.assert_called_once_with(
req.context, image_meta['id'], update_data,
from_state='saving')
mock_initiate_del.assert_called_once_with(
req, {'url': location, 'status': 'active',
'metadata': {}}, image_meta['id'])
mock_safe_kill.assert_called_once_with(
req, image_meta['id'], 'saving')
@mock.patch.object(registry, 'update_image_metadata',
side_effect=exception.NotAuthenticated)
@mock.patch.object(upload_utils, 'initiate_deletion')
def test_activate_image_with_expired_token(
self, mocked_delete, mocked_update):
"""Test token expiration during image upload.
If users token expired before image was uploaded then if auth error
was caught from registry during changing image status from 'saving'
to 'active' then it's required to delete all image data.
"""
context = mock.Mock()
req = mock.Mock()
req.context = context
with self._get_store_and_notifier() as (location, checksum, image_meta,
image_data, store, notifier,
update_data):
self.assertRaises(webob.exc.HTTPUnauthorized,
upload_utils.upload_data_to_store,
req, image_meta, image_data, store, notifier)
self.assertEqual(2, mocked_update.call_count)
mocked_delete.assert_called_once_with(
req,
{'url': 'file://foo/bar', 'status': 'active', 'metadata': {}},
'c80a1a6c-bd1f-41c5-90ee-81afedb1d58d')