tests/testInterfaces.py

   1 #!/usr/bin/python
   2 import sys
   3 import os
   4 import sfa.util.xmlrpcprotocol as xmlrpc
   5 import unittest
   6 from unittest import TestCase
   7 from optparse import OptionParser
   8 from sfa.util.xmlrpcprotocol import ServerException
   9 from sfa.util.namespace import *
  10 from sfa.util.config import *
  11 from sfa.trust.certificate import *
  12 from sfa.client import sfi
  13
  14 class Client:
  15     registry = None
  16     aggregate = None
  17     sm = None
  18     cm = None
  19     user = None
  20     key = None
  21     cert = None
  22     credential = None
  23     type = None
  24     def __init__(self, options):
  25         try: self.config = config = Config(options.config_file)
  26         except:
  27             print "failed to read config_file %s" % options.config_file
  28             sys.exit(1)
  29         key_path = os.path.dirname(options.config_file)
  30         user_name = self.config.SFI_USER.split('.')[-1:][0]
  31         key_file = key_path + os.sep + user_name + '.pkey'
  32         cert_file = key_path + os.sep + user_name + '.cert'
  33         self.key = Keypair(filename=key_file)
  34         self.cert = Certificate(subject=self.config.SFI_USER)
  35         self.cert.set_pubkey(self.key)
  36         self.cert.set_issuer(self.key, self.config.SFI_USER)
  37         self.cert.sign()
  38         self.cert.save_to_file(cert_file)
  39         SFI_AGGREGATE = config.SFI_SM.replace('12347', '12346')
  40         SFI_CM = 'http://' + options.cm_host + ':12346'
  41         self.registry = xmlrpc.get_server(config.SFI_REGISTRY, key_file, cert_file)
  42         self.aggregate = xmlrpc.get_server(SFI_AGGREGATE, key_file, cert_file)
  43         self.sm = xmlrpc.get_server(config.SFI_SM, key_file, cert_file)
  44         self.cm = xmlrpc.get_server(SFI_CM, key_file, cert_file)
  45         self.user = config.SFI_USER
  46         # XX defaulting to user, but this should be configurable so we can
  47         # test from components persepctive
  48         self.type = 'user'
  49
  50     def get_credential(self, hrn = None, type = 'user'):
  51         if not hrn: hrn = self.user
  52         if hrn == self.user:
  53             cert = self.cert.save_to_string(save_parents=True)
  54             request_hash = self.key.compute_hash([cert, 'user', hrn])
  55             self.credential = self.registry.get_self_credential(cert, type, hrn, request_hash)
  56             return self.credential
  57         else:
  58             if not self.credential:
  59                 self.get_credential(self.user, 'user')
  60             return self.registry.get_credential(self.credential, type, hrn)
  61
  62
  63
  64 class BasicTestCase(unittest.TestCase):
  65     def __init__(self, testname, client):
  66         unittest.TestCase.__init__(self, testname)
  67         self.client = client
  68
  69     def setUp(self):
  70         self.registry = self.client.registry
  71         self.aggregate = self.client.aggregate
  72         self.sm = self.client.sm
  73         self.cm = self.client.cm
  74         self.credential = self.client.credential
  75         self.hrn = self.client.user
  76         self.type = self.client.type
  77
  78 # Registry tests
  79 class RegistryTest(BasicTestCase):
  80
  81     def setUp(self):
  82         """
  83         Make sure test records dont exsit
  84         """
  85         BasicTestCase.setUp(self)
  86
  87     def testGetSelfCredential(self):
  88         self.client.get_credential()
  89
  90     def testRegister(self):
  91         assert True
  92
  93     def testRegisterPeerObject(self):
  94         assert True
  95
  96     def testUpdate(self):
  97         assert True
  98
  99     def testResolve(self):
 100         self.registry.resolve(self.credential, self.hrn)
 101         assert True
 102
 103     def testRemove(self):
 104         assert True
 105
 106     def testRemovePeerObject(self):
 107         assert True
 108
 109     def testList(self):
 110         authority = get_authority(self.client.user)
 111         self.registry.list(self.credential, authority)
 112
 113     def testGetRegistries(self):
 114         self.registry.get_registries(self.credential)
 115
 116     def testGetAggregates(self):
 117         self.registry.get_aggregates(self.credential)
 118
 119     def testGetTrustedCerts(self):
 120         # this should fail unless we are a node
 121         callable = self.registry.get_trusted_certs
 122         server_exception = False
 123         try:
 124             callable(self.credential)
 125         except ServerException:
 126             server_exception = True
 127         finally:
 128             if self.type in ['user'] and not server_exception:
 129                 assert False
 130
 131
 132
 133
 134 class AggregateTest(BasicTestCase):
 135     def setup(self):
 136         BasicTestCase.setUp(self)
 137
 138     def testGetSlices(self):
 139         self.aggregate.get_slices(self.credential)
 140
 141     def testGetResources(self):
 142         self.aggregate.get_resources(self.credential)
 143
 144     def testCreateSlice(self):
 145         assert True
 146
 147     def testDeleteSlice(self):
 148         assert True
 149
 150     def testGetTicket(self):
 151         assert True
 152
 153 class SlicemgrTest(AggregateTest):
 154     def setup(self):
 155         AggregateTest.setUp(self)
 156
 157 class ComponentTest(BasicTestCase):
 158     def setup(self):
 159         BasicTestCase.setUp(self)
 160
 161     def testStartSlice(self):
 162         assert True
 163
 164     def testStopSlice(self):
 165         assert True
 166
 167     def testDeleteSlice(self):
 168         assert True
 169
 170     def testRestartSlice(self):
 171         assert True
 172
 173     def testGetSlices(self):
 174         assert True
 175
 176     def testRedeemTicket(self):
 177         assert True
 178
 179
 180 def test_names(testcase):
 181     return [name for name in dir(testcase) if name.startswith('test')]
 182
 183 if __name__ == '__main__':
 184
 185     args = sys.argv
 186     prog_name = args[0]
 187     default_config_dir = os.path.expanduser('~/.sfi/sfi_config')
 188     default_cm = "echo.cs.princeton.edu"
 189     parser = OptionParser(usage="%(prog_name)s [options]" % locals())
 190     parser.add_option('-f', '--config_file', dest='config_file', default=default_config_dir,
 191                       help='config file. default is %s' % default_config_dir)
 192     parser.add_option('-r', '--registry', dest='registry', action='store_true',
 193                       default=False, help='run registry tests')
 194     parser.add_option('-a', '--aggregate', dest='aggregate', action='store_true',
 195                       default=False, help='run aggregate tests')
 196     parser.add_option('-s', '--slicemgr', dest='slicemgr', action='store_true',
 197                       default=False, help='run slicemgr tests')
 198     parser.add_option('-c', '--component', dest='component', action='store_true',
 199                       default=False, help='run component tests')
 200     parser.add_option('-d', '--cm_host', dest='cm_host', default=default_cm,
 201                       help='dns name of component to test. default is %s' % default_cm)
 202     parser.add_option('-A', '--all', dest='all', action='store_true',
 203                       default=False, help='run component tests')
 204
 205     options, args = parser.parse_args()
 206     suite = unittest.TestSuite()
 207     client = Client(options)
 208     # bootstrap the users credential
 209     client.get_credential()
 210
 211     if options.registry or options.all:
 212         for name in test_names(RegistryTest):
 213             suite.addTest(RegistryTest(name, client))
 214
 215     if options.aggregate or options.all:
 216         for name in test_names(AggregateTest):
 217             suite.addTest(AggregateTest(name, client))
 218
 219     if options.slicemgr or options.all:
 220         for name in test_names(SlicemgrTest):
 221             suite.addTest(SlicemgrTest(name, client))
 222
 223     if options.component or options.all:
 224         for name in test_names(ComponentTest):
 225             suite.addTest(ComponentTest(name, client))
 226
 227     unittest.TextTestRunner(verbosity=2).run(suite)
 228