<s> import mock <EOL> from django . core . urlresolvers import reverse <EOL> from django import http <EOL> from mox3 . mox import IsA <EOL> from openstack_dashboard import api <EOL> from openstack_dashboard . dashboards . admin . aggregates import constants <EOL> from openstack_dashboard . dashboards . admin . aggregates import workflows <EOL> from openstack_dashboard . test import helpers as test <EOL> class BaseAggregateWorkflowTests ( test . BaseAdminViewTests ) : <EOL> def _get_create_workflow_data ( self , aggregate , hosts = None ) : <EOL> aggregate_info = { "<STR_LIT:name>" : aggregate . name , <EOL> "<STR_LIT>" : aggregate . availability_zone } <EOL> if hosts : <EOL> compute_hosts = [ ] <EOL> for host in hosts : <EOL> if host . service == '<STR_LIT>' : <EOL> compute_hosts . append ( host ) <EOL> host_field_name = '<STR_LIT>' <EOL> aggregate_info [ host_field_name ] = [ h . host_name for h in compute_hosts ] <EOL> return aggregate_info <EOL> def _get_manage_workflow_data ( self , aggregate , hosts = None , ) : <EOL> aggregate_info = { "<STR_LIT:id>" : aggregate . id } <EOL> if hosts : <EOL> compute_hosts = [ ] <EOL> for host in hosts : <EOL> if host . service == '<STR_LIT>' : <EOL> compute_hosts . append ( host ) <EOL> host_field_name = '<STR_LIT>' <EOL> aggregate_info [ host_field_name ] = [ h . host_name for h in compute_hosts ] <EOL> return aggregate_info <EOL> class CreateAggregateWorkflowTests ( BaseAggregateWorkflowTests ) : <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , ) , } ) <EOL> def test_workflow_get ( self ) : <EOL> api . nova . host_list ( IsA ( http . HttpRequest ) ) . AndReturn ( self . hosts . list ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> url = reverse ( constants . AGGREGATES_CREATE_URL ) <EOL> res = self . client . get ( url ) <EOL> workflow = res . context [ '<STR_LIT>' ] <EOL> self . assertTemplateUsed ( res , constants . AGGREGATES_CREATE_VIEW_TEMPLATE ) <EOL> self . assertEqual ( workflow . name , workflows . CreateAggregateWorkflow . name ) <EOL> self . assertQuerysetEqual ( <EOL> workflow . steps , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) , } ) <EOL> def _test_generic_create_aggregate ( self , workflow_data , aggregate , <EOL> error_count = <NUM_LIT:0> , <EOL> expected_error_message = None ) : <EOL> api . nova . host_list ( IsA ( http . HttpRequest ) ) . AndReturn ( self . hosts . list ( ) ) <EOL> api . nova . aggregate_details_list ( IsA ( http . HttpRequest ) ) . AndReturn ( [ ] ) <EOL> if not expected_error_message : <EOL> api . nova . aggregate_create ( <EOL> IsA ( http . HttpRequest ) , <EOL> name = workflow_data [ '<STR_LIT:name>' ] , <EOL> availability_zone = workflow_data [ '<STR_LIT>' ] , <EOL> ) . AndReturn ( aggregate ) <EOL> self . mox . ReplayAll ( ) <EOL> url = reverse ( constants . AGGREGATES_CREATE_URL ) <EOL> res = self . client . post ( url , workflow_data ) <EOL> if not expected_error_message : <EOL> self . assertNoFormErrors ( res ) <EOL> self . assertRedirectsNoFollow ( <EOL> res , reverse ( constants . AGGREGATES_INDEX_URL ) ) <EOL> else : <EOL> self . assertFormErrors ( res , error_count , expected_error_message ) <EOL> def test_create_aggregate ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> workflow_data = self . _get_create_workflow_data ( aggregate ) <EOL> self . _test_generic_create_aggregate ( workflow_data , aggregate ) <EOL> def test_create_aggregate_fails_missing_fields ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> workflow_data = self . _get_create_workflow_data ( aggregate ) <EOL> workflow_data [ '<STR_LIT:name>' ] = '<STR_LIT>' <EOL> workflow_data [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . _test_generic_create_aggregate ( workflow_data , aggregate , <NUM_LIT:1> , <EOL> u'<STR_LIT>' ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) , } ) <EOL> def test_create_aggregate_with_hosts ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> hosts = self . hosts . list ( ) <EOL> api . nova . host_list ( IsA ( http . HttpRequest ) ) . AndReturn ( self . hosts . list ( ) ) <EOL> api . nova . aggregate_details_list ( IsA ( http . HttpRequest ) ) . AndReturn ( [ ] ) <EOL> workflow_data = self . _get_create_workflow_data ( aggregate , hosts ) <EOL> api . nova . aggregate_create ( <EOL> IsA ( http . HttpRequest ) , <EOL> name = workflow_data [ '<STR_LIT:name>' ] , <EOL> availability_zone = workflow_data [ '<STR_LIT>' ] , <EOL> ) . AndReturn ( aggregate ) <EOL> compute_hosts = [ ] <EOL> for host in hosts : <EOL> if host . service == '<STR_LIT>' : <EOL> compute_hosts . append ( host ) <EOL> for host in compute_hosts : <EOL> api . nova . add_host_to_aggregate ( <EOL> IsA ( http . HttpRequest ) , <EOL> aggregate . id , host . host_name ) . InAnyOrder ( ) <EOL> self . mox . ReplayAll ( ) <EOL> url = reverse ( constants . AGGREGATES_CREATE_URL ) <EOL> res = self . client . post ( url , workflow_data ) <EOL> self . assertNoFormErrors ( res ) <EOL> self . assertRedirectsNoFollow ( res , <EOL> reverse ( constants . AGGREGATES_INDEX_URL ) ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , '<STR_LIT>' , ) , } ) <EOL> def test_host_list_nova_compute ( self ) : <EOL> hosts = self . hosts . list ( ) <EOL> compute_hosts = [ ] <EOL> for host in hosts : <EOL> if host . service == '<STR_LIT>' : <EOL> compute_hosts . append ( host ) <EOL> api . nova . host_list ( IsA ( http . HttpRequest ) ) . AndReturn ( self . hosts . list ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> url = reverse ( constants . AGGREGATES_CREATE_URL ) <EOL> res = self . client . get ( url ) <EOL> workflow = res . context [ '<STR_LIT>' ] <EOL> step = workflow . get_step ( "<STR_LIT>" ) <EOL> field_name = step . get_member_field_name ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( step . action . fields [ field_name ] . choices ) , <EOL> len ( compute_hosts ) ) <EOL> class AggregatesViewTests ( test . BaseAdminViewTests ) : <EOL> @ mock . patch ( '<STR_LIT>' , <EOL> mock . Mock ( return_value = False ) ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , <EOL> '<STR_LIT>' , ) , <EOL> api . cinder : ( '<STR_LIT>' , ) } ) <EOL> def test_panel_not_available ( self ) : <EOL> api . cinder . tenant_absolute_limits ( IsA ( http . HttpRequest ) ) . MultipleTimes ( ) . AndReturn ( self . cinder_limits [ '<STR_LIT>' ] ) <EOL> self . mox . ReplayAll ( ) <EOL> self . patchers [ '<STR_LIT>' ] . stop ( ) <EOL> res = self . client . get ( reverse ( '<STR_LIT>' ) ) <EOL> self . assertNotIn ( b'<STR_LIT>' , res . content ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , <EOL> '<STR_LIT>' , ) } ) <EOL> def test_index ( self ) : <EOL> api . nova . aggregate_details_list ( IsA ( http . HttpRequest ) ) . AndReturn ( self . aggregates . list ( ) ) <EOL> api . nova . availability_zone_list ( IsA ( http . HttpRequest ) , detailed = True ) . AndReturn ( self . availability_zones . list ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> res = self . client . get ( reverse ( constants . AGGREGATES_INDEX_URL ) ) <EOL> self . assertTemplateUsed ( res , constants . AGGREGATES_INDEX_VIEW_TEMPLATE ) <EOL> self . assertItemsEqual ( res . context [ '<STR_LIT>' ] . data , <EOL> self . aggregates . list ( ) ) <EOL> self . assertItemsEqual ( res . context [ '<STR_LIT>' ] . data , <EOL> self . availability_zones . list ( ) ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , '<STR_LIT>' , ) , } ) <EOL> def _test_generic_update_aggregate ( self , form_data , aggregate , <EOL> error_count = <NUM_LIT:0> , <EOL> expected_error_message = None ) : <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . AndReturn ( aggregate ) <EOL> if not expected_error_message : <EOL> az = form_data [ '<STR_LIT>' ] <EOL> aggregate_data = { '<STR_LIT:name>' : form_data [ '<STR_LIT:name>' ] , <EOL> '<STR_LIT>' : az } <EOL> api . nova . aggregate_update ( IsA ( http . HttpRequest ) , str ( aggregate . id ) , <EOL> aggregate_data ) <EOL> self . mox . ReplayAll ( ) <EOL> res = self . client . post ( reverse ( constants . AGGREGATES_UPDATE_URL , <EOL> args = [ aggregate . id ] ) , <EOL> form_data ) <EOL> if not expected_error_message : <EOL> self . assertNoFormErrors ( res ) <EOL> self . assertRedirectsNoFollow ( <EOL> res , reverse ( constants . AGGREGATES_INDEX_URL ) ) <EOL> else : <EOL> self . assertFormErrors ( res , error_count , expected_error_message ) <EOL> def test_update_aggregate ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> form_data = { '<STR_LIT:id>' : aggregate . id , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _test_generic_update_aggregate ( form_data , aggregate ) <EOL> def test_update_aggregate_fails_missing_fields ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> form_data = { '<STR_LIT:id>' : aggregate . id } <EOL> self . _test_generic_update_aggregate ( form_data , aggregate , <NUM_LIT:1> , <EOL> u'<STR_LIT>' ) <EOL> class ManageHostsTests ( test . BaseAdminViewTests ) : <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , '<STR_LIT>' ) } ) <EOL> def test_manage_hosts ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . AndReturn ( aggregate ) <EOL> api . nova . host_list ( IsA ( http . HttpRequest ) ) . AndReturn ( self . hosts . list ( ) ) <EOL> self . mox . ReplayAll ( ) <EOL> res = self . client . get ( reverse ( constants . AGGREGATES_MANAGE_HOSTS_URL , <EOL> args = [ aggregate . id ] ) ) <EOL> self . assertEqual ( res . status_code , <NUM_LIT:200> ) <EOL> self . assertTemplateUsed ( res , <EOL> constants . AGGREGATES_MANAGE_HOSTS_TEMPLATE ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) } ) <EOL> def test_manage_hosts_update_add_remove_not_empty_aggregate ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> aggregate . hosts = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> host = self . hosts . list ( ) [ <NUM_LIT:0> ] <EOL> form_data = { '<STR_LIT>' : <EOL> [ host . host_name ] } <EOL> api . nova . remove_host_from_aggregate ( IsA ( http . HttpRequest ) , <EOL> str ( aggregate . id ) , <EOL> '<STR_LIT>' ) . InAnyOrder ( ) <EOL> api . nova . remove_host_from_aggregate ( IsA ( http . HttpRequest ) , <EOL> str ( aggregate . id ) , <EOL> '<STR_LIT>' ) . InAnyOrder ( ) <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . AndReturn ( aggregate ) <EOL> api . nova . host_list ( IsA ( http . HttpRequest ) ) . AndReturn ( self . hosts . list ( ) ) <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . AndReturn ( aggregate ) <EOL> api . nova . add_host_to_aggregate ( IsA ( http . HttpRequest ) , <EOL> str ( aggregate . id ) , host . host_name ) <EOL> self . mox . ReplayAll ( ) <EOL> res = self . client . post ( reverse ( constants . AGGREGATES_MANAGE_HOSTS_URL , <EOL> args = [ aggregate . id ] ) , <EOL> form_data ) <EOL> self . assertNoFormErrors ( res ) <EOL> self . assertRedirectsNoFollow ( res , <EOL> reverse ( constants . AGGREGATES_INDEX_URL ) ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) } ) <EOL> def test_manage_hosts_update_add_not_empty_aggregate_should_fail ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> aggregate . hosts = [ '<STR_LIT>' ] <EOL> host1 = self . hosts . list ( ) [ <NUM_LIT:0> ] <EOL> host3 = self . hosts . list ( ) [ <NUM_LIT:2> ] <EOL> form_data = { '<STR_LIT>' : <EOL> [ host1 . host_name , host3 . host_name ] } <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . InAnyOrder ( ) . AndReturn ( aggregate ) <EOL> api . nova . host_list ( IsA ( http . HttpRequest ) ) . InAnyOrder ( ) . AndReturn ( self . hosts . list ( ) ) <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . InAnyOrder ( ) . AndReturn ( aggregate ) <EOL> api . nova . add_host_to_aggregate ( IsA ( http . HttpRequest ) , <EOL> str ( aggregate . id ) , host3 . host_name ) . InAnyOrder ( ) . AndRaise ( self . exceptions . nova ) <EOL> self . mox . ReplayAll ( ) <EOL> res = self . client . post ( reverse ( constants . AGGREGATES_MANAGE_HOSTS_URL , <EOL> args = [ aggregate . id ] ) , <EOL> form_data ) <EOL> self . assertNoFormErrors ( res ) <EOL> self . assertMessageCount ( error = <NUM_LIT:2> ) <EOL> self . assertRedirectsNoFollow ( res , <EOL> reverse ( constants . AGGREGATES_INDEX_URL ) ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) } ) <EOL> def test_manage_hosts_update_clean_not_empty_aggregate_should_fail ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> aggregate . hosts = [ '<STR_LIT>' ] <EOL> form_data = { '<STR_LIT>' : <EOL> [ ] } <EOL> api . nova . remove_host_from_aggregate ( IsA ( http . HttpRequest ) , <EOL> str ( aggregate . id ) , <EOL> '<STR_LIT>' ) . AndRaise ( self . exceptions . nova ) <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . AndReturn ( aggregate ) <EOL> api . nova . host_list ( IsA ( http . HttpRequest ) ) . AndReturn ( self . hosts . list ( ) ) <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . AndReturn ( aggregate ) <EOL> self . mox . ReplayAll ( ) <EOL> res = self . client . post ( reverse ( constants . AGGREGATES_MANAGE_HOSTS_URL , <EOL> args = [ aggregate . id ] ) , <EOL> form_data ) <EOL> self . assertNoFormErrors ( res ) <EOL> self . assertMessageCount ( error = <NUM_LIT:2> ) <EOL> self . assertRedirectsNoFollow ( res , <EOL> reverse ( constants . AGGREGATES_INDEX_URL ) ) <EOL> @ test . create_stubs ( { api . nova : ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) } ) <EOL> def _test_manage_hosts_update ( self , <EOL> host , <EOL> aggregate , <EOL> form_data , <EOL> addAggregate = False , <EOL> cleanAggregates = False ) : <EOL> if cleanAggregates : <EOL> api . nova . remove_host_from_aggregate ( IsA ( http . HttpRequest ) , <EOL> str ( aggregate . id ) , <EOL> '<STR_LIT>' ) . InAnyOrder ( ) <EOL> api . nova . remove_host_from_aggregate ( IsA ( http . HttpRequest ) , <EOL> str ( aggregate . id ) , <EOL> '<STR_LIT>' ) . InAnyOrder ( ) <EOL> api . nova . remove_host_from_aggregate ( IsA ( http . HttpRequest ) , <EOL> str ( aggregate . id ) , <EOL> '<STR_LIT>' ) . InAnyOrder ( ) <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . AndReturn ( aggregate ) <EOL> api . nova . host_list ( IsA ( http . HttpRequest ) ) . AndReturn ( self . hosts . list ( ) ) <EOL> api . nova . aggregate_get ( IsA ( http . HttpRequest ) , str ( aggregate . id ) ) . AndReturn ( aggregate ) <EOL> if addAggregate : <EOL> api . nova . add_host_to_aggregate ( IsA ( http . HttpRequest ) , <EOL> str ( aggregate . id ) , <EOL> host . host_name ) <EOL> self . mox . ReplayAll ( ) <EOL> res = self . client . post ( reverse ( constants . AGGREGATES_MANAGE_HOSTS_URL , <EOL> args = [ aggregate . id ] ) , <EOL> form_data ) <EOL> self . assertNoFormErrors ( res ) <EOL> self . assertRedirectsNoFollow ( res , <EOL> reverse ( constants . AGGREGATES_INDEX_URL ) ) <EOL> def test_manage_hosts_update_nothing_not_empty_aggregate ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> host = self . hosts . list ( ) [ <NUM_LIT:0> ] <EOL> aggregate . hosts = [ host . host_name ] <EOL> form_data = { '<STR_LIT>' : <EOL> [ host . host_name ] } <EOL> self . _test_manage_hosts_update ( host , <EOL> aggregate , <EOL> form_data , <EOL> addAggregate = False ) <EOL> def test_manage_hosts_update_nothing_empty_aggregate ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> aggregate . hosts = [ ] <EOL> form_data = { '<STR_LIT>' : <EOL> [ ] } <EOL> self . _test_manage_hosts_update ( None , <EOL> aggregate , <EOL> form_data , <EOL> addAggregate = False ) <EOL> def test_manage_hosts_update_add_empty_aggregate ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> aggregate . hosts = [ ] <EOL> host = self . hosts . list ( ) [ <NUM_LIT:0> ] <EOL> form_data = { '<STR_LIT>' : <EOL> [ host . host_name ] } <EOL> self . _test_manage_hosts_update ( host , <EOL> aggregate , <EOL> form_data , <EOL> addAggregate = True ) <EOL> def test_manage_hosts_update_add_not_empty_aggregate ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> aggregate . hosts = [ '<STR_LIT>' ] <EOL> host1 = self . hosts . list ( ) [ <NUM_LIT:0> ] <EOL> host3 = self . hosts . list ( ) [ <NUM_LIT:2> ] <EOL> form_data = { '<STR_LIT>' : <EOL> [ host1 . host_name , host3 . host_name ] } <EOL> self . _test_manage_hosts_update ( host3 , <EOL> aggregate , <EOL> form_data , <EOL> addAggregate = True ) <EOL> def test_manage_hosts_update_clean_not_empty_aggregate ( self ) : <EOL> aggregate = self . aggregates . first ( ) <EOL> aggregate . hosts = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> form_data = { '<STR_LIT>' : <EOL> [ ] } <EOL> self . _test_manage_hosts_update ( None , <EOL> aggregate , <EOL> form_data , <EOL> addAggregate = False , <EOL> cleanAggregates = True ) </s>
