#!/usr/bin/python # # Test script utility class # # Mark Huang # Copyright (C) 2006 The Trustees of Princeton University # # $Id: Test.py 5574 2007-10-25 20:33:17Z thierry $ # from pprint import pprint from string import letters, digits, punctuation from traceback import print_exc from optparse import OptionParser import socket import base64 import struct import os import xmlrpclib from PLC.Shell import Shell from random import Random random = Random() def randfloat(min = 0.0, max = 1.0): return float(min) + (random.random() * (float(max) - float(min))) def randint(min = 0, max = 1): return int(randfloat(min, max + 1)) # See "2.2 Characters" in the XML specification: # # #x9 | #xA | #xD | [#x20-#xD7FF] | [#xE000-#xFFFD] # avoiding # [#x7F-#x84], [#x86-#x9F], [#xFDD0-#xFDDF] # ascii_xml_chars = map(unichr, [0x9, 0xA]) # xmlrpclib uses xml.parsers.expat, which always converts either '\r' # (#xD) or '\n' (#xA) to '\n'. So avoid using '\r', too, if this is # still the case. if xmlrpclib.loads(xmlrpclib.dumps(('\r',)))[0][0] == '\r': ascii_xml_chars.append('\r') ascii_xml_chars += map(unichr, xrange(0x20, 0x7F - 1)) low_xml_chars = list(ascii_xml_chars) low_xml_chars += map(unichr, xrange(0x84 + 1, 0x86 - 1)) low_xml_chars += map(unichr, xrange(0x9F + 1, 0xFF)) valid_xml_chars = list(low_xml_chars) valid_xml_chars += map(unichr, xrange(0xFF + 1, 0xD7FF)) valid_xml_chars += map(unichr, xrange(0xE000, 0xFDD0 - 1)) valid_xml_chars += map(unichr, xrange(0xFDDF + 1, 0xFFFD)) def randstr(length, pool = valid_xml_chars, encoding = "utf-8"): sample = random.sample(pool, min(length, len(pool))) while True: s = u''.join(sample) bytes = len(s.encode(encoding)) if bytes > length: sample.pop() elif bytes < length: sample += random.sample(pool, min(length - bytes, len(pool))) random.shuffle(sample) else: break return s def randhostname(): # 1. Each part begins and ends with a letter or number. # 2. Each part except the last can contain letters, numbers, or hyphens. # 3. Each part is between 1 and 64 characters, including the trailing dot. # 4. At least two parts. # 5. Last part can only contain between 2 and 6 letters. hostname = 'a' + randstr(61, letters + digits + '-') + '1.' + \ 'b' + randstr(61, letters + digits + '-') + '2.' + \ 'c' + randstr(5, letters) return hostname def randpath(length): parts = [] for i in range(randint(1, 10)): parts.append(randstr(randint(1, 30), ascii_xml_chars)) return u'/'.join(parts)[0:length] def randemail(): return (randstr(100, letters + digits) + "@" + randhostname()).lower() def randkey(bits = 2048): ssh_key_types = ["ssh-dss", "ssh-rsa"] key_type = random.sample(ssh_key_types, 1)[0] return ' '.join([key_type, base64.b64encode(''.join(randstr(bits / 8).encode("utf-8"))), randemail()]) def random_site(): return { 'name': randstr(254), 'abbreviated_name': randstr(50), 'login_base': randstr(20, letters).lower(), 'latitude': int(randfloat(-90.0, 90.0) * 1000) / 1000.0, 'longitude': int(randfloat(-180.0, 180.0) * 1000) / 1000.0, } def random_address_type(): return { 'name': randstr(20), 'description': randstr(254), } def random_address(): return { 'line1': randstr(254), 'line2': randstr(254), 'line3': randstr(254), 'city': randstr(254), 'state': randstr(254), 'postalcode': randstr(64), 'country': randstr(128), } def random_person(): return { 'first_name': randstr(128), 'last_name': randstr(128), 'email': randemail(), 'bio': randstr(254), # Accounts are disabled by default 'enabled': False, 'password': randstr(254), } def random_key(key_types): return { 'key_type': random.sample(key_types, 1)[0], 'key': randkey() } def random_nodegroup(): return { 'name': randstr(50), 'description': randstr(200), } def random_node(boot_states): return { 'hostname': randhostname(), 'boot_state': random.sample(boot_states, 1)[0], 'model': randstr(255), 'version': randstr(64), } def random_nodenetwork(method, type): nodenetwork_fields = { 'method': method, 'type': type, 'bwlimit': randint(500000, 10000000), } if method != 'dhcp': ip = randint(0, 0xffffffff) netmask = (0xffffffff << randint(2, 31)) & 0xffffffff network = ip & netmask broadcast = ((ip & netmask) | ~netmask) & 0xffffffff gateway = randint(network + 1, broadcast - 1) dns1 = randint(0, 0xffffffff) for field in 'ip', 'netmask', 'network', 'broadcast', 'gateway', 'dns1': nodenetwork_fields[field] = socket.inet_ntoa(struct.pack('>L', locals()[field])) return nodenetwork_fields def random_pcu(): return { 'hostname': randhostname(), 'ip': socket.inet_ntoa(struct.pack('>L', randint(0, 0xffffffff))), 'protocol': randstr(16), 'username': randstr(254), 'password': randstr(254), 'notes': randstr(254), 'model': randstr(32), } def random_conf_file(): return { 'enabled': bool(randint()), 'source': randpath(255), 'dest': randpath(255), 'file_permissions': "%#o" % randint(0, 512), 'file_owner': randstr(32, letters + '_' + digits), 'file_group': randstr(32, letters + '_' + digits), 'preinstall_cmd': randpath(100), 'postinstall_cmd': randpath(100), 'error_cmd': randpath(100), 'ignore_cmd_errors': bool(randint()), 'always_update': bool(randint()), } def random_attribute_type(role_ids): return { 'name': randstr(100), 'description': randstr(254), 'min_role_id': random.sample(role_ids, 1)[0], } def random_slice(login_base): return { 'name': login_base + "_" + randstr(11, letters).lower(), 'url': "http://" + randhostname() + "/", 'description': randstr(2048), } class Test: tiny = { 'sites': 1, 'address_types': 1, 'addresses_per_site': 1, 'persons_per_site': 1, 'keys_per_person': 1, 'nodegroups': 1, 'nodes_per_site': 1, 'nodenetworks_per_node': 1, 'pcus_per_site': 1, 'conf_files': 1, 'attribute_types': 1, 'slices_per_site': 1, 'attributes_per_slice': 1, } default = { 'sites': 10, 'address_types': 2, 'addresses_per_site': 2, 'persons_per_site': 10, 'keys_per_person': 2, 'nodegroups': 10, 'nodes_per_site': 2, 'nodenetworks_per_node': 1, 'pcus_per_site': 1, 'conf_files': 10, 'attribute_types': 10, 'slices_per_site': 10, 'attributes_per_slice': 2, } def __init__(self, api, check = True, verbose = True): self.api = api self.check = check self.verbose = verbose self.site_ids = [] self.address_type_ids = [] self.address_ids = [] self.person_ids = [] self.key_ids = [] self.nodegroup_ids = [] self.node_ids = [] self.nodenetwork_ids = [] self.pcu_ids = [] self.conf_file_ids = [] self.attribute_type_ids = [] self.slice_ids = [] self.slice_attribute_ids = [] def Run(self, **kwds): """ Run a complete database and API consistency test. Populates the database with a set of random entities, updates them, then deletes them. Examples: test.Run() # Defaults test.Run(**Test.default) # Defaults test.Run(**Test.tiny) # Tiny set test.Run(sites = 123, slices_per_site = 4) # Defaults with overrides """ try: self.Add(**kwds) self.Update() finally: self.Delete() def Add(self, **kwds): """ Populate the database with a set of random entities. Examples: test.populate() # Defaults test.populate(Test.tiny) # Tiny set test.populate(sites = 123, slices_per_site = 4) # Defaults with overrides """ params = self.default.copy() params.update(kwds) self.AddSites(params['sites']) self.AddAddressTypes(params['address_types']) self.AddAddresses(params['addresses_per_site']) self.AddPersons(params['persons_per_site']) self.AddKeys(params['keys_per_person']) self.AddNodeGroups(params['nodegroups']) self.AddNodes(params['nodes_per_site']) self.AddNodeNetworks(params['nodenetworks_per_node']) self.AddPCUs(params['pcus_per_site']) self.AddConfFiles(params['conf_files']) self.AddSliceAttributeTypes(params['attribute_types']) self.AddSlices(params['slices_per_site']) self.AddSliceAttributes(params['attributes_per_slice']) def Update(self): self.UpdateSites() self.UpdateAddressTypes() self.UpdateAddresses() self.UpdatePersons() self.UpdateKeys() self.UpdateNodeGroups() self.UpdateNodes() self.UpdateNodeNetworks() self.UpdatePCUs() self.UpdateConfFiles() self.UpdateSliceAttributeTypes() self.UpdateSlices() self.UpdateSliceAttributes() def Delete(self): self.DeleteSliceAttributes() self.DeleteSlices() self.DeleteSliceAttributeTypes() self.DeleteKeys() self.DeleteConfFiles() self.DeletePCUs() self.DeleteNodeNetworks() self.DeleteNodes() self.DeletePersons() self.DeleteNodeGroups() self.DeleteAddresses() self.DeleteAddressTypes() self.DeleteSites() def AddSites(self, n = 10): """ Add a number of random sites. """ for i in range(n): # Add site site_fields = random_site() site_id = self.api.AddSite(site_fields) # Should return a unique site_id assert site_id not in self.site_ids self.site_ids.append(site_id) # Enable slice creation site_fields['max_slices'] = randint(1, 10) self.api.UpdateSite(site_id, site_fields) if self.check: # Check site site = self.api.GetSites([site_id])[0] for field in site_fields: assert site[field] == site_fields[field] if self.verbose: print "Added site", site_id def UpdateSites(self): """ Make random changes to any sites we may have added. """ for site_id in self.site_ids: # Update site site_fields = random_site() # Do not change login_base if 'login_base' in site_fields: del site_fields['login_base'] self.api.UpdateSite(site_id, site_fields) if self.check: # Check site site = self.api.GetSites([site_id])[0] for field in site_fields: assert site[field] == site_fields[field] if self.verbose: print "Updated site", site_id def DeleteSites(self): """ Delete any random sites we may have added. """ for site_id in self.site_ids: self.api.DeleteSite(site_id) if self.check: assert not self.api.GetSites([site_id]) if self.verbose: print "Deleted site", site_id if self.check: assert not self.api.GetSites(self.site_ids) self.site_ids = [] def AddAddressTypes(self, n = 2): """ Add a number of random address types. """ for i in range(n): address_type_fields = random_address_type() address_type_id = self.api.AddAddressType(address_type_fields) # Should return a unique address_type_id assert address_type_id not in self.address_type_ids self.address_type_ids.append(address_type_id) if self.check: # Check address type address_type = self.api.GetAddressTypes([address_type_id])[0] for field in address_type_fields: assert address_type[field] == address_type_fields[field] if self.verbose: print "Added address type", address_type_id def UpdateAddressTypes(self): """ Make random changes to any address types we may have added. """ for address_type_id in self.address_type_ids: # Update address_type address_type_fields = random_address_type() self.api.UpdateAddressType(address_type_id, address_type_fields) if self.check: # Check address type address_type = self.api.GetAddressTypes([address_type_id])[0] for field in address_type_fields: assert address_type[field] == address_type_fields[field] if self.verbose: print "Updated address_type", address_type_id def DeleteAddressTypes(self): """ Delete any random address types we may have added. """ for address_type_id in self.address_type_ids: self.api.DeleteAddressType(address_type_id) if self.check: assert not self.api.GetAddressTypes([address_type_id]) if self.verbose: print "Deleted address type", address_type_id if self.check: assert not self.api.GetAddressTypes(self.address_type_ids) self.address_type_ids = [] def AddAddresses(self, per_site = 2): """ Add a number of random addresses to each site. """ for site_id in self.site_ids: for i in range(per_site): address_fields = random_address() address_id = self.api.AddSiteAddress(site_id, address_fields) # Should return a unique address_id assert address_id not in self.address_ids self.address_ids.append(address_id) # Add random address type if self.address_type_ids: for address_type_id in random.sample(self.address_type_ids, 1): self.api.AddAddressTypeToAddress(address_type_id, address_id) if self.check: # Check address address = self.api.GetAddresses([address_id])[0] for field in address_fields: assert address[field] == address_fields[field] if self.verbose: print "Added address", address_id, "to site", site_id def UpdateAddresses(self): """ Make random changes to any addresses we may have added. """ for address_id in self.address_ids: # Update address address_fields = random_address() self.api.UpdateAddress(address_id, address_fields) if self.check: # Check address address = self.api.GetAddresses([address_id])[0] for field in address_fields: assert address[field] == address_fields[field] if self.verbose: print "Updated address", address_id def DeleteAddresses(self): """ Delete any random addresses we may have added. """ for address_id in self.address_ids: # Remove address types address = self.api.GetAddresses([address_id])[0] for address_type_id in address['address_type_ids']: self.api.DeleteAddressTypeFromAddress(address_type_id, address_id) if self.check: address = self.api.GetAddresses([address_id])[0] assert not address['address_type_ids'] self.api.DeleteAddress(address_id) if self.check: assert not self.api.GetAddresses([address_id]) if self.verbose: print "Deleted address", address_id if self.check: assert not self.api.GetAddresses(self.address_ids) self.address_ids = [] def AddPersons(self, per_site = 10): """ Add a number of random users to each site. """ for site_id in self.site_ids: for i in range(per_site): # Add user person_fields = random_person() person_id = self.api.AddPerson(person_fields) # Should return a unique person_id assert person_id not in self.person_ids self.person_ids.append(person_id) if self.check: # Check user person = self.api.GetPersons([person_id])[0] for field in person_fields: if field != 'password': assert person[field] == person_fields[field] auth = {'AuthMethod': "password", 'Username': person_fields['email'], 'AuthString': person_fields['password']} if self.check: # Check that user is disabled try: assert not self.api.AuthCheck(auth) except: pass # Add random set of roles role_ids = random.sample([20, 30, 40], randint(1, 3)) for role_id in role_ids: self.api.AddRoleToPerson(role_id, person_id) if self.check: person = self.api.GetPersons([person_id])[0] assert set(role_ids) == set(person['role_ids']) # Enable user self.api.UpdatePerson(person_id, {'enabled': True}) if self.check: # Check that user is enabled assert self.api.AuthCheck(auth) # Associate user with site self.api.AddPersonToSite(person_id, site_id) self.api.SetPersonPrimarySite(person_id, site_id) if self.check: person = self.api.GetPersons([person_id])[0] assert person['site_ids'][0] == site_id if self.verbose: print "Added user", person_id, "to site", site_id def UpdatePersons(self): """ Make random changes to any users we may have added. """ for person_id in self.person_ids: # Update user person_fields = random_person() # Keep them enabled person_fields['enabled'] = True self.api.UpdatePerson(person_id, person_fields) if self.check: # Check user person = self.api.GetPersons([person_id])[0] for field in person_fields: if field != 'password': assert person[field] == person_fields[field] if self.verbose: print "Updated person", person_id person = self.api.GetPersons([person_id])[0] # Associate user with a random set of sites site_ids = random.sample(self.site_ids, randint(0, len(self.site_ids))) for site_id in (set(site_ids) - set(person['site_ids'])): self.api.AddPersonToSite(person_id, site_id) for site_id in (set(person['site_ids']) - set(site_ids)): self.api.DeletePersonFromSite(person_id, site_id) if site_ids: self.api.SetPersonPrimarySite(person_id, site_ids[0]) if self.check: person = self.api.GetPersons([person_id])[0] assert set(site_ids) == set(person['site_ids']) if self.verbose: print "Updated person", person_id, "to sites", site_ids def DeletePersons(self): """ Delete any random users we may have added. """ for person_id in self.person_ids: # Remove from site person = self.api.GetPersons([person_id])[0] for site_id in person['site_ids']: self.api.DeletePersonFromSite(person_id, site_id) if self.check: person = self.api.GetPersons([person_id])[0] assert not person['site_ids'] # Revoke roles for role_id in person['role_ids']: self.api.DeleteRoleFromPerson(role_id, person_id) if self.check: person = self.api.GetPersons([person_id])[0] assert not person['role_ids'] # Disable account self.api.UpdatePerson(person_id, {'enabled': False}) if self.check: person = self.api.GetPersons([person_id])[0] assert not person['enabled'] # Delete account self.api.DeletePerson(person_id) if self.check: assert not self.api.GetPersons([person_id]) if self.verbose: print "Deleted user", person_id if self.check: assert not self.api.GetPersons(self.person_ids) self.person_ids = [] def AddKeys(self, per_person = 2): """ Add a number of random keys to each user. """ key_types = self.api.GetKeyTypes() if not key_types: raise Exception, "No key types" for person_id in self.person_ids: for i in range(per_person): # Add key key_fields = random_key(key_types) key_id = self.api.AddPersonKey(person_id, key_fields) # Should return a unique key_id assert key_id not in self.key_ids self.key_ids.append(key_id) if self.check: # Check key key = self.api.GetKeys([key_id])[0] for field in key_fields: assert key[field] == key_fields[field] # Add and immediately blacklist a key key_fields = random_key(key_types) key_id = self.api.AddPersonKey(person_id, key_fields) self.api.BlacklistKey(key_id) # Is effectively deleted assert not self.api.GetKeys([key_id]) # Cannot be added again try: key_id = self.api.AddPersonKey(person_id, key_fields) assert False except Exception, e: pass if self.verbose: print "Added key", key_id, "to user", person_id def UpdateKeys(self): """ Make random changes to any keys we may have added. """ key_types = self.api.GetKeyTypes() if not key_types: raise Exception, "No key types" for key_id in self.key_ids: # Update key key_fields = random_key(key_types) self.api.UpdateKey(key_id, key_fields) if self.check: # Check key key = self.api.GetKeys([key_id])[0] for field in key_fields: assert key[field] == key_fields[field] if self.verbose: print "Updated key", key_id def DeleteKeys(self): """ Delete any random keys we may have added. """ for key_id in self.key_ids: self.api.DeleteKey(key_id) if self.check: assert not self.api.GetKeys([key_id]) if self.verbose: print "Deleted key", key_id if self.check: assert not self.api.GetKeys(self.key_ids) self.key_ids = [] def AddNodeGroups(self, n = 10): """ Add a number of random node groups. """ for i in range(n): # Add node group nodegroup_fields = random_nodegroup() nodegroup_id = self.api.AddNodeGroup(nodegroup_fields) # Should return a unique nodegroup_id assert nodegroup_id not in self.nodegroup_ids self.nodegroup_ids.append(nodegroup_id) if self.check: # Check node group nodegroup = self.api.GetNodeGroups([nodegroup_id])[0] for field in nodegroup_fields: assert nodegroup[field] == nodegroup_fields[field] if self.verbose: print "Added node group", nodegroup_id def UpdateNodeGroups(self): """ Make random changes to any node groups we may have added. """ for nodegroup_id in self.nodegroup_ids: # Update nodegroup nodegroup_fields = random_nodegroup() self.api.UpdateNodeGroup(nodegroup_id, nodegroup_fields) if self.check: # Check nodegroup nodegroup = self.api.GetNodeGroups([nodegroup_id])[0] for field in nodegroup_fields: assert nodegroup[field] == nodegroup_fields[field] if self.verbose: print "Updated node group", nodegroup_id def DeleteNodeGroups(self): """ Delete any random node groups we may have added. """ for nodegroup_id in self.nodegroup_ids: self.api.DeleteNodeGroup(nodegroup_id) if self.check: assert not self.api.GetNodeGroups([nodegroup_id]) if self.verbose: print "Deleted node group", nodegroup_id if self.check: assert not self.api.GetNodeGroups(self.nodegroup_ids) self.nodegroup_ids = [] def AddNodes(self, per_site = 2): """ Add a number of random nodes to each site. Each node will also be added to a random node group if AddNodeGroups() was previously run. """ boot_states = self.api.GetBootStates() if not boot_states: raise Exception, "No boot states" for site_id in self.site_ids: for i in range(per_site): # Add node node_fields = random_node(boot_states) node_id = self.api.AddNode(site_id, node_fields) # Should return a unique node_id assert node_id not in self.node_ids self.node_ids.append(node_id) # Add to a random set of node groups nodegroup_ids = random.sample(self.nodegroup_ids, randint(0, len(self.nodegroup_ids))) for nodegroup_id in nodegroup_ids: self.api.AddNodeToNodeGroup(node_id, nodegroup_id) if self.check: # Check node node = self.api.GetNodes([node_id])[0] for field in node_fields: assert node[field] == node_fields[field] if self.verbose: print "Added node", node_id def UpdateNodes(self): """ Make random changes to any nodes we may have added. """ boot_states = self.api.GetBootStates() if not boot_states: raise Exception, "No boot states" for node_id in self.node_ids: # Update node node_fields = random_node(boot_states) self.api.UpdateNode(node_id, node_fields) node = self.api.GetNodes([node_id])[0] # Add to a random set of node groups nodegroup_ids = random.sample(self.nodegroup_ids, randint(0, len(self.nodegroup_ids))) for nodegroup_id in (set(nodegroup_ids) - set(node['nodegroup_ids'])): self.api.AddNodeToNodeGroup(node_id, nodegroup_id) for nodegroup_id in (set(node['nodegroup_ids']) - set(nodegroup_ids)): self.api.DeleteNodeFromNodeGroup(node_id, nodegroup_id) if self.check: # Check node node = self.api.GetNodes([node_id])[0] for field in node_fields: assert node[field] == node_fields[field] assert set(nodegroup_ids) == set(node['nodegroup_ids']) if self.verbose: print "Updated node", node_id print "Added node", node_id, "to node groups", nodegroup_ids def DeleteNodes(self): """ Delete any random nodes we may have added. """ for node_id in self.node_ids: # Remove from node groups node = self.api.GetNodes([node_id])[0] for nodegroup_id in node['nodegroup_ids']: self.api.DeleteNodeFromNodeGroup(node_id, nodegroup_id) if self.check: node = self.api.GetNodes([node_id])[0] assert not node['nodegroup_ids'] self.api.DeleteNode(node_id) if self.check: assert not self.api.GetNodes([node_id]) if self.verbose: print "Deleted node", node_id if self.check: assert not self.api.GetNodes(self.node_ids) self.node_ids = [] def AddNodeNetworks(self, per_node = 1): """ Add a number of random network interfaces to each node. """ network_methods = self.api.GetNetworkMethods() if not network_methods: raise Exception, "No network methods" network_types = self.api.GetNetworkTypes() if not network_types: raise Exception, "No network types" for node_id in self.node_ids: for i in range(per_node): method = random.sample(network_methods, 1)[0] type = random.sample(network_types, 1)[0] # Add node network nodenetwork_fields = random_nodenetwork(method, type) nodenetwork_id = self.api.AddNodeNetwork(node_id, nodenetwork_fields) # Should return a unique nodenetwork_id assert nodenetwork_id not in self.nodenetwork_ids self.nodenetwork_ids.append(nodenetwork_id) if self.check: # Check node network nodenetwork = self.api.GetNodeNetworks([nodenetwork_id])[0] for field in nodenetwork_fields: assert nodenetwork[field] == nodenetwork_fields[field] if self.verbose: print "Added node network", nodenetwork_id, "to node", node_id def UpdateNodeNetworks(self): """ Make random changes to any network interfaces we may have added. """ network_methods = self.api.GetNetworkMethods() if not network_methods: raise Exception, "No network methods" network_types = self.api.GetNetworkTypes() if not network_types: raise Exception, "No network types" for nodenetwork_id in self.nodenetwork_ids: method = random.sample(network_methods, 1)[0] type = random.sample(network_types, 1)[0] # Update nodenetwork nodenetwork_fields = random_nodenetwork(method, type) self.api.UpdateNodeNetwork(nodenetwork_id, nodenetwork_fields) if self.check: # Check nodenetwork nodenetwork = self.api.GetNodeNetworks([nodenetwork_id])[0] for field in nodenetwork_fields: assert nodenetwork[field] == nodenetwork_fields[field] if self.verbose: print "Updated node network", nodenetwork_id def DeleteNodeNetworks(self): """ Delete any random network interfaces we may have added. """ for nodenetwork_id in self.nodenetwork_ids: self.api.DeleteNodeNetwork(nodenetwork_id) if self.check: assert not self.api.GetNodeNetworks([nodenetwork_id]) if self.verbose: print "Deleted node network", nodenetwork_id if self.check: assert not self.api.GetNodeNetworks(self.nodenetwork_ids) self.nodenetwork_ids = [] def AddPCUs(self, per_site = 1): """ Add a number of random PCUs to each site. Each node at the site will be added to a port on the PCU if AddNodes() was previously run. """ for site_id in self.site_ids: for i in range(per_site): # Add PCU pcu_fields = random_pcu() pcu_id = self.api.AddPCU(site_id, pcu_fields) # Should return a unique pcu_id assert pcu_id not in self.pcu_ids self.pcu_ids.append(pcu_id) # Add each node at this site to a different port on this PCU site = self.api.GetSites([site_id])[0] port = randint(1, 10) for node_id in site['node_ids']: self.api.AddNodeToPCU(node_id, pcu_id, port) port += 1 if self.check: # Check PCU pcu = self.api.GetPCUs([pcu_id])[0] for field in pcu_fields: assert pcu[field] == pcu_fields[field] if self.verbose: print "Added PCU", pcu_id, "to site", site_id def UpdatePCUs(self): """ Make random changes to any PCUs we may have added. """ for pcu_id in self.pcu_ids: # Update PCU pcu_fields = random_pcu() self.api.UpdatePCU(pcu_id, pcu_fields) if self.check: # Check PCU pcu = self.api.GetPCUs([pcu_id])[0] for field in pcu_fields: assert pcu[field] == pcu_fields[field] if self.verbose: print "Updated PCU", pcu_id def DeletePCUs(self): """ Delete any random nodes we may have added. """ for pcu_id in self.pcu_ids: # Remove nodes from PCU pcu = self.api.GetPCUs([pcu_id])[0] for node_id in pcu['node_ids']: self.api.DeleteNodeFromPCU(node_id, pcu_id) if self.check: pcu = self.api.GetPCUs([pcu_id])[0] assert not pcu['node_ids'] self.api.DeletePCU(pcu_id) if self.check: assert not self.api.GetPCUs([pcu_id]) if self.verbose: print "Deleted PCU", pcu_id if self.check: assert not self.api.GetPCUs(self.pcu_ids) self.pcu_ids = [] def AddConfFiles(self, n = 10): """ Add a number of random global configuration files. """ conf_files = [] for i in range(n): # Add a random configuration file conf_files.append(random_conf_file()) if n: # Add a nodegroup override file nodegroup_conf_file = conf_files[0].copy() nodegroup_conf_file['source'] = randpath(255) conf_files.append(nodegroup_conf_file) # Add a node override file node_conf_file = conf_files[0].copy() node_conf_file['source'] = randpath(255) conf_files.append(node_conf_file) for conf_file_fields in conf_files: conf_file_id = self.api.AddConfFile(conf_file_fields) # Should return a unique conf_file_id assert conf_file_id not in self.conf_file_ids self.conf_file_ids.append(conf_file_id) # Add to nodegroup if conf_file_fields == nodegroup_conf_file and self.nodegroup_ids: nodegroup_id = random.sample(self.nodegroup_ids, 1)[0] self.api.AddConfFileToNodeGroup(conf_file_id, nodegroup_id) else: nodegroup_id = None # Add to node if conf_file_fields == node_conf_file and self.node_ids: node_id = random.sample(self.node_ids, 1)[0] self.api.AddConfFileToNode(conf_file_id, node_id) else: node_id = None if self.check: # Check configuration file conf_file = self.api.GetConfFiles([conf_file_id])[0] for field in conf_file_fields: assert conf_file[field] == conf_file_fields[field] if self.verbose: print "Added configuration file", conf_file_id, if nodegroup_id is not None: print "to node group", nodegroup_id, elif node_id is not None: print "to node", node_id, print def UpdateConfFiles(self): """ Make random changes to any configuration files we may have added. """ for conf_file_id in self.conf_file_ids: # Update configuration file conf_file_fields = random_conf_file() # Do not update dest so that it remains an override if set if 'dest' in conf_file_fields: del conf_file_fields['dest'] self.api.UpdateConfFile(conf_file_id, conf_file_fields) if self.check: # Check configuration file conf_file = self.api.GetConfFiles([conf_file_id])[0] for field in conf_file_fields: assert conf_file[field] == conf_file_fields[field] if self.verbose: print "Updated configuration file", conf_file_id def DeleteConfFiles(self): """ Delete any random configuration files we may have added. """ for conf_file_id in self.conf_file_ids: self.api.DeleteConfFile(conf_file_id) if self.check: assert not self.api.GetConfFiles([conf_file_id]) if self.verbose: print "Deleted configuration file", conf_file_id if self.check: assert not self.api.GetConfFiles(self.conf_file_ids) self.conf_file_ids = [] def AddSliceAttributeTypes(self, n = 10): """ Add a number of random slice attribute types. """ roles = self.api.GetRoles() if not roles: raise Exception, "No roles" role_ids = [role['role_id'] for role in roles] for i in range(n): attribute_type_fields = random_attribute_type(role_ids) attribute_type_id = self.api.AddSliceAttributeType(attribute_type_fields) # Should return a unique attribute_type_id assert attribute_type_id not in self.attribute_type_ids self.attribute_type_ids.append(attribute_type_id) if self.check: # Check slice attribute type attribute_type = self.api.GetSliceAttributeTypes([attribute_type_id])[0] for field in attribute_type_fields: assert attribute_type[field] == attribute_type_fields[field] if self.verbose: print "Added slice attribute type", attribute_type_id def UpdateSliceAttributeTypes(self): """ Make random changes to any slice attribute types we may have added. """ roles = self.api.GetRoles() if not roles: raise Exception, "No roles" role_ids = [role['role_id'] for role in roles] for attribute_type_id in self.attribute_type_ids: # Update slice attribute type attribute_type_fields = random_attribute_type(role_ids) self.api.UpdateSliceAttributeType(attribute_type_id, attribute_type_fields) if self.check: # Check slice attribute type attribute_type = self.api.GetSliceAttributeTypes([attribute_type_id])[0] for field in attribute_type_fields: assert attribute_type[field] == attribute_type_fields[field] if self.verbose: print "Updated slice attribute type", attribute_type_id def DeleteSliceAttributeTypes(self): """ Delete any random slice attribute types we may have added. """ for attribute_type_id in self.attribute_type_ids: self.api.DeleteSliceAttributeType(attribute_type_id) if self.check: assert not self.api.GetSliceAttributeTypes([attribute_type_id]) if self.verbose: print "Deleted slice attribute type", attribute_type_id if self.check: assert not self.api.GetSliceAttributeTypes(self.attribute_type_ids) self.attribute_type_ids = [] def AddSlices(self, per_site = 10): """ Add a number of random slices per site. """ for site in self.api.GetSites(self.site_ids): for i in range(min(per_site, site['max_slices'])): # Add slice slice_fields = random_slice(site['login_base']) slice_id = self.api.AddSlice(slice_fields) # Should return a unique slice_id assert slice_id not in self.slice_ids self.slice_ids.append(slice_id) # Add slice to a random set of nodes node_ids = random.sample(self.node_ids, randint(0, len(self.node_ids))) if node_ids: self.api.AddSliceToNodes(slice_id, node_ids) # Add random set of site users to slice person_ids = random.sample(site['person_ids'], randint(0, len(site['person_ids']))) for person_id in person_ids: self.api.AddPersonToSlice(person_id, slice_id) if self.check: # Check slice slice = self.api.GetSlices([slice_id])[0] for field in slice_fields: assert slice[field] == slice_fields[field] assert set(node_ids) == set(slice['node_ids']) assert set(person_ids) == set(slice['person_ids']) if self.verbose: print "Added slice", slice_id, "to site", site['site_id'], if node_ids: print "and nodes", node_ids, print if person_ids: print "Added users", site['person_ids'], "to slice", slice_id def UpdateSlices(self): """ Make random changes to any slices we may have added. """ for slice_id in self.slice_ids: # Update slice slice_fields = random_slice("unused") # Cannot change slice name if 'name' in slice_fields: del slice_fields['name'] self.api.UpdateSlice(slice_id, slice_fields) slice = self.api.GetSlices([slice_id])[0] # Add slice to a random set of nodes node_ids = random.sample(self.node_ids, randint(0, len(self.node_ids))) self.api.AddSliceToNodes(slice_id, list(set(node_ids) - set(slice['node_ids']))) self.api.DeleteSliceFromNodes(slice_id, list(set(slice['node_ids']) - set(node_ids))) # Add random set of users to slice person_ids = random.sample(self.person_ids, randint(0, len(self.person_ids))) for person_id in (set(person_ids) - set(slice['person_ids'])): self.api.AddPersonToSlice(person_id, slice_id) for person_id in (set(slice['person_ids']) - set(person_ids)): self.api.DeletePersonFromSlice(person_id, slice_id) if self.check: slice = self.api.GetSlices([slice_id])[0] for field in slice_fields: assert slice[field] == slice_fields[field] assert set(node_ids) == set(slice['node_ids']) assert set(person_ids) == set(slice['person_ids']) if self.verbose: print "Updated slice", slice_id print "Added nodes", node_ids, "to slice", slice_id print "Added persons", person_ids, "to slice", slice_id def DeleteSlices(self): """ Delete any random slices we may have added. """ for slice_id in self.slice_ids: self.api.DeleteSlice(slice_id) if self.check: assert not self.api.GetSlices([slice_id]) if self.verbose: print "Deleted slice", slice_id if self.check: assert not self.api.GetSlices(self.slice_ids) self.slice_ids = [] def AddSliceAttributes(self, per_slice = 2): """ Add a number of random slices per site. """ if not self.attribute_type_ids: return for slice_id in self.slice_ids: slice = self.api.GetSlices([slice_id])[0] for i in range(per_slice): # Set a random slice/sliver attribute for attribute_type_id in random.sample(self.attribute_type_ids, 1): value = randstr(16, letters + '_' + digits) # Make it a sliver attribute with 50% probability if slice['node_ids']: node_id = random.sample(slice['node_ids'] + [None] * len(slice['node_ids']), 1)[0] else: node_id = None # Add slice attribute if node_id is None: slice_attribute_id = self.api.AddSliceAttribute(slice_id, attribute_type_id, value) else: slice_attribute_id = self.api.AddSliceAttribute(slice_id, attribute_type_id, value, node_id) # Should return a unique slice_attribute_id assert slice_attribute_id not in self.slice_attribute_ids self.slice_attribute_ids.append(slice_attribute_id) if self.check: # Check slice attribute slice_attribute = self.api.GetSliceAttributes([slice_attribute_id])[0] for field in 'attribute_type_id', 'slice_id', 'node_id', 'slice_attribute_id', 'value': assert slice_attribute[field] == locals()[field] if self.verbose: print "Added slice attribute", slice_attribute_id, "of type", attribute_type_id, if node_id is not None: print "to node", node_id, print def UpdateSliceAttributes(self): """ Make random changes to any slice attributes we may have added. """ for slice_attribute_id in self.slice_attribute_ids: # Update slice attribute value = randstr(16, letters + '_' + digits) self.api.UpdateSliceAttribute(slice_attribute_id, value) # Check slice attribute again slice_attribute = self.api.GetSliceAttributes([slice_attribute_id])[0] assert slice_attribute['value'] == value if self.verbose: print "Updated slice attribute", slice_attribute_id def DeleteSliceAttributes(self): """ Delete any random slice attributes we may have added. """ for slice_attribute_id in self.slice_attribute_ids: self.api.DeleteSliceAttribute(slice_attribute_id) if self.check: assert not self.api.GetSliceAttributes([slice_attribute_id]) if self.verbose: print "Deleted slice attribute", slice_attribute_id if self.check: assert not self.api.GetSliceAttributes(self.slice_attribute_ids) self.slice_attribute_ids = [] def main(): parser = OptionParser() parser.add_option("-c", "--check", action = "store_true", default = False, help = "Check most actions (default: %default)") parser.add_option("-q", "--quiet", action = "store_true", default = False, help = "Be quiet (default: %default)") parser.add_option("-t", "--tiny", action = "store_true", default = False, help = "Run a tiny test (default: %default)") (options, args) = parser.parse_args() test = Test(api = Shell(), check = options.check, verbose = not options.quiet) if options.tiny: params = Test.tiny else: params = Test.default test.Run(**params) if __name__ == "__main__": main()