python类dict()的实例源码

def test_get_jira_user_multi(self):
        """
        Tests the _get_jira_user method when JIRA returns more than one
        matching user.
        """
        self.mock_auth.search_users = Mock(return_value=[self.mock_jira_user_1,
                                                         self.mock_jira_user_2])
        with patch('ambassador.transport.Transport.get_key_cert',
                   return_value=self.mock_cert):
            with patch('platforms.jira.handlers.jira.JIRA',
                       return_value=self.mock_auth):
                with patch.dict('platforms.jira.handlers.settings.JIRA',
                                self.mock_settings):
                    handler_w_user = IssueAPI(endpoint=self.endpoint,
                                              user=self.user)
                    actual = handler_w_user._get_jira_user()
                    expected = None
                    self.assertEqual(actual, expected)

def test_trims_path(self, virtual_env):
        # pylint: disable=no-self-use
        venv_path = '/Y1w4sD/DELETE/THIS/ISUy0r'
        values = {
            'VIRTUAL_ENV': venv_path,
            'PATH': ':'.join(('/0ho8Ke/hL9DsW/ymH81W',
                              '/L51pua',
                              '/Vngp3V/G2m7Ih/05m7qW/LyrYkK/l5NuwA/oq1DPp',
                              venv_path,
                              '/Zpdhu4/bjvuqt',
                              '/STGtcb/FhAnWH/HwTvOr/gngGiB',
                              '/Zizj4D/szncsv/O5wO6X/joFHVT'))
        }
        with patch.dict('os.environ', values=values, clear=True):
            env = virtual_env.sanitized_env()

        assert venv_path not in env['PATH']
        assert env['PATH'] == (
            '/0ho8Ke/hL9DsW/ymH81W:/L51pua:/Vngp3V/G2m7Ih/05m7qW/LyrYkK/'
            'l5NuwA/oq1DPp:/Zpdhu4/bjvuqt:/STGtcb/FhAnWH/HwTvOr/gngGiB:'
            '/Zizj4D/szncsv/O5wO6X/joFHVT')

def test_multipletg_1(request, env):
    """Verify pytest_multipletg plugin creates MultipleTG instance and modifies env.tg dict.

    """
    # Verify env.tg before
    assert list(env.tg.keys()) == [1, 2]
    tg = MultipleTGClass(env)
    tg.setup()
    # Verify env.tg after setup
    assert list(env.tg.keys()) == [1]
    assert isinstance(env.tg[1], multiple_tg.MultipleTG)
    assert env.tg[1].id == "0102"
    assert env.tg[1].type == "ixiahl"
    assert not env.tg[1].port_list  # port_list should be empty
    assert sorted(env.tg[1].ports) == [
        multiple_tg.Port("01", (1, 1, 1)),
        multiple_tg.Port("01", (1, 1, 2)),
        multiple_tg.Port("02", (1, 2, 1)),
        multiple_tg.Port("02", (1, 2, 2))]

def setUp(self):
        """Initialise the class for the tests."""
        self.dbus_mock = MagicMock()
        self.mainloop_mock = MagicMock()
        self.gobject_mock = MagicMock()

        modules = {
            'dbus': self.dbus_mock,
            'dbus.mainloop.glib': self.mainloop_mock,
            'gi.repository': self.gobject_mock,
        }
        self.dbus_mock.Interface.return_value.GetManagedObjects.return_value = tests.obj_data.full_ubits
        self.dbus_mock.Interface.return_value.Get = mock_get
        self.dbus_mock.Interface.return_value.Set = mock_set
        self.module_patcher = patch.dict('sys.modules', modules)
        self.module_patcher.start()
        from bluezero import central
        self.module_under_test = central
        self.path = '/org/bluez/hci0/dev_E4_43_33_7E_54_1C/service002a'
        self.adapter_path = '/org/bluez/hci0'
        self.dev_name = 'BBC micro:bit [zezet]'
        self.adapter_addr = '00:00:00:00:5A:AD'
        self.device_addr = 'E4:43:33:7E:54:1C'
        self.service_uuid = 'e95dd91d-251d-470a-a062-fa1922dfa9a8'

def setUp(self):
        """
        Patch the DBus module
        :return:
        """
        self.dbus_mock = MagicMock()
        self.mainloop_mock = MagicMock()
        self.gobject_mock = MagicMock()
        self.process_mock = MagicMock()

        modules = {
            'dbus': self.dbus_mock,
            'dbus.mainloop.glib': self.mainloop_mock,
            'gi.repository': self.gobject_mock,
            'subprocess': self.process_mock
        }
        self.module_patcher = patch.dict('sys.modules', modules)
        self.module_patcher.start()
        from bluezero import async_tools
        self.module_under_test = async_tools.EventLoop()

def setUp(self):
        """
        Patch the DBus module
        :return:
        """
        self.dbus_mock = MagicMock()
        self.mainloop_mock = MagicMock()
        self.gobject_mock = MagicMock()

        modules = {
            'dbus': self.dbus_mock,
            'dbus.mainloop.glib': self.mainloop_mock,
            'gi.repository': self.gobject_mock,
        }
        self.dbus_mock.Interface.return_value.GetManagedObjects.return_value = tests.obj_data.full_ubits
        self.module_patcher = patch.dict('sys.modules', modules)
        self.module_patcher.start()
        from bluezero import tools
        self.module_under_test = tools

def setUp(self):
        """
        Patch the DBus module
        :return:
        """
        self.dbus_mock = MagicMock()
        self.mainloop_mock = MagicMock()
        self.gobject_mock = MagicMock()
        self.process_mock = MagicMock()

        modules = {
            'dbus': self.dbus_mock,
            'dbus.mainloop.glib': self.mainloop_mock,
            'gi.repository': self.gobject_mock,
            'subprocess': self.process_mock
        }
        self.dbus_mock.Interface.return_value.GetManagedObjects.return_value = tests.obj_data.full_ubits
        self.process_mock.check_output = self.get_bluetooth_service
        self.process_mock.Popen.return_value.communicate.return_value = (b'5.43\n', None)
        self.module_patcher = patch.dict('sys.modules', modules)
        self.module_patcher.start()
        from bluezero import dbus_tools
        self.module_under_test = dbus_tools

def test_put_log_messages_no_token(self):
        ''' test put_log_messages() when no sequence token given '''

        messages = [dict(a='abc'), dict(b='xyz', __CURSOR=sentinel.cursor)]
        events = [sentinel.msg1, sentinel.msg2]
        with patch.object(self.client, 'make_message', side_effect=events, autospec=True) as make_message:
            with patch.object(self.client, 'save_cursor', autospec=True) as save_cursor:
                self.client.put_log_messages(sentinel.group, sentinel.stream, None, messages)

        self.cwl.put_log_events.assert_called_once_with(
            logGroupName=sentinel.group,
            logStreamName=sentinel.stream,
            logEvents=events,
        )
        # saves the cursor once finished
        save_cursor.assert_called_once_with(sentinel.cursor)

def setup():
    """setup temporary env for tests"""
    global tmp
    tmp = tempfile.mkdtemp()
    patchers[:] = [
        patch.dict(os.environ, {
            'HOME': tmp,
            # Let tests work with --user install when HOME is changed:
            'PYTHONPATH': os.pathsep.join(sys.path),
        }),
    ]
    for p in patchers:
        p.start()

    # install IPython in the temp home:
    install(user=True)

def setup_method(self, method):
        """ setup any state tied to the execution of the given method in a
        class.  setup_method is invoked for every test method of a class.
        """

        self.journal_mock = Mock()
        self.cw_mock = Mock()
        self.cw_mock.CloudwatchLogger.return_value = 'hello'
        self.es_mock = Mock()
        self.es_mock.ElasticsearchLogger.return_value = 'bye'

        modules = {
            'systemdlogger.journal': self.journal_mock,
            'systemdlogger.cloudwatch': self.cw_mock,
            'systemdlogger.elasticsearch': self.es_mock
        }
        self.module_patcher = patch.dict('sys.modules', modules)
        self.module_patcher.start()
        from systemdlogger.runner import Runner
        self.Runner = Runner

def test_setting():
    s = settings.Setting(TEST_SETTING)
    assert s.get() is None

    with patch.dict(environ, {TEST_SETTING: 'hello'}):
        assert s.get() is None
        s.clear()
        assert s.get() == 'hello'

    s = settings.Setting(TEST_SETTING, default='nope')
    assert s.get() is 'nope'

    with patch.dict(environ, {TEST_SETTING: 'hello'}):
        assert s.get() == 'nope'
        s.clear()
        assert s.get() == 'hello'

def test_get_blind_sample(self):
        """
        Tests the get_sample method.
        """
        codebook = CodeBook.objects.get_by_natural_key('Acme')
        with patch.dict('teasers.models.settings.TEASERS',
                        self.mock_teaser_settings):
            with patch.dict('codebooks.models.settings.CODEBOOKS',
                            self.mock_codebooks_settings):
                actual = self.teaser.get_blind_sample(self.data, codebook)
                location = actual.pop('location')
                expected = {
                    'collection': 'elasticsearch.test_index.acme',
                    'author': '*FORGE*',
                    'title': '*PEAK* & *',
                    'content': 12345678910123,
                    'date': self.now,
                }
                self.assertEqual(actual, expected)
                self.assertEqual(location.coords, (-148.000, 32.000))

def test_format_descr_w_notes(self):
        """
        Tests the _format_description method when the alert has notes.
        """
        self.mock_settings['INCLUDE_FULL_DESCRIPTION'] = False
        with patch('ambassador.transport.Transport.get_key_cert',
                   return_value=self.mock_cert):
            with patch('platforms.jira.handlers.jira.JIRA',
                       return_value=self.mock_auth):
                with patch.dict('platforms.jira.handlers.settings.JIRA',
                                self.mock_settings):
                    handler_w_user = IssueAPI(endpoint=self.endpoint,
                                              user=self.user)
                    alert = Alert.objects.get(pk=3)
                    actual = handler_w_user._format_description(alert)
                    expected = 'Some example notes.'
        self.assertEqual(actual, expected)

def test_format_descr_wo_notes(self):
        """
        Tests the _format_description method when the alert has no notes.
        """
        self.mock_settings['INCLUDE_FULL_DESCRIPTION'] = False
        with patch('ambassador.transport.Transport.get_key_cert',
                   return_value=self.mock_cert):
            with patch('platforms.jira.handlers.jira.JIRA',
                       return_value=self.mock_auth):
                with patch.dict('platforms.jira.handlers.settings.JIRA',
                                self.mock_settings):
                    handler_w_user = IssueAPI(endpoint=self.endpoint,
                                              user=self.user)
                    alert = Alert.objects.get(pk=2)
                    actual = handler_w_user._format_description(alert)
                    expected = ''
        self.assertEqual(actual, expected)

def test_allowed_file(self):
        """
        Tests the get_file_path function for an allowed file type.
        """
        self.msg.attach(self.image)
        attachment = get_first_attachment(self.msg)
        mock_uuid = Mock()
        mock_uuid.hex = self.uuid
        with patch('sifter.mailsifter.attachments.uuid.uuid4',
                   return_value=mock_uuid):
            with patch.dict('sifter.mailsifter.attachments.settings.MAILSIFTER',
                            self.mock_mailsifter_settings):
                actual = attachments.get_file_path(attachment, self.company)
                expected = '%s/test-attachments/%s.jpeg' \
                           % (self.company.uuid.hex, self.uuid)
                self.assertEqual(actual, expected)

def test_unallowed_file(self):
        """
        Tests the get_file_path function for an unallowed file type.
        """
        self.mock_mailsifter_settings['ALLOWED_EMAIL_ATTACHMENTS'] = ('application/java',)
        with patch.dict('sifter.mailsifter.accessors.settings.MAILSIFTER',
                        self.mock_mailsifter_settings):
            self.msg.attach(self.java)
            attachment = get_first_attachment(self.msg)
            with patch('sifter.mailsifter.attachments.settings',
                       return_value=self.mock_settings):
                with LogCapture() as log_capture:
                    actual = attachments.get_file_path(attachment)
                    expected = None
                    self.assertEqual(actual, expected)
                    msg = 'The attachment %s is not an allowed file type' \
                          % self.java_file
                    log_capture.check(
                        ('sifter.mailsifter.attachments', 'WARNING', msg),
                    )

def test_match_with_default(self):
        """
        Tests the process_email receiver for an email that matches an
        existing MailChute.
        """
        doc_obj = self.doc_obj
        doc_obj.data['Subject'] = 'critical alert'
        mock_config = {
            'DEFAULT_MUNGER': 'default_mail',
            'DEFAULT_MUNGER_ENABLED': True
        }
        with patch('distilleries.models.Distillery.save_data',
                   return_value='id_123') as mock_save:
            with patch.dict('sifter.mailsifter.mailchutes.models.conf.MAILSIFTER',
                            mock_config):
                with patch('sifter.mailsifter.mailchutes.models.MailChuteManager._process_with_default') \
                        as mock_catch_email:
                    MailChute.objects.process(doc_obj)
                    self.assertIs(mock_save.called, True)
                    self.assertIs(mock_catch_email.called, False)

def test_no_match_without_default(self):
        """
        Tests the process_email receiver for an email that doesn't match
        an existing MailChute when a default MailMunger is not enabled.
        """
        doc_obj = self.doc_obj
        doc_obj.data['Subject'] = 'nothing to see here'
        mock_config = {
            'DEFAULT_MUNGER': 'default_mail',
            'DEFAULT_MUNGER_ENABLED': False
        }
        with patch('distilleries.models.Distillery.save_data') as mock_save:
            with patch.dict('sifter.mailsifter.mailchutes.models.conf.MAILSIFTER',
                            mock_config):
                with patch('sifter.mailsifter.mailchutes.models.MailChuteManager._process_with_default') \
                        as mock_catch_email:
                    MailChute.objects.process(doc_obj)
                    self.assertIs(mock_save.called, False)
                    self.assertIs(mock_catch_email.called, False)

def test_no_match_with_default(self):
        """
        Tests the process_email receiver for an email that doesn't match
        an existing MailChute when a default MailMunger is enabled.
        """
        doc_obj = self.doc_obj
        doc_obj.data['Subject'] = 'nothing to see here'
        mock_config = {
            'DEFAULT_MUNGER': 'default_mail',
            'DEFAULT_MUNGER_ENABLED': True
        }
        with patch.dict('sifter.mailsifter.mailchutes.models.conf.MAILSIFTER',
                        mock_config):
            with patch('sifter.mailsifter.mailchutes.models.MailChuteManager._process_with_default') \
                    as mock_default_process:
                MailChute.objects.process(doc_obj)
                mock_default_process.assert_called_once_with(doc_obj)

def test_no_match_missing_munger(self):
        """
        Tests the process_email receiver for an email that doesn't match
        an existing MailChute when a default MailChute is enabled but
        the defaul MailMunger can't be found.
        """
        doc_obj = self.doc_obj
        doc_obj.data['Subject'] = 'nothing to see here'
        mock_config = {
            'DEFAULT_MUNGER': 'missing_munger',
            'DEFAULT_MUNGER_ENABLED': True
        }
        with patch.dict('sifter.mailsifter.mailchutes.models.conf.MAILSIFTER',
                        mock_config):
            with LogCapture() as log_capture:
                msg = 'Default MailMunger "missing_munger" is not configured.'
                MailChute.objects.process(doc_obj)
                log_capture.check(
                    ('sifter.chutes.models', 'ERROR', msg),
                )

def test_get_default_no_chute(self):
        """
        Tests the _default_munger function when the default LogMunger
        does not exist.
        """
        mock_config = {
            'DEFAULT_MUNGER': 'dummy_munger',
            'DEFAULT_MUNGER_ENABLED': True
        }
        with patch.dict('sifter.logsifter.logchutes.models.conf.LOGSIFTER',
                        mock_config):
            with LogCapture() as log_capture:
                actual = LogChute.objects._default_munger
                expected = None
                self.assertEqual(actual, expected)
                self.assertFalse(LogChute.objects._default_munger_enabled)
                log_capture.check(
                    ('sifter.chutes.models',
                     'ERROR',
                     'Default LogMunger "dummy_munger" is not configured.'),
                )

def test_id_not_set(self):
        with patch.dict(
            'os.environ',
            {'SHIBBOLETH_ROOT': "http://rooturl.com"}
        ):
            session = self.client.session
            session.pop("user_id")
            session.save()

            res = self.client.get('/dashboard/')
            self.assertRedirects(
                res,
                "http://rooturl.com/Login?target="
                "http%3A//testserver/dashboard/user/login.callback",
                fetch_redirect_response=False,
            )

def test_config_set_by_code_in_container_env(self):
        # Fixture
        path = "some/path"
        container_id = "someId"

        with patch.dict('os.environ', {'HOST_CW_DIR': path,
                                       "HOSTNAME": container_id}):

            # Test
            actual = self.sut.get_tb_config(self.toolbelt_root,
                                            self.extensions_mock)

            # Assert
            self.assertEqual('container', actual['host'])
            self.assertEqual(path, actual['module_root_in_docker_host'])
            self.assertEqual(container_id, actual['container_id'])

def test_change_request_product_no_item_price(self, mock_product):
        # make fake object for receive result from Product.objects.get()
        res = MagicMock()
        res.price = 5
        # fake object Product.objects.get() return fake pr.price ( this 5 )
        mock_product.objects.get.return_value = res

        request = MagicMock()
        request.POST = {}  # must have

        fake_self = MagicMock()
        fake_self.kwargs = {}
        fake_self.total_deal_price = 0

        product = 'products-0-product'
        item_price = 'products-0-item_price'
        qty = 'products-0-qty'
        delete = 'products-0-DELETE'

        with patch.dict(request.POST,
                        {'products-TOTAL_FORMS': '1', item_price: '0', product : '3', qty :'50', delete: 'on' }), \
                        patch.dict(fake_self.kwargs, {'pk': '3'}):
            request = DealUpdateView.change_request_product(fake_self, request)
            self.assertEqual(request.POST['products-0-deal'], '3')
            self.assertEqual(request.POST['products-0-total_price'], '250')

def SETTINGS():
    from jenkins_epo.settings import DEFAULTS, SETTINGS
    patcher = patch.dict('jenkins_epo.settings.SETTINGS', DEFAULTS)
    patcher.start()
    SETTINGS.DEBUG = 0
    SETTINGS.DRY_RUN = 0
    yield SETTINGS
    patcher.stop()

def test_init_prompt(self):
        self.assertRegexpMatches(
            sys.ps1, '\001\033\[1;3[23]m\002>>> \001\033\[0m\002'
        )
        self.assertEqual(sys.ps2, '\001\033[1;31m\002... \001\033[0m\002')

        with patch.dict(os.environ,
                        {'SSH_CONNECTION': '1.1.1.1 10240 127.0.0.1 22'}):
            self.pymp.init_prompt()
            self.assertIn('[127.0.0.1]>>> ', sys.ps1)
            self.assertIn('[127.0.0.1]... ', sys.ps2)

def test_raw_input_edit_cmd(self, ignored):
        mocked_cmd = Mock()
        with patch.dict(self.pymp.commands, {'\e': mocked_cmd}):
            self.pymp.raw_input('>>> ')
            mocked_cmd.assert_called_once_with('code')

def test_raw_input_list_cmd0(self, ignored):
        mocked_cmd = Mock()
        with patch.dict(self.pymp.commands, {'\l': mocked_cmd}):
            ret = self.pymp.raw_input('>>> ')
            mocked_cmd.assert_called_once_with('shutil')

def test_raw_input_list_cmd1(self, ignored):
        mocked_cmd = Mock()
        with patch.dict(self.pymp.commands, {'\l': mocked_cmd}):
            self.pymp.raw_input('>>> ')
            mocked_cmd.assert_called_once_with('global')

def test_sh_exec1(self, mocked_chdir):
        """Test sh exec with cd, user home and shell variable"""
        self.pymp.locals['path'] = "~/${RUNTIME}/location"
        with patch.dict(pythonrc.os.environ, {'RUNTIME': 'dummy',
                                              'HOME': '/home/me/'}):
            self.pymp.process_sh_cmd('cd {path}')
            mocked_chdir.assert_called_once_with('/home/me/dummy/location')