python类MagicMock()的实例源码

def setUp(self):
        client = mock.MagicMock()
        bootstrap = factory.TrailingOrdersFactory().make_fake_bootstrap(self.INITIAL_SETUP)

        logging_patch = mock.patch('trading_system.systems.trailing_orders.system.logging')
        self.addCleanup(logging_patch.stop)
        logging_patch.start()

        self.system = TrailingOrders(client, bootstrap)
        self.system.get_pending_orders = mock.MagicMock(return_value=[])

        self.set_next_operation = mock.MagicMock()
        next_operation_patcher = mock.patch(
            'trading_system.systems.trailing_orders.system.TrailingOrders.set_next_operation', self.set_next_operation
        )
        self.addCleanup(next_operation_patcher.stop)
        next_operation_patcher.start()

def test_decrypt(self, mock_import_module):
        mock_crypter_module = mock.MagicMock()
        mock_crypter_module.__name__ = 'secretcrypt.mock_crypter'

        def mock_import_side_effect(*args, **kwargs):
            self.assertEqual(kwargs['package'], secretcrypt.__name__)
            if args[0] == '.mock_crypter':
                return mock_crypter_module
            raise Exception('Importing wrong module')
        mock_import_module.side_effect = mock_import_side_effect

        secret = StrictSecret('mock_crypter:key=value&key2=value2:myciphertext')
        self.assertEqual(secret._decrypt_params, dict(key='value', key2='value2'))
        self.assertEqual(secret._ciphertext, b'myciphertext')

        secret.decrypt()
        secret.decrypt()
        mock_crypter_module.decrypt.assert_called_with(
            b'myciphertext',
            key='value',
            key2='value2',
        )

def test_eager_decrypt(self, mock_import_module):
        mock_crypter_module = mock.MagicMock()
        mock_crypter_module.decrypt.side_effect = lambda *args, **kwargs: b'plaintext'
        mock_crypter_module.__name__ = 'secretcrypt.mock_crypter'

        def mock_import_side_effect(*args, **kwargs):
            self.assertEqual(kwargs['package'], secretcrypt.__name__)
            if args[0] == '.mock_crypter':
                return mock_crypter_module
            raise Exception('Importing wrong module')
        mock_import_module.side_effect = mock_import_side_effect

        secret = Secret('mock_crypter:key=value&key2=value2:myciphertext')
        mock_crypter_module.decrypt.assert_called_with(
            b'myciphertext',
            key='value',
            key2='value2',
        )
        mock_crypter_module.reset_mock()
        plaintext = secret.get()
        self.assertEqual(b'plaintext', plaintext)
        mock_crypter_module.assert_not_called()

def test_run_with_existing_remote_build_container(self, skipper_runner_run_mock, requests_get_mock):
        requests_response_class_mock = mock.MagicMock(spec='requests.Response')
        requests_response_mock = requests_response_class_mock.return_value
        requests_response_mock.json.return_value = {
            'name': 'my_image',
            'tags': ['latest', 'aaaaaaa', 'bbbbbbb', 'build-container-tag']
        }
        requests_response_mock.status_code = http_client.OK
        requests_get_mock.return_value = requests_response_mock

        command = ['ls', '-l']
        run_params = command
        self._invoke_cli(
            global_params=self.global_params,
            subcmd='run',
            subcmd_params=run_params
        )
        expected_image_name = 'registry.io:5000/build-container-image:build-container-tag'
        skipper_runner_run_mock.assert_called_once_with(command, fqdn_image=expected_image_name, environment=[],
                                                        interactive=False, name=None, net='host', volumes=None,
                                                        workdir=None, use_cache=False)

def test_run_with_non_existing_build_container(self, requests_get_mock):
        requests_response_class_mock = mock.MagicMock(spec='requests.Response')
        requests_response_mock = requests_response_class_mock.return_value
        requests_response_mock.json.return_value = {
            'name': 'my_image',
            'tags': ['latest', 'aaaaaaa', 'bbbbbbb']
        }

        requests_get_mock.return_value = requests_response_mock
        command = ['ls', '-l']
        run_params = command
        ret = self._invoke_cli(
            global_params=self.global_params,
            subcmd='run',
            subcmd_params=run_params
        )
        self.assertIsInstance(ret.exception, click.exceptions.ClickException)

def test_cli_with_custom_login(self, mocked_method):
        mocked = mock.MagicMock()
        mocked.login.return_value = {"Status": "Login Succeeded"}
        mocked.push.return_value = [
            {"stream": "In process"},
            {"status": "Successfully pushed"}]
        mocked_method.return_value = mocked
        with FakeProjectDirectory() as tmpdir:
            add_sh_fake_config(tmpdir)
            runner = CliRunner()
            result = runner.invoke(
                cli, ["dev", "--version", "test", "--username", "user",
                      "--password", "pass", "--email", "mail"])
            assert result.exit_code == 0
            mocked.login.assert_called_with(
                email=u'mail', password=u'pass',
                reauth=False, registry='registry', username=u'user')
            mocked.push.assert_called_with(
                'registry/user/project:test', decode=True,
                insecure_registry=False, stream=True)

def test_cli_with_insecure_registry(self, mocked_method):
        mocked = mock.MagicMock()
        mocked.login.return_value = {"Status": "Login Succeeded"}
        mocked.push.return_value = [
            {"stream": "In process"},
            {"status": "Successfully pushed"}]
        mocked_method.return_value = mocked
        with FakeProjectDirectory() as tmpdir:
            add_sh_fake_config(tmpdir)
            runner = CliRunner()
            result = runner.invoke(
                cli, ["dev", "--version", "test", "--insecure"])
            assert result.exit_code == 0
            assert not mocked.login.called
            mocked.push.assert_called_with(
                'registry/user/project:test', decode=True,
                insecure_registry=True, stream=True)

def test_cli(self, test_mock, mocked_method):
        mocked = mock.MagicMock()
        mocked.build.return_value = [
            {"stream": "all is ok"},
            {"stream": "Successfully built 12345"}]
        mocked_method.return_value = mocked
        with FakeProjectDirectory() as tmpdir:
            add_scrapy_fake_config(tmpdir)
            add_sh_fake_config(tmpdir)
            add_fake_dockerfile(tmpdir)
            setup_py_path = os.path.join(tmpdir, 'setup.py')
            assert not os.path.isfile(setup_py_path)
            runner = CliRunner()
            result = runner.invoke(cli, ["dev", "-d"])
            assert result.exit_code == 0
            mocked.build.assert_called_with(
                decode=True, path=tmpdir, tag='registry/user/project:1.0')
            assert os.path.isfile(setup_py_path)
            test_mock.assert_called_with("dev", None)

def test_cli_custom_version(self, test_mock, mocked_method):
        mocked = mock.MagicMock()
        mocked.build.return_value = [
            {"stream": "all is ok"},
            {"stream": "Successfully built 12345"}]
        mocked_method.return_value = mocked
        with FakeProjectDirectory() as tmpdir:
            add_scrapy_fake_config(tmpdir)
            add_sh_fake_config(tmpdir)
            add_fake_dockerfile(tmpdir)
            runner = CliRunner()
            result = runner.invoke(cli, ["dev", "--version", "test"])
            assert result.exit_code == 0
            mocked.build.assert_called_with(
                decode=True, path=tmpdir, tag='registry/user/project:test')
            test_mock.assert_called_with("dev", "test")

def test_request_backend(self, options, config):
        driver = self._get_driver(options, config)
        driver._relay_request = mock.MagicMock()
        response = requests.Response()
        response.status_code = requests.codes.ok
        driver._relay_request.return_value = response

        context = mock.MagicMock()
        data_dict = {}
        obj_name = 'object'
        action = 'TEST'

        code, message = driver._request_backend(context, data_dict, obj_name,
                                                action)

        self.assertEqual(requests.codes.ok, code)
        self.assertEqual({'message': None}, message)
        driver._relay_request.assert_called()

def test_create_resource(self, options, config):
        driver = self._get_driver(options, config)
        driver._request_backend = mock.MagicMock()
        status_code = requests.codes.ok
        context = mock.MagicMock()
        res_type = 'network'
        res_data = {
            'id': 'network-123',
            'tenant_id': ATTR_NOT_SPECIFIED,
        }
        driver._request_backend.return_value = (status_code, res_data)

        response = driver._create_resource(res_type,
                                           context,
                                           {res_type: res_data})

        self.assertEqual(response, res_data)
        driver._request_backend.assert_called_with(context, mock.ANY,
                                                   res_type, 'CREATE')

def test_get_resource(self, options, config):
        driver = self._get_driver(options, config)
        driver._request_backend = mock.MagicMock()
        status_code = requests.codes.ok
        context = mock.MagicMock()
        res_type = 'network'
        res_data = {
            'id': 'network-123',
        }
        fields = ['id']
        driver._request_backend.return_value = (status_code, res_data)

        response = driver._get_resource(res_type,
                                        context, res_data['id'], fields)

        self.assertEqual(response, res_data)
        driver._request_backend.assert_called_with(context, mock.ANY,
                                                   res_type, 'READ')

def test_update_resource(self, options, config):
        driver = self._get_driver(options, config)
        driver._request_backend = mock.MagicMock()
        status_code = requests.codes.ok
        context = mock.MagicMock()
        res_type = 'network'
        res_data = {
            'id': 'network-123',
        }
        driver._request_backend.return_value = (status_code, res_data)

        response = driver._update_resource(res_type, context, res_data['id'],
                                           {res_type: res_data})

        self.assertEqual(response, res_data)
        driver._request_backend.assert_called_with(context, mock.ANY,
                                                   res_type, 'UPDATE')

def test_list_resource(self, options, config):
        driver = self._get_driver(options, config)
        driver._request_backend = mock.MagicMock()
        status_code = requests.codes.ok
        context = mock.MagicMock()
        res_type = 'network'
        res_data = {
            'id': 'network-123',
        }
        fields = ['id']
        driver._request_backend.return_value = (status_code, [res_data])

        response = driver._list_resource(res_type, context, None, fields)

        self.assertEqual(response, [res_data])
        driver._request_backend.assert_called_with(context, mock.ANY,
                                                   res_type, 'READALL')

def test_add_router_interface(self, options, config):
        driver = self._get_driver(options, config)
        driver._request_backend = mock.MagicMock()
        status_code = requests.codes.ok
        context = mock.MagicMock()
        res_type = 'router'
        res_data = {
            'id': 'router-123',
        }
        res_id = res_data['id']
        driver._request_backend.return_value = (status_code, res_data)

        response = driver.add_router_interface(context, res_id, res_data)

        self.assertEqual(response, res_data)
        driver._request_backend.assert_called_with(context, mock.ANY,
                                                   res_type, 'ADDINTERFACE')

def test_remove_router_interface(self, options, config):
        driver = self._get_driver(options, config)
        driver._request_backend = mock.MagicMock()
        status_code = requests.codes.ok
        context = mock.MagicMock()
        res_type = 'router'
        res_data = {
            'id': 'router-123',
        }
        res_id = res_data['id']
        driver._request_backend.return_value = (status_code, res_data)

        response = driver.remove_router_interface(context, res_id, res_data)

        self.assertEqual(response, res_data)
        driver._request_backend.assert_called_with(context, mock.ANY,
                                                   res_type, 'DELINTERFACE')

def test_prepared_query_not_found(self):
        session = self.make_session()
        pool = session._pools.get.return_value
        connection = Mock(spec=Connection)
        pool.borrow_connection.return_value = (connection, 1)

        rf = self.make_response_future(session)
        rf.send_request()

        session.cluster._prepared_statements = MagicMock(dict)
        prepared_statement = session.cluster._prepared_statements.__getitem__.return_value
        prepared_statement.query_string = "SELECT * FROM foobar"
        prepared_statement.keyspace = "FooKeyspace"
        rf._connection.keyspace = "FooKeyspace"

        result = Mock(spec=PreparedQueryNotFound, info='a' * 16)
        rf._set_result(None, None, None, result)

        self.assertTrue(session.submit.call_args)
        args, kwargs = session.submit.call_args
        self.assertEqual(rf._reprepare, args[-5])
        self.assertIsInstance(args[-4], PrepareMessage)
        self.assertEqual(args[-4].query, "SELECT * FROM foobar")

def test_prepared_query_not_found_bad_keyspace(self):
        session = self.make_session()
        pool = session._pools.get.return_value
        connection = Mock(spec=Connection)
        pool.borrow_connection.return_value = (connection, 1)

        rf = self.make_response_future(session)
        rf.send_request()

        session.cluster._prepared_statements = MagicMock(dict)
        prepared_statement = session.cluster._prepared_statements.__getitem__.return_value
        prepared_statement.query_string = "SELECT * FROM foobar"
        prepared_statement.keyspace = "FooKeyspace"
        rf._connection.keyspace = "BarKeyspace"

        result = Mock(spec=PreparedQueryNotFound, info='a' * 16)
        rf._set_result(None, None, None, result)
        self.assertRaises(ValueError, rf.result)

def runTest(self, mock_template, mock_yaml, mock_isdir, mock_isfile, mock_open):
        ''' Execute test '''
        self.config = mock.MagicMock(spec_set={'runbook_path' : '/path/'})
        mock_isfile.return_value = True
        mock_isdir.return_value = True
        mock_template = mock.MagicMock(**{
            'render.return_value' : """
                '*':
                  - book
                'target':
                  - book1
                  - book2
            """
        })
        mock_yaml.return_value = {'*':['book'], 'target':['book1', 'book2']}
        mock_open.return_value = mock.MagicMock(spec=file)
        result = cache_runbooks(self.config, self.logger)
        self.assertEqual(result.keys(), ['*', 'target'], "Expected dictionary keys not found")
        self.assertTrue(mock_open.called, "open not called")
        self.assertTrue(mock_yaml.called, "yaml.safe_load not called")
        self.assertTrue(mock_isdir.called, "os.path.isdir not called")
        self.assertEqual(mock_isfile.call_count, 4,
            "mock_open.call_count {0} is not 4".format(mock_open.call_count))

def runTest(self, mock_run, mock_put, mock_local, mock_hide, mock_env, mock_set_env):
        ''' Execute test '''
        # Set mock_env to empty dict
        mock_env = mock.MagicMock(spec={})
        mock_set_env.return_value = mock_env
        mock_local.return_value = mock.MagicMock(**{ 'succeeded': True})
        mock_run.return_value = mock.MagicMock(**{ 'succeeded': True})
        mock_put = True
        mock_hide = True
        action = {
            'type' : 'cmd',
            'execute_from' : 'remote',
            'cmd' : "bash"
        }
        results = execute_runbook(action, self.target, self.config, self.logger)
        self.assertTrue(results)
        self.assertFalse(self.logger.warn.called)
        self.assertTrue(mock_local.called)
        mock_local.assert_called_with("bash", capture=True)