qaapi/qa/tests/api_unit_test.py

   1 #!/usr/bin/env /usr/share/plc_api/plcsh
   2 #
   3 # Test script example
   4 #
   5 # Copyright (C) 2006 The Trustees of Princeton University
   6 #
   7 #
   8
   9 from pprint import pprint
  10 from string import letters, digits, punctuation
  11 from traceback import print_exc
  12 import base64
  13 import os, sys
  14 import socket
  15 import xmlrpclib
  16 import time
  17
  18 from Test import Test
  19 from qa import utils
  20 from qa.Config import Config
  21 from qa.logger import Logfile, log
  22 from random import Random
  23
  24 random = Random()
  25
  26 config = Config()
  27
  28 try: boot_states = GetBootStates()
  29 except: boot_states = [u'boot', u'dbg', u'inst', u'new', u'rcnf', u'rins']
  30
  31 try: roles = [role['name'] for role in GetRoles()]
  32 except: roles = [u'admin', u'pi', u'user', u'tech']
  33
  34 try: methods = GetNetworkMethods()
  35 except: methods = [u'static', u'dhcp', u'proxy', u'tap', u'ipmi', u'unknown']
  36
  37 try: key_types = GetKeyTypes()
  38 except: key_types = [u'ssh']
  39
  40 try:types = GetNetworkTypes()
  41 except: types = [u'ipv4']
  42
  43 def randfloat(min = 0.0, max = 1.0):
  44     return float(min) + (random.random() * (float(max) - float(min)))
  45
  46 def randint(min = 0, max = 1):
  47     return int(randfloat(min, max + 1))
  48
  49 # See "2.2 Characters" in the XML specification:
  50 #
  51 # #x9 | #xA | #xD | [#x20-#xD7FF] | [#xE000-#xFFFD]
  52 # avoiding
  53 # [#x7F-#x84], [#x86-#x9F], [#xFDD0-#xFDDF]
  54
  55 ascii_xml_chars = map(unichr, [0x9, 0xA, 0xD])
  56 ascii_xml_chars += map(unichr, xrange(0x20, 0x7F - 1))
  57 low_xml_chars = list(ascii_xml_chars)
  58 low_xml_chars += map(unichr, xrange(0x84 + 1, 0x86 - 1))
  59 low_xml_chars += map(unichr, xrange(0x9F + 1, 0xFF))
  60 valid_xml_chars = list(low_xml_chars)
  61 valid_xml_chars += map(unichr, xrange(0xFF + 1, 0xD7FF))
  62 valid_xml_chars += map(unichr, xrange(0xE000, 0xFDD0 - 1))
  63 valid_xml_chars += map(unichr, xrange(0xFDDF + 1, 0xFFFD))
  64
  65 def randstr(length, pool = valid_xml_chars, encoding = "utf-8"):
  66     sample = random.sample(pool, min(length, len(pool)))
  67     while True:
  68         s = u''.join(sample)
  69         bytes = len(s.encode(encoding))
  70         if bytes > length:
  71             sample.pop()
  72         elif bytes < length:
  73             sample += random.sample(pool, min(length - bytes, len(pool)))
  74             random.shuffle(sample)
  75         else:
  76             break
  77     return s
  78
  79 def randhostname():
  80     # 1. Each part begins and ends with a letter or number.
  81     # 2. Each part except the last can contain letters, numbers, or hyphens.
  82     # 3. Each part is between 1 and 64 characters, including the trailing dot.
  83     # 4. At least two parts.
  84     # 5. Last part can only contain between 2 and 6 letters.
  85     hostname = 'a' + randstr(61, letters + digits + '-') + '1.' + \
  86                'b' + randstr(61, letters + digits + '-') + '2.' + \
  87                'c' + randstr(5, letters)
  88     return hostname
  89
  90 def randpath(length):
  91     parts = []
  92     for i in range(randint(1, 10)):
  93         parts.append(randstr(randint(1, 30), ascii_xml_chars))
  94     return os.sep.join(parts)[0:length]
  95
  96 def randemail():
  97     return (randstr(100, letters + digits) + "@" + randhostname()).lower()
  98
  99 def randkey(bits = 2048):
 100     key_types = ["ssh-dss", "ssh-rsa"]
 101     key_type = random.sample(key_types, 1)[0]
 102     return ' '.join([key_type,
 103                      base64.b64encode(''.join(randstr(bits / 8).encode("utf-8"))),
 104                      randemail()])
 105
 106 def random_site():
 107     return {
 108         'name': randstr(254),
 109         'abbreviated_name': randstr(50),
 110         'login_base': randstr(20, letters).lower(),
 111         'latitude': int(randfloat(-90.0, 90.0) * 1000) / 1000.0,
 112         'longitude': int(randfloat(-180.0, 180.0) * 1000) / 1000.0,
 113         }
 114
 115 def random_address_type():
 116     return {
 117         'name': randstr(20),
 118         'description': randstr(254),
 119         }
 120
 121 def random_address():
 122     return {
 123         'line1': randstr(254),
 124         'line2': randstr(254),
 125         'line3': randstr(254),
 126         'city': randstr(254),
 127         'state': randstr(254),
 128         'postalcode': randstr(64),
 129         'country': randstr(128),
 130         }
 131
 132 def random_person():
 133     return {
 134         'first_name': randstr(128),
 135         'last_name': randstr(128),
 136         'email': randemail(),
 137         'bio': randstr(254),
 138         # Accounts are disabled by default
 139         'enabled': False,
 140         'password': randstr(254),
 141         }
 142
 143 def random_key():
 144     return {
 145         'key_type': random.sample(key_types, 1)[0],
 146         'key': randkey()
 147         }
 148
 149 def random_slice():
 150     return {
 151         'name': site['login_base'] + "_" + randstr(11, letters).lower(),
 152         'url': "http://" + randhostname() + "/",
 153         'description': randstr(2048),
 154         }
 155
 156 def random_nodegroup():
 157     return {
 158         'name': randstr(50),
 159         'description': randstr(200),
 160         }
 161
 162 def random_node():
 163    return {
 164        'hostname': randhostname(),
 165        'boot_state': random.sample(boot_states, 1)[0],
 166        'model': randstr(255),
 167        'version': randstr(64),
 168        }
 169
 170 def random_nodenetwork():
 171     nodenetwork_fields = {
 172         'method': random.sample(methods, 1)[0],
 173         'type': random.sample(types, 1)[0],
 174         'bwlimit': randint(500000, 10000000),
 175         }
 176
 177     if method != 'dhcp':
 178         ip = randint(0, 0xffffffff)
 179         netmask = (0xffffffff << randint(2, 31)) & 0xffffffff
 180         network = ip & netmask
 181         broadcast = ((ip & netmask) | ~netmask) & 0xffffffff
 182         gateway = randint(network + 1, broadcast - 1)
 183         dns1 = randint(0, 0xffffffff)
 184
 185         for field in 'ip', 'netmask', 'network', 'broadcast', 'gateway', 'dns1':
 186             nodenetwork_fields[field] = socket.inet_ntoa(struct.pack('>L', locals()[field]))
 187
 188     return nodenetwork_fields
 189
 190 def random_pcu():
 191     return {
 192         'hostname': randhostname(),
 193         'ip': socket.inet_ntoa(struct.pack('>L', randint(0, 0xffffffff))),
 194         'protocol': randstr(16),
 195         'username': randstr(254),
 196         'password': randstr(254),
 197         'notes': randstr(254),
 198         'model': randstr(32),
 199         }
 200
 201 def random_conf_file():
 202     return {
 203         'enabled': bool(randint()),
 204         'source': randpath(255),
 205         'dest': randpath(255),
 206         'file_permissions': "%#o" % randint(0, 512),
 207         'file_owner': randstr(32, letters + '_' + digits),
 208         'file_group': randstr(32, letters + '_' + digits),
 209         'preinstall_cmd': randpath(100),
 210         'postinstall_cmd': randpath(100),
 211         'error_cmd': randpath(100),
 212         'ignore_cmd_errors': bool(randint()),
 213         'always_update': bool(randint()),
 214         }
 215
 216 def random_attribute_type():
 217     return {
 218         'name': randstr(100),
 219         'description': randstr(254),
 220         'min_role_id': random.sample(roles.values(), 1)[0],
 221         }
 222
 223 def isequal(object_fields, expected_fields):
 224     try:
 225         for field in expected_fields:
 226             assert field in object_fields
 227             assert object_fields[field] == expected_fields[field]
 228     except:
 229         return False
 230     return True
 231
 232 def islistequal(list1, list2):
 233     try:
 234         assert set(list1) == set(list2)
 235     except:
 236         return False
 237     return True
 238
 239 def isunique(id, id_list):
 240     try:
 241         assert id not in id_list
 242     except:
 243         return False
 244     return True
 245
 246 class api_unit_test(Test):
 247
 248     def call(self,
 249              sites = 2,
 250              boot_sates = 2,
 251              conf_files = 3,
 252              nodes = 4,
 253              address_types = 3,
 254              addresses = 2,
 255              persons = 4,
 256              keys = 3,
 257              key_types = 3,
 258              slices = 4,
 259              initscripts = 4,
 260             ):
 261         self.all_methods = set(system.listMethods())
 262         self.methods_tested = set()
 263         self.methods_failed = set()
 264
 265         # Begin testing methods
 266         try:
 267             try:
 268                 #self.boot_state_ids = self.BootStates(boot_states)
 269                 self.site_ids = self.Sites(sites)
 270                 #self.peer_ids = self.Peers(peers)
 271                 self.address_type_ids = self.AddressTypes(address_types)
 272                 self.address_ids = self.Addresses(addresses)
 273                 #self.conf_files = self.ConfFiles(conf_files)
 274                 #self.network_method_ids = self.NetworkMethods()
 275                 #self.network_type_ids = self.NetworkTypes()
 276                 #self.nodegroup_ids = self.NodeGroups()
 277                 self.node_ids = self.Nodes(nodes)
 278                 #self.node_network_ids = self.NodeNetworks(node_networks)
 279                 #self.node_network_setting_type_ids = self.NodeNetworkSettingsTypes(node_network_settings_types)
 280                 #self.node_network_setting_ids = self.NodeNetworkSettings(node_network_settings)
 281                 #self.pcu_protocol_types_ids = self.PCUProtocolTypes(pcu_protocol_types)
 282                 #self.pcus_ids = self.PCUs(pcus)
 283                 #self.pcu_types_ids = self.PCUTypes(pcu_types)
 284                 #self.role_ids = self.Roles(roles)
 285                 #self.key_types = self.KeyTypes(key_types)
 286                 #self.slice_attribute_type_ids = self.SliceAttributeTypes(slice_attribute_types)
 287                 #self.slice_instantiation_ids = self.SliceInstantiations(slice_instantiations)
 288                 self.slice_ids = self.Slices(slices)
 289                 #self.slice_attribute_ids = self.SliceAttributes(slice_attributes)
 290                 #self.initscript_ids = self.InitScripts(initscripts)
 291                 self.person_ids = self.Persons(persons)
 292                 self.key_ids = self.Keys(keys)
 293                 # self.message_ids = self.Messages(messages)
 294                 #self.NotifPersons()
 295                 # Test GetEventObject only
 296                 #self.event_object_ids = self.GetEventObjects()
 297                 #self.event_ids = self.GetEvents()
 298             except:
 299                 print_exc()
 300         finally:
 301             try:
 302                 self.cleanup()
 303             finally:
 304
 305                 logfile = Logfile("api-unittest.log")
 306                 methods_ok = list(self.methods_tested.difference(self.methods_failed))
 307                 methods_failed = list(self.methods_failed)
 308                 methods_untested = list(self.all_methods.difference(self.methods_tested))
 309                 methods_ok.sort()
 310                 methods_failed.sort()
 311                 methods_untested.sort()
 312                 print >> logfile, "\n".join([m+": [OK]" for m in  methods_ok])
 313                 print >> logfile, "\n".join([m+": [FAILED]" for m in methods_failed])
 314                 print >> logfile, "\n".join([m+": [Not Tested]" for m in  methods_untested])
 315
 316     def isequal(self, object_fields, expected_fields, method_name):
 317         try:
 318             for field in expected_fields:
 319                 assert field in object_fields
 320                 assert object_fields[field] == expected_fields[field]
 321         except:
 322             self.methods_failed.update([method_name])
 323             return False
 324         return True
 325
 326     def islistequal(self, list1, list2, method_name):
 327         try: assert set(list1) == set(list2)
 328         except:
 329             self.methods_failed.update([method_name])
 330             return False
 331         return True
 332
 333     def isunique(self, id, id_list, method_name):
 334         try: assert id not in id_list
 335         except:
 336             self.methods_failed.update([method_name])
 337             return False
 338         return True
 339
 340     def debug(self, method, method_name=None):
 341         if method_name is None:
 342              method_name = method.name
 343
 344         self.methods_tested.update([method_name])
 345         def wrapper(*args, **kwds):
 346             try:
 347                 return method(*args, **kwds)
 348             except:
 349                 self.methods_failed.update([method_name])
 350                 return None
 351
 352         return wrapper
 353
 354     def cleanup(self):
 355         if hasattr(self, 'initscript_ids'): self.DeleteInitScripts()
 356         if hasattr(self, 'slice_attribute_ids'): self.DeleteSliceAttributes()
 357         if hasattr(self, 'slice_ids'): self.DeleteSlices()
 358         if hasattr(self, 'slice_instantiation_ids'): self.DeleteSliceInstantiations()
 359         if hasattr(self, 'slice_attribute_type_ids'): self.DeleteSliceAttributeTypes()
 360         if hasattr(self, 'slice_attribute_ids'): self.DeleteSliceAttributes()
 361         if hasattr(self, 'key_type_ids'): self.DeleteKeyTypes()
 362         if hasattr(self, 'key_ids'): self.DeleteKeys()
 363         if hasattr(self, 'person_ids'): self.DeletePersons()
 364         if hasattr(self, 'role_ids'): self.DeleteRoles()
 365         if hasattr(self, 'pcu_type_ids'): self.DeletePCUTypes()
 366         if hasattr(self, 'pcu_ids'): self.DeletePCUs()
 367         if hasattr(self, 'pcu_protocol_type_ids'): self.DeleteProtocolTypes()
 368         if hasattr(self, 'node_network_setting_ids'): self.DeleteNodeNetworkSettings()
 369         if hasattr(self, 'address_ids'): self.DeleteAddresses()
 370         if hasattr(self, 'attress_type_ids'): self.DeleteAddressTypes()
 371         if hasattr(self, 'node_ids'): self.DeleteNodes()
 372         if hasattr(self, 'site_ids'): self.DeleteSites()
 373
 374
 375     def Sites(self, n=4):
 376         site_ids = []
 377         for i in range(n):
 378             # Add Site
 379             site_fields = random_site()
 380             AddSite = self.debug(shell.AddSite)
 381             site_id = AddSite(site_fields)
 382             if site_id is None: continue
 383
 384             # Should return a unique id
 385             self.isunique(site_id, site_ids, 'AddSite - isunique')
 386             site_ids.append(site_id)
 387             GetSites = self.debug(shell.GetSites)
 388             sites = GetSites([site_id])
 389             if sites is None: continue
 390             site = sites[0]
 391             self.isequal(site, site_fields, 'AddSite - isequal')
 392
 393             # Update site
 394             site_fields = random_site()
 395             UpdateSite = self.debug(shell.UpdateSite)
 396             result = UpdateSite(site_id, site_fields)
 397
 398             # Check again
 399             sites = GetSites([site_id])
 400             if sites is None: continue
 401             site = sites[0]
 402             self.isequal(site, site_fields, 'UpdateSite - isequal')
 403
 404         sites = GetSites(site_ids)
 405         if sites is not None:
 406             self.islistequal(site_ids, [site['site_id'] for site in sites], 'GetSites - isequal')
 407
 408         if self.config.verbose:
 409             utils.header("Added sites: %s" % site_ids)
 410
 411         return site_ids
 412
 413
 414     def DeleteSites(self):
 415         # Delete all sites
 416         DeleteSite = self.debug(shell.DeleteSite)
 417         for site_id in self.site_ids:
 418             result = DeleteSite(site_id)
 419
 420         # Check if sites are deleted
 421         GetSites = self.debug(shell.GetSites)
 422         sites = GetSites(self.site_ids)
 423         self.islistequal(sites, [], 'DeleteSite - check')
 424
 425         if self.config.verbose:
 426             utils.header("Deleted sites: %s" % self.site_ids)
 427
 428         self.site_ids = []
 429
 430     def Nodes(self, n=4):
 431         node_ids = []
 432         for i in range(n):
 433             # Add Node
 434             node_fields = random_node()
 435             site_id = random.sample(self.site_ids, 1)[0]
 436             AddNode = self.debug(shell.AddNode)
 437             node_id = AddNode(site_id, node_fields)
 438             if node_id is None: continue
 439
 440             # Should return a unique id
 441             self.isunique(node_id, node_ids, 'AddNode - isunique')
 442             node_ids.append(node_id)
 443
 444             # Check nodes
 445             GetNodes = self.debug(shell.GetNodes)
 446             nodes = GetNodes([node_id])
 447             if nodes is None: continue
 448             node = nodes[0]
 449             self.isequal(node, node_fields, 'AddNode - isequal')
 450
 451             # Update node
 452             node_fields = random_node()
 453             UpdateNode = self.debug(shell.UpdateNode)
 454             result = UpdateNode(node_id, node_fields)
 455
 456             # Check again
 457             nodes = GetNodes([node_id])
 458             if nodes is None: continue
 459             node = nodes[0]
 460             self.isequal(node, node_fields, 'UpdateNode - isequal')
 461
 462         nodes = GetNodes(node_ids)
 463         if nodes is not None:
 464             self.islistequal(node_ids, [node['node_id'] for node in nodes], 'GetNodes - isequal')
 465
 466         if self.config.verbose:
 467             utils.header("Added nodes: %s" % node_ids)
 468
 469         return node_ids
 470
 471     def DeleteNodes(self):
 472         DeleteNode = self.debug(shell.DeleteNode)
 473         for node_id in self.node_ids:
 474             result = DeleteNode(node_id)
 475
 476         # Check if nodes are deleted
 477         GetNodes = self.debug(shell.GetNodes)
 478         nodes = GetNodes(self.node_ids)
 479         self.islistequal(nodes, [], 'DeleteNode Check')
 480
 481         if self.config.verbose:
 482             utils.header("Deleted nodes: %s" % self.node_ids)
 483
 484         self.node_ids = []
 485
 486     def AddressTypes(self, n = 3):
 487         address_type_ids = []
 488         for i in range(n):
 489             address_type_fields = random_address_type()
 490             AddAddressType = self.debug(shell.AddAddressType)
 491             address_type_id = AddAddressType(address_type_fields)
 492             if address_type_id is None: continue
 493
 494             # Should return a unique address_type_id
 495             self.isunique(address_type_id, address_type_ids, 'AddAddressType - isunique')
 496             address_type_ids.append(address_type_id)
 497
 498             # Check address type
 499             GetAddressTypes = self.debug(shell.GetAddressTypes)
 500             address_types = GetAddressTypes([address_type_id])
 501             if address_types is None: continue
 502             address_type = address_types[0]
 503             self.isequal(address_type, address_type_fields, 'AddAddressType - isequal')
 504
 505             # Update address type
 506             address_type_fields = random_address_type()
 507             UpdateAddressType = self.debug(shell.UpdateAddressType)
 508             result = UpdateAddressType(address_type_id, address_type_fields)
 509             if result is None: continue
 510
 511             # Check address type again
 512             address_types = GetAddressTypes([address_type_id])
 513             if address_types is None: continue
 514             address_type = address_types[0]
 515             self.isequal(address_type, address_type_fields, 'UpdateAddressType - isequal')
 516
 517         # Check get all address types
 518         address_types = GetAddressTypes(address_type_ids)
 519         if address_types is not None:
 520             self.islistequal(address_type_ids, [address_type['address_type_id'] for address_type in address_types], 'GetAddressTypes - isequal')
 521
 522         if self.config.verbose:
 523             utils.header("Added address types: %s " % address_type_ids)
 524
 525         return address_type_ids
 526
 527     def DeleteAddressTypes(self):
 528
 529         DeleteAddressType = self.debug(shell.DeleteAddressType)
 530         for address_type_id in self.address_type_ids:
 531             DeleteAddressType(ddress_type_id)
 532
 533         GetAddressTypes = self.debug(shell.GetAddressTypes)
 534         address_types = GetAddressTypes(self.address_type_ids)
 535         self.islistequal(address_types, [], 'DeleteAddressType - check')
 536
 537         if self.config.verbose:
 538             utils.header("Deleted address types: " % self.address_type_ids)
 539
 540         self.address_type_ids = []
 541
 542     def Addresses(self, n = 3):
 543         address_ids = []
 544         for i in range(n):
 545             address_fields = random_address()
 546             site_id = random.sample(self.site_ids, 1)[0]
 547             AddSiteAddress = self.debug(shell.AddSiteAddress)
 548             address_id = AddSiteAddress(site_id, address_fields)
 549             if address_id is None: continue
 550
 551             # Should return a unique address_id
 552             self.isunique(address_id, address_ids, 'AddSiteAddress - isunique')
 553             address_ids.append(address_id)
 554
 555             # Check address
 556             GetAddresses = self.debug(shell.GetAddresses)
 557             addresses = GetAddresses([address_id])
 558             if addresses is None: continue
 559             address = addresses[0]
 560             self.isequal(address, address_fields, 'AddSiteAddress - isequal')
 561
 562             # Update address
 563             address_fields = random_address()
 564             UpdateAddress = self.debug(shell.UpdateAddress)
 565             result = UpdateAddress(address_id, address_fields)
 566
 567             # Check again
 568             addresses = GetAddresses([address_id])
 569             if addresses is None: continue
 570             address = addresses[0]
 571             self.isequal(address, address_fields, 'UpdateAddress - isequal')
 572
 573         addresses = GetAddresses(address_ids)
 574         if addresses is not None:
 575             self.islistequal(address_ids, [ad['address_id'] for ad in addresses], 'GetAddresses - isequal')
 576
 577         if self.config.verbose:
 578             utils.header("Added addresses: %s" % address_ids)
 579
 580         return address_ids
 581
 582     def DeleteAddresses(self):
 583
 584         DeleteAddress = self.debug(shell.DeleteAddress)
 585         # Delete site addresses
 586         for address_id in self.address_ids:
 587             result = DeleteAddress(address_id)
 588
 589         # Check
 590         GetAddresses = self.debug(shell.GetAddresses)
 591         addresses = GetAddresses(self.address_ids)
 592         self.islistequal(addresses, [], 'DeleteAddress - check')
 593         if self.config.verbose:
 594             utils.header("Deleted addresses: %s" % self.address_ids)
 595
 596         self.address_ids = []
 597
 598     def Slices(self, n = 3):
 599         slice_ids = []
 600         for i in range(n):
 601             # Add Site
 602             slice_fields = random_slice()
 603             AddSlice = self.debug(shell.Slice)
 604             slice_id = AddSlice(slice_fields)
 605             if slice_id is None: continue
 606
 607             # Should return a unique id
 608             self.isunique(slice_id, slice_ids, 'AddSlicel - isunique')
 609             slice_ids.append(slice_id)
 610             GetSlices = self.debug(shell.GetSlices)
 611             slices = GetSlices([slice_id])
 612             if slices is None: continue
 613             slice = slices[0]
 614             self.isequal(slice, slice_fields, 'AddSlice - isequal')
 615
 616             # Update slice
 617             slice_fields = random_slice()
 618             UpdateSite = self.debug(shell.UpdateSlice)
 619             result = UpdateSlice(slice_id, slice_fields)
 620
 621             # Check again
 622             slices = GetSites([slice_id])
 623             if slices is None: continue
 624             slice = slices[0]
 625             self.isequal(slice, slice_fields, 'UpdateSlice - isequal')
 626
 627             # XX Add attribute
 628
 629             # XX Add node
 630
 631         slices = GetSites(slice_ids)
 632         if slices is not None:
 633             self.islistequal(slice_ids, [slice['slice_id'] for slice in slices], 'GetSlices - isequal')
 634
 635         if self.config.verbose:
 636             utils.header("Added slices: %s" % slice_ids)
 637
 638         return slice_ids
 639
 640     def DeleteSlices(self):
 641
 642         # XX manually delete attributes for first slice
 643         slices = GetSlices(self.slice_ids, ['slice_attribute_ids', 'node_ids'])
 644
 645
 646         DeleteSlice = self.debug(shell.DeleteSlice)
 647         # Have DeleteSlice automatically delete attriubtes for the rest
 648         for slice_id in self.slice_ids:
 649             # Delete account
 650             DeleteSlice(slice_id)
 651
 652         # Check if persons are deleted
 653         GetSlices = self.debug(shell.GetSlices)
 654         slices = GetSlices(self.slice_ids)
 655         self.islistequal(slices, [], 'DeleteSlice - check')
 656
 657         if self.verbose:
 658             utils.header("Deleted slices: %s" % self.slice_ids)
 659
 660         self.slice_ids = []
 661
 662
 663     def Persons(self, n = 3):
 664
 665         person_ids = []
 666         for i in range(n):
 667
 668             # Add account
 669             person_fields = random_person()
 670             AddPerson = self.debug(shell.AddPerson)
 671             person_id = AddPerson(person_fields)
 672             if person_id is None: continue
 673
 674             # Should return a unique person_id
 675             self.isunique(person_id, person_ids, 'AddPerson - isunique')
 676             person_ids.append(person_id)
 677             GetPersons = self.debug(shell.GetPersons)
 678             persons = GetPersons([person_id])
 679             if persons is None: continue
 680             person = persons[0]
 681             self.isequal(person, person_fields, 'AddPerson - isequal')
 682
 683             # Update account
 684             person_fields = random_person()
 685             person_fields['enabled'] = True
 686             UpdatePerson = self.debug(shell.UpdatePerson)
 687             result = UpdatePerson(person_id, person_fields)
 688
 689             # Add random role
 690             AddRoleToPerson = self.debug(shell.AddRoleToPerson)
 691             role = random.sample(roles, 1)[0]
 692             result = AddRoleToPerson(role, person_id)
 693
 694             # Add key to person
 695             key = random_key()
 696             key_id = AddPersonKey = self.debug(shell.AddPersonKey)
 697             AddPersonKey(person_id, key)
 698
 699             # Add person to site
 700             site_id = random.sample(self.site_ids, 1)[0]
 701             AddPersonToSite = self.debug(shell.AddPersonToSite)
 702             AddPersonToSite(person_id, site_id)
 703
 704             # Add person to slice
 705             slice_id = random.sample(self.slice_ids, 1)[0]
 706             AddPersonToSlice = self.debug(self.AddPersonToSlice)
 707             AddPersonToSlice(person_id, slice_id)
 708
 709             # check role, key, site, slice
 710             persons = GetPersons([person_id], ['roles', 'key_ids', 'site_ids', 'slice_ids'])
 711             if persons is None or not persons: continue
 712             person = persons[0]
 713             self.islistequal([role], person['roles'], 'AddRoleToPerson - check')
 714             self.islistequal([key_id], person['key_ids'], 'AddPersonKey - check')
 715             self.islistequal([site_id], person['site_ids'], 'AddPersonToSite - check')
 716             self.islistequal([slice_id], person['slice_ids'], 'AddPersonToSlice - check')
 717
 718         persons = GetPersons(person_ids)
 719         if persons is not None:
 720             self.islistequal(person_ids, [p['person_id'] for p in persons], 'GetPersons - isequal')
 721
 722         if self.verbose:
 723             utils.header("Added users: %s" % self.person_ids)
 724
 725     def DeletePersons(self):
 726
 727         # Delete attributes manually for first person
 728         persons = GetPersons(self.person_ids, ['person_id' , 'key_ids', 'site_ids', 'slice_ids'])
 729         if persons is None or not persons: return 0
 730         person = persons[0]
 731
 732         # Delete role
 733         DeleteRoleFromPerson = self.debug(shell.DeleteRoleFromPerson)
 734         DeleteRoleFromPerson(person['role'], person['person_id'])
 735
 736         # Delete key
 737         DeleteKey = self.debug(shell.DeleteKey)
 738         DeleteKey(person['key_id'])
 739
 740         # Remove person from site
 741         DeletePersonFromSite = self.debug(shell.DeletePersonFromSite)
 742         DeletePersonFromSite(person['person_id'], person['site_id'])
 743
 744         # Remove person from slice
 745         DeletePersonFromSlice = self.debug(shell.DeletePersonFromSlice)
 746         DeletePersonFromSlice(person['person_id'], person['slice_id'])
 747
 748         # check role, key, site, slice
 749         persons = GetPersons([person['person_id']], ['roles', 'key_ids', 'site_ids', 'slice_ids'])
 750         if persons is None or not persons: return 0
 751         person = persons[0]
 752         self.islistequal([], person['roles'], 'DeleteRoleFromPerson - check')
 753         self.islistequal([], person['key_ids'], 'DeleteKey - check')
 754         self.islistequal([], person['site_ids'], 'DeletePersonFromSite - check')
 755         self.islistequal([], person['slice_ids'], 'DeletePersonFromSlice - check')
 756
 757         DeletePerson = self.debug(shell.DeletePerson)
 758         # Have DeletePeson automatically delete attriubtes for all other persons
 759         for person_id in self.person_ids:
 760             # Delete account
 761             DeletePerson(person_id)
 762
 763         # Check if persons are deleted
 764         GetPersons = self.debug(shell.GetPersons)
 765         persons = GetPersons(self.person_ids)
 766         self.islistequal(persons, [], 'DeletePerson - check')
 767
 768         if self.verbose:
 769             utils.header("Deleted users: %s" % self.person_ids)
 770
 771         self.person_ids = []
 772
 773
 774 if __name__ == '__main__':
 775     args = tuple(sys.argv[1:])
 776     api_unit_test()(*args)