#!/usr/bin/env plcsh
#
+# WARNING: as opposed to the rest of the python code in this repo
+# the current script runs on top of plcsh and so it is for now
+# pinned as python2 code
+#
+#
# Test script utility class
#
# Mark Huang <mlhuang@cs.princeton.edu>
# Copyright (C) 2006 The Trustees of Princeton University
#
-# $Id$
+
+# NOTE on porting to python3
+#
+# this file gets fed to plcsh on the tested myplc, so
+# it needs to remain python2 for now
#
from pprint import pprint
import base64
import struct
import os
-import xmlrpclib
+import xmlrpc.client
from PLC.Shell import Shell
# [#x7F-#x84], [#x86-#x9F], [#xFDD0-#xFDDF]
#
-ascii_xml_chars = map(unichr, [0x9, 0xA])
+ascii_xml_chars = list(map(chr, [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':
+if xmlrpc.client.loads(xmlrpc.client.dumps(('\r',)))[0][0] == '\r':
ascii_xml_chars.append('\r')
-ascii_xml_chars += map(unichr, xrange(0x20, 0x7F - 1))
+ascii_xml_chars += list(map(chr, range(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))
+low_xml_chars += list(map(chr, range(0x84 + 1, 0x86 - 1)))
+low_xml_chars += list(map(chr, range(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))
+valid_xml_chars += list(map(chr, range(0xFF + 1, 0xD7FF)))
+valid_xml_chars += list(map(chr, range(0xE000, 0xFDD0 - 1)))
+valid_xml_chars += list(map(chr, range(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)
+ s = ''.join(sample)
bytes = len(s.encode(encoding))
if bytes > length:
sample.pop()
hostname = 'a' + randstr(namelengths['hostname1'], letters + digits + '-') + '1.' + \
'b' + randstr(namelengths['hostname1'], letters + digits + '-') + '2.' + \
'c' + randstr(namelengths['hostname2'], letters)
- return hostname
+ return hostname.lower()
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]
+ return '/'.join(parts)[0:length]
def randemail(namelengths):
return (randstr(namelengths['email'], letters + digits) + "@" + randhostname(namelengths)).lower()
}
def random_site(namelengths):
- try:
- sitename=randstr(namelengths['sitename'],namelengths['sitename_contents'])
- except:
- sitename=randstr(namelengths['sitename'])
- try:
- abbreviated_name=randstr(namelengths['abbreviated_name'],namelengths['abbreviated_name_contents'])
- except:
- abbreviated_name=randstr(namelengths['abbreviated_name'])
+ sitename=randstr(namelengths['sitename'],namelengths['sitename_contents'])
+ abbreviated_name=randstr(namelengths['abbreviated_name'],namelengths['abbreviated_name_contents'])
+ print('nl[a] in random_site',namelengths['abbreviated_name'],'actual',len(abbreviated_name))
return {
'name': sitename,
'abbreviated_name': abbreviated_name,
'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),
def random_tag_type (role_ids):
return {'tagname': randstr(12,letters+digits),
'category':randstr(4,letters+digits)+'/'+randstr(6,letters+digits),
- 'min_role_id': random.sample(role_ids, 1)[0],
'description' : randstr(128,letters+digits+whitespace+punctuation),
}
def random_nodegroup():
return {'groupname' : randstr(30, letters+digits+whitespace) }
+def random_roles(role_ids):
+ nb_roles=len(role_ids)
+ return random.sample(role_ids,random.choice(list(range(1,nb_roles+1))))
+
tag_fields=['arch']
def random_node(node_types,boot_states,namelengths):
return {
'hostname2':5,
'login_base':20,
'sitename':254,
+ 'sitename_contents':letters+digits,
'abbreviated_name':50,
+ 'abbreviated_name_contents':letters+digits+whitespace+punctuation,
'model':255,
'first_name':128,
'last_name':128,
'hostname2':3,
'login_base':8,
'sitename':64,
- 'sitename_contents':letters+digits+whitespace+punctuation,
+ 'sitename_contents':letters+digits,
'abbreviated_name':24,
'abbreviated_name_contents':letters+digits+whitespace+punctuation,
'model':40,
self.verbose = verbose
self.preserve = preserve
self.federating = federating
-
+
self.site_ids = []
self.address_type_ids = []
self.address_ids = []
self.slice_tag_ids = []
def Cardinals (self):
- return [len(x) for x in (
+ return [len(x) for x in (
self.api.GetNodes({},['node_id']),
self.api.GetSites({},['site_id']),
self.api.GetPersons({},['person_id']),
"""
cardinals_before=self.Cardinals()
- print 'Cardinals before test (n,s,p,sl)',cardinals_before
+ print('Cardinals before test (n,s,p,sl)',cardinals_before)
self.Add(**kwds)
# if federating : we're done
if self.federating or self.preserve:
- print 'Preserving - update & delete skipped'
+ print('Preserving - update & delete skipped')
else:
self.Update()
self.Delete()
cardinals_after=self.Cardinals()
- print 'Cardinals after test (n,s,p,sl)',cardinals_after
+ print('Cardinals after test (n,s,p,sl)',cardinals_after)
if cardinals_before != cardinals_after:
- raise Exception, 'cardinals before and after differ - check deletion mechanisms'
+ raise Exception('cardinals before and after differ - check deletion mechanisms')
def Add(self, **kwds):
"""
self.AddConfFiles(sizes['conf_files'])
self.AddSlices(sizes['slices_per_site'])
self.AddSliceTags(sizes['attributes_per_slice'])
-
+
else:
self.RecordStatus()
self.AddSites(sizes['sites'])
self.DeleteAddressTypes()
self.DeleteSites()
- # record current (old) objects
+ # record current (old) objects
def RecordStatus (self):
self.old_site_ids = [ s['site_id'] for s in self.api.GetSites({},['site_id']) ]
self.old_person_ids = [ s['person_id'] for s in self.api.GetPersons({},['person_id']) ]
peer_id=self.api.AddPeer (random_peer())
peer = GetPeers([peer_id])[0]
if self.verbose:
- print "Added peer",peer_id
+ print("Added peer",peer_id)
# add new sites (the ones not in self.site_ids) in the peer
# cheating a bit
assert site[field] == site_fields[field]
if self.verbose:
- print "Added site", site_id
+ print("Added site", site_id)
def UpdateSites(self):
"""
assert site[field] == site_fields[field]
if self.verbose:
- print "Updated site", site_id
+ print("Updated site", site_id)
def DeleteSites(self):
"""
assert not self.api.GetSites([site_id])
if self.verbose:
- print "Deleted site", site_id
+ print("Deleted site", site_id)
if self.check:
assert not self.api.GetSites(self.site_ids)
assert address_type[field] == address_type_fields[field]
if self.verbose:
- print "Added address type", address_type_id
+ print("Added address type", address_type_id)
def UpdateAddressTypes(self):
"""
assert address_type[field] == address_type_fields[field]
if self.verbose:
- print "Updated address_type", address_type_id
+ print("Updated address_type", address_type_id)
def DeleteAddressTypes(self):
"""
assert not self.api.GetAddressTypes([address_type_id])
if self.verbose:
- print "Deleted address type", address_type_id
+ print("Deleted address type", address_type_id)
if self.check:
assert not self.api.GetAddressTypes(self.address_type_ids)
assert address[field] == address_fields[field]
if self.verbose:
- print "Added address", address_id, "to site", site_id
+ print("Added address", address_id, "to site", site_id)
def UpdateAddresses(self):
"""
assert address[field] == address_fields[field]
if self.verbose:
- print "Updated address", address_id
+ print("Updated address", address_id)
def DeleteAddresses(self):
"""
assert not self.api.GetAddresses([address_id])
if self.verbose:
- print "Deleted address", address_id
+ print("Deleted address", address_id)
if self.check:
assert not self.api.GetAddresses(self.address_ids)
assert person['site_ids'][0] == site_id
if self.verbose:
- print "Added user", person_id, "to site", site_id
+ print("Added user", person_id, "to site", site_id)
def UpdatePersons(self):
"""
assert person[field] == person_fields[field]
if self.verbose:
- print "Updated person", person_id
+ print("Updated person", person_id)
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
+ print("Updated person", person_id, "to sites", site_ids)
def DeletePersons(self):
"""
self.api.DeletePerson(person_id)
if self.check:
- assert not self.api.GetPersons([person_id])
+ assert not self.api.GetPersons([person_id])
if self.verbose:
- print "Deleted user", person_id
+ print("Deleted user", person_id)
if self.check:
assert not self.api.GetPersons(self.person_ids)
key_types = self.api.GetKeyTypes()
if not key_types:
- raise Exception, "No key types"
+ raise Exception("No key types")
for person_id in self.person_ids:
for i in range(per_person):
try:
key_id = self.api.AddPersonKey(person_id, key_fields)
assert False
- except Exception, e:
+ except Exception as e:
pass
if self.verbose:
- print "Added key", key_id, "to user", person_id
+ print("Added key", key_id, "to user", person_id)
def UpdateKeys(self):
"""
key_types = self.api.GetKeyTypes()
if not key_types:
- raise Exception, "No key types"
+ raise Exception("No key types")
for key_id in self.key_ids:
# Update key
assert key[field] == key_fields[field]
if self.verbose:
- print "Updated key", key_id
+ print("Updated key", key_id)
def DeleteKeys(self):
"""
assert not self.api.GetKeys([key_id])
if self.verbose:
- print "Deleted key", key_id
+ print("Deleted key", key_id)
if self.check:
assert not self.api.GetKeys(self.key_ids)
# locate tag type
tag_type_id = self.nodegroup_type_ids[i]
tagname=self.api.GetTagTypes([tag_type_id])[0]['tagname']
-
+
# Add node group
groupname = random_nodegroup() ['groupname']
value = 'yes'
assert nodegroup['value'] == value
if self.verbose:
- print "Added node group", nodegroup_id
+ print("Added node group", nodegroup_id)
def UpdateNodeGroups(self):
"""
assert nodegroup[field] == nodegroup_fields[field]
if self.verbose:
- print "Updated node group", nodegroup_id
+ print("Updated node group", nodegroup_id)
def DeleteNodeGroups(self):
"""
assert not self.api.GetNodeGroups([nodegroup_id])
if self.verbose:
- print "Deleted node group", nodegroup_id
+ print("Deleted node group", nodegroup_id)
if self.check:
assert not self.api.GetNodeGroups(self.nodegroup_ids)
be added to a random node group if AddNodeGroups() was
previously run.
"""
-
+
node_types = self.api.GetNodeTypes()
if not node_types:
- raise Exception, "No node types"
+ raise Exception("No node types")
boot_states = self.api.GetBootStates()
if not boot_states:
- raise Exception, "No boot states"
+ raise Exception("No boot states")
for site_id in self.site_ids:
for i in range(per_site):
assert node[field] == node_fields[field]
if self.verbose:
- print "Added node", node_id
+ print("Added node", node_id)
def UpdateNodes(self):
"""
node_types = self.api.GetNodeTypes()
if not node_types:
- raise Exception, "No node types"
+ raise Exception("No node types")
boot_states = self.api.GetBootStates()
if not boot_states:
- raise Exception, "No boot states"
+ raise Exception("No boot states")
for node_id in self.node_ids:
# Update node
node_fields = random_node(node_types,boot_states,self.namelengths)
self.api.UpdateNode(node_id, node_fields)
-
+
node = self.api.GetNodes([node_id])[0]
# Add to a random set of node groups
for field in node_fields:
if field not in tag_fields:
if node[field] != node_fields[field]:
- raise Exception, "Unexpected field %s in node after GetNodes()"%field
+ raise Exception("Unexpected field %s in node after GetNodes()"%field)
assert set(nodegroup_ids) == set(node['nodegroup_ids'])
# again but we are now fetching 'arch' explicitly
- node2 = self.api.GetNodes([node_id],node_fields.keys())[0]
+ node2 = self.api.GetNodes([node_id],list(node_fields.keys()))[0]
for field in node_fields:
if node2[field] != node_fields[field]:
- raise Exception, "Unexpected field %s in node after GetNodes(tags)"%field
+ raise Exception("Unexpected field %s in node after GetNodes(tags)"%field)
if self.verbose:
- print "Updated node", node_id
+ print("Updated node", node_id)
def DeleteNodes(self):
"""
for node_id in self.node_ids:
# Remove from node groups
node = self.api.GetNodes([node_id])[0]
- for node_tag in GetNodeTags ( {'node_id': node_id} ):
+ for node_tag in self.api.GetNodeTags ( {'node_id': node_id} ):
self.api.UpdateNodeTag(node_tag['node_tag_id'],'')
if self.check:
assert not self.api.GetNodes([node_id])
if self.verbose:
- print "Deleted node", node_id
+ print("Deleted node", node_id)
if self.check:
assert not self.api.GetNodes(self.node_ids)
network_methods = self.api.GetNetworkMethods()
if not network_methods:
- raise Exception, "No network methods"
-
+ raise Exception("No network methods")
+
network_types = self.api.GetNetworkTypes()
if not network_types:
- raise Exception, "No network types"
+ raise Exception("No network types")
for node_id in self.node_ids:
for i in range(per_node):
assert interface[field] == interface_fields[field]
if self.verbose:
- print "Added interface", interface_id, "to node", node_id
+ print("Added interface", interface_id, "to node", node_id)
def UpdateInterfaces(self):
"""
network_methods = self.api.GetNetworkMethods()
if not network_methods:
- raise Exception, "No network methods"
-
+ raise Exception("No network methods")
+
network_types = self.api.GetNetworkTypes()
if not network_types:
- raise Exception, "No network types"
+ raise Exception("No network types")
for interface_id in self.interface_ids:
method = random.sample(network_methods, 1)[0]
assert interface[field] == interface_fields[field]
if self.verbose:
- print "Updated interface", interface_id
+ print("Updated interface", interface_id)
def DeleteInterfaces(self):
"""
assert not self.api.GetInterfaces([interface_id])
if self.verbose:
- print "Deleted interface", interface_id
+ print("Deleted interface", interface_id)
if self.check:
assert not self.api.GetInterfaces(self.interface_ids)
self.interface_ids = []
-
+
def AddIlinks (self, n):
"""
Add random links between interfaces.
self.ilink_ids.append(ilink_id)
if self.verbose:
- print 'Added Ilink',ilink_id,' - attached interface',src,'to',dst
+ print('Added Ilink',ilink_id,' - attached interface',src,'to',dst)
if self.check:
retrieve=GetIlinks({'src_interface_id':src,'dst_interface_id':dst,
assert ilink['value'] == new_value
if self.verbose:
- print 'Updated Ilink',ilink_id
+ print('Updated Ilink',ilink_id)
def DeleteIlinks (self):
for ilink_id in self.ilink_ids:
assert not self.api.GetIlinks({'ilink_id':ilink_id})
if self.verbose:
- print 'Deleted Ilink',ilink_id
+ print('Deleted Ilink',ilink_id)
if self.check:
assert not self.api.GetIlinks(self.ilink_ids)
assert pcu[field] == pcu_fields[field]
if self.verbose:
- print "Added PCU", pcu_id, "to site", site_id
+ print("Added PCU", pcu_id, "to site", site_id)
def UpdatePCUs(self):
"""
assert pcu[field] == pcu_fields[field]
if self.verbose:
- print "Updated PCU", pcu_id
+ print("Updated PCU", pcu_id)
def DeletePCUs(self):
"""
assert not self.api.GetPCUs([pcu_id])
if self.verbose:
- print "Deleted PCU", pcu_id
+ print("Deleted PCU", pcu_id)
if self.check:
assert not self.api.GetPCUs(self.pcu_ids)
assert conf_file[field] == conf_file_fields[field]
if self.verbose:
- print "Added configuration file", conf_file_id,
+ print("Added configuration file", conf_file_id, end=' ')
if nodegroup_id is not None:
- print "to node group", nodegroup_id,
+ print("to node group", nodegroup_id, end=' ')
elif node_id is not None:
- print "to node", node_id,
- print
+ print("to node", node_id, end=' ')
+ print()
def UpdateConfFiles(self):
"""
assert conf_file[field] == conf_file_fields[field]
if self.verbose:
- print "Updated configuration file", conf_file_id
+ print("Updated configuration file", conf_file_id)
def DeleteConfFiles(self):
"""
assert not self.api.GetConfFiles([conf_file_id])
if self.verbose:
- print "Deleted configuration file", conf_file_id
+ print("Deleted configuration file", conf_file_id)
if self.check:
assert not self.api.GetConfFiles(self.conf_file_ids)
def AddTagTypes(self,n_sa,n_ng,n_il):
"""
- Add as many tag types as there are nodegroups,
+ Add as many tag types as there are nodegroups,
will use value=yes for each nodegroup
"""
roles = self.api.GetRoles()
if not roles:
- raise Exception, "No roles"
+ raise Exception("No roles")
role_ids = [role['role_id'] for role in roles]
for i in range (n_sa + n_ng + n_il):
self.slice_type_ids + \
self.nodegroup_type_ids + \
self.ilink_type_ids
-
+
+ tt_role_ids=random_roles(role_ids)
+ for tt_role_id in tt_role_ids:
+ self.api.AddRoleToTagType(tt_role_id,tag_type_id)
+
if i < n_sa:
self.slice_type_ids.append(tag_type_id)
elif i < n_sa+n_ng :
tag_type = self.api.GetTagTypes([tag_type_id])[0]
for field in tag_type_fields:
assert tag_type[field] == tag_type_fields[field]
+ for tt_role_id in tt_role_ids:
+ assert tt_role_id in tag_type['role_ids']
if self.verbose:
- print "Updated slice attribute type", tag_type_id
+ print("Created tag type", tag_type_id)
def UpdateTagTypes(self):
"""
roles = self.api.GetRoles()
if not roles:
- raise Exception, "No roles"
+ raise Exception("No roles")
role_ids = [role['role_id'] for role in roles]
for tag_type_id in self.slice_type_ids + self.nodegroup_type_ids + self.ilink_type_ids:
for field in tag_type_fields:
assert tag_type[field] == tag_type_fields[field]
if self.verbose:
- print "Updated slice attribute type", tag_type_id
+ print("Updated tag type", tag_type_id)
def DeleteTagTypes(self):
"""
assert not self.api.GetTagTypes([tag_type_id])
if self.verbose:
- print "Deleted slice attribute type", tag_type_id
+ print("Deleted tag type", tag_type_id)
if self.check:
assert not self.api.GetTagTypes(self.slice_type_ids+self.nodegroup_type_ids+self.ilink_type_ids)
assert set(person_ids) == set(slice['person_ids'])
if self.verbose:
- print "Added slice", slice_id, "to site", site['site_id'],
+ print("Added slice", slice_id, "to site", site['site_id'], end=' ')
if node_ids:
- print "and nodes", node_ids,
- print
+ print("and nodes", node_ids, end=' ')
+ print()
if person_ids:
- print "Added users", site['person_ids'], "to slice", slice_id
+ print("Added users", site['person_ids'], "to slice", slice_id)
def UpdateSlices(self):
"""
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
+ 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):
"""
assert not self.api.GetSlices([slice_id])
if self.verbose:
- print "Deleted slice", slice_id
+ print("Deleted slice", slice_id)
if self.check:
assert not self.api.GetSlices(self.slice_ids)
assert slice_tag[field] == locals()[field]
if self.verbose:
- print "Added slice attribute", slice_tag_id, "of type", tag_type_id,
+ print("Added slice attribute", slice_tag_id, "of type", tag_type_id, end=' ')
if node_id is not None:
- print "to node", node_id,
- print
-
+ print("to node", node_id, end=' ')
+ print()
+
def UpdateSliceTags(self):
"""
Make random changes to any slice attributes we may have added.
assert slice_tag['value'] == value
if self.verbose:
- print "Updated slice attribute", slice_tag_id
+ print("Updated slice attribute", slice_tag_id)
def DeleteSliceTags(self):
"""
assert not self.api.GetSliceTags([slice_tag_id])
if self.verbose:
- print "Deleted slice attribute", slice_tag_id
+ print("Deleted slice attribute", slice_tag_id)
if self.check:
assert not self.api.GetSliceTags(self.slice_tag_ids)
self.slice_tag_ids = []
+ # convenience for cleaning up
+ # not exactly accurate -- use on test plcs only
+ def WipeSitesFromLength(self):
+ for site in self.api.GetSites():
+ abbrev=site['abbreviated_name']
+# print 'matching',len(abbrev),'against',self.namelengths['abbreviated_name']
+ if len(abbrev)==self.namelengths['abbreviated_name']:
+# if len(abbrev)==17:
+ print('wiping site %d (%s)'%(site['site_id'],site['name']))
+ self.api.DeleteSite(site['site_id'])
+
def main():
parser = OptionParser()
- parser.add_option("-c", "--check", action = "store_true", default = False,
+ 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,
+ parser.add_option("-q", "--quiet", action = "store_true", default = False,
help = "Be quiet (default: %default)")
parser.add_option("-p","--preserve", action="store_true", default =False,
help = "Do not delete created objects")
- parser.add_option("-t", "--tiny", action = "store_true", default = False,
+ parser.add_option("-t", "--tiny", action = "store_true", default = False,
help = "Run a tiny test (default: %default)")
- parser.add_option("-l", "--large", action = "store_true", default = False,
+ parser.add_option("-l", "--large", action = "store_true", default = False,
help = "Run a large test (default: %default)")
- parser.add_option("-x", "--xlarge", action = "store_true", default = False,
+ parser.add_option("-x", "--xlarge", action = "store_true", default = False,
help = "Run an XL test (default: %default)")
- parser.add_option("-s", "--short-names", action="store_true", dest="short_names", default = False,
+ parser.add_option("-s", "--short-names", action="store_true", dest="short_names", default = False,
help = "Generate smaller names for checking UI rendering")
parser.add_option ("-f", "--foreign", action="store_true", dest="federating", default = False,
help = "Create a fake peer and add items in it (no update, no delete)")
+ parser.add_option ("-w", "--wipe", action="store_true", dest="wipe", default = False,
+ help = "Wipe sites whose abbrev matches what the tests created")
(options, args) = parser.parse_args()
test = Test(api = Shell(),
else:
test.namelengths = Test.namelengths_default
+ if options.wipe:
+ test.WipeSitesFromLength()
+ return
+
if options.tiny:
sizes = Test.sizes_tiny
elif options.large:
sizes = Test.sizes_xlarge
else:
sizes = Test.sizes_default
-
test.Run(**sizes)
if __name__ == "__main__":
git://git.onelab.eu / tests.git / blobdiff