import random
import string
import unittest
-import sfa.util.xmlrpcprotocol as xmlrpc
+
from unittest import TestCase
from optparse import OptionParser
-from sfa.util.xmlrpcprotocol import ServerException
-from sfa.util.namespace import *
+from sfa.util.xrn import get_authority
from sfa.util.config import *
from sfa.trust.certificate import *
from sfa.trust.credential import *
-from sfa.util.sfaticket import *
-from sfa.util.rspec import *
+from sfa.trust.sfaticket import SfaTicket
from sfa.client import sfi
+from sfa.client.sfaserverproxy import SfaServerProxy, ServerException
def random_string(size):
return "".join(random.sample(string.letters, size))
self.cert.save_to_file(cert_file)
SFI_AGGREGATE = config.SFI_SM.replace('12347', '12346')
SFI_CM = 'http://' + options.cm_host + ':12346'
- self.registry = xmlrpc.get_server(config.SFI_REGISTRY, key_file, cert_file)
- self.aggregate = xmlrpc.get_server(SFI_AGGREGATE, key_file, cert_file)
- self.sm = xmlrpc.get_server(config.SFI_SM, key_file, cert_file)
- self.cm = xmlrpc.get_server(SFI_CM, key_file, cert_file)
+ self.registry = SfaServerProxy(config.SFI_REGISTRY, key_file, cert_file)
+ self.aggregate = SfaServerProxy(SFI_AGGREGATE, key_file, cert_file)
+ self.sm = SfaServerProxy(config.SFI_SM, key_file, cert_file)
+ self.cm = SfaServerProxy(SFI_CM, key_file, cert_file)
self.hrn = config.SFI_USER
# XX defaulting to user, but this should be configurable so we can
# test from components persepctive
self.type = 'user'
- self.credential = self.get_credential(self.hrn)
+ self.credential = self.GetCredential(self.hrn)
- def get_credential(self, hrn = None, type = 'user'):
+ def GetCredential(self, hrn = None, type = 'user'):
if not hrn: hrn = self.hrn
if hrn == self.hrn:
cert = self.cert.save_to_string(save_parents=True)
return credential
else:
if not self.credential:
- self.credential = self.get_credential(self.hrn, 'user')
- return self.registry.get_credential(self.credential, type, hrn)
+ self.credential = self.GetCredential(self.hrn, 'user')
+ return self.registry.GetCredential(self.credential, type, hrn)
class BasicTestCase(unittest.TestCase):
def __init__(self, testname, client, test_slice=None):
BasicTestCase.setUp(self)
def testGetSelfCredential(self):
- cred = self.client.get_credential()
+ cred = self.client.GetCredential()
# this will raise an openssl error if the credential string isnt valid
Credential(string=cred)
def testRegister(self):
authority = get_authority(self.hrn)
- auth_cred = self.client.get_credential(authority, 'authority')
+ auth_cred = self.client.GetCredential(authority, 'authority')
auth_record = {'hrn': '.'.join([authority, random_string(10).lower()]),
'type': 'authority'}
node_record = {'hrn': '.'.join([authority, random_string(10)]),
all_records = [auth_record, node_record, slice_record, user_record]
for record in all_records:
try:
- self.registry.register(auth_cred, record)
- self.registry.resolve(self.credential, record['hrn'])
+ self.registry.Register(auth_cred, record)
+ self.registry.Resolve(self.credential, record['hrn'])
except:
raise
finally:
- try: self.registry.remove(auth_cred, record['type'], record['hrn'])
+ try: self.registry.Remove(auth_cred, record['type'], record['hrn'])
except: pass
def testUpdate(self):
authority = get_authority(self.hrn)
- auth_cred = self.client.get_credential(authority, 'authority')
- records = self.registry.resolve(self.credential, self.hrn)
+ auth_cred = self.client.GetCredential(authority, 'authority')
+ records = self.registry.Resolve(self.credential, self.hrn)
if not records: assert False
record = records[0]
self.registry.update(auth_cred, record)
def testResolve(self):
authority = get_authority(self.hrn)
- self.registry.resolve(self.credential, self.hrn)
+ self.registry.Resolve(self.credential, self.hrn)
def testRemove(self):
authority = get_authority(self.hrn)
- auth_cred = self.client.get_credential(authority, 'authority')
+ auth_cred = self.client.GetCredential(authority, 'authority')
record = {'hrn': ".".join([authority, random_string(10)]),
'type': 'slice'}
- self.registry.register(auth_cred, record)
- self.registry.remove(auth_cred, record['type'], record['hrn'])
+ self.registry.Register(auth_cred, record)
+ self.registry.Remove(auth_cred, record['type'], record['hrn'])
# should generate an exception
try:
- self.registry.resolve(self.credential, record['hrn'])
+ self.registry.Resolve(self.credential, record['hrn'])
assert False
except:
assert True
def testList(self):
authority = get_authority(self.client.hrn)
- self.registry.list(self.credential, authority)
+ self.registry.List(self.credential, authority)
def testGetRegistries(self):
self.registry.get_registries(self.credential)
BasicTestCase.setUp(self)
def testGetSlices(self):
- self.aggregate.get_slices(self.credential)
+ self.aggregate.ListSlices(self.credential)
def testGetResources(self):
# available resources
def testCreateSlice(self):
# get availabel resources
rspec = self.aggregate.get_resources(self.credential)
- slice_credential = self.client.get_credential(self.slice['hrn'], 'slice')
- self.aggregate.create_slice(slice_credential, self.slice['hrn'], rspec)
+ slice_credential = self.client.GetCredential(self.slice['hrn'], 'slice')
+ self.aggregate.CreateSliver(slice_credential, self.slice['hrn'], rspec)
def testDeleteSlice(self):
- slice_credential = self.client.get_credential(self.slice['hrn'], 'slice')
- self.aggregate.delete_slice(slice_credential, self.slice['hrn'])
+ slice_credential = self.client.GetCredential(self.slice['hrn'], 'slice')
+ self.aggregate.DeleteSliver(slice_credential, self.slice['hrn'],"call-id-delete-slice")
def testGetTicket(self):
- slice_credential = self.client.get_credential(self.slice['hrn'], 'slice')
+ slice_credential = self.client.GetCredential(self.slice['hrn'], 'slice')
rspec = self.aggregate.get_resources(self.credential)
- ticket = self.aggregate.get_ticket(slice_credential, self.slice['hrn'], rspec)
+ ticket = self.aggregate.GetTicket(slice_credential, self.slice['hrn'], rspec)
# will raise an exception if the ticket inst valid
SfaTicket(string=ticket)
class ComponentTest(BasicTestCase):
def setUp(self):
BasicTestCase.setUp(self)
- self.slice_cred = self.client.get_credential(self.slice['hrn'], 'slice')
+ self.slice_cred = self.client.GetCredential(self.slice['hrn'], 'slice')
def testStartSlice(self):
self.cm.start_slice(self.slice_cred, self.slice['hrn'])
self.cm.stop_slice(self.slice_cred, self.slice['hrn'])
def testDeleteSlice(self):
- self.cm.delete_slice(self.slice_cred, self.slice['hrn'])
+ self.cm.DeleteSliver(self.slice_cred, self.slice['hrn'],"call-id-delete-slice-cm")
def testRestartSlice(self):
self.cm.restart_slice(self.slice_cred, self.slice['hrn'])
def testGetSlices(self):
- self.cm.get_slices(self.slice_cred, self.slice['hrn'])
+ self.cm.ListSlices(self.slice_cred, self.slice['hrn'])
def testRedeemTicket(self):
rspec = self.aggregate.get_resources(self.credential)
- ticket = self.aggregate.get_ticket(slice_cred, self.slice['hrn'], rspec)
+ ticket = self.aggregate.GetTicket(slice_cred, self.slice['hrn'], rspec)
self.cm.redeem_ticket(slice_cred, ticket)
def test_names(testcase):
return [name for name in dir(testcase) if name.startswith('test')]
-def create_slice(client):
+def CreateSliver(client):
# register a slice that will be used for some test
authority = get_authority(client.hrn)
- auth_cred = client.get_credential(authority, 'authority')
+ auth_cred = client.GetCredential(authority, 'authority')
slice_record = {'hrn': ".".join([authority, random_string(10)]),
'type': 'slice', 'researcher': [client.hrn]}
- client.registry.register(auth_cred, slice_record)
+ client.registry.Register(auth_cred, slice_record)
return slice_record
-def delete_slice(cleint, slice):
+def DeleteSliver(client, slice):
authority = get_authority(client.hrn)
- auth_cred = client.get_credential(authority, 'authority')
+ auth_cred = client.GetCredential(authority, 'authority')
if slice:
- client.registry.remove(auth_cred, 'slice', slice['hrn'])
+ client.registry.Remove(auth_cred, 'slice', slice['hrn'])
if __name__ == '__main__':
# create the test slice if necessary
if options.all or options.slicemgr or options.aggregate \
or options.component:
- test_slice = create_slice(client)
+ test_slice = CreateSliver(client)
if options.registry or options.all:
for name in test_names(RegistryTest):
unittest.TextTestRunner(verbosity=2).run(suite)
# remove teset slice
- delete_slice(client, test_slice)
+ DeleteSliver(client, test_slice)