<s> '''<STR_LIT>''' <EOL> import sys <EOL> from pprint import pformat <EOL> from os . path import join , dirname <EOL> import time <EOL> import subprocess <EOL> import shlex <EOL> PYTHON_DIR = join ( dirname ( __file__ ) , "<STR_LIT>" ) <EOL> sys . path . append ( PYTHON_DIR ) <EOL> from unittest import TestCase <EOL> import xmlrpclib <EOL> from openflow . tests import test_settings as settings <EOL> from openflow . tests . helpers import parse_rspec , create_random_resv , kill_old_procs <EOL> from expedient . common . utils . certtransport import SafeTransportWithCert <EOL> from expedient . common . tests . commands import call_env_command , Env <EOL> ch_env = Env ( settings . CH_PROJECT_DIR ) <EOL> ch_env . switch_to ( ) <EOL> if settings . SHOW_PROCESSES_IN_XTERM : <EOL> from expedient . common . tests . utils import run_cmd_in_xterm as run_cmd <EOL> else : <EOL> from expedient . common . tests . utils import run_cmd <EOL> from expedient . common . tests . utils import wrap_xmlrpc_call , drop_to_shell <EOL> from expedient . common . tests . client import Browser <EOL> import logging <EOL> logger = logging . getLogger ( "<STR_LIT>" ) <EOL> SCHEME = "<STR_LIT>" if settings . USE_HTTPS else "<STR_LIT:http>" <EOL> def get_base_url ( path ) : <EOL> return SCHEME + "<STR_LIT>" % ( <EOL> settings . HOST , settings . CH_PORT , settings . PREFIX , path ) <EOL> class GAPITests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def create_ch_slice ( self ) : <EOL> """<STR_LIT>""" <EOL> import gcf . sfa . trust . credential as cred <EOL> slice_cred_string = wrap_xmlrpc_call ( <EOL> self . ch_client . CreateSlice , [ ] , { } , settings . TIMEOUT ) <EOL> slice_credential = cred . Credential ( string = slice_cred_string ) <EOL> slice_gid = slice_credential . get_gid_object ( ) <EOL> slice_urn = slice_gid . get_urn ( ) <EOL> credentials = [ slice_cred_string ] <EOL> return ( slice_urn , credentials ) <EOL> def setup_dummy_oms ( self ) : <EOL> from openflow . dummyom . models import DummyOM <EOL> from django . contrib . auth . models import User <EOL> from django . core . urlresolvers import reverse <EOL> for i in range ( settings . NUM_DUMMY_OMS ) : <EOL> username = "<STR_LIT>" % i <EOL> password = "<STR_LIT>" <EOL> User . objects . create_user ( <EOL> username = username , email = "<STR_LIT>" , password = password ) <EOL> om = DummyOM . objects . create ( ) <EOL> om . populate_links ( settings . NUM_SWITCHES_PER_AGG , <EOL> settings . NUM_LINKS_PER_AGG / <NUM_LIT:2> ) <EOL> logger . debug ( "<STR_LIT>" % ( i + <NUM_LIT:1> ) ) <EOL> response = self . browser . get_and_post_form ( <EOL> url = get_base_url ( reverse ( "<STR_LIT>" ) ) , <EOL> params = dict ( <EOL> name = "<STR_LIT>" % i , <EOL> description = "<STR_LIT>" , <EOL> location = "<STR_LIT>" , <EOL> usage_agreement = "<STR_LIT>" , <EOL> username = username , <EOL> password = password , <EOL> url = get_base_url ( reverse ( "<STR_LIT>" , args = [ om . id ] ) ) , <EOL> ) , <EOL> del_params = [ "<STR_LIT>" ] , <EOL> ) <EOL> self . assertEqual ( <EOL> response . geturl ( ) , <EOL> get_base_url ( <EOL> reverse ( "<STR_LIT>" , args = [ i + <NUM_LIT:1> ] ) ) , <EOL> "<STR_LIT>" <EOL> % response . read ( ) , <EOL> ) <EOL> def create_users ( self ) : <EOL> from django . contrib . auth . models import User <EOL> User . objects . create_superuser ( <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> import time , httplib , os <EOL> call_env_command ( settings . CH_PROJECT_DIR , "<STR_LIT>" , interactive = False ) <EOL> call_env_command ( settings . CH_PROJECT_DIR , "<STR_LIT>" , interactive = False ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> from django . core . urlresolvers import reverse <EOL> from django . conf import settings as djangosettings <EOL> from openflow . plugin . models import OpenFlowAggregate <EOL> self . assertEqual ( OpenFlowAggregate . objects . all ( ) . count ( ) , <NUM_LIT:0> ) <EOL> self . create_users ( ) <EOL> self . browser = Browser ( ) <EOL> self . browser . login ( get_base_url ( reverse ( "<STR_LIT>" ) ) , <EOL> username = "<STR_LIT>" , <EOL> password = "<STR_LIT>" ) <EOL> self . setup_dummy_oms ( ) <EOL> self . before = os . listdir ( djangosettings . XMLRPC_TRUSTED_CA_PATH ) <EOL> cmd = "<STR_LIT>" % settings . SSL_DIR <EOL> run_cmd ( cmd ) . wait ( ) <EOL> try : <EOL> os . unlink ( join ( settings . APACHE_CERTS_DIR , "<STR_LIT>" ) ) <EOL> except OSError as e : <EOL> if "<STR_LIT>" in "<STR_LIT:%s>" % e : <EOL> pass <EOL> else : <EOL> raise <EOL> os . symlink ( <EOL> os . path . abspath ( join ( settings . SSL_DIR , "<STR_LIT>" ) ) , <EOL> join ( settings . APACHE_CERTS_DIR , "<STR_LIT>" ) <EOL> ) <EOL> subprocess . call ( shlex . split ( "<STR_LIT>" % settings . APACHE_CERTS_DIR ) ) <EOL> kill_old_procs ( settings . GCH_PORT , settings . GAM_PORT ) <EOL> cmd = "<STR_LIT>" % ( <EOL> join ( settings . GCF_DIR , "<STR_LIT>" ) , <EOL> join ( settings . SSL_DIR , "<STR_LIT>" ) , <EOL> join ( settings . SSL_DIR , "<STR_LIT>" ) , <EOL> join ( settings . SSL_DIR , "<STR_LIT>" ) , join ( settings . SSL_DIR , "<STR_LIT>" ) , <EOL> settings . GCH_PORT , <EOL> ) <EOL> self . ch_proc = run_cmd ( cmd , pause = True ) <EOL> ch_host = "<STR_LIT>" % ( settings . HOST , settings . GCH_PORT ) <EOL> cert_transport = SafeTransportWithCert ( <EOL> keyfile = join ( settings . SSL_DIR , "<STR_LIT>" ) , <EOL> certfile = join ( settings . SSL_DIR , "<STR_LIT>" ) ) <EOL> self . ch_client = xmlrpclib . ServerProxy ( <EOL> "<STR_LIT>" + ch_host + "<STR_LIT:/>" , <EOL> transport = cert_transport ) <EOL> am_host = "<STR_LIT>" % ( settings . HOST , settings . CH_PORT ) <EOL> cert_transport = SafeTransportWithCert ( <EOL> keyfile = join ( settings . SSL_DIR , "<STR_LIT>" ) , <EOL> certfile = join ( settings . SSL_DIR , "<STR_LIT>" ) ) <EOL> self . am_client = xmlrpclib . ServerProxy ( <EOL> "<STR_LIT>" + am_host + "<STR_LIT>" , <EOL> transport = cert_transport ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> def tearDown ( self ) : <EOL> from django . conf import settings as djangosettings <EOL> import os <EOL> after = os . listdir ( djangosettings . XMLRPC_TRUSTED_CA_PATH ) <EOL> for path in after : <EOL> if path not in self . before : <EOL> os . unlink ( os . path . join ( djangosettings . XMLRPC_TRUSTED_CA_PATH , path ) ) <EOL> if settings . PAUSE_AFTER_TESTS : <EOL> raw_input ( "<STR_LIT>" ) <EOL> try : <EOL> del self . switches <EOL> except : <EOL> pass <EOL> try : <EOL> del self . links <EOL> except : <EOL> pass <EOL> try : <EOL> self . am_proc . terminate ( ) <EOL> except : <EOL> pass <EOL> try : <EOL> self . ch_proc . terminate ( ) <EOL> except : <EOL> pass <EOL> import time <EOL> time . sleep ( <NUM_LIT:4> ) <EOL> def test_GetVersion ( self ) : <EOL> """<STR_LIT>""" <EOL> ret = wrap_xmlrpc_call ( <EOL> self . am_client . GetVersion , [ ] , { } , settings . TIMEOUT ) <EOL> self . assertEqual ( ret [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> def test_ListResources ( self , zipped = False ) : <EOL> """<STR_LIT>""" <EOL> from openflow . dummyom . models import DummyOM <EOL> slice_urn , cred = self . create_ch_slice ( ) <EOL> options = dict ( geni_compressed = zipped , geni_available = True ) <EOL> rspec = wrap_xmlrpc_call ( <EOL> self . am_client . ListResources , <EOL> [ cred , options ] , { } , settings . TIMEOUT ) <EOL> logger . debug ( "<STR_LIT>" % rspec ) <EOL> if zipped : <EOL> import zlib , base64 <EOL> rspec = zlib . decompress ( base64 . b64decode ( rspec ) ) <EOL> self . switches , self . links = parse_rspec ( rspec ) <EOL> num_links = sum ( [ len ( d . get_switches ( ) ) for d in DummyOM . objects . all ( ) ] ) <EOL> self . assertEqual ( len ( self . switches ) , <EOL> num_links ) <EOL> self . assertEqual ( len ( self . links ) , <EOL> settings . NUM_LINKS_PER_AGG * settings . NUM_DUMMY_OMS ) <EOL> def test_ZippedListResources ( self ) : <EOL> """<STR_LIT>""" <EOL> self . test_ListResources ( zipped = True ) <EOL> def test_topoChange_ListResources ( self ) : <EOL> """<STR_LIT>""" <EOL> from openflow . dummyom . models import DummyOM <EOL> slice_urn , cred = self . create_ch_slice ( ) <EOL> options = dict ( geni_compressed = False , geni_available = True ) <EOL> rspec = wrap_xmlrpc_call ( <EOL> self . am_client . ListResources , <EOL> [ cred , options ] , { } , settings . TIMEOUT ) <EOL> self . switches , self . links = parse_rspec ( rspec ) <EOL> num_links = sum ( [ len ( d . get_switches ( ) ) for d in DummyOM . objects . all ( ) ] ) <EOL> self . assertEqual ( len ( self . switches ) , <EOL> num_links ) <EOL> self . assertEqual ( len ( self . links ) , <EOL> settings . NUM_LINKS_PER_AGG * settings . NUM_DUMMY_OMS ) <EOL> killed_dpids = [ ] <EOL> for om in DummyOM . objects . all ( ) : <EOL> killed_dpids . append ( om . kill_dpid ( ) ) <EOL> om . dummycallbackproxy . call_back ( ) <EOL> options = dict ( geni_compressed = False , geni_available = True ) <EOL> rspec = self . am_client . ListResources ( cred , options ) <EOL> self . switches , self . links = parse_rspec ( rspec ) <EOL> num_links = sum ( [ len ( d . get_switches ( ) ) for d in DummyOM . objects . all ( ) ] ) <EOL> self . assertEqual ( len ( self . switches ) , <EOL> num_links ) <EOL> for s in self . switches : <EOL> for d in killed_dpids : <EOL> self . assertNotEqual ( str ( s . dpid ) , str ( d ) ) <EOL> def test_CreateSliver ( self ) : <EOL> """<STR_LIT>""" <EOL> from openflow . plugin . models import OpenFlowSwitch <EOL> from openflow . dummyom . models import DummyOMSlice <EOL> from expedient . clearinghouse . slice . models import Slice <EOL> slice_urn , cred = self . create_ch_slice ( ) <EOL> options = dict ( geni_compressed = False , geni_available = True ) <EOL> rspec = wrap_xmlrpc_call ( <EOL> self . am_client . ListResources , <EOL> [ cred , options ] , { } , settings . TIMEOUT ) <EOL> self . switches , self . links = parse_rspec ( rspec ) <EOL> resv_rspec , flowspaces = create_random_resv ( <NUM_LIT:20> , self . switches ) <EOL> users = [ { '<STR_LIT:key>' : '<STR_LIT>' } ] <EOL> ret = self . am_client . CreateSliver ( slice_urn , cred , resv_rspec , users ) <EOL> self . assertEqual ( resv_rspec , ret ) <EOL> slice = Slice . objects . get ( gapislice__slice_urn = slice_urn ) <EOL> switches = OpenFlowSwitch . objects . filter ( <EOL> openflowinterface__slice_set = slice ) . distinct ( ) <EOL> dpids = [ ] <EOL> for fs in flowspaces : <EOL> for switch in fs . switches : <EOL> dpids . append ( switch . dpid ) <EOL> dpids = set ( dpids ) <EOL> self . assertEqual ( len ( dpids ) , len ( switches ) ) <EOL> self . assertEqual ( len ( DummyOMSlice . objects . all ( ) ) , <EOL> settings . NUM_DUMMY_OMS ) <EOL> def test_CreateDeleteSliver ( self ) : <EOL> """<STR_LIT>""" <EOL> from expedient . clearinghouse . slice . models import Slice <EOL> from openflow . dummyom . models import DummyOMSlice <EOL> slice_urn , cred = self . create_ch_slice ( ) <EOL> options = dict ( geni_compressed = False , geni_available = True ) <EOL> rspec = wrap_xmlrpc_call ( <EOL> self . am_client . ListResources , <EOL> [ cred , options ] , { } , settings . TIMEOUT ) <EOL> self . switches , self . links = parse_rspec ( rspec ) <EOL> resv_rspec , flowspaces = create_random_resv ( <NUM_LIT:20> , self . switches ) <EOL> users = [ { '<STR_LIT:key>' : '<STR_LIT>' } ] <EOL> self . am_client . CreateSliver ( slice_urn , cred , resv_rspec , users ) <EOL> self . assertTrue ( self . am_client . DeleteSliver ( slice_urn , cred ) ) <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> self . assertTrue ( Slice . objects . all ( ) . count ( ) == <NUM_LIT:0> , <EOL> "<STR_LIT>" ) <EOL> self . assertTrue ( DummyOMSlice . objects . all ( ) . count ( ) == <NUM_LIT:0> , <EOL> "<STR_LIT>" ) <EOL> def test_parse_slice ( self ) : <EOL> from openflow . plugin . gapi . rspec import parse_slice <EOL> slice_urn , cred = self . create_ch_slice ( ) <EOL> options = dict ( geni_compressed = False , geni_available = True ) <EOL> rspec = wrap_xmlrpc_call ( <EOL> self . am_client . ListResources , <EOL> [ cred , options ] , { } , settings . TIMEOUT ) <EOL> logger . debug ( "<STR_LIT>" % rspec ) <EOL> self . switches , self . links = parse_rspec ( rspec ) <EOL> vals = dict ( <EOL> firstname = "<STR_LIT>" , lastname = "<STR_LIT>" , <EOL> email = "<STR_LIT>" , password = "<STR_LIT>" , <EOL> proj_name = "<STR_LIT>" , <EOL> proj_desc = "<STR_LIT>" , <EOL> slice_name = "<STR_LIT>" , <EOL> slice_desc = "<STR_LIT>" , <EOL> ctrl_url = "<STR_LIT>" ) <EOL> resv_rspec , flowspaces = create_random_resv ( <NUM_LIT:2> , self . switches , ** vals ) <EOL> project_name , project_desc , slice_name , slice_desc , controller_url , email , password , iface_fs_map = parse_slice ( resv_rspec ) <EOL> self . assertEqual ( project_name , vals [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( project_desc , vals [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( slice_name , vals [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( slice_desc , vals [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( controller_url , vals [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( email , vals [ "<STR_LIT:email>" ] ) <EOL> self . assertEqual ( password , vals [ "<STR_LIT:password>" ] ) <EOL> dpid_fs_map = { } <EOL> for fs in flowspaces : <EOL> for sw in fs . switches : <EOL> if sw . dpid not in dpid_fs_map : <EOL> dpid_fs_map [ sw . dpid ] = [ ] <EOL> dpid_fs_map [ sw . dpid ] . append ( fs ) <EOL> logger . debug ( iface_fs_map ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import unittest <EOL> unittest . main ( ) </s>
