<s> import getpass , os <EOL> import re <EOL> from StringIO import StringIO <EOL> from paramiko . rsakey import RSAKey <EOL> from django . test import TestCase <EOL> from django . contrib . auth . models import User <EOL> from django . core . urlresolvers import reverse <EOL> from expedient . clearinghouse . slice . models import Slice <EOL> from expedient . common . permissions . models import ExpedientPermission <EOL> from expedient . clearinghouse . project . models import Project <EOL> from expedient . common . tests . client import test_get_and_post_form <EOL> from sshaggregate . views import aggregate_add_servers , aggregate_crud <EOL> from sshaggregate . models import SSHAggregate , SSHServer , SSHServerSliver , SSHSliceInfo <EOL> class Tests ( TestCase ) : <EOL> CREATED_USER_FNAME = "<STR_LIT>" <EOL> '''<STR_LIT>''' <EOL> DELETED_USER_FNAME = "<STR_LIT>" <EOL> '''<STR_LIT>''' <EOL> PUBKEY_USER_FNAME = "<STR_LIT>" <EOL> '''<STR_LIT>''' <EOL> TEST_SSH_KEY_NAME = "<STR_LIT>" <EOL> TEST_KEY_COMMENT = "<STR_LIT>" <EOL> def create_test_ssh_key ( self ) : <EOL> self . test_key = RSAKey . generate ( <NUM_LIT> ) <EOL> self . test_pubkey = "<STR_LIT>" % ( self . test_key . get_base64 ( ) , Tests . TEST_KEY_COMMENT ) <EOL> f = open ( os . path . expanduser ( "<STR_LIT>" ) , "<STR_LIT:a>" ) <EOL> f . write ( self . test_pubkey ) <EOL> f . close ( ) <EOL> def delete_test_ssh_key ( self ) : <EOL> f = open ( os . path . expanduser ( "<STR_LIT>" ) , "<STR_LIT:r>" ) <EOL> auth_keys = f . read ( ) <EOL> f . close ( ) <EOL> new_auth_keys = re . sub ( <EOL> r"<STR_LIT>" % Tests . TEST_KEY_COMMENT , "<STR_LIT>" , auth_keys ) <EOL> f = open ( os . path . expanduser ( "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> f . write ( new_auth_keys ) <EOL> f . close ( ) <EOL> def setUp ( self ) : <EOL> self . su = User . objects . create_superuser ( <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:password>" ) <EOL> ExpedientPermission . objects . disable_checks ( ) <EOL> for f in Tests . CREATED_USER_FNAME , Tests . DELETED_USER_FNAME , Tests . PUBKEY_USER_FNAME : <EOL> if os . access ( f , os . F_OK ) : <EOL> os . unlink ( f ) <EOL> self . create_test_ssh_key ( ) <EOL> self . client . login ( username = "<STR_LIT>" , password = "<STR_LIT:password>" ) <EOL> def tearDown ( self ) : <EOL> for f in Tests . CREATED_USER_FNAME , Tests . DELETED_USER_FNAME , Tests . PUBKEY_USER_FNAME : <EOL> if os . access ( f , os . F_OK ) : <EOL> os . unlink ( f ) <EOL> self . delete_test_ssh_key ( ) <EOL> ExpedientPermission . objects . enable_checks ( ) <EOL> def test_add_aggregate ( self ) : <EOL> self . assertEqual ( SSHAggregate . objects . count ( ) , <NUM_LIT:0> ) <EOL> pkey_f = StringIO ( ) <EOL> self . test_key . write_private_key ( pkey_f ) <EOL> pkey = pkey_f . getvalue ( ) <EOL> pkey_f . close ( ) <EOL> response = test_get_and_post_form ( <EOL> self . client , <EOL> url = reverse ( aggregate_crud ) , <EOL> params = dict ( <EOL> name = "<STR_LIT>" , <EOL> description = "<STR_LIT>" , <EOL> location = "<STR_LIT>" , <EOL> admin_username = getpass . getuser ( ) , <EOL> private_key = pkey , <EOL> add_user_command = "<STR_LIT>" + <EOL> Tests . CREATED_USER_FNAME + "<STR_LIT:'>" , <EOL> del_user_command = "<STR_LIT>" + <EOL> Tests . DELETED_USER_FNAME + "<STR_LIT:'>" , <EOL> add_pubkey_user_command = "<STR_LIT>" + <EOL> Tests . PUBKEY_USER_FNAME + "<STR_LIT:'>" , <EOL> ) , <EOL> ) <EOL> self . assertEqual ( SSHAggregate . objects . count ( ) , <NUM_LIT:1> ) <EOL> next_url = reverse ( aggregate_add_servers , args = [ <NUM_LIT:1> ] ) <EOL> self . assertRedirects ( <EOL> response , next_url ) <EOL> response = test_get_and_post_form ( <EOL> self . client , <EOL> url = next_url , <EOL> params = { <EOL> "<STR_LIT>" : "<STR_LIT:localhost>" , <EOL> "<STR_LIT>" : "<STR_LIT:127.0.0.1>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> ) <EOL> self . assertRedirects ( <EOL> response , reverse ( aggregate_add_servers , args = [ <NUM_LIT:1> ] ) , <EOL> msg_prefix = "<STR_LIT>" % response ) <EOL> self . assertEqual ( SSHServer . objects . count ( ) , <NUM_LIT:1> ) <EOL> def test_create_delete_slice ( self ) : <EOL> self . test_add_aggregate ( ) <EOL> agg = SSHAggregate . objects . all ( ) [ <NUM_LIT:0> ] <EOL> self . client . post ( <EOL> reverse ( "<STR_LIT>" ) , <EOL> data = dict ( name = "<STR_LIT>" , description = "<STR_LIT>" ) , <EOL> ) <EOL> project = Project . objects . get ( name = "<STR_LIT>" ) <EOL> url = reverse ( "<STR_LIT>" , args = [ project . id ] ) <EOL> response = self . client . post ( <EOL> path = url , <EOL> data = { "<STR_LIT:id>" : agg . id } , <EOL> ) <EOL> self . assertTrue ( project . aggregates . count ( ) == <NUM_LIT:1> ) <EOL> self . assertRedirects ( response , url ) <EOL> self . client . post ( <EOL> reverse ( "<STR_LIT>" , args = [ project . id ] ) , <EOL> data = dict ( <EOL> name = "<STR_LIT>" , <EOL> description = "<STR_LIT>" , <EOL> ) <EOL> ) <EOL> slice = Slice . objects . get ( name = "<STR_LIT>" ) <EOL> slice_add_agg_url = reverse ( "<STR_LIT>" , args = [ slice . id ] ) <EOL> response = self . client . post ( <EOL> path = slice_add_agg_url , <EOL> data = { "<STR_LIT:id>" : agg . id } , <EOL> ) <EOL> self . assertRedirects ( response , slice_add_agg_url ) <EOL> self . assertEqual ( <EOL> slice . aggregates . count ( ) , <NUM_LIT:1> , <EOL> "<STR_LIT>" ) <EOL> server = SSHServer . objects . all ( ) [ <NUM_LIT:0> ] <EOL> SSHServerSliver . objects . create ( resource = server , slice = slice ) <EOL> SSHSliceInfo . objects . create ( <EOL> slice = slice , public_key = self . test_pubkey ) <EOL> self . client . post ( reverse ( "<STR_LIT>" , args = [ slice . id ] ) ) <EOL> self . assertTrue ( os . access ( Tests . CREATED_USER_FNAME , os . F_OK ) ) <EOL> f = open ( Tests . CREATED_USER_FNAME , "<STR_LIT:r>" ) <EOL> username = f . read ( ) <EOL> f . close ( ) <EOL> self . assertEqual ( <EOL> username . strip ( ) , self . su . username , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % ( self . su . username , Tests . CREATED_USER_FNAME , username ) ) <EOL> f = open ( Tests . PUBKEY_USER_FNAME , "<STR_LIT:r>" ) <EOL> pubkey = f . read ( ) <EOL> f . close ( ) <EOL> self . assertEqual ( <EOL> username . strip ( ) , self . su . username , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % ( self . test_pubkey , Tests . PUBKEY_USER_FNAME , pubkey ) ) <EOL> self . client . post ( reverse ( "<STR_LIT>" , args = [ slice . id ] ) ) <EOL> f = open ( Tests . DELETED_USER_FNAME , "<STR_LIT:r>" ) <EOL> username = f . read ( ) <EOL> f . close ( ) <EOL> self . assertEqual ( <EOL> username . strip ( ) , self . su . username , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % ( self . su . username , Tests . DELETED_USER_FNAME , username ) ) </s>
