python类DEFAULT的实例源码

def test_train_0(self):
        with patch.multiple(self.wb,
                            _extract_features=MOCK_DEFAULT,
                            _model=MOCK_DEFAULT):
            self.wb.train(([], []), None)

def test_train_0(self):
        with patch("dsenser.wang.wangbase.GridSearchCV"):
            with patch.multiple(self.wb,
                                _generate_ts=lambda *x: (self.TRAIN_X, Y),
                                _model=MOCK_DEFAULT):
                self.wb.train(([], []), None)

def test_train_1(self):
        rels = [REL1] * NFOLDS
        parses = [PARSE1] * NFOLDS
        with patch("dsenser.wang.wangbase.GridSearchCV"):
            with patch.multiple(self.wb,
                                _generate_ts=lambda *x: (self.TRAIN_X, Y),
                                _model=MOCK_DEFAULT):
                self.wb.train((rels, parses),
                              (rels, parses))

def test_nrpe_dependency_installed(self):
        with patch.multiple(ceph_hooks,
                            apt_install=DEFAULT,
                            rsync=DEFAULT,
                            log=DEFAULT,
                            write_file=DEFAULT,
                            nrpe=DEFAULT) as mocks:
            ceph_hooks.update_nrpe_config()
        mocks["apt_install"].assert_called_once_with(
            ["python-dbus", "lockfile-progs"])

def test_cached(self):
        mock_t = Mock()
        mock_std = Mock()
        mock_stpp = Mock()
        mock_stm = Mock()
        mock_mct = Mock()
        mock_mbs = Mock()
        mock_mos = Mock()
        with patch.multiple(
            pb,
            autospec=True,
            _transactions=DEFAULT,
            _scheduled_transactions_date=DEFAULT,
            _scheduled_transactions_per_period=DEFAULT,
            _scheduled_transactions_monthly=DEFAULT,
            _make_combined_transactions=DEFAULT,
            _make_budget_sums=DEFAULT,
            _make_overall_sums=DEFAULT
        ) as mocks:
            mocks['_transactions'].return_value.all.return_value = mock_t
            mocks['_scheduled_transactions_date'
                  ''].return_value.all.return_value = mock_std
            mocks['_scheduled_transactions_per_period'
                  ''].return_value.all.return_value = mock_stpp
            mocks['_scheduled_transactions_monthly'
                  ''].return_value.all.return_value = mock_stm
            mocks['_make_combined_transactions'].return_value = mock_mct
            mocks['_make_budget_sums'].return_value = mock_mbs
            mocks['_make_overall_sums'].return_value = mock_mos
            self.cls._data_cache = {'foo': 'bar'}
            res = self.cls._data
        assert res == {'foo': 'bar'}
        assert mocks['_transactions'].mock_calls == []
        assert mocks['_scheduled_transactions_date'].mock_calls == []
        assert mocks['_scheduled_transactions_per_period'].mock_calls == []
        assert mocks['_scheduled_transactions_monthly'].mock_calls == []
        assert mocks['_make_combined_transactions'].mock_calls == []
        assert mocks['_make_budget_sums'].mock_calls == []
        assert mocks['_make_overall_sums'].mock_calls == []

def test_insert_violations_with_error(self):
        """Test insert_violations handles errors during insert.

        Setup:
            * Create mocks:
                * self.dao.conn
                * self.dao.get_latest_snapshot_timestamp
                * self.dao.create_snapshot_table
            * Create side effect for one violation to raise an error.

        Expect:
            * Log MySQLError when table insert error occurs and return list
              of errors.
            * Return a tuple of (num_violations-1, [violation])
        """
        resource_name = 'policy_violations'
        self.dao.get_latest_snapshot_timestamp = mock.MagicMock(
            return_value=self.fake_snapshot_timestamp)
        self.dao.create_snapshot_table = mock.MagicMock(
            return_value=self.fake_table_name)
        violation_dao.LOGGER = mock.MagicMock()

        def insert_violation_side_effect(*args, **kwargs):
            if args[2] == self.expected_fake_violations[1]:
                raise MySQLdb.DataError(
                    self.resource_name, mock.MagicMock())
            else:
                return mock.DEFAULT

        self.dao.execute_sql_with_commit = mock.MagicMock(
            side_effect=insert_violation_side_effect)

        actual = self.dao.insert_violations(
            self.fake_flattened_violations,
            self.resource_name)

        expected = (2, [self.expected_fake_violations[1]])

        self.assertEqual(expected, actual)
        self.assertEquals(1, violation_dao.LOGGER.error.call_count)

def test_rebuild_indices(self):

        with patch.multiple(
            Command, _create=DEFAULT, _delete=DEFAULT, _populate=DEFAULT
        ) as handles:
            handles['_delete'].return_value = True
            call_command('search_index', stdout=self.out, action='rebuild')
            handles['_delete'].assert_called()
            handles['_create'].assert_called()
            handles['_populate'].assert_called()

def test_rebuild_indices_aborted(self):

        with patch.multiple(
            Command, _create=DEFAULT, _delete=DEFAULT, _populate=DEFAULT
        ) as handles:
            handles['_delete'].return_value = False
            call_command('search_index', stdout=self.out, action='rebuild')
            handles['_delete'].assert_called()
            handles['_create'].assert_not_called()
            handles['_populate'].assert_not_called()

def test_with_statement(self):
        with patch.multiple(WireMockServer, start=DEFAULT, stop=DEFAULT) as mocks:

            with WireMockServer() as wm:
                self.assertIsInstance(wm, WireMockServer)
                mocks['start'].assert_called_once_with()

            mocks['stop'].assert_called_once_with()

def test_ioctl_fn_ptr_r(self, ioctl_mock):
        def _handle_ioctl(fd, request, int_ptr):
            assert fd == 12
            assert request == 32
            assert type(int_ptr) == ctypes.POINTER(ctypes.c_int)
            int_ptr.contents.value = 42
            return mock.DEFAULT
        ioctl_mock.side_effect = _handle_ioctl

        fn = ioctl.ioctl_fn_ptr_r(32, ctypes.c_int)
        res = fn(12)
        assert res == 42

def test_ioctl_fn_ptr_w(self, ioctl_mock):
        def _handle_ioctl(fd, request, int_ptr):
            assert fd == 12
            assert request == 32
            assert type(int_ptr) == ctypes.POINTER(ctypes.c_int)
            assert int_ptr.contents.value == 42
            return mock.DEFAULT
        ioctl_mock.side_effect = _handle_ioctl

        fn = ioctl.ioctl_fn_ptr_w(32, ctypes.c_int)
        fn(12, 42)

def test_ioctl_fn_ptr_wr(self, ioctl_mock):
        def _handle_ioctl(fd, request, int_ptr):
            assert fd == 12
            assert request == 32
            assert type(int_ptr) == ctypes.POINTER(ctypes.c_int)
            assert int_ptr.contents.value == 24
            int_ptr.contents.value = 42
            return mock.DEFAULT
        ioctl_mock.side_effect = _handle_ioctl

        fn = ioctl.ioctl_fn_ptr_wr(32, ctypes.c_int)
        res = fn(12, 24)
        assert res == 42

def test_ioctl_fn_w(self, ioctl_mock):
        def _handle_ioctl(fd, request, int_val):
            assert fd == 12
            assert request == 32
            assert type(int_val) == ctypes.c_int
            assert int_val.value == 42
            return mock.DEFAULT
        ioctl_mock.side_effect = _handle_ioctl

        fn = ioctl.ioctl_fn_w(32, ctypes.c_int)
        fn(12, 42)

def _patch_object(
        target, attribute, new=mock.DEFAULT, spec=None,
        create=False, mocksignature=False, spec_set=None, autospec=False,
        new_callable=None, **kwargs
):
    getter = lambda: target
    return _make_patch_async(
        getter, attribute, new, spec, create, mocksignature, spec_set, autospec, new_callable,
        kwargs
    )

def _maybe_wrap_new(new):
    """If the mock replacement cannot have attributes set on it, wraps it in a function.

    Also, if the replacement object is a method, applies the async() decorator.

    This is needed so that we support patch(..., x.method) where x.method is an instancemethod
    object, because instancemethods do not support attribute assignment.

    """
    if new is mock.DEFAULT:
        return new

    if inspect.isfunction(new) or isinstance(new, (classmethod, staticmethod)):
        return asynq(sync_fn=new)(new)
    elif not callable(new):
        return new

    try:
        new._maybe_wrap_new_test_attribute = None
        del new._maybe_wrap_new_test_attribute
    except (AttributeError, TypeError):
        # setting something on a bound method raises AttributeError, setting something on a
        # Cythonized class raises TypeError
        should_wrap = True
    else:
        should_wrap = False

    if should_wrap:
        # we can't just use a lambda because that overrides __get__ and creates bound methods we
        # don't want, so we make a wrapper class that overrides __call__
        class Wrapper(object):
            def __call__(self, *args, **kwargs):
                return new(*args, **kwargs)
        return Wrapper()
    else:
        return new

def __init__(self, obj, attr, new=mock.DEFAULT, **kwargs):
        self.obj = obj
        self.attr = attr
        self.kwargs = kwargs
        self.new = new

def test_attach_volume_not_formatted(self, mock_execute):
        self.encryptor._get_key = mock.MagicMock()
        self.encryptor._get_key.return_value = \
                test_cryptsetup.fake__get_key(None)

        mock_execute.side_effect = [
                processutils.ProcessExecutionError(exit_code=1),  # luksOpen
                processutils.ProcessExecutionError(exit_code=1),  # isLuks
                mock.DEFAULT,  # luksFormat
                mock.DEFAULT,  # luksOpen
                mock.DEFAULT,  # ln
        ]

        self.encryptor.attach_volume(None)

        mock_execute.assert_has_calls([
            mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
                      self.dev_name, process_input='0' * 32,
                      run_as_root=True, check_exit_code=True),
            mock.call('cryptsetup', 'isLuks', '--verbose', self.dev_path,
                      run_as_root=True, check_exit_code=True),
            mock.call('cryptsetup', '--batch-mode', 'luksFormat',
                      '--key-file=-', self.dev_path, process_input='0' * 32,
                      run_as_root=True, check_exit_code=True, attempts=3),
            mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
                      self.dev_name, process_input='0' * 32,
                      run_as_root=True, check_exit_code=True),
            mock.call('ln', '--symbolic', '--force',
                      '/dev/mapper/%s' % self.dev_name, self.symlink_path,
                      run_as_root=True, check_exit_code=True),
        ], any_order=False)
        self.assertEqual(5, mock_execute.call_count)

def test_get_rdp_console(self):
        mock_get_host_ip = self.rdpconsoleops._hostops.get_host_ip_addr
        mock_get_rdp_port = (
            self.rdpconsoleops._rdpconsoleutils.get_rdp_console_port)
        mock_get_vm_id = self.rdpconsoleops._vmutils.get_vm_id

        connect_info = self.rdpconsoleops.get_rdp_console(mock.DEFAULT)

        self.assertEqual(mock_get_host_ip.return_value, connect_info.host)
        self.assertEqual(mock_get_rdp_port.return_value, connect_info.port)
        self.assertEqual(mock_get_vm_id.return_value,
                         connect_info.internal_access_path)

def test_spawn_no_admin_permissions(self):
        self._vmops._vmutils.check_admin_permissions.side_effect = (
            os_win_exc.HyperVException)
        self.assertRaises(os_win_exc.HyperVException,
                          self._vmops.spawn,
                          self.context, mock.DEFAULT, mock.DEFAULT,
                          [mock.sentinel.FILE], mock.sentinel.PASSWORD,
                          mock.sentinel.INFO, mock.sentinel.DEV_INFO)

def test_get_volume_connector(self):
        mock_instance = mock.DEFAULT
        initiator = self._volumeops._volutils.get_iscsi_initiator.return_value
        expected = {'ip': CONF.my_ip,
                    'host': CONF.host,
                    'initiator': initiator}

        response = self._volumeops.get_volume_connector(instance=mock_instance)

        self._volumeops._volutils.get_iscsi_initiator.assert_called_once_with()
        self.assertEqual(expected, response)