stress_test_osg_basic.py

#!/usr/bin/python

import time
from eucaops import Eucaops
from eutester.eutestcase import EutesterTestCase
from boto.s3.key import Key
import os

class OSGBasics(EutesterTestCase):
    def __init__(self):
        #### Pre-conditions
        self.setuptestcase()
        self.setup_parser()
        self.get_args()
        # Setup basic eutester object
        self.tester = Eucaops( credpath=self.args.credpath)
        user_account = "osgstress1"
        user_name = "admin"
        access_key = "AKI0UCUKYD73XIU9ZAY9"
        secret_key = "m0WQz2MIuo7hB5Uau0JGJL2LBbfhHhEt5AtKlIvh"
        self.osg_tester = Eucaops(aws_access_key_id=access_key,
                                        aws_secret_access_key=secret_key,
                                        ec2_ip=self.tester.ec2.host, s3_ip=self.tester.s3.host,
                                        s3_path=self.tester.get_s3_path(),
                                        username=user_name, account=user_account)
        
        

    def clean_method(self):
        """
        Description: Attempts to clean up resources created in this test
        """
        self.tester.cleanup_artifacts()

    def MyTest(self):
        """
        Description: This test will simply run an instance and check that it is reachable via ssh.
        """
        max_keys = 500
        test_bucket = "stress-bucket-" + str(int(time.time()))
        self.osg_tester.create_bucket(test_bucket)
        bucket = self.osg_tester.get_bucket_by_name(test_bucket)
        for i in range(max_keys):
            k = Key(bucket)
            k.key = test_bucket + "_object" + str(i)
            k.set_contents_from_string('This is a test of S3' + str(i))
        keys = bucket.get_all_keys(max_keys=max_keys)
        while keys:
            with ThreadPoolExecutor(max_workers=50) as executor:
                print "Number of keys to delete: " + str(len(keys))
                sys.stdout.flush()
                for key in keys:
                    executor.submit(key.delete)
                if len(keys) == max_keys:
                    keys = bucket.get_all_keys(max_keys=max_keys,marker=keys.next_marker)
                else:
                    keys = []
        bucket.delete()
        print "Bucket purged"
     
if __name__ == "__main__":
    testcase = OSGBasics()
    ### Use the list of tests passed from config/command line to determine what subset of tests to run
    ### or use a predefined list
    list = testcase.args.tests or [ "MyTest"]

    ### Convert test suite methods to EutesterUnitTest objects
    unit_list = [ ]
    for test in list:
        unit_list.append( testcase.create_testunit_by_name(test) )

    ### Run the EutesterUnitTest objects
    result = testcase.run_test_case_list(unit_list,clean_on_exit=True)
    exit(result)