<s> import mock <EOL> from mox3 import mox <EOL> from nova . compute import manager as compute_manager <EOL> from nova import context <EOL> from nova import db <EOL> from nova import exception <EOL> from nova import objects <EOL> from nova import test <EOL> from nova . virt import fake <EOL> from nova . virt import virtapi <EOL> class VirtAPIBaseTest ( test . NoDBTestCase , test . APICoverage ) : <EOL> cover_api = virtapi . VirtAPI <EOL> def setUp ( self ) : <EOL> super ( VirtAPIBaseTest , self ) . setUp ( ) <EOL> self . context = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . set_up_virtapi ( ) <EOL> def set_up_virtapi ( self ) : <EOL> self . virtapi = virtapi . VirtAPI ( ) <EOL> def assertExpected ( self , method , * args , ** kwargs ) : <EOL> self . assertRaises ( NotImplementedError , <EOL> getattr ( self . virtapi , method ) , self . context , <EOL> * args , ** kwargs ) <EOL> def test_wait_for_instance_event ( self ) : <EOL> self . assertExpected ( '<STR_LIT>' , <EOL> '<STR_LIT>' , [ '<STR_LIT>' ] ) <EOL> class FakeVirtAPITest ( VirtAPIBaseTest ) : <EOL> cover_api = fake . FakeVirtAPI <EOL> def set_up_virtapi ( self ) : <EOL> self . virtapi = fake . FakeVirtAPI ( ) <EOL> def assertExpected ( self , method , * args , ** kwargs ) : <EOL> if method == '<STR_LIT>' : <EOL> run = False <EOL> with self . virtapi . wait_for_instance_event ( * args , ** kwargs ) : <EOL> run = True <EOL> self . assertTrue ( run ) <EOL> return <EOL> self . mox . StubOutWithMock ( db , method ) <EOL> if method in ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> e_args = tuple ( [ args [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ] + list ( args [ <NUM_LIT:1> : ] ) ) <EOL> elif method == '<STR_LIT>' : <EOL> e_args = tuple ( [ args [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ] + list ( args [ <NUM_LIT:1> : ] ) ) <EOL> else : <EOL> e_args = args <EOL> getattr ( db , method ) ( self . context , * e_args , ** kwargs ) . AndReturn ( <EOL> '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> result = getattr ( self . virtapi , method ) ( self . context , * args , ** kwargs ) <EOL> self . assertEqual ( result , '<STR_LIT>' ) <EOL> class FakeCompute ( object ) : <EOL> def __init__ ( self ) : <EOL> self . conductor_api = mox . MockAnything ( ) <EOL> self . db = mox . MockAnything ( ) <EOL> self . _events = [ ] <EOL> self . instance_events = mock . MagicMock ( ) <EOL> self . instance_events . prepare_for_instance_event . side_effect = self . _prepare_for_instance_event <EOL> def _event_waiter ( self ) : <EOL> event = mock . MagicMock ( ) <EOL> event . status = '<STR_LIT>' <EOL> return event <EOL> def _prepare_for_instance_event ( self , instance , event_name ) : <EOL> m = mock . MagicMock ( ) <EOL> m . instance = instance <EOL> m . event_name = event_name <EOL> m . wait . side_effect = self . _event_waiter <EOL> self . _events . append ( m ) <EOL> return m <EOL> class ComputeVirtAPITest ( VirtAPIBaseTest ) : <EOL> cover_api = compute_manager . ComputeVirtAPI <EOL> def set_up_virtapi ( self ) : <EOL> self . compute = FakeCompute ( ) <EOL> self . virtapi = compute_manager . ComputeVirtAPI ( self . compute ) <EOL> def assertExpected ( self , method , * args , ** kwargs ) : <EOL> self . mox . StubOutWithMock ( self . compute . conductor_api , method ) <EOL> getattr ( self . compute . conductor_api , method ) ( <EOL> self . context , * args , ** kwargs ) . AndReturn ( '<STR_LIT>' ) <EOL> self . mox . ReplayAll ( ) <EOL> result = getattr ( self . virtapi , method ) ( self . context , * args , ** kwargs ) <EOL> self . assertEqual ( result , '<STR_LIT>' ) <EOL> def test_wait_for_instance_event ( self ) : <EOL> and_i_ran = '<STR_LIT>' <EOL> event_1_tag = objects . InstanceExternalEvent . make_key ( <EOL> '<STR_LIT>' ) <EOL> event_2_tag = objects . InstanceExternalEvent . make_key ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> events = { <EOL> '<STR_LIT>' : event_1_tag , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) : event_2_tag , <EOL> } <EOL> with self . virtapi . wait_for_instance_event ( '<STR_LIT>' , events . keys ( ) ) : <EOL> and_i_ran = '<STR_LIT>' <EOL> self . assertEqual ( '<STR_LIT>' , and_i_ran ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( self . compute . _events ) ) <EOL> for event in self . compute . _events : <EOL> self . assertEqual ( '<STR_LIT>' , event . instance ) <EOL> self . assertIn ( event . event_name , events . values ( ) ) <EOL> event . wait . assert_called_once_with ( ) <EOL> def test_wait_for_instance_event_failed ( self ) : <EOL> def _failer ( ) : <EOL> event = mock . MagicMock ( ) <EOL> event . status = '<STR_LIT>' <EOL> return event <EOL> @ mock . patch . object ( self . virtapi . _compute , '<STR_LIT>' , _failer ) <EOL> def do_test ( ) : <EOL> with self . virtapi . wait_for_instance_event ( '<STR_LIT>' , [ '<STR_LIT:foo>' ] ) : <EOL> pass <EOL> self . assertRaises ( exception . NovaException , do_test ) <EOL> def test_wait_for_instance_event_failed_callback ( self ) : <EOL> def _failer ( ) : <EOL> event = mock . MagicMock ( ) <EOL> event . status = '<STR_LIT>' <EOL> return event <EOL> @ mock . patch . object ( self . virtapi . _compute , '<STR_LIT>' , _failer ) <EOL> def do_test ( ) : <EOL> callback = mock . MagicMock ( ) <EOL> with self . virtapi . wait_for_instance_event ( '<STR_LIT>' , [ '<STR_LIT:foo>' ] , <EOL> error_callback = callback ) : <EOL> pass <EOL> callback . assert_called_with ( '<STR_LIT:foo>' , '<STR_LIT>' ) <EOL> do_test ( ) <EOL> def test_wait_for_instance_event_timeout ( self ) : <EOL> class TestException ( Exception ) : <EOL> pass <EOL> def _failer ( ) : <EOL> raise TestException ( ) <EOL> @ mock . patch . object ( self . virtapi . _compute , '<STR_LIT>' , _failer ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def do_test ( timeout ) : <EOL> with self . virtapi . wait_for_instance_event ( '<STR_LIT>' , [ '<STR_LIT:foo>' ] ) : <EOL> pass <EOL> self . assertRaises ( TestException , do_test ) </s>
