python类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