我们从Python开源项目中，提取了以下28个代码示例，用于说明如何使用mock.mock.MagicMock()。
def test_create_from_volume_id_with_unexpected_status_1( self, mock_docker_volume): fake_volume_name = 'fake_vol' fake_volume_args = {'volume_id': DEFAULT_VOLUME_ID, 'status': 'attaching'} fake_cinder_volume = fake_object.FakeCinderVolume(**fake_volume_args) self.cinderprovider._get_docker_volume = mock.MagicMock() self.cinderprovider._get_docker_volume.return_value \ = (fake_cinder_volume, consts.UNKNOWN) self.cinderprovider.cinderclient.volumes.get = mock.MagicMock() self.cinderprovider.cinderclient.volumes.get.return_value = \ fake_cinder_volume self.assertRaises(exceptions.FuxiException, self.cinderprovider.create, fake_volume_name, {'volume_id': DEFAULT_VOLUME_ID})
def test_create_from_volume_id_with_unexpected_status_2(self): fake_server_id = 'fake_server_123' fake_host_name = 'attached_to_other' fake_volume_name = 'fake_vol' fake_volume_args = {'volume_id': DEFAULT_VOLUME_ID, 'status': 'in-use', 'multiattach': False, 'attachments': [{'server_id': fake_server_id, 'host_name': fake_host_name}]} fake_cinder_volume = fake_object.FakeCinderVolume(**fake_volume_args) self.cinderprovider._get_docker_volume = mock.MagicMock() self.cinderprovider._get_docker_volume.return_value \ = (fake_cinder_volume, consts.UNKNOWN) self.cinderprovider.cinderclient.volumes.get = mock.MagicMock() self.cinderprovider.cinderclient.volumes.get.return_value = \ fake_cinder_volume self.assertRaises(exceptions.FuxiException, self.cinderprovider.create, fake_volume_name, {'volume_id': DEFAULT_VOLUME_ID})
def test_create_with_volume_attach_to_this(self): fake_server_id = 'fake_server_123' fake_host_name = 'attached_to_this' fake_volume_args = {'id': DEFAULT_VOLUME_ID, 'status': 'in-use', 'attachments': [{'server_id': fake_server_id, 'host_name': fake_host_name}] } fake_cinder_volume = fake_object.FakeCinderVolume(**fake_volume_args) self.cinderprovider._get_docker_volume = mock.MagicMock() self.cinderprovider._get_docker_volume.return_value \ = (fake_cinder_volume, consts.ATTACH_TO_THIS) self.cinderprovider.cinderclient.volumes.get = mock.MagicMock() self.cinderprovider.cinderclient.volumes.get.return_value = \ fake_cinder_volume fake_result = self.cinderprovider.create('fake-vol', {}) self.assertEqual(os.path.join(volume_link_dir, DEFAULT_VOLUME_ID), fake_result['path'])
def test_delete(self, mock_execute): fd, tmpfname = tempfile.mkstemp() attachments = [{u'server_id': u'123', u'attachment_id': u'123', u'attached_at': u'2016-05-20T09:19:57.000000', u'host_name': utils.get_hostname(), u'device': None, u'id': u'123'}] self.cinderprovider._get_docker_volume = mock.MagicMock() self.cinderprovider._get_docker_volume.return_value = ( fake_object.FakeCinderVolume(id=tmpfname, attachments=attachments), consts.ATTACH_TO_THIS) self.cinderprovider._delete_volume = mock.MagicMock() self.assertTrue(self.cinderprovider.delete('fake-vol'))
def test_delete_failed(self, mock_execute, mock_delete): fd, tmpfname = tempfile.mkstemp() attachments = [{u'server_id': u'123', u'attachment_id': u'123', u'attached_at': u'2016-05-20T09:19:57.000000', u'host_name': utils.get_hostname(), u'device': None, u'id': u'123'}] self.cinderprovider._get_docker_volume = mock.MagicMock() self.cinderprovider._get_docker_volume.return_value = ( fake_object.FakeCinderVolume(id=tmpfname, attachments=attachments), consts.ATTACH_TO_THIS) self.assertRaises(cinder_exception.ClientException, self.cinderprovider.delete, 'fake-vol')
def test_delete_timeout(self, mock_execute): consts.DESTROY_VOLUME_TIMEOUT = 4 fd, tmpfname = tempfile.mkstemp() attachments = [{u'server_id': u'123', u'attachment_id': u'123', u'attached_at': u'2016-05-20T09:19:57.000000', u'host_name': utils.get_hostname(), u'device': None, u'id': u'123'}] self.cinderprovider._get_docker_volume = mock.MagicMock() self.cinderprovider._get_docker_volume.return_value = ( fake_object.FakeCinderVolume(id=tmpfname, attachments=attachments), consts.ATTACH_TO_THIS) self.assertRaises(exceptions.TimeoutException, self.cinderprovider.delete, 'fake-vol')
def test_cookie_dict(): c = RetsHttpClient('login_url', 'username', 'password') c._session = mock.MagicMock() jar = RequestsCookieJar() c1 = Cookie(1, 'name1', 'value1', 80, 80, 'domain', 'domain_specified', 'domain_initial_dot', 'path', 'path_specified', True, True, False, 'comment', 'comment_url', 'rest') c2 = Cookie(1, 'name2', 'value2', 80, 80, 'domain', 'domain_specified', 'domain_initial_dot', 'path', 'path_specified', True, True, False, 'comment', 'comment_url', 'rest') c3 = Cookie(1, 'name1', 'value1', 80, 80, 'domain', 'domain_specified3', 'domain_initial_dot3', 'path3', 'path_specified3', True, True, False, 'comment', 'comment_url', 'rest') jar.set_cookie(c1) jar.set_cookie(c2) jar.set_cookie(c3) c._session.cookies = jar assert c.cookie_dict == {'name1': 'value1', 'name2': 'value2'}
def test_create_with_volume_no_attach(self): fake_cinder_volume = fake_object.FakeCinderVolume() self.cinderprovider._get_docker_volume = mock.MagicMock() self.cinderprovider._get_docker_volume.return_value \ = (fake_cinder_volume, consts.NOT_ATTACH) fake_result = self.cinderprovider.create('fake-vol', {}) self.assertEqual(os.path.join(volume_link_dir, DEFAULT_VOLUME_ID), fake_result['path'])
def test_check_exists(self): self.cinderprovider._get_docker_volume = mock.MagicMock() self.cinderprovider._get_docker_volume.return_value = ( None, consts.UNKNOWN) result = self.cinderprovider.check_exist('fake-vol') self.assertFalse(result) self.cinderprovider._get_docker_volume.return_value = ( fake_object.FakeCinderVolume(), consts.NOT_ATTACH) result = self.cinderprovider.check_exist('fake-vol') self.assertTrue(result)
def test_load_stt_plugin(self): # Test getting default stt ol = OrderListener() stt1 = Stt(name="default-stt", parameters=dict()) stt2 = Stt(name="second-stt", parameters=dict()) stt3 = Stt(name="third-stt", parameters=dict()) resources = Resources(stt_folder="/tmp") ol.settings = mock.MagicMock(default_stt_name="default-stt", stts=[stt1,stt2,stt3], resources=resources) callback = mock.MagicMock() ol.callback = callback with mock.patch("kalliope.core.Utils.get_dynamic_class_instantiation") as mock_get_dynamic_class_instantiation: mock_get_dynamic_class_instantiation.return_value = 'class_instance' self.assertEqual(ol.load_stt_plugin(), "class_instance", "Fail getting the proper value") mock_get_dynamic_class_instantiation.assert_called_once_with(package_name= "stt", module_name= "Default-stt", parameters={'callback' : callback}, resources_dir= "/tmp")
def test_extract_services(self): html_file = open(os.path.join(os.path.dirname(__file__), 'resources/clientarea_services.html'), 'r') data = html_file.read() mock = MagicMock(ClientArea) mock.clientarea_url = '' services = ClientArea._extract_services(mock, data) self.assertEqual(services, [ {'status': 'active', 'product': u'Basic', 'url': '?action=productdetails&id=8961', 'price': u'$4.99 USD', 'term': u'Monthly', 'next_due_date': u'2017-06-19', 'id': '8961'}, {'status': 'cancelled', 'product': u'Basic', 'url': '?action=productdetails&id=9019', 'price': u'$4.99 USD', 'term': u'Monthly', 'next_due_date': u'2017-05-24', 'id': '9019'} ])
def test_execute_purchase(self): self._mock_options([self._create_option()]) RockHoster.purchase = MagicMock() command = ["purchase", "rockhoster", "-f", "-c", "config_test.cfg", "-rp", "asdf", "0"] cmdline.execute(command) RockHoster.purchase.assert_called_once()
def _mock_options(items=None): if items is None: items = [] RockHoster.options = MagicMock(return_value=items) return RockHoster.options
def test_hoster_print(self): hoster = cloudomate.vps.hoster.Hoster() options = [self._create_option()] hoster.configurations = options wallet.get_rates = MagicMock(return_value={'USD': 1.1}) hoster.gateway = MagicMock() hoster.gateway.estimate_price.return_value = 1.2 hoster.print_configurations() wallet.get_rates.assert_called_once() hoster.gateway.estimate_price.assert_called_once()
def setUp(self): self.client = mock.MagicMock() self.files = Files(self.client)
def test_remove_services(self, cinder_manage_remove): self.action_get.return_value = 'sv1host' svc1_mock = mock.MagicMock() svc1_mock.binary = "svc1bin" svc1_mock.host = "svc1host" query_mock = mock.MagicMock() query_mock.filter().all.return_value = [svc1_mock] cinder.db.sqlalchemy.api.model_query.return_value = query_mock self.os_release.return_value = 'liberty' cinder_manage.remove_services('arg') cinder_manage_remove.assert_called_once_with('svc1bin', 'svc1host') self.action_set.assert_called_once_with({'removed': 'svc1host'})
def test_remove_services_kilo(self, cinder_manage_remove): self.action_get.return_value = 'sv1host' svc1_mock = mock.MagicMock() svc1_mock.binary = "svc1bin" svc1_mock.host = "svc1host" svc1_mock.id = 42 cinder.context.get_admin_context.return_value = 'admctxt' query_mock = mock.MagicMock() query_mock.filter().all.return_value = [svc1_mock] cinder.db.sqlalchemy.api.model_query.return_value = query_mock self.os_release.return_value = 'kilo' cinder_manage.remove_services('arg') cinder.db.service_destroy.assert_called_once_with('admctxt', 42) self.action_set.assert_called_once_with({'removed': 'svc1host'})
def test_remove_services_fail(self, cinder_manage_remove): cinder_manage_remove.side_effect = Exception() self.action_get.return_value = 'sv1host' svc1_mock = mock.MagicMock() svc1_mock.binary = "svc1bin" svc1_mock.host = "svc1host" query_mock = mock.MagicMock() query_mock.filter().all.return_value = [svc1_mock] cinder.db.sqlalchemy.api.model_query.return_value = query_mock self.os_release.return_value = 'liberty' cinder_manage.remove_services('arg') cinder_manage_remove.assert_called_once_with('svc1bin', 'svc1host') self.action_fail.assert_called_once_with( 'Cannot remove service: svc1host')
def test__rename_volume_host(self, cinder_manage_volume_update_host): self.action_get.return_value = 'myhost' query_mock = mock.MagicMock() query_mock.filter().all.return_value = ['myhost'] cinder.db.sqlalchemy.api.model_query.return_value = query_mock cinder.db.sqlalchemy.api.model_query.return_value = query_mock cinder_manage._rename_volume_host('a', 'b') cinder_manage_volume_update_host.assert_called_once_with('a', 'b')
def test__rename_volume_host_missing(self, cinder_manage_volume_update_host): self.action_get.return_value = 'myhost' query_mock = mock.MagicMock() query_mock.filter().all.return_value = [] cinder.db.sqlalchemy.api.model_query.return_value = query_mock cinder_manage._rename_volume_host('a', 'b') self.assertFalse(cinder_manage_volume_update_host.called) self.action_fail.assert_called_once_with( 'Cannot update host attribute from a, a not found')
def check_call(self): xp = self.link.xp y = chainer.Variable(xp.random.uniform( -1, 1, (5, 7)).astype(numpy.float32)) self.link.predictor = mock.MagicMock(return_value=y) x = chainer.Variable(xp.asarray(self.x)) t = chainer.Variable(xp.asarray(self.t)) if self.x_num == 1: loss = self.link(x, t) self.link.predictor.assert_called_with(x) elif self.x_num == 2: x_ = chainer.Variable(xp.asarray(self.x.copy())) loss = self.link(x, x_, t) self.link.predictor.assert_called_with(x, x_) self.assertTrue(hasattr(self.link, 'y')) self.assertIsNotNone(self.link.y) self.assertTrue(hasattr(self.link, 'loss')) xp.testing.assert_allclose(self.link.loss.data, loss.data) self.assertTrue(hasattr(self.link, 'accuracy')) if self.compute_accuracy: self.assertIsNotNone(self.link.accuracy) else: self.assertIsNone(self.link.accuracy)
def test__bug1258464_when_condition_for_kill_is_true(mocker): mock_pymysql = mocker.patch('proxysql_tools.util.bug1258464.pymysql') mock_cursor = mock.MagicMock() mock_cursor.fetchone.return_value = [101] mock_pymysql.connect.return_value.cursor.return_value.__enter__.return_value = mock_cursor mock_get_my_cnf = mocker.patch('proxysql_tools.util.bug1258464.get_my_cnf') mock_get_my_cnf.return_value = 'some path' mock_get_pid = mocker.patch('proxysql_tools.util.bug1258464.get_pid') mock_get_pid.return_value = 300 mock_kill_process = mocker.patch('proxysql_tools.util.bug1258464.kill_process') assert bug1258464('some path')
def test__bug1258464_when_condition_for_kill_is_false(mocker): mock_pymysql = mocker.patch('proxysql_tools.util.bug1258464.pymysql') mock_cursor = mock.MagicMock() mock_cursor.fetchone.return_value = [100] mock_pymysql.connect.return_value.cursor.return_value.__enter__.return_value = mock_cursor mock_get_my_cnf = mocker.patch('proxysql_tools.util.bug1258464.get_my_cnf') mock_get_my_cnf.return_value = 'some path' mock_get_pid = mocker.patch('proxysql_tools.util.bug1258464.get_pid') mock_get_pid.return_value = 300 mock_kill_process = mocker.patch('proxysql_tools.util.bug1258464.kill_process') assert not bug1258464('some path')
def test__bug1258464_when_raise_NotImplementedError(mocker): mock_pymysql = mocker.patch('proxysql_tools.util.bug1258464.pymysql') mock_cursor = mock.MagicMock() mock_cursor.fetchone.return_value = [100] mock_pymysql.connect.return_value.cursor.return_value.__enter__.return_value = mock_cursor mock_get_my_cnf = mocker.patch('proxysql_tools.util.bug1258464.get_my_cnf') mock_get_my_cnf.return_value = 'some path' mock_get_my_cnf.side_effect = NotImplementedError assert not bug1258464('some path')
def test__bug1258464_when_raise_OSError(mocker): mock_pymysql = mocker.patch('proxysql_tools.util.bug1258464.pymysql') mock_cursor = mock.MagicMock() mock_cursor.fetchone.return_value = [100] mock_pymysql.connect.return_value.cursor.return_value.__enter__.return_value = mock_cursor mock_get_my_cnf = mocker.patch('proxysql_tools.util.bug1258464.get_my_cnf') mock_get_my_cnf.return_value = 'some path' mock_get_pid = mocker.patch('proxysql_tools.util.bug1258464.get_pid') mock_get_pid.side_effect = OSError assert not bug1258464('some path')
def test__bug1258464_when_raise_OperationalError(mocker): mock_pymysql = mocker.patch('proxysql_tools.util.bug1258464.pymysql') mock_cursor = mock.MagicMock() mock_cursor.fetchone.return_value = [100] mock_pymysql.connect.return_value.cursor.return_value.__enter__.return_value = mock_cursor mock_pymysql.connect.side_effect = OperationalError assert not bug1258464('some path')
def _mocked_service_browser(zc, search, listener): assert isinstance(zc, zeroconf.Zeroconf) assert search == "_soundtouch._tcp.local." mock_zeroconf = mock.MagicMock() service_info = mock.MagicMock() service_info.port = 8090 mock_zeroconf.get_service_info.return_value = service_info listener.add_service(mock_zeroconf, '', 'device.tcp')
def mock_index(): fake_index = mock.MagicMock() fake_index.metadata_types.get_by_name.return_value = mock.MagicMock(spec=MetadataType) # Check is performed validating the name of query fields fake_index.datasets.get_field_names.return_value = {'time', 'source_filter'} fake_index.products.get_by_name.return_value.measurements = {'red': { 'name': 'mock_measurement', 'dtype': 'int8', 'nodata': -999, 'units': '1'}} fake_index.metadata_types.get_by_name.return_value = MetadataType( { 'name': 'eo', 'dataset': dict( id=['id'], label=['ga_label'], creation_time=['creation_dt'], measurements=['image', 'bands'], sources=['lineage', 'source_datasets'] ) }, dataset_search_fields={} ) return fake_index
