create functional test for register_image

Change-Id: Ie5407ced8bdce6c0e0fd7565cc26a6976c067efe
This commit is contained in:
Andrey Pavlov 2015-08-07 13:39:35 +03:00
parent 6a8ee565ce
commit b385f567e4
6 changed files with 119 additions and 3 deletions

2
.gitignore vendored
View File

@ -5,6 +5,6 @@
ec2_api.egg-info
.tox
.testrepository
/functional_tests.conf
/functional_tests.conf*
/buckets
.venv

View File

@ -143,9 +143,20 @@ if [[ ! -f $TEST_CONFIG_DIR/$TEST_CONFIG ]]; then
run_long_tests="False"
fi
# TODO(andey-mp): make own code
# copy some variables from tempest.conf
tempest_conf="../tempest/etc/tempest.conf"
if [[ -f $tempest_conf ]]; then
s3_materials_path=`grep ^s3_materials_path $tempest_conf | awk '{split($0,a,"="); print a[2]}'`
aki_manifest=`grep ^aki_manifest $tempest_conf | awk '{split($0,a,"="); print a[2]}'`
ami_manifest=`grep ^ami_manifest $tempest_conf | awk '{split($0,a,"="); print a[2]}'`
ari_manifest=`grep ^ari_manifest $tempest_conf | awk '{split($0,a,"="); print a[2]}'`
fi
sudo bash -c "cat > $TEST_CONFIG_DIR/$TEST_CONFIG <<EOF
[aws]
ec2_url = $EC2_URL
s3_url = $S3_URL
aws_access = $EC2_ACCESS_KEY
aws_secret = $EC2_SECRET_KEY
image_id = $image_id
@ -155,6 +166,10 @@ build_timeout = $timeout
run_long_tests = $run_long_tests
instance_type = m1.ec2api
instance_type_alt = m1.ec2api-alt
s3_materials_path=$s3_materials_path
aki_manifest=$aki_manifest
ami_manifest=$ami_manifest
ari_manifest=$ari_manifest
EOF"
fi

View File

@ -13,6 +13,9 @@
# License for the specific language governing permissions and limitations
# under the License.
import os
import re
from tempest_lib.common.utils import data_utils
import testtools
@ -279,3 +282,75 @@ class ImageTest(base.EC2TestCase):
data = self.client.deregister_image(ImageId=image_id)
self.cancelResourceCleanUp(image_clean)
class ImageRegisterTest(base.EC2TestCase):
@classmethod
@base.safe_setup
def setUpClass(cls):
super(ImageRegisterTest, cls).setUpClass()
cls.materials_path = CONF.aws.s3_materials_path
if not cls.materials_path:
raise cls.skipException('Image materials are not ready')
cls.bucket_name = data_utils.rand_name("bucket")
cls.ami_manifest = CONF.aws.ami_manifest
cls.aki_manifest = CONF.aws.aki_manifest
cls.ari_manifest = CONF.aws.ari_manifest
cls.ami_path = cls.materials_path + os.sep + cls.ami_manifest
cls.aki_path = cls.materials_path + os.sep + cls.aki_manifest
cls.ari_path = cls.materials_path + os.sep + cls.ari_manifest
cls.s3_client.create_bucket(Bucket=cls.bucket_name, ACL='public-read')
cls.addResourceCleanUpStatic(cls.s3_client.delete_bucket,
Bucket=cls.bucket_name)
cls._s3_upload_dir(cls.bucket_name, cls.materials_path)
@classmethod
def _s3_upload_dir(cls, bucket_name, path, prefix=""):
for root, _, files in os.walk(path):
for file in files:
source = root + os.sep + file
target = re.sub("^" + re.escape(path) + "?/", prefix, source)
if os.sep != '/':
target = re.sub(re.escape(os.sep), '/', target)
hfile = open(source, 'r')
cls.s3_client.put_object(Bucket=bucket_name, Key=target,
Body=hfile)
valid_image_state = set(('available', 'pending', 'failed'))
def _register_get_deregister_image(self, image_type, manifest):
image_name = data_utils.rand_name(image_type + "-name")
image_location = self.bucket_name + "/" + manifest
data = self.client.register_image(
Name=image_name, ImageLocation=image_location)
image_id = data['ImageId']
image_clean = self.addResourceCleanUp(self.client.deregister_image,
ImageId=image_id)
self.assertEqual(image_id[0:3], image_type)
data = self.client.describe_images(ImageIds=[image_id])
self.assertEqual(1, len(data['Images']))
image = data['Images'][0]
self.assertEqual(image_name, image['Name'])
self.assertEqual(image_id, image['ImageId'])
self.assertIn(image['State'], self.valid_image_state)
self.get_image_waiter().wait_available(image_id)
self.client.deregister_image(ImageId=image_id)
self.assertRaises('InvalidAMIID.NotFound',
self.client.describe_images,
ImageIds=[image_id])
self.cancelResourceCleanUp(image_clean)
def test_register_get_deregister_ami_image(self):
self._register_get_deregister_image('ami', self.ami_manifest)
def test_register_get_deregister_aki_image(self):
self._register_get_deregister_image('aki', self.aki_manifest)
def test_register_get_deregister_ari_image(self):
self._register_get_deregister_image('ari', self.ari_manifest)

View File

@ -218,6 +218,9 @@ class EC2TestCase(base.BaseTestCase):
cls.client = botocoreclient._get_ec2_client(
CONF.aws.ec2_url, CONF.aws.aws_region,
CONF.aws.aws_access, CONF.aws.aws_secret)
cls.s3_client = botocoreclient._get_s3_client(
CONF.aws.s3_url, CONF.aws.aws_region,
CONF.aws.aws_access, CONF.aws.aws_secret)
TesterStateHolder().ec2_client = cls.client
@classmethod

View File

@ -16,12 +16,20 @@
import botocore.session
def _get_ec2_client(url, region, access, secret):
def _get_client(client_name, url, region, access, secret):
connection_data = {
'config_file': (None, 'AWS_CONFIG_FILE', None, None),
'region': ('region', 'BOTO_DEFAULT_REGION', region, None),
}
session = botocore.session.get_session(connection_data)
return session.create_client(
'ec2', region_name=region, endpoint_url=url,
client_name, region_name=region, endpoint_url=url,
aws_access_key_id=access, aws_secret_access_key=secret)
def _get_ec2_client(url, region, access, secret):
return _get_client('ec2', url, region, access, secret)
def _get_s3_client(url, region, access, secret):
return _get_client('s3', url, region, access, secret)

View File

@ -34,6 +34,9 @@ AWSGroup = [
cfg.StrOpt('ec2_url',
default="http://localhost:8788/",
help="EC2 URL"),
cfg.StrOpt('s3_url',
default="http://localhost:3334/",
help="S3 URL"),
cfg.StrOpt('aws_secret',
default=None,
help="AWS Secret Key",
@ -80,6 +83,18 @@ AWSGroup = [
cfg.BoolOpt('run_long_tests',
default=False,
help='Will run all long tests also.'),
cfg.StrOpt('s3_materials_path',
default=None,
help="S3 Materials Path"),
cfg.StrOpt('ari_manifest',
default=None,
help="ARI Ramdisk Image manifest"),
cfg.StrOpt('ami_manifest',
default=None,
help="AMI Machine Image manifest"),
cfg.StrOpt('aki_manifest',
default=None,
help="AKI Kernel Image manifest"),
]