| import httplib2 |
| import json |
| import os |
| import re |
| |
| from kong import openstack |
| from kong import tests |
| |
| |
| class TestImagesThroughCompute(tests.FunctionalTest): |
| |
| def setUp(self): |
| super(TestImagesThroughCompute, self).setUp() |
| self.os = openstack.Manager(self.nova) |
| |
| def _assert_image_links(self, image): |
| image_id = str(image['id']) |
| |
| mgmt_url = self.os.nova.management_url |
| mgmt_url = re.sub(r'1//', r'1/', mgmt_url) # TODO: is this a bug in Nova? |
| bmk_url = re.sub(r'v1.1\/', r'', mgmt_url) |
| self_link = {'rel': 'self', |
| 'href': os.path.join(mgmt_url, 'images', image_id)} |
| bookmark_link = {'rel': 'bookmark', |
| 'href': os.path.join(bmk_url, 'images', image_id)} |
| |
| self.assertIn(bookmark_link, image['links']) |
| self.assertIn(self_link, image['links']) |
| |
| def _assert_image_entity_basic(self, image): |
| actual_keys = set(image.keys()) |
| expected_keys = set(( |
| 'id', |
| 'name', |
| 'links', |
| )) |
| self.assertEqual(actual_keys, expected_keys) |
| |
| self._assert_image_links(image) |
| |
| def _assert_image_entity_detailed(self, image): |
| keys = image.keys() |
| if 'server' in keys: |
| keys.remove('server') |
| actual_keys = set(keys) |
| expected_keys = set(( |
| 'created', |
| 'id', |
| 'links', |
| 'metadata', |
| 'minDisk', |
| 'minRam', |
| 'name', |
| 'progress', |
| 'status', |
| 'updated', |
| )) |
| self.assertEqual(actual_keys, expected_keys) |
| |
| self._assert_image_links(image) |
| |
| def test_index(self): |
| """List all images""" |
| |
| response, body = self.os.nova.request('GET', '/images') |
| |
| self.assertEqual(response['status'], '200') |
| resp_body = json.loads(body) |
| self.assertEqual(resp_body.keys(), ['images']) |
| |
| for image in resp_body['images']: |
| self._assert_image_entity_basic(image) |
| test_index.tags = ['nova', 'glance'] |
| |
| def test_detail(self): |
| """List all images in detail""" |
| |
| response, body = self.os.nova.request('GET', '/images/detail') |
| self.assertEqual(response['status'], '200') |
| resp_body = json.loads(body) |
| self.assertEqual(resp_body.keys(), ['images']) |
| |
| for image in resp_body['images']: |
| self._assert_image_entity_detailed(image) |
| test_detail.tags = ['nova', 'glance'] |
| |
| |
| class TestGlanceAPI(tests.FunctionalTest): |
| |
| def setUp(self): |
| super(TestGlanceAPI, self).setUp() |
| self.base_url = "http://%s:%s/%s/images" % (self.glance['host'], |
| self.glance['port'], |
| self.glance['apiver']) |
| |
| def test_upload_ami_style_image(self): |
| """Uploads a three-part ami-style image""" |
| aki_location = self.config['environment']['aki_location'] |
| headers = {'x-image-meta-is-public': 'true', |
| 'x-image-meta-name': 'test-kernel', |
| 'x-image-meta-disk-format': 'aki', |
| 'x-image-meta-container-format': 'aki', |
| 'Content-Length': '%d' % os.path.getsize(aki_location), |
| 'Content-Type': 'application/octet-stream'} |
| image_file = open(aki_location, "rb") |
| http = httplib2.Http() |
| response, content = http.request(self.base_url, 'POST', |
| headers=headers,body=image_file) |
| image_file.close() |
| self.assertEqual(201, response.status) |
| data = json.loads(content) |
| self.assertEqual(data['image']['name'], "test-kernel") |
| self.assertEqual(data['image']['checksum'], |
| self._md5sum_file(aki_location)) |
| kernel_id = data['image']['id'] |
| |
| ari_location = self.config['environment'].get('ari_location') |
| if ari_location: |
| headers = {'x-image-meta-is-public': 'true', |
| 'x-image-meta-name': 'test-ramdisk', |
| 'x-image-meta-disk-format': 'ari', |
| 'x-image-meta-container-format': 'ari', |
| 'Content-Length': '%d' % os.path.getsize(ari_location), |
| 'Content-Type': 'application/octet-stream'} |
| image_file = open(ari_location, "rb") |
| http = httplib2.Http() |
| response, content = http.request(self.base_url, 'POST', |
| headers=headers, body=image_file) |
| image_file.close() |
| self.assertEqual(201, response.status) |
| data = json.loads(content) |
| self.assertEqual(data['image']['name'], "test-ramdisk") |
| self.assertEqual(data['image']['checksum'], |
| self._md5sum_file(ari_location)) |
| ramdisk_id = data['image']['id'] |
| else: |
| ramdisk_id = None |
| |
| ami_location = self.config['environment']['ami_location'] |
| upload_data = "" |
| for chunk in self._read_in_chunks(ami_location): |
| upload_data += chunk |
| headers = {'x-image-meta-is-public': 'true', |
| 'x-image-meta-name': 'test-image', |
| 'x-image-meta-disk-format': 'ami', |
| 'x-image-meta-container-format': 'ami', |
| 'x-image-meta-property-kernel_id': kernel_id, |
| 'Content-Length': '%d' % os.path.getsize(ami_location), |
| 'Content-Type': 'application/octet-stream'} |
| |
| if ari_location: |
| headers['x-image-meta-property-ramdisk_id'] = ramdisk_id |
| |
| http = httplib2.Http() |
| response, content = http.request(self.base_url, 'POST', |
| headers=headers, body=upload_data) |
| self.assertEqual(201, response.status) |
| data = json.loads(content) |
| self.assertEqual(data['image']['name'], "test-image") |
| self.assertEqual(data['image']['checksum'], |
| self._md5sum_file(ami_location)) |
| machine_id = data['image']['id'] |
| |
| # now ensure we can modify the image properties |
| headers = {'X-Image-Meta-Property-distro': 'Ubuntu', |
| 'X-Image-Meta-Property-arch': 'x86_64', |
| 'X-Image-Meta-Property-kernel_id': kernel_id} |
| if ari_location: |
| headers['X-Image-Meta-Property-ramdisk_id'] = ramdisk_id |
| |
| http = httplib2.Http() |
| url = '%s/%s' % (self.base_url, machine_id) |
| response, content = http.request(url, 'PUT', headers=headers) |
| self.assertEqual(response.status, 200) |
| data = json.loads(content) |
| properties = data['image']['properties'] |
| self.assertEqual(properties['arch'], "x86_64") |
| self.assertEqual(properties['distro'], "Ubuntu") |
| self.assertEqual(properties['kernel_id'], kernel_id) |
| if ari_location: |
| self.assertEqual(properties['ramdisk_id'], ramdisk_id) |
| |
| # list the metadata to ensure the new values stuck |
| http = httplib2.Http() |
| response, content = http.request(url, 'HEAD') |
| self.assertEqual(response.status, 200) |
| self.assertEqual(response['x-image-meta-name'], "test-image") |
| self.assertEqual(response['x-image-meta-checksum'], |
| self._md5sum_file(ami_location)) |
| self.assertEqual(response['x-image-meta-container_format'], "ami") |
| self.assertEqual(response['x-image-meta-disk_format'], "ami") |
| self.assertEqual(response['x-image-meta-property-arch'], "x86_64") |
| self.assertEqual(response['x-image-meta-property-distro'], "Ubuntu") |
| self.assertEqual(response['x-image-meta-property-kernel_id'], |
| kernel_id) |
| if ari_location: |
| self.assertEqual(response['x-image-meta-property-ramdisk_id'], |
| ramdisk_id) |
| |
| # delete images for which we have non-None ids |
| delete_ids = filter(lambda x: x, (kernel_id, ramdisk_id, machine_id)) |
| for image_id in delete_ids: |
| http = httplib2.Http() |
| url = '%s/%s' % (self.base_url, image_id) |
| response, content = http.request(url, 'DELETE') |
| |
| test_upload_ami_style_image.tags = ['glance'] |