<s> import sys <EOL> import mock <EOL> from oslo_config import cfg <EOL> from oslo_utils import encodeutils <EOL> from nova import context <EOL> from nova import exception <EOL> from nova import test <EOL> from nova . tests . unit . virt . libvirt import fakelibvirt <EOL> from nova import utils <EOL> from nova . virt . libvirt import config as vconfig <EOL> from nova . virt . libvirt import guest as libvirt_guest <EOL> from nova . virt . libvirt import host <EOL> host . libvirt = fakelibvirt <EOL> libvirt_guest . libvirt = fakelibvirt <EOL> CONF = cfg . CONF <EOL> if sys . version_info > ( <NUM_LIT:3> , ) : <EOL> long = int <EOL> class GuestTestCase ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( GuestTestCase , self ) . setUp ( ) <EOL> self . useFixture ( fakelibvirt . FakeLibvirtFixture ( ) ) <EOL> self . host = host . Host ( "<STR_LIT>" ) <EOL> self . context = context . get_admin_context ( ) <EOL> self . domain = mock . Mock ( spec = fakelibvirt . virDomain ) <EOL> self . guest = libvirt_guest . Guest ( self . domain ) <EOL> def test_repr ( self ) : <EOL> self . domain . ID . return_value = <NUM_LIT> <EOL> self . domain . UUIDString . return_value = "<STR_LIT>" <EOL> self . domain . name . return_value = "<STR_LIT:foo>" <EOL> self . assertEqual ( "<STR_LIT>" , repr ( self . guest ) ) <EOL> @ mock . patch . object ( fakelibvirt . Connection , '<STR_LIT>' ) <EOL> def test_create ( self , mock_define ) : <EOL> libvirt_guest . Guest . create ( "<STR_LIT>" , self . host ) <EOL> mock_define . assert_called_once_with ( "<STR_LIT>" ) <EOL> @ mock . patch . object ( fakelibvirt . Connection , '<STR_LIT>' ) <EOL> def test_create_exception ( self , mock_define ) : <EOL> mock_define . side_effect = test . TestingException <EOL> self . assertRaises ( test . TestingException , <EOL> libvirt_guest . Guest . create , <EOL> "<STR_LIT:foo>" , self . host ) <EOL> def test_launch ( self ) : <EOL> self . guest . launch ( ) <EOL> self . domain . createWithFlags . assert_called_once_with ( <NUM_LIT:0> ) <EOL> def test_launch_and_pause ( self ) : <EOL> self . guest . launch ( pause = True ) <EOL> self . domain . createWithFlags . assert_called_once_with ( <EOL> fakelibvirt . VIR_DOMAIN_START_PAUSED ) <EOL> def test_shutdown ( self ) : <EOL> self . domain . shutdown = mock . MagicMock ( ) <EOL> self . guest . shutdown ( ) <EOL> self . domain . shutdown . assert_called_once_with ( ) <EOL> @ mock . patch . object ( encodeutils , '<STR_LIT>' ) <EOL> def test_launch_exception ( self , mock_safe_decode ) : <EOL> self . domain . createWithFlags . side_effect = test . TestingException <EOL> mock_safe_decode . return_value = "<STR_LIT>" <EOL> self . assertRaises ( test . TestingException , self . guest . launch ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_safe_decode . called ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' ) <EOL> @ mock . patch . object ( libvirt_guest . Guest , '<STR_LIT>' ) <EOL> def test_enable_hairpin ( self , mock_get_interfaces , mock_execute ) : <EOL> mock_get_interfaces . return_value = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> self . guest . enable_hairpin ( ) <EOL> mock_execute . assert_has_calls ( [ <EOL> mock . call ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> run_as_root = True , process_input = '<STR_LIT:1>' , check_exit_code = [ <NUM_LIT:0> , <NUM_LIT:1> ] ) , <EOL> mock . call ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> run_as_root = True , process_input = '<STR_LIT:1>' , check_exit_code = [ <NUM_LIT:0> , <NUM_LIT:1> ] ) ] ) <EOL> @ mock . patch . object ( encodeutils , '<STR_LIT>' ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' ) <EOL> @ mock . patch . object ( libvirt_guest . Guest , '<STR_LIT>' ) <EOL> def test_enable_hairpin_exception ( self , mock_get_interfaces , <EOL> mock_execute , mock_safe_decode ) : <EOL> mock_get_interfaces . return_value = [ "<STR_LIT:foo>" ] <EOL> mock_execute . side_effect = test . TestingException ( '<STR_LIT>' ) <EOL> self . assertRaises ( test . TestingException , self . guest . enable_hairpin ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_safe_decode . called ) <EOL> def test_get_interfaces ( self ) : <EOL> self . domain . XMLDesc . return_value = """<STR_LIT>""" <EOL> self . assertEqual ( [ "<STR_LIT>" , "<STR_LIT>" ] , self . guest . get_interfaces ( ) ) <EOL> def test_get_interfaces_exception ( self ) : <EOL> self . domain . XMLDesc . return_value = "<STR_LIT>" <EOL> self . assertEqual ( [ ] , self . guest . get_interfaces ( ) ) <EOL> def test_poweroff ( self ) : <EOL> self . guest . poweroff ( ) <EOL> self . domain . destroy . assert_called_once_with ( ) <EOL> def test_resume ( self ) : <EOL> self . guest . resume ( ) <EOL> self . domain . resume . assert_called_once_with ( ) <EOL> def test_get_vcpus_info ( self ) : <EOL> self . domain . vcpus . return_value = ( [ ( <NUM_LIT:0> , <NUM_LIT:1> , int ( <NUM_LIT> ) , <NUM_LIT:2> ) ] , <EOL> [ ( True , True ) ] ) <EOL> vcpus = list ( self . guest . get_vcpus_info ( ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , vcpus [ <NUM_LIT:0> ] . id ) <EOL> self . assertEqual ( <NUM_LIT:2> , vcpus [ <NUM_LIT:0> ] . cpu ) <EOL> self . assertEqual ( <NUM_LIT:1> , vcpus [ <NUM_LIT:0> ] . state ) <EOL> self . assertEqual ( int ( <NUM_LIT> ) , vcpus [ <NUM_LIT:0> ] . time ) <EOL> def test_delete_configuration ( self ) : <EOL> self . guest . delete_configuration ( ) <EOL> self . domain . undefineFlags . assert_called_once_with ( <EOL> fakelibvirt . VIR_DOMAIN_UNDEFINE_MANAGED_SAVE ) <EOL> def test_delete_configuration_exception ( self ) : <EOL> self . domain . undefineFlags . side_effect = fakelibvirt . libvirtError ( <EOL> '<STR_LIT>' ) <EOL> self . domain . ID . return_value = <NUM_LIT:1> <EOL> self . guest . delete_configuration ( ) <EOL> self . domain . undefine . assert_called_once_with ( ) <EOL> def test_attach_device ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> self . guest . attach_device ( conf ) <EOL> self . domain . attachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = <NUM_LIT:0> ) <EOL> def test_attach_device_persistent ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> self . guest . attach_device ( conf , persistent = True ) <EOL> self . domain . attachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = fakelibvirt . VIR_DOMAIN_AFFECT_CONFIG ) <EOL> def test_attach_device_live ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> self . guest . attach_device ( conf , live = True ) <EOL> self . domain . attachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = fakelibvirt . VIR_DOMAIN_AFFECT_LIVE ) <EOL> def test_attach_device_persistent_live ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> self . guest . attach_device ( conf , persistent = True , live = True ) <EOL> self . domain . attachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = ( fakelibvirt . VIR_DOMAIN_AFFECT_CONFIG | <EOL> fakelibvirt . VIR_DOMAIN_AFFECT_LIVE ) ) <EOL> def test_detach_device ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> self . guest . detach_device ( conf ) <EOL> self . domain . detachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = <NUM_LIT:0> ) <EOL> def test_detach_device_persistent ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> self . guest . detach_device ( conf , persistent = True ) <EOL> self . domain . detachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = fakelibvirt . VIR_DOMAIN_AFFECT_CONFIG ) <EOL> def test_detach_device_live ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> self . guest . detach_device ( conf , live = True ) <EOL> self . domain . detachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = fakelibvirt . VIR_DOMAIN_AFFECT_LIVE ) <EOL> def test_detach_device_persistent_live ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> self . guest . detach_device ( conf , persistent = True , live = True ) <EOL> self . domain . detachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = ( fakelibvirt . VIR_DOMAIN_AFFECT_CONFIG | <EOL> fakelibvirt . VIR_DOMAIN_AFFECT_LIVE ) ) <EOL> def test_detach_device_with_retry_detach_success ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> get_config = mock . Mock ( ) <EOL> get_config . side_effect = [ conf , conf , conf , None ] <EOL> dev_path = "<STR_LIT>" <EOL> retry_detach = self . guest . detach_device_with_retry ( <EOL> get_config , dev_path , persistent = True , live = True , <EOL> inc_sleep_time = <NUM_LIT> ) <EOL> self . domain . detachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = ( fakelibvirt . VIR_DOMAIN_AFFECT_CONFIG | <EOL> fakelibvirt . VIR_DOMAIN_AFFECT_LIVE ) ) <EOL> get_config . assert_called_with ( dev_path ) <EOL> self . domain . detachDeviceFlags . reset_mock ( ) <EOL> retry_detach ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , self . domain . detachDeviceFlags . call_count ) <EOL> def test_detach_device_with_retry_detach_failure ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestDevice ) <EOL> conf . to_xml . return_value = "<STR_LIT>" <EOL> get_config = mock . Mock ( return_value = conf ) <EOL> retry_detach = self . guest . detach_device_with_retry ( <EOL> get_config , "<STR_LIT>" , persistent = True , live = True , <EOL> inc_sleep_time = <NUM_LIT> , max_retry_count = <NUM_LIT:3> ) <EOL> self . domain . detachDeviceFlags . assert_called_once_with ( <EOL> "<STR_LIT>" , flags = ( fakelibvirt . VIR_DOMAIN_AFFECT_CONFIG | <EOL> fakelibvirt . VIR_DOMAIN_AFFECT_LIVE ) ) <EOL> self . domain . detachDeviceFlags . reset_mock ( ) <EOL> self . assertRaises ( exception . DeviceDetachFailed , retry_detach ) <EOL> self . assertEqual ( <NUM_LIT:4> , self . domain . detachDeviceFlags . call_count ) <EOL> def test_detach_device_with_retry_device_not_found ( self ) : <EOL> get_config = mock . Mock ( return_value = None ) <EOL> self . assertRaises ( <EOL> exception . DeviceNotFound , self . guest . detach_device_with_retry , <EOL> get_config , "<STR_LIT>" , persistent = True , live = True ) <EOL> def test_get_xml_desc ( self ) : <EOL> self . guest . get_xml_desc ( ) <EOL> self . domain . XMLDesc . assert_called_once_with ( flags = <NUM_LIT:0> ) <EOL> def test_get_xml_desc_dump_inactive ( self ) : <EOL> self . guest . get_xml_desc ( dump_inactive = True ) <EOL> self . domain . XMLDesc . assert_called_once_with ( <EOL> flags = fakelibvirt . VIR_DOMAIN_XML_INACTIVE ) <EOL> def test_get_xml_desc_dump_sensitive ( self ) : <EOL> self . guest . get_xml_desc ( dump_sensitive = True ) <EOL> self . domain . XMLDesc . assert_called_once_with ( <EOL> flags = fakelibvirt . VIR_DOMAIN_XML_SECURE ) <EOL> def test_get_xml_desc_dump_inactive_dump_sensitive ( self ) : <EOL> self . guest . get_xml_desc ( dump_inactive = True , dump_sensitive = True ) <EOL> self . domain . XMLDesc . assert_called_once_with ( <EOL> flags = ( fakelibvirt . VIR_DOMAIN_XML_INACTIVE | <EOL> fakelibvirt . VIR_DOMAIN_XML_SECURE ) ) <EOL> def test_get_xml_desc_dump_migratable ( self ) : <EOL> self . guest . get_xml_desc ( dump_migratable = True ) <EOL> self . domain . XMLDesc . assert_called_once_with ( <EOL> flags = fakelibvirt . VIR_DOMAIN_XML_MIGRATABLE ) <EOL> def test_has_persistent_configuration ( self ) : <EOL> self . assertTrue ( <EOL> self . guest . has_persistent_configuration ( ) ) <EOL> self . domain . isPersistent . assert_called_once_with ( ) <EOL> def test_save_memory_state ( self ) : <EOL> self . guest . save_memory_state ( ) <EOL> self . domain . managedSave . assert_called_once_with ( <NUM_LIT:0> ) <EOL> def test_get_block_device ( self ) : <EOL> disk = '<STR_LIT>' <EOL> gblock = self . guest . get_block_device ( disk ) <EOL> self . assertEqual ( disk , gblock . _disk ) <EOL> self . assertEqual ( self . guest , gblock . _guest ) <EOL> def test_set_user_password ( self ) : <EOL> self . guest . set_user_password ( "<STR_LIT:foo>" , "<STR_LIT>" ) <EOL> self . domain . setUserPassword . assert_called_once_with ( "<STR_LIT:foo>" , "<STR_LIT>" , <NUM_LIT:0> ) <EOL> def test_get_devices ( self ) : <EOL> xml = """<STR_LIT>""" <EOL> self . domain . XMLDesc . return_value = xml <EOL> devs = self . guest . get_all_devices ( ) <EOL> self . assertEqual ( <NUM_LIT:6> , len ( devs ) ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:0> ] , vconfig . LibvirtConfigGuestDisk ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:1> ] , vconfig . LibvirtConfigGuestDisk ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:2> ] , vconfig . LibvirtConfigGuestDisk ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:3> ] , vconfig . LibvirtConfigGuestHostdev ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:4> ] , vconfig . LibvirtConfigGuestHostdev ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:5> ] , vconfig . LibvirtConfigGuestInterface ) <EOL> devs = self . guest . get_all_devices ( vconfig . LibvirtConfigGuestDisk ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( devs ) ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:0> ] , vconfig . LibvirtConfigGuestDisk ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:1> ] , vconfig . LibvirtConfigGuestDisk ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:2> ] , vconfig . LibvirtConfigGuestDisk ) <EOL> devs = self . guest . get_all_disks ( ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( devs ) ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:0> ] , vconfig . LibvirtConfigGuestDisk ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:1> ] , vconfig . LibvirtConfigGuestDisk ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:2> ] , vconfig . LibvirtConfigGuestDisk ) <EOL> devs = self . guest . get_all_devices ( vconfig . LibvirtConfigGuestHostdev ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( devs ) ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:0> ] , vconfig . LibvirtConfigGuestHostdev ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:1> ] , vconfig . LibvirtConfigGuestHostdev ) <EOL> devs = self . guest . get_all_devices ( vconfig . LibvirtConfigGuestInterface ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( devs ) ) <EOL> self . assertIsInstance ( devs [ <NUM_LIT:0> ] , vconfig . LibvirtConfigGuestInterface ) <EOL> self . assertIsNotNone ( <EOL> self . guest . get_interface_by_mac ( '<STR_LIT>' ) ) <EOL> self . assertIsNone ( self . guest . get_interface_by_mac ( None ) ) <EOL> def test_get_info ( self ) : <EOL> self . domain . info . return_value = ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> ) <EOL> self . domain . ID . return_value = <NUM_LIT:6> <EOL> info = self . guest . get_info ( self . host ) <EOL> self . domain . info . assert_called_once_with ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , info . state ) <EOL> self . assertEqual ( <NUM_LIT:2> , info . max_mem_kb ) <EOL> self . assertEqual ( <NUM_LIT:3> , info . mem_kb ) <EOL> self . assertEqual ( <NUM_LIT:4> , info . num_cpu ) <EOL> self . assertEqual ( <NUM_LIT:5> , info . cpu_time_ns ) <EOL> self . assertEqual ( <NUM_LIT:6> , info . id ) <EOL> def test_get_power_state ( self ) : <EOL> self . domain . info . return_value = ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> ) <EOL> power = self . guest . get_power_state ( self . host ) <EOL> self . assertEqual ( <NUM_LIT:1> , power ) <EOL> def test_is_active_when_domain_is_active ( self ) : <EOL> with mock . patch . object ( self . domain , "<STR_LIT>" , return_value = True ) : <EOL> self . assertTrue ( self . guest . is_active ( ) ) <EOL> def test_is_active_when_domain_not_active ( self ) : <EOL> with mock . patch . object ( self . domain , "<STR_LIT>" , return_value = False ) : <EOL> self . assertFalse ( self . guest . is_active ( ) ) <EOL> def test_freeze_filesystems ( self ) : <EOL> self . guest . freeze_filesystems ( ) <EOL> self . domain . fsFreeze . assert_called_once_with ( ) <EOL> def test_thaw_filesystems ( self ) : <EOL> self . guest . thaw_filesystems ( ) <EOL> self . domain . fsThaw . assert_called_once_with ( ) <EOL> def _conf_snapshot ( self ) : <EOL> conf = mock . Mock ( spec = vconfig . LibvirtConfigGuestSnapshotDisk ) <EOL> conf . to_xml . return_value = '<STR_LIT>' <EOL> return conf <EOL> def test_snapshot ( self ) : <EOL> conf = self . _conf_snapshot ( ) <EOL> self . guest . snapshot ( conf ) <EOL> self . domain . snapshotCreateXML ( '<STR_LIT>' , flags = <NUM_LIT:0> ) <EOL> conf . to_xml . assert_called_once_with ( ) <EOL> def test_snapshot_no_metadata ( self ) : <EOL> conf = self . _conf_snapshot ( ) <EOL> self . guest . snapshot ( conf , no_metadata = True ) <EOL> self . domain . snapshotCreateXML ( <EOL> '<STR_LIT>' , <EOL> flags = fakelibvirt . VIR_DOMAIN_SNAPSHOT_CREATE_NO_METADATA ) <EOL> conf . to_xml . assert_called_once_with ( ) <EOL> def test_snapshot_disk_only ( self ) : <EOL> conf = self . _conf_snapshot ( ) <EOL> self . guest . snapshot ( conf , disk_only = True ) <EOL> self . domain . snapshotCreateXML ( <EOL> '<STR_LIT>' , flags = fakelibvirt . VIR_DOMAIN_SNAPSHOT_CREATE_DISK_ONLY ) <EOL> conf . to_xml . assert_called_once_with ( ) <EOL> def test_snapshot_reuse_ext ( self ) : <EOL> conf = self . _conf_snapshot ( ) <EOL> self . guest . snapshot ( conf , reuse_ext = True ) <EOL> self . domain . snapshotCreateXML ( <EOL> '<STR_LIT>' , flags = fakelibvirt . VIR_DOMAIN_SNAPSHOT_CREATE_REUSE_EXT ) <EOL> conf . to_xml . assert_called_once_with ( ) <EOL> def test_snapshot_quiesce ( self ) : <EOL> conf = self . _conf_snapshot ( ) <EOL> self . guest . snapshot ( conf , quiesce = True ) <EOL> self . domain . snapshotCreateXML ( <EOL> '<STR_LIT>' , flags = fakelibvirt . VIR_DOMAIN_SNAPSHOT_CREATE_QUIESCE ) <EOL> conf . to_xml . assert_called_once_with ( ) <EOL> def test_snapshot_all ( self ) : <EOL> conf = self . _conf_snapshot ( ) <EOL> self . guest . snapshot ( conf , no_metadata = True , <EOL> disk_only = True , reuse_ext = True , <EOL> quiesce = True ) <EOL> self . domain . snapshotCreateXML ( <EOL> '<STR_LIT>' , flags = ( <EOL> fakelibvirt . VIR_DOMAIN_SNAPSHOT_CREATE_REUSE_EXT <EOL> | fakelibvirt . VIR_DOMAIN_SNAPSHOT_CREATE_DISK_ONLY <EOL> | fakelibvirt . VIR_DOMAIN_SNAPSHOT_CREATE_NO_METADATA <EOL> | fakelibvirt . VIR_DOMAIN_SNAPSHOT_CREATE_QUIESCE ) ) <EOL> conf . to_xml . assert_called_once_with ( ) <EOL> def test_pause ( self ) : <EOL> self . guest . pause ( ) <EOL> self . domain . suspend . assert_called_once_with ( ) <EOL> class GuestBlockTestCase ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( GuestBlockTestCase , self ) . setUp ( ) <EOL> self . useFixture ( fakelibvirt . FakeLibvirtFixture ( ) ) <EOL> self . host = host . Host ( "<STR_LIT>" ) <EOL> self . context = context . get_admin_context ( ) <EOL> self . domain = mock . Mock ( spec = fakelibvirt . virDomain ) <EOL> self . guest = libvirt_guest . Guest ( self . domain ) <EOL> self . gblock = self . guest . get_block_device ( '<STR_LIT>' ) <EOL> def test_abort_job ( self ) : <EOL> self . gblock . abort_job ( ) <EOL> self . domain . blockJobAbort . assert_called_once_with ( '<STR_LIT>' , flags = <NUM_LIT:0> ) <EOL> def test_abort_job_async ( self ) : <EOL> self . gblock . abort_job ( async = True ) <EOL> self . domain . blockJobAbort . assert_called_once_with ( <EOL> '<STR_LIT>' , flags = fakelibvirt . VIR_DOMAIN_BLOCK_JOB_ABORT_ASYNC ) <EOL> def test_abort_job_pivot ( self ) : <EOL> self . gblock . abort_job ( pivot = True ) <EOL> self . domain . blockJobAbort . assert_called_once_with ( <EOL> '<STR_LIT>' , flags = fakelibvirt . VIR_DOMAIN_BLOCK_JOB_ABORT_PIVOT ) <EOL> def test_get_job_info ( self ) : <EOL> self . domain . blockJobInfo . return_value = { <EOL> "<STR_LIT:type>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT:end>" : <NUM_LIT:100> } <EOL> info = self . gblock . get_job_info ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , info . job ) <EOL> self . assertEqual ( <NUM_LIT> , info . bandwidth ) <EOL> self . assertEqual ( <NUM_LIT> , info . cur ) <EOL> self . assertEqual ( <NUM_LIT:100> , info . end ) <EOL> self . domain . blockJobInfo . assert_called_once_with ( '<STR_LIT>' , flags = <NUM_LIT:0> ) <EOL> def test_resize ( self ) : <EOL> self . gblock . resize ( <NUM_LIT:10> ) <EOL> self . domain . blockResize . assert_called_once_with ( '<STR_LIT>' , <NUM_LIT:10> ) <EOL> def test_rebase ( self ) : <EOL> self . gblock . rebase ( "<STR_LIT:foo>" ) <EOL> self . domain . blockRebase . assert_called_once_with ( <EOL> '<STR_LIT>' , "<STR_LIT:foo>" , <NUM_LIT:0> , flags = <NUM_LIT:0> ) <EOL> def test_rebase_shallow ( self ) : <EOL> self . gblock . rebase ( "<STR_LIT:foo>" , shallow = True ) <EOL> self . domain . blockRebase . assert_called_once_with ( <EOL> '<STR_LIT>' , "<STR_LIT:foo>" , <NUM_LIT:0> , flags = fakelibvirt . VIR_DOMAIN_BLOCK_REBASE_SHALLOW ) <EOL> def test_rebase_reuse_ext ( self ) : <EOL> self . gblock . rebase ( "<STR_LIT:foo>" , reuse_ext = True ) <EOL> self . domain . blockRebase . assert_called_once_with ( <EOL> '<STR_LIT>' , "<STR_LIT:foo>" , <NUM_LIT:0> , <EOL> flags = fakelibvirt . VIR_DOMAIN_BLOCK_REBASE_REUSE_EXT ) <EOL> def test_rebase_copy ( self ) : <EOL> self . gblock . rebase ( "<STR_LIT:foo>" , copy = True ) <EOL> self . domain . blockRebase . assert_called_once_with ( <EOL> '<STR_LIT>' , "<STR_LIT:foo>" , <NUM_LIT:0> , <EOL> flags = fakelibvirt . VIR_DOMAIN_BLOCK_REBASE_COPY ) <EOL> def test_rebase_relative ( self ) : <EOL> self . gblock . rebase ( "<STR_LIT:foo>" , relative = True ) <EOL> self . domain . blockRebase . assert_called_once_with ( <EOL> '<STR_LIT>' , "<STR_LIT:foo>" , <NUM_LIT:0> , <EOL> flags = fakelibvirt . VIR_DOMAIN_BLOCK_REBASE_RELATIVE ) <EOL> def test_commit ( self ) : <EOL> self . gblock . commit ( "<STR_LIT:foo>" , "<STR_LIT>" ) <EOL> self . domain . blockCommit . assert_called_once_with ( <EOL> '<STR_LIT>' , "<STR_LIT:foo>" , "<STR_LIT>" , <NUM_LIT:0> , flags = <NUM_LIT:0> ) <EOL> def test_commit_relative ( self ) : <EOL> self . gblock . commit ( "<STR_LIT:foo>" , "<STR_LIT>" , relative = True ) <EOL> self . domain . blockCommit . assert_called_once_with ( <EOL> '<STR_LIT>' , "<STR_LIT:foo>" , "<STR_LIT>" , <NUM_LIT:0> , <EOL> flags = fakelibvirt . VIR_DOMAIN_BLOCK_COMMIT_RELATIVE ) <EOL> def test_wait_for_job ( self ) : <EOL> self . domain . blockJobInfo . return_value = { <EOL> "<STR_LIT:type>" : <NUM_LIT:4> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT:end>" : <NUM_LIT:100> } <EOL> in_progress = self . gblock . wait_for_job ( ) <EOL> self . assertTrue ( in_progress ) <EOL> self . domain . blockJobInfo . return_value = { <EOL> "<STR_LIT:type>" : <NUM_LIT:4> , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT:100> , <EOL> "<STR_LIT:end>" : <NUM_LIT:100> } <EOL> in_progress = self . gblock . wait_for_job ( ) <EOL> self . assertFalse ( in_progress ) <EOL> self . domain . blockJobInfo . return_value = { "<STR_LIT:type>" : <NUM_LIT:0> } <EOL> in_progress = self . gblock . wait_for_job ( wait_for_job_clean = True ) <EOL> self . assertFalse ( in_progress ) <EOL> def test_wait_for_job_arbort_on_error ( self ) : <EOL> self . domain . blockJobInfo . return_value = - <NUM_LIT:1> <EOL> self . assertRaises ( <EOL> exception . NovaException , <EOL> self . gblock . wait_for_job , abort_on_error = True ) </s>
