<s> from django . test import TestCase , TransactionTestCase <EOL> from django . test . client import Client <EOL> from systems . models import System <EOL> from systems . tests . utils import create_fake_host <EOL> from core . service . models import Service , Dependency <EOL> from core . site . models import Site <EOL> import simplejson as json <EOL> class BaseServiceTests ( object ) : <EOL> def setUp ( self ) : <EOL> self . client = Client ( ) <EOL> self . site = Site . objects . create ( full_name = '<STR_LIT:foo>' ) <EOL> self . s1 = Service . objects . create ( name = '<STR_LIT>' , site = self . site ) <EOL> self . sys1 = create_fake_host ( hostname = '<STR_LIT>' ) <EOL> self . sys2 = create_fake_host ( hostname = '<STR_LIT>' ) <EOL> self . sys3 = create_fake_host ( hostname = '<STR_LIT>' ) <EOL> self . s1 . systems . add ( self . sys1 ) <EOL> self . s1 . systems . add ( self . sys2 ) <EOL> self . s1 . systems . add ( self . sys3 ) <EOL> self . s1 . save ( ) <EOL> self . s2 = Service . objects . create ( name = '<STR_LIT>' ) <EOL> sys4 = create_fake_host ( hostname = '<STR_LIT>' ) <EOL> sys5 = create_fake_host ( hostname = '<STR_LIT>' ) <EOL> sys6 = create_fake_host ( hostname = '<STR_LIT>' ) <EOL> self . s2 . systems . add ( sys4 ) <EOL> self . s2 . systems . add ( sys5 ) <EOL> self . s2 . systems . add ( sys6 ) <EOL> self . s2 . save ( ) <EOL> self . s3 = Service . objects . create ( name = '<STR_LIT>' , site = self . site ) <EOL> Dependency . objects . get_or_create ( <EOL> dependant = self . s1 , provider = self . s2 <EOL> ) <EOL> Dependency . objects . get_or_create ( <EOL> dependant = self . s1 , provider = self . s3 <EOL> ) <EOL> def export_services ( self , query ) : <EOL> resp = self . client . get ( <EOL> '<STR_LIT>' , { '<STR_LIT>' : query } , follow = True <EOL> ) <EOL> return json . loads ( resp . content ) , resp <EOL> def import_services ( self , data ) : <EOL> resp = self . client . post ( <EOL> '<STR_LIT>' , data = json . dumps ( data ) , <EOL> content_type = '<STR_LIT:application/json>' <EOL> ) <EOL> return json . loads ( resp . content ) , resp <EOL> def get_service ( self , iql_stmt ) : <EOL> sblob , resp = self . export_services ( iql_stmt ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , len ( sblob [ '<STR_LIT>' ] ) , "<STR_LIT>" <EOL> ) <EOL> return sblob <EOL> class ServiceAPITests ( BaseServiceTests , TestCase ) : <EOL> def test_export ( self ) : <EOL> actual1 = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> self . assertEqual ( <EOL> actual1 [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> ) <EOL> actual2 = self . get_service ( self . s2 . iql_stmt ( ) ) <EOL> self . assertEqual ( <EOL> actual2 [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> ) <EOL> def test_import_new_service ( self ) : <EOL> services_blob = json . loads ( """<STR_LIT>""" . strip ( ) . format ( self . s1 . iql_stmt ( ) , self . site . full_name , self . sys1 ) ) <EOL> _ , resp = self . import_services ( services_blob ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> a = Service . objects . get ( name = '<STR_LIT>' ) <EOL> b = Service . objects . get ( name = '<STR_LIT>' ) <EOL> self . assertEqual ( a . parent_service , self . s1 ) <EOL> self . assertFalse ( b . parent_service ) <EOL> def test_import_create_delete_create ( self ) : <EOL> original = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> systems_count = System . objects . count ( ) <EOL> self . s1 . delete ( ) <EOL> self . assertEqual ( <EOL> systems_count , System . objects . count ( ) , <EOL> "<STR_LIT>" <EOL> ) <EOL> self . s1 = Service . objects . create ( name = '<STR_LIT>' , site = self . site ) <EOL> original [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> original [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = self . s1 . pk <EOL> _ , resp = self . import_services ( original ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> new = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> self . assertEqual ( original , new ) <EOL> def test_add_parent_service ( self ) : <EOL> original = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> self . assertEqual ( <EOL> original [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT:None>' <EOL> ) <EOL> original [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = self . s2 . iql_stmt ( ) <EOL> _ , resp = self . import_services ( original ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> self . assertEqual ( <EOL> Service . objects . get ( pk = self . s1 . pk ) . parent_service , <EOL> self . s2 <EOL> ) <EOL> new = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> self . assertEqual ( <EOL> new [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> self . s2 . iql_stmt ( ) <EOL> ) <EOL> new [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> _ , resp = self . import_services ( new ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> self . assertEqual ( <EOL> Service . objects . get ( pk = self . s1 . pk ) . parent_service , <EOL> None <EOL> ) <EOL> new_again = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> self . assertEqual ( <EOL> new_again [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> '<STR_LIT:None>' <EOL> ) <EOL> def test_import_add_system ( self ) : <EOL> original = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> s = create_fake_host ( hostname = '<STR_LIT>' ) <EOL> original [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] . append ( s . hostname ) <EOL> self . assertFalse ( self . s1 . systems . filter ( hostname = s . hostname ) . exists ( ) ) <EOL> _ , resp = self . import_services ( original ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> self . assertTrue ( self . s1 . systems . filter ( hostname = s . hostname ) . exists ( ) ) <EOL> def test_import_remove_system ( self ) : <EOL> sblob = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> s = create_fake_host ( hostname = '<STR_LIT>' ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] . append ( s . hostname ) <EOL> self . assertFalse ( self . s1 . systems . filter ( hostname = s . hostname ) . exists ( ) ) <EOL> _ , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> self . assertTrue ( self . s1 . systems . filter ( hostname = s . hostname ) . exists ( ) ) <EOL> sblob = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> self . assertTrue ( s . hostname in sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] . remove ( s . hostname ) <EOL> _ , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> self . assertFalse ( self . s1 . systems . filter ( hostname = s . hostname ) . exists ( ) ) <EOL> def test_add_deps ( self ) : <EOL> s4 = Service . objects . create ( name = '<STR_LIT:foo>' , site = self . site ) <EOL> sblob = self . get_service ( s4 . iql_stmt ( ) ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = [ <EOL> self . s1 . iql_stmt ( ) , <EOL> self . s2 . iql_stmt ( ) , <EOL> self . s3 . iql_stmt ( ) <EOL> ] <EOL> _ , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> pk_list = map ( lambda d : d . provider . pk , s4 . providers . all ( ) ) <EOL> self . assertTrue ( self . s1 . pk in pk_list ) <EOL> self . assertTrue ( self . s2 . pk in pk_list ) <EOL> self . assertTrue ( self . s3 . pk in pk_list ) <EOL> def test_import_remove_add_dependant_service ( self ) : <EOL> sblob = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> old_dep_count = Dependency . objects . count ( ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:2> , len ( sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] . remove ( self . s2 . iql_stmt ( ) ) <EOL> self . assertTrue ( <EOL> self . s3 . iql_stmt ( ) in sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> ) <EOL> _ , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> sblob = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , len ( sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> ) <EOL> self . assertTrue ( <EOL> self . s3 . iql_stmt ( ) in sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> ) <EOL> new_dep_count = Dependency . objects . count ( ) <EOL> self . assertEqual ( <EOL> old_dep_count - <NUM_LIT:1> , new_dep_count , <EOL> "<STR_LIT>" <EOL> ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] . append ( self . s2 . iql_stmt ( ) ) <EOL> _ , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , resp . content ) <EOL> sblob = self . get_service ( self . s1 . iql_stmt ( ) ) <EOL> self . assertTrue ( <EOL> self . s2 . iql_stmt ( ) in sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> ) <EOL> def test_bad_impact_option ( self ) : <EOL> sblob , resp = self . export_services ( <EOL> '<STR_LIT>' . format ( self . s1 . name ) <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , len ( sblob [ '<STR_LIT>' ] ) , "<STR_LIT>" <EOL> ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> resp_json , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( <EOL> resp . status_code , <NUM_LIT> , "<STR_LIT>" <EOL> "<STR_LIT>" . format ( resp . content ) <EOL> ) <EOL> self . assertTrue ( <EOL> '<STR_LIT>' in resp_json , "<STR_LIT>" <EOL> ) <EOL> def test_create_duplicate_service ( self ) : <EOL> sblob , resp = self . export_services ( <EOL> '<STR_LIT>' . format ( self . s1 . name ) <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , len ( sblob [ '<STR_LIT>' ] ) , "<STR_LIT>" <EOL> ) <EOL> del sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> resp_json , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> def test_implicit_update ( self ) : <EOL> sblob , resp = self . export_services ( <EOL> '<STR_LIT>' . format ( self . s1 . name ) <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , len ( sblob [ '<STR_LIT>' ] ) , "<STR_LIT>" <EOL> ) <EOL> del sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> resp_json , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( '<STR_LIT>' , Service . objects . get ( pk = self . s1 . pk ) . notes ) <EOL> def test_explicit_update ( self ) : <EOL> sblob , resp = self . export_services ( <EOL> '<STR_LIT>' . format ( self . s1 . name ) <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , len ( sblob [ '<STR_LIT>' ] ) , "<STR_LIT>" <EOL> ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] = '<STR_LIT>' <EOL> resp_json , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( '<STR_LIT>' , Service . objects . get ( pk = self . s1 . pk ) . name ) <EOL> def test_non_existant_parent_service ( self ) : <EOL> sblob , resp = self . export_services ( <EOL> '<STR_LIT>' . format ( self . s1 . name ) <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , len ( sblob [ '<STR_LIT>' ] ) , "<STR_LIT>" <EOL> ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> resp_json , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( <EOL> resp . status_code , <NUM_LIT> , "<STR_LIT>" <EOL> "<STR_LIT>" . format ( resp . content ) <EOL> ) <EOL> self . assertTrue ( <EOL> '<STR_LIT>' in resp_json , "<STR_LIT>" <EOL> ) <EOL> def test_non_existant_site ( self ) : <EOL> sblob , resp = self . export_services ( <EOL> '<STR_LIT>' . format ( self . s1 . name ) <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , len ( sblob [ '<STR_LIT>' ] ) , "<STR_LIT>" <EOL> ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> resp_json , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( <EOL> resp . status_code , <NUM_LIT> , "<STR_LIT>" <EOL> "<STR_LIT>" . format ( resp . content ) <EOL> ) <EOL> self . assertTrue ( <EOL> '<STR_LIT>' in resp_json , "<STR_LIT>" <EOL> ) <EOL> class ServiceMangleAPITests ( BaseServiceTests , TransactionTestCase ) : <EOL> def tearDown ( self ) : <EOL> Site . objects . all ( ) . delete ( ) <EOL> System . objects . all ( ) . delete ( ) <EOL> Service . objects . all ( ) . delete ( ) <EOL> def test_non_existant_system ( self ) : <EOL> system = create_fake_host ( hostname = '<STR_LIT>' ) <EOL> sblob , resp = self . export_services ( <EOL> '<STR_LIT>' . format ( self . s1 . name ) <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:200> , resp . status_code , resp . content ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , len ( sblob [ '<STR_LIT>' ] ) , "<STR_LIT>" <EOL> ) <EOL> sblob [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] += [ system . hostname , '<STR_LIT>' ] <EOL> resp_json , resp = self . import_services ( sblob ) <EOL> self . assertEqual ( <EOL> resp . status_code , <NUM_LIT> , "<STR_LIT>" <EOL> "<STR_LIT>" . format ( resp . content ) <EOL> ) <EOL> self . assertTrue ( <EOL> '<STR_LIT>' in resp_json , "<STR_LIT>" <EOL> ) <EOL> system = system . refresh ( ) <EOL> self . assertFalse ( system . service_set . exists ( ) ) </s>
