<s> import os , datetime , json <EOL> from twisted . trial import unittest <EOL> from twisted . internet import defer , task <EOL> from opennsa import config , nsa , database <EOL> from opennsa . topology import nml <EOL> from opennsa . backends import ncsvpn <EOL> from . import common <EOL> class NCSVPNBackendTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . clock = task . Clock ( ) <EOL> tcf = os . path . expanduser ( '<STR_LIT>' ) <EOL> tc = json . load ( open ( tcf ) ) <EOL> ncs_config = { <EOL> config . NCS_SERVICES_URL : tc [ '<STR_LIT>' ] , <EOL> config . NCS_USER : tc [ '<STR_LIT>' ] , <EOL> config . NCS_PASSWORD : tc [ '<STR_LIT>' ] <EOL> } <EOL> self . requester = common . DUDRequester ( ) <EOL> self . backend = ncsvpn . NCSVPNBackend ( '<STR_LIT>' , self . sr , self . requester , ncs_config ) <EOL> self . backend . scheduler . clock = self . clock <EOL> self . backend . startService ( ) <EOL> database . setupDatabase ( tc [ '<STR_LIT>' ] , tc [ '<STR_LIT>' ] , tc [ '<STR_LIT>' ] ) <EOL> self . requester_nsa = nsa . NetworkServiceAgent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . provider_nsa = nsa . NetworkServiceAgent ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> source_stp = nsa . STP ( '<STR_LIT>' , '<STR_LIT>' , labels = [ nsa . Label ( nml . ETHERNET_VLAN , '<STR_LIT>' ) ] ) <EOL> dest_stp = nsa . STP ( '<STR_LIT>' , '<STR_LIT>' , labels = [ nsa . Label ( nml . ETHERNET_VLAN , '<STR_LIT>' ) ] ) <EOL> start_time = datetime . datetime . utcnow ( ) + datetime . timedelta ( seconds = <NUM_LIT:2> ) <EOL> end_time = datetime . datetime . utcnow ( ) + datetime . timedelta ( seconds = <NUM_LIT:30> ) <EOL> bandwidth = <NUM_LIT:200> <EOL> self . service_params = nsa . ServiceParameters ( start_time , end_time , source_stp , dest_stp , bandwidth ) <EOL> @ defer . inlineCallbacks <EOL> def tearDown ( self ) : <EOL> from opennsa . backends . common import simplebackend <EOL> yield simplebackend . Simplebackendconnection . deleteAll ( ) <EOL> yield self . backend . stopService ( ) <EOL> @ defer . inlineCallbacks <EOL> def testActivation ( self ) : <EOL> _ , _ , cid , sp = yield self . reserve ( self . requester_nsa , self . provider_nsa , None , None , None , None , self . service_params ) <EOL> yield self . backend . reserveCommit ( self . requester_nsa , self . provider_nsa , None , cid ) <EOL> yield self . backend . provision ( self . requester_nsa , self . provider_nsa , None , cid ) <EOL> self . clock . advance ( <NUM_LIT:3> ) <EOL> connection_id , active , version_consistent , version , timestamp = yield d_up <EOL> self . failUnlessEqual ( cid , connection_id ) <EOL> self . failUnlessEqual ( active , True ) <EOL> self . failUnlessEqual ( version_consistent , True ) <EOL> yield self . backend . terminate ( self . requester_nsa , self . provider_nsa , None , cid ) <EOL> connection_id , active , version_consistent , version , timestamp = yield d_down <EOL> self . failUnlessEqual ( cid , connection_id ) <EOL> self . failUnlessEqual ( active , False ) <EOL> self . failUnlessEqual ( version_consistent , True ) <EOL> testActivation . skip = '<STR_LIT>' </s>
