Source code for pOCCI.CORE

import re
import sys
import time
import urlparse

from occi_libs import *
import transport
import occi


renderer = None
renderer_big = None
renderer_httpheaders = None
connection = None

#'Category: offline;scheme="http://schemas.ogf.org/occi/infrastructure/storagelink/action#";class="action";title="deactivate storagelink"'
required_categories = [
    occi.Category({'scheme': 'http://schemas.ogf.org/occi/core#', 'term': 'entity'}),
    occi.Category({'scheme': 'http://schemas.ogf.org/occi/core#', 'term': 'resource'}),
    occi.Category({'scheme': 'http://schemas.ogf.org/occi/core#', 'term': 'link'}),
]

example_attributes = {
    'occi.core.id': occi.Attribute({'name': 'occi.core.id', 'value': '1'}),
    'occi.core.title': occi.Attribute({'name': 'occi.core.title', 'value': 'Test_title_%d' % time.time()}),
    'occi.storage.size': occi.Attribute({'name': 'occi.storage.size', 'type': 'number', 'value': 0.1}),
    'occi.storage.state': occi.Attribute({'name': 'occi.storage.state', 'type': 'enum', 'value': "inactive"}),
    'occi.storagelink.deviceid': occi.Attribute({'name': 'occi.storagelink.deviceid', 'type': 'string', 'value': "/dev/blk0"})
}


[docs]class Test: """Base class for OCCI compliance tests""" objective = None categories = [] @classmethod
[docs] def test(self=None): return [False, ['Not implemented']]
@classmethod
[docs] def fetch_categories(self, err_msg): categories = [] check_parse = True body, response_headers, http_status, content_type = connection.get(mimetype=occi_config['mimetype.big']) if body is not None: try: categories = renderer_big.parse_categories(body, response_headers) except occi.ParseError as pe: check_parse = False err_msg.append(str(pe)) if occi_config['curlverbose']: print '==== CATEGORIES ====' if check_parse: for category in categories: print category else: print 'ERROR: %s' % err_msg[-1] print '====================' # for OpenStack # TODO: needed? for category in categories: if 'location' not in category: category['location'] = '/' + category['term'] + '/' Test.categories = categories return [check_parse, body, response_headers, http_status, content_type]
@classmethod
[docs] def get_category(self, uri): for cat in Test.categories: if cat['scheme'] + cat['term'] == uri: return cat return None
@classmethod
[docs] def clear_categories(self): Test.categories = []
@classmethod
[docs] def search_category(self, filter): for cat in Test.categories: if match_category(cat, filter): return cat return None
@classmethod
[docs] def pretest_http_status(self, http_ok_status, err_msg, force=False): check_pretest = True check_categories = True body = None response_headers = None http_status = None content_type = None if not Test.categories or force: check_categories, body, response_headers, http_status, content_type = Test.fetch_categories(err_msg) check_pretest, tmp_err_msg = check_http_status(http_ok_status, http_status) err_msg += tmp_err_msg return [body, response_headers, http_status, content_type, check_pretest and check_categories]
[docs]def testsuite_init(): """Initialize OCCI testsuite Renderers from occi_libs needs to be initialized first. """ self = sys.modules[__name__] if 'pOCCI.occi_libs' in sys.modules: occi_libs = sys.modules['pOCCI.occi_libs'] else: occi_libs = sys.modules['occi_libs'] self.renderer = occi_libs.renderer self.renderer_big = occi_libs.renderer_big self.renderer_httpheaders = occi_libs.renderer_httpheaders if not self.renderer: print >> sys.stderr, 'No renderer (invalid mimetype?)' sys.exit(2) self.connection = transport.Transport(occi_config)
[docs]def check_content_type(content_type): if content_type in ['text/occi', 'text/plain', 'application/occi+json']: return [True, []] else: return [False, ['Wrong Content-Type in response']]
[docs]def check_requested_content_type(content_type, big=False, headers=False): if big: requested = occi_config['mimetype.big'] elif headers: requested = 'text/occi' else: requested = occi_config['mimetype'] if content_type == requested: return [True, []] else: return [False, ['Result mime type differs']]
[docs]def check_http_status(http_expected_status, http_status): if not re.match(r'^HTTP/.* %s' % http_expected_status, http_status): return [False, ['HTTP status is not %s (%s)' % (http_expected_status, http_status)]] else: return [True, []]
[docs]def match_category(category, filter): for key, value in filter.items(): if not (key in category and category[key] == value): return False return True
[docs]def match_entity(attributes, filter): adict = {} for a in attributes: adict[a['name']] = a for key, value in filter.items(): a2 = occi.Attribute({'value': value}) if key not in adict or not occi.Attribute.equals(adict[key], a2): #print '[match_entity] bad key %s' % key return False return True
[docs]def check_body_entities(body, headers, err_msg=[]): entities = [] try: entities = renderer_httpheaders.parse_locations(body, headers) except occi.ParseError as pe: err_msg.append(str(pe)) return None if not entities: try: entities = renderer.parse_locations(body, headers) except occi.ParseError as pe: err_msg.append(str(pe)) err_msg.append('HTTP Body doesn\'t contain the OCCI Compute Resource description') return None return entities
[docs]def gen_id(prefix): return '%s_%d' % (prefix, time.time())
[docs]class CORE_DISCOVERY001(Test): objective = 'Retrieving all OCCI Categories supported by the OCCI Server' @classmethod
[docs] def test(self=None): check_cat = False err_msg = [] body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg, force=True) check_ct, tmp_err_msg = check_content_type(content_type) err_msg += tmp_err_msg check_rct, tmp_err_msg = check_requested_content_type(content_type, big=True) err_msg += tmp_err_msg count = 0 for category in required_categories: if Test.search_category(category) is not None: count += 1 if count == len(required_categories): check_cat = True else: err_msg.append('Body doesn\'t contain appropriate categories') return [check_pretest and check_ct and check_rct and check_cat, err_msg]
[docs]class CORE_DISCOVERY002(Test): objective = 'Retrieving the OCCI Categories with an OCCI Category filter from the OCCI Server' @classmethod
[docs] def test(self=None): err_msg = [] filtered_categories = [] check = True body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) if not Test.categories: err_msg += ['No categories returned'] return [False, err_msg] filter = occi.Category({ 'term': Test.categories[0]['term'], 'scheme': Test.categories[0]['scheme'], 'class': Test.categories[0]['class'], }) cat_in = [] cat_in.append('Content-Type: text/occi') cat_in += renderer_httpheaders.render_category(filter)[1] body, response_headers, http_status, content_type = connection.get(headers=cat_in) check_ct, err_msg = check_content_type(content_type) if not re.match(r'^HTTP/.* 200 OK', http_status): check = False err_msg.append('HTTP status on filter not 200 OK (%s)' % http_status) check_rct, tmp_err_msg = check_requested_content_type(content_type) err_msg += tmp_err_msg try: filtered_categories = renderer.parse_categories(body, response_headers) except occi.ParseError as pe: check = False err_msg.append(str(pe)) check_filter = False for cat in filtered_categories: if match_category(cat, filter): check_filter = True break if not check_filter: err_msg.append('Category "%s" (scheme "%s") not in filtered result' % (filter['term'], filter['scheme'])) return [check and check_pretest and check_ct and check_rct and check_filter, err_msg]
[docs]class CORE_CREATE001(Test): """Create an OCCI Resource Unsupported test: Creating compute instances without os_tpl is not supported. It can be called by:: pOCCI -t 'CORE/CREATE/001' """ objective = 'Create an OCCI Resource' @classmethod
[docs] def test(self=None): err_msg = [] body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) if not check_pretest: return [False, err_msg] #kind = Test.search_category({'class': 'kind'}) kind = Test.search_category({'class': 'kind', 'term': 'compute'}) #print kind if kind: for item in ['location', 'term', 'scheme']: if item not in kind.keys(): err_msg.append('No %s in OCCI Kind' % item) return [False, err_msg] else: err_msg.append('No OCCI Kind found') return [False, err_msg] new_cat_s, new_cat_h = renderer.render_resource( categories=[ occi.Category({ 'term': 'compute', 'scheme': 'http://schemas.ogf.org/occi/infrastructure#', 'class': 'kind', 'title': 'titulek', }), ], attributes=[ occi.Attribute({ 'name': 'occi.core.id', 'value': gen_id('Compute'), }), occi.Attribute({ 'name': 'occi.core.title', 'value': 'titulek', }), occi.Attribute({ 'name': 'occi.core.summary', 'value': 'sumarko', }), occi.Attribute({ 'name': 'occi.compute.architecture', 'value': 'arch', }), ] ) body, response_headers, http_status, content_type = connection.post(url=occi_config['url'] + kind['location'], headers=['Content-Type: %s' % occi_config['mimetype']] + new_cat_h, body=new_cat_s) Test.clear_categories() check_create, tmp_err_msg = check_http_status("201 Created", http_status) err_msg += tmp_err_msg if not check_create: print >> sys.stderr, body return [check_create, err_msg]
[docs]class CORE_CREATE006(Test): """Add an OCCI Mixin definition Unsupported test: Not implemented. It can be called by:: pOCCI -t 'CORE/CREATE/006' """ objective = 'Add an OCCI mixin definition' @classmethod
[docs] def test(self=None): err_msg = [] new_mixin = 'Category: stufik; scheme="http://example.com/occi/my_stuff#"; class="mixin"; location="/mixin/resource_tpl/extra_large/"; rel="http://schemas.ogf.org/occi/infrastructure#resource_tpl"' body, response_headers, http_status, content_type = connection.post(url=occi_config['url'] + '/-/', headers=['Content-Type: text/plain'], body=new_mixin) check_create, tmp_err_msg = check_http_status("200 OK", http_status) err_msg += tmp_err_msg if not check_create: print >> sys.stderr, body return [check_create, err_msg]
[docs]def CORE_READ_URL(filter): err_msg = [] check_200ok = False check = True body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) if content_type: check, tmp_err_msg = check_content_type(content_type) err_msg += tmp_err_msg if not check_pretest: return [False, err_msg, None] mixin = Test.search_category(filter) #kind = Test.search_category({'class': 'kind'}) if not mixin: err_msg.append('No required OCCI Category found') return [False, err_msg, None] for category in [mixin]: body, response_headers, http_status, content_type = connection.get(url=occi_config['url'] + category['location']) try: locations = renderer.parse_locations(body, response_headers) except occi.ParseError as pe: locations = [] check = False err_msg.append(str(pe)) if re.match(r'^HTTP/.* 200 OK', http_status): check_200ok = True else: err_msg.append('Returned HTTP status is not 200 OK (%s)' % http_status) return [check_pretest and check and check_200ok, err_msg, locations]
[docs]class CORE_READ001(Test): objective = 'Retrieve the URLs of all OCCI Entities belonging to an OCCI Kind or OCCI Mixin' @classmethod
[docs] def test(self=None): check, err_msg, urls = CORE_READ_URL(occi_config['occi.tests.category']) return [check, err_msg]
[docs]def CORE_READ002_COMMON(category, links=[]): check_url = True check_200ok = False err_msg = [] headers = [] headers.append('Content-Type: text/occi') headers += renderer_httpheaders.render_category(occi.Category({'term': category['term'], 'scheme': category['scheme'], 'class': category['class']}))[1] body, response_headers, http_status, content_type = connection.get(url=occi_config['url'] + category['location'], headers=headers) try: locations = renderer.parse_locations(body, response_headers) except occi.ParseError as pe: locations = [] check_url = False err_msg.append(str(pe)) links[0:] = locations if re.match(r'^HTTP/.* 200 OK', http_status): check_200ok = True else: err_msg.append('Returned HTTP status is not 200 OK (%s)' % http_status) check_ct2, tmp_err_msg = check_content_type(content_type) err_msg += tmp_err_msg return [check_url and check_ct2 and check_200ok, err_msg]
[docs]class CORE_READ002(Test): objective = 'Retrieve the URLs of the OCCI Entities belonging to an OCCI Kind or OCCI Mixin and related to an OCCI Category filter' @classmethod
[docs] def test(self=None): err_msg = [] check = True body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) if content_type: check, tmp_err_msg = check_content_type(content_type) err_msg += tmp_err_msg if not check_pretest: return [False, err_msg] kinds = Test.search_category({'class': 'kind'}) found = False check_read = False for category in kinds: filter = Test.search_category({'rel': '%s%s' % ('http://schemas.ogf.org/occi/core#', 'resource')}) if filter is None: continue else: found = True check_read, tmp_err_msg = CORE_READ002_COMMON(category=filter) if not found: err_msg += ['No desired OCCI Mixin found'] return [check and check_read, err_msg]
[docs]def CORE_READ_DESCRIPTION(filter=None): """Read OCCI Entity Description :param string filter{}: required attribute names and values :return: status, err_msg, url, entity description :rtype: [bool, string[], string, occi.Category[], occi.Link[], occi.Attribute[]] """ categories = [] links = [] attributes = [] found = False entity_url = None check, err_msg, urls = CORE_READ_URL(occi_config['occi.tests.category']) if not check: return [False, err_msg, entity_url, None, None, None] #print urls for entity_url in urls: #print base_url, url body, headers, http_status, content_type = connection.get(url=entity_url) check1, tmp_err_msg = check_http_status("200 OK", http_status) err_msg += tmp_err_msg if content_type: check2, tmp_err_msg = check_requested_content_type(content_type) err_msg += tmp_err_msg if not check1 or not check2: return [False, err_msg, entity_url, None, None, None] try: categories, links, attributes = renderer.parse_resource(body, headers) except occi.ParseError as pe: err_msg += [str(pe)] return [False, err_msg, entity_url, None, None, None] if not categories: err_msg += ['HTTP Response doesn\'t contain categories'] return [False, err_msg, entity_url, None, None, None] #print 'Got OCCI Entity: ' + str(categories[0]) if match_entity(attributes, filter): #print 'Hit!' found = True break if not found: err_msg.append('No required OCCI Entity instance found') return [found, err_msg, entity_url, categories, links, attributes]
[docs]class CORE_READ007(Test): objective = 'Retrieve the description of an OCCI Entity' @classmethod
[docs] def test(self=None): check, err_msg, url, categories, links, attributes = CORE_READ_DESCRIPTION({}) return [check, err_msg]
[docs]def get_attributes(attribute_definitions, attributes, err_msg): """Fill attribute values from example_attributes for all required attributes. :param occi.AttributeDefinition attribute_definitions[]: attribute definitions :param occi.Attribute attributes{}: result attributes dictionary :param string err_msg[]: list of errors to append :return: all required attributes has a value :rtype: bool """ has_all_attributes = True #print 'list of attributes: %s' % attributes for ad in attribute_definitions: #print 'attribute: %s' % ad if ad.isrequired() and not ad.isimmutable() and not ad['name'] in attributes: if ad['name'] not in example_attributes: err_msg.append('Tests error: unknown attribute %s' % ad['name']) has_all_attributes = False continue #print 'adding attribute: %s' % ad attributes[ad['name']] = example_attributes[ad['name']] return has_all_attributes
[docs]class CORE_DELETE001(Test): objective = 'Delete an OCCI Entity' @classmethod
[docs] def test(self=None): err_msg = [] body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) category = occi.Category(occi_config['occi.tests.category']) check, err_msg = INFRA_CREATE_COMMON('compute', [category], [], err_msg) if not check: return [False, err_msg] check, err_msg, tmp_urls = CORE_READ_URL(occi_config['occi.tests.category']) if not tmp_urls: err_msg += ["OCCI entity URL not found!"] return [False, err_msg] url = tmp_urls[0] if occi_config['curlverbose']: print '[OCCI/CORE/DELETE/001] entity URL: %s' % str(url) body, response_headers, http_status, content_type = connection.get(url=url) check_exist1, tmp_err_msg = check_http_status("200 OK", http_status) err_msg += tmp_err_msg if occi_config['curlverbose']: print '[OCCI/CORE/DELETE/001] entity body: %s' % str(body) body, response_headers, http_status, content_type = connection.delete(url=url) check_delete1, tmp_err_msg = check_http_status('2\d\d ', http_status) err_msg += tmp_err_msg # It takes some time to delete machine, second delete action force it # Not testing result of the operation (various backends have different behaviour) body, response_headers, http_status, content_type = connection.delete(url=url) body, response_headers, http_status, content_type = connection.get(url=url) check_exist2, tmp_err_msg = check_http_status("404 Not Found", http_status) err_msg += tmp_err_msg Test.clear_categories() return [check_exist1 and check_exist2 and check_delete1, err_msg]
[docs]class CORE_UPDATE001(Test): """Full update of a specific OCCI Entity Requires existing compute machine. OpenNebula issues: * https://github.com/EGI-FCTF/rOCCI-server/issues/125: poweroff state required * https://github.com/EGI-FCTF/rOCCI-server/issues/126: not all attributes implemented * https://github.com/EGI-FCTF/rOCCI-server/issues/128: parse error """ objective = 'Full update of a specific OCCI Entity' @classmethod
[docs] def test(self=None): err_msg = [] check_response = True check, err_msg, urls = CORE_READ_URL(occi_config['occi.tests.category']) if not urls: err_msg.append('No OCCI Entity instance found') return [False, err_msg] entity_url = urls[0] if occi_config['curlverbose']: print '[OCCI/CORE/UPDATE/001] all entities: %s' % str(urls) print '[OCCI/CORE/UPDATE/001] selected entity: %s' % str(entity_url) body, response_headers, http_status, content_type = connection.get(url=entity_url) categories, links, attributes = renderer.parse_resource(body, response_headers) if occi_config['curlverbose']: print '[OCCI/CORE/UPDATE/001] resource categories: %s' % str(categories) print '[OCCI/CORE/UPDATE/001] resource links: %s' % str(links) print '[OCCI/CORE/UPDATE/001] resource attributes: %s' % str(attributes) # change one attribute a = None for a in attributes: if a['name'] == 'occi.core.title': break if a is None or a['name'] != 'occi.core.title': a = occi.Attribute({'name': 'occi.core.title', 'value': gen_id('c_pOCCI')}) attributes.append(a) else: a['value'] = gen_id(a['value']) body, headers = renderer.render_resource(categories, links, attributes) # update body, response_headers, http_status, content_type = connection.put(url=entity_url, headers=['Content-Type: %s' % occi_config['mimetype']] + headers, body=body) #print body #print http_status Test.clear_categories() check_ct, tmp_err_msg = check_content_type(content_type) err_msg += tmp_err_msg if content_type != occi_config['mimetype']: err_msg += ['Content-Type is not requested "%s"' % occi_config['mimetype']] check_ct = False check, tmp_err_msg = check_http_status("200 OK", http_status) if check: # response contains OCCI Entity description response_categories, response_links, response_attributes = renderer.parse_resource(body, response_headers) if not response_categories or not response_attributes: err_msg += ['HTTP Response doesn\'t contain categories or attributes'] check_response = False else: check, tmp_err_msg = check_http_status("201 Created", http_status) if check: response_urls = renderer_httpheaders.parse_locations(None, response_headers) #print 'Entity URL: %s\n' % entity_url #print 'Response URLs:\n' #print response_urls check_response = False for url in response_urls: if url == entity_url: check_response = True if not check_response: err_msg += ['HTTP Location headers has not OCCI Entity URL (%s)' % entity_url] if not check: err_msg.append('HTTP status is neither 200 OK nor 201 Created (%s)' % http_status) print >> sys.stderr, body return [check and check_ct and check_response, err_msg]
[docs]class CORE_MISC001(Test): objective = 'Trigger OCCI Action on existing OCCI Entity' @classmethod
[docs] def test(self=None): err_msg = [] actions = set() check, err_msg, entity_url, categories, links, attributes = CORE_READ_DESCRIPTION(filter=occi_config['occi.tests.entity']) if not check: return [False, err_msg] for cat in categories: model_cat = Test.get_category(cat['scheme'] + cat['term']) #print model_cat['term'] if 'actions' in model_cat: #print model_cat['actions'] actions |= set(model_cat['actions']) if occi_config['curlverbose']: print '[OCCI/CORE/MISC/001] actions: ' + str(list(actions)) # select appropriate action, fallback to the first one action = list(actions)[0] action_start = action action_stop = action for act in actions: if re.search('#start$', act): action_start = act if re.search('#suspend$', act): action_stop = act for a in attributes: if a['name'] == 'occi.compute.state': break if a['name'] == 'occi.compute.state': if occi_config['curlverbose']: print '[OCCI/CORE/MISC/001] state: %s' % a['value'] if a['value'] == 'active': action = action_stop else: action = action_start if occi_config['curlverbose']: print '[OCCI/CORE/MISC/001] selected action: %s' % action action_cat = Test.get_category(action) if not action_cat: return [False, 'Action "%s" not found in OCCI Model' % action] #print action_cat url = entity_url url += '?action=' + action_cat['term'] if occi_config['curlverbose']: print '[OCCI/CORE/MISC/001] calling action: %s' % url body, headers = renderer.render_category(action_cat) body, response_request, http_status, content_type = connection.post(url=url, headers=['Content-Type: %s' % occi_config['mimetype']] + headers, body=body) check, tmp_err_msg = check_http_status("200 OK", http_status) err_msg += tmp_err_msg return [check, err_msg]
[docs]def INFRA_CREATE_COMMON(resource, categories, additional_attributes, err_msg): """Generic help function to create OCCI Infrastructure resources. HTTP Headers renderer is always used. :param string resource: OCCI Category term (compute, storage, network) :param occi.Category categories[]: OCCI Categories to add to rendering :param occi.AttributeDefinition additional_attributes[]: additional attributes to set from example defaults :param string err_msg[]: error messages list to append :return: status and error message list :rtype: [bool, string[]] """ has_kind = True has_all_attributes = True all_attributes = [] attributes = {} kind = Test.search_category({'class': 'kind', 'term': resource, 'scheme': 'http://schemas.ogf.org/occi/infrastructure#'}) #print kind if not kind: has_kind = False err_msg.append('No OCCI Kind found') if not kind: return [False, err_msg] for item in ['location', 'term', 'scheme']: if item not in kind.keys(): has_kind = False err_msg.append('No %s in OCCI Kind' % item) if 'attributes' in kind: all_attributes += kind['attributes'] if additional_attributes is not None: all_attributes += additional_attributes #print 'list of attributes: %s' % repr(all_attributes) if not get_attributes(all_attributes, attributes, err_msg): has_all_attributes = False #print 'list of result attribute keys: %s' % repr(attributes.keys()) new_cat_s, new_cat_h = renderer.render_resource(categories, None, attributes.values()) body, response_request, http_status, content_type = connection.post(url=occi_config['url'] + kind['location'], headers=['Content-Type: %s' % occi_config['mimetype']] + new_cat_h, body=new_cat_s) check_create, tmp_err_msg = check_http_status("(200 OK|201 Created)", http_status) err_msg += tmp_err_msg check_ct, tmp_err_msg = check_content_type(content_type) err_msg += tmp_err_msg entities = check_body_entities(body, response_request, err_msg) check_br = (entities is not None and entities) check_rct, tmp_err_msg = check_requested_content_type(content_type) err_msg += tmp_err_msg if not check_create: print >> sys.stderr, body Test.clear_categories() body, response_headers, http_status, content_type = connection.get(url=occi_config['url'] + kind['location']) entities2 = check_body_entities(body, response_headers, err_msg) # check if the entity is really created (just check the first: entities[0]) check_created = False if entities: for line in entities2: if line == entities[0]: check_created = True break if not check_created: err_msg.append('OCCI %s Resource hasn\'t been successfully created' % resource.title()) return [has_kind and has_all_attributes and check_create and check_created and check_ct and check_br and check_rct and check_created, err_msg]
[docs]class INFRA_CREATE001(Test): objective = 'Create an OCCI Compute Resource' @classmethod
[docs] def test(self=None): err_msg = [] category = occi.Category({'term': 'compute', 'scheme': 'http://schemas.ogf.org/occi/infrastructure#', 'class': 'kind'}) body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) if not check_pretest: return [False, err_msg] return INFRA_CREATE_COMMON('compute', [category], [], err_msg)
[docs]class INFRA_CREATE002(Test): objective = 'Create an OCCI Storage Resource' @classmethod
[docs] def test(self=None): err_msg = [] category = occi.Category({'term': 'storage', 'scheme': 'http://schemas.ogf.org/occi/infrastructure#', 'class': 'kind'}) additional_attributes = [ occi.AttributeDefinition({"name": "occi.core.title", "required": True}), occi.AttributeDefinition({"name": "occi.storage.size", "required": True}), ] body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) if not check_pretest: return [False, err_msg] return INFRA_CREATE_COMMON('storage', [category], additional_attributes, err_msg)
[docs]class INFRA_CREATE003(Test): objective = 'Create an OCCI Network Resource' @classmethod
[docs] def test(self=None): err_msg = [] category = occi.Category({'term': 'network', 'scheme': 'http://schemas.ogf.org/occi/infrastructure#', 'class': 'kind'}) additional_attributes = [occi.AttributeDefinition({"name": "occi.core.title", "required": True})] body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) if not check_pretest: return [False, err_msg] return INFRA_CREATE_COMMON('network', [category], additional_attributes, err_msg)
[docs]class INFRA_CREATE004(Test): objective = 'Create an OCCI Compute Resource using an OS and resource template' @classmethod
[docs] def test(self=None): err_msg = [] categories = [ occi.Category({'term': 'compute', 'scheme': 'http://schemas.ogf.org/occi/infrastructure#', 'class': 'kind'}) ] body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) if not check_pretest: return [False, err_msg] # OS template os_tpl = Test.search_category({'class': 'mixin', 'rel': 'http://schemas.ogf.org/occi/infrastructure#os_tpl'}) if occi_config['curlverbose']: print '[OCCI/INFRA/CREATE/004] os_tpl: %s' % str(os_tpl) if not os_tpl: err_msg.append('No OCCI OS Template found') return [False, err_msg] # 'term': 'uuid_ttylinux_0', 'scheme': 'http://occi.myriad5.zcu.cz/occi/infrastructure/os_tpl#', 'class': 'mixin' categories.append(occi.Category({'term': os_tpl['term'], 'scheme': os_tpl['scheme'], 'class': 'mixin'})) if 'attributes' in os_tpl: tpl_attributes = os_tpl['attributes'] else: tpl_attributes = [] # resource template resource_tpl = Test.search_category({'class': 'mixin', 'rel': 'http://schemas.ogf.org/occi/infrastructure#resource_tpl'}) if occi_config['curlverbose']: print '[OCCI/INFRA/CREATE/004] resource_tpl: %s' % str(resource_tpl) if not resource_tpl: err_msg.append('No OCCI Resource Template found') return [False, err_msg] if resource_tpl: categories.append(occi.Category({'term': resource_tpl['term'], 'scheme': resource_tpl['scheme'], 'class': 'mixin'})) if 'attributes' in resource_tpl: tpl_attributes += resource_tpl['attributes'] if occi_config['curlverbose']: print '[OCCI/INFRA/CREATE/004] attributes: %s' % str(tpl_attributes) return INFRA_CREATE_COMMON('compute', categories, tpl_attributes, err_msg)
[docs]class INFRA_CREATE005(Test): """ Unsupported test, os_tpl required. It can be called by pOCCI -t 'INFRA/CREATE/005' """ objective = 'Create an OCCI Compute Resource with an OCCI Storagelink and an OCCI Networkinterface' @classmethod
[docs] def test(self=None): network_links = [] storage_links = [] err_msg = [] check = True body, response_headers, http_status, content_type, check_pretest = Test.pretest_http_status("200 OK", err_msg) if content_type: check, tmp_err_msg = check_content_type(content_type) err_msg += tmp_err_msg if not check_pretest: return [False, err_msg] storage = Test.search_category({'term': 'storage', 'scheme': 'http://schemas.ogf.org/occi/infrastructure#'}) network = Test.search_category({'term': 'network', 'scheme': 'http://schemas.ogf.org/occi/infrastructure#'}) check_read, tmp_err_msg = CORE_READ002_COMMON(category=storage, links=storage_links) if not check_read: check = False err_msg += tmp_err_msg check_read, tmp_err_msg = CORE_READ002_COMMON(category=network, links=network_links) if not check_read: check = False err_msg += tmp_err_msg print storage_links print network_links if not storage_links or not network_links: if not storage_links: err_msg.append('No storage found') if not network_links: err_msg.append('No network found') return [False, err_msg] compute = Test.search_category({'term': 'compute', 'scheme': 'http://schemas.ogf.org/occi/infrastructure#'}) new_compute = 'Category: %s; scheme="%s"; class="%s"\n\r\ Link: <%s>; rel="%s"; category="%s"\n\r\ Link: <%s>; rel="%s"; category="%s"\n\r\ ' % (compute['term'], compute['scheme'], compute['class'], storage_links[0], storage['scheme'] + storage['term'], 'http://schemas.ogf.org/occi/infrastructure#storagelink', network_links[0], network['scheme'] + network['term'], 'http://schemas.ogf.org/occi/infrastructure#networkinterface') body, response_headers, http_status, content_type = connection.post(url=occi_config['url'] + compute['location'], headers=['Content-Type: text/plain'], body=new_compute) check_create, tmp_err_msg = check_http_status("201 Created", http_status) err_msg += tmp_err_msg if not check_create: print >> sys.stderr, body Test.clear_categories() return [check and check_create, err_msg]
[docs]class INFRA_CREATE006(Test): """ Opennebula requires running compute instance. """ objective = 'Create an OCCI Storagelink between an existing OCCI Compute and OCCI Storage Resource' @classmethod
[docs] def test(self=None): return INFRA_CREATE_LINK('storage', 'link')
[docs]class INFRA_CREATE007(Test): """ Opennebula requires running compute instance. """ objective = 'Create an OCCI Networkinterface between an existing OCCI Compute and OCCI Network Resource' @classmethod
[docs] def test(self=None): return INFRA_CREATE_LINK('network', 'interface')