python类mock_open()的实例源码

def patch_open_read(files):
    """
    Patch open() command and mock read() results

    :type files: dict
    """
    files = {os.path.abspath(path): content for path, content in files.items()}

    def mock_open_wrapper(path, *args, **kwargs):
        __tracebackhide__ = True  # pylint: disable=unused-variable

        assert path in files, 'try to open a non-mocked path\n    desired={desired!r}\n    mocked={mocked!r}'.format(
            desired=path, mocked=files.keys())

        open_mock = mock.mock_open(read_data=files[path])
        return open_mock(path, *args, **kwargs)

    return mock.patch(__get_open_ref(), mock_open_wrapper)

def test_load_ok(self):
        in_config = json.dumps({'command': '/bin/true',
                                'config_files': {}})

        mo = mock.mock_open(read_data=in_config)
        with mock.patch.object(set_configs, 'open', mo):
            config = set_configs.load_config()
            set_configs.copy_config(config)
            self.assertEqual([
                mock.call('/var/lib/kolla/config_files/config.json'),
                mock.call().__enter__(),
                mock.call().read(),
                mock.call().__exit__(None, None, None),
                mock.call('/run_command', 'w+'),
                mock.call().__enter__(),
                mock.call().write(u'/bin/true'),
                mock.call().__exit__(None, None, None)], mo.mock_calls)

def test_saving_image(monkeypatch):
    # This still has some mocks, but they are more localized and do not
    # have to be monkey-patched into standard library modules (always a
    # risky business).
    mock_file_open = mock_open()

    fake_uuid = '123e4567-e89b-12d3-a456-426655440000'

    def mock_uuidgen():
        return fake_uuid

    fake_image_bytes = b'fake-image-bytes'
    fake_request_stream = io.BytesIO(fake_image_bytes)
    storage_path = 'fake-storage-path'
    store = look.images.ImageStore(
        storage_path,
        uuidgen=mock_uuidgen,
        fopen=mock_file_open
    )

    assert store.save(fake_request_stream, 'image/png') == fake_uuid + '.png'
    assert call().write(fake_image_bytes) in mock_file_open.mock_calls

def _test_missing_param(self, params, image_id, flavor_id):
        with mock.patch('functest.opnfv_tests.openstack.tempest.'
                        'conf_utils.ConfigParser.RawConfigParser.'
                        'set') as mset, \
            mock.patch('functest.opnfv_tests.openstack.tempest.'
                       'conf_utils.ConfigParser.RawConfigParser.'
                       'read') as mread, \
            mock.patch('functest.opnfv_tests.openstack.tempest.'
                       'conf_utils.ConfigParser.RawConfigParser.'
                       'write') as mwrite, \
            mock.patch('__builtin__.open', mock.mock_open()), \
            mock.patch('functest.opnfv_tests.openstack.tempest.'
                       'conf_utils.backup_tempest_config'), \
            mock.patch('functest.utils.functest_utils.yaml.safe_load',
                       return_value={'validation': {'ssh_timeout': 300}}):
            CONST.__setattr__('OS_ENDPOINT_TYPE', None)
            conf_utils.configure_tempest_update_params(
                'test_conf_file', image_id=image_id, flavor_id=flavor_id)
            mset.assert_any_call(params[0], params[1], params[2])
            self.assertTrue(mread.called)
            self.assertTrue(mwrite.called)

def test_pull_image_success(self, mock_find_image, mock_download_image,
                                mock_should_pull_image, mock_search_on_host):
        mock_should_pull_image.return_value = True
        mock_search_on_host.return_value = {'image': 'nginx', 'path': 'xyz',
                                            'checksum': 'xxx'}
        image_meta = mock.MagicMock()
        image_meta.id = '1234'
        mock_find_image.return_value = image_meta
        mock_download_image.return_value = 'content'
        CONF.set_override('images_directory', self.test_dir, group='glance')
        out_path = os.path.join(self.test_dir, '1234' + '.tar')
        mock_open_file = mock.mock_open()
        with mock.patch('zun.image.glance.driver.open', mock_open_file):
            ret = self.driver.pull_image(None, 'image', 'latest', 'always')
        mock_open_file.assert_any_call('xyz', 'rb')
        mock_open_file.assert_any_call(out_path, 'wb')
        self.assertTrue(mock_search_on_host.called)
        self.assertTrue(mock_should_pull_image.called)
        self.assertTrue(mock_find_image.called)
        self.assertTrue(mock_download_image.called)
        self.assertEqual(({'image': 'image', 'path': out_path}, False), ret)

def test_fetch_file_local_dir_specified(OSF_project, os_path_exists,
                                        os_makedirs):
    # check that `osf fetch` opens the right files with the right name
    # and mode when specifying a local filename
    args = MockArgs(project='1234', remote='osfstorage/a/a/a',
                    local='subdir/foobar.txt')

    mock_open_func = mock_open()

    with patch('osfclient.cli.open', mock_open_func):
        fetch(args)

    OSF_project.assert_called_once_with('1234')
    # check that the project and the files have been accessed
    store = OSF_project.return_value.storages[0]
    assert store._name_mock.return_value == 'osfstorage'

    assert (mock.call('subdir/foobar.txt', 'wb') in
            mock_open_func.mock_calls)
    assert mock.call('subdir', exist_ok=True) in os_makedirs.mock_calls

def test_readlines_data(self):
        # Test that emulating a file that ends in a newline character works
        mock = mock_open(read_data='foo\nbar\nbaz\n')
        with patch('%s.open' % __name__, mock, create=True):
            h = open('bar')
            result = h.readlines()
        self.assertEqual(result, ['foo\n', 'bar\n', 'baz\n'])

        # Test that files without a final newline will also be correctly
        # emulated
        mock = mock_open(read_data='foo\nbar\nbaz')
        with patch('%s.open' % __name__, mock, create=True):
            h = open('bar')
            result = h.readlines()

        self.assertEqual(result, ['foo\n', 'bar\n', 'baz'])

def test_tarball_download(self, mock_requests, mock_temp):
        url = 'http://localhost:8879/charts/mariadb-0.1.0.tgz'
        mock_temp.mkstemp.return_value = (None, '/tmp/armada')
        mock_response = mock.Mock()
        mock_response.content = 'some string'
        mock_requests.get.return_value = mock_response

        mock_open = mock.mock_open()
        with mock.patch.object(source, 'open', mock_open, create=True):
            source.download_tarball(url)

        mock_temp.mkstemp.assert_called_once()
        mock_requests.get.assert_called_once_with(url, verify=False)
        mock_open.assert_called_once_with('/tmp/armada', 'wb')
        mock_open().write.assert_called_once_with(
            mock_requests.get(url).content)

def test_migrate_ovs_default_file(self, mock_restart):
        # Tests that the /etc/default/openvswitch-switch file is/isn't
        # migrated on the upgrade-charm hook and that no restarts are
        # attempted of the openvswitch-switch service.
        tests = [
            ('package-provided-openvswitch-switch', True),
            ('16.07-dpdk-openvswitch-switch', True),
            ('16.10-openvswitch-switch', False),
        ]
        for sample, should_migrate in tests:
            self.CONFIGS.write.reset_mock()
            with open('unit_tests/%s' % sample, 'r') as f:
                content = f.read()

            with patch('builtins.open', mock_open(read_data=content),
                       create=True):
                self._call_hook('upgrade-charm')
                if should_migrate:
                    self.CONFIGS.write.assert_called_with(utils.OVS_DEFAULT)
                else:
                    self.CONFIGS.write.assert_not_called()
                self.assertEqual(0, mock_restart.call_count)

def test_copy_expert(self):
        m = mock.mock_open(read_data='{"some": "json"}')
        with mock.patch('airflow.hooks.postgres_hook.open', m, create=True) as m:
            statement = "SQL"
            filename = "filename"

            self.cur.fetchall.return_value = None
            f = m(filename, 'w')
            def test_open(filename, mode):
                return f

            self.assertEqual(None, self.db_hook.copy_expert(statement, filename, open=test_open))

            self.conn.close.assert_called_once()
            self.cur.close.assert_called_once()
            self.cur.copy_expert.assert_called_once_with(statement, f)

def test_options_write_config_per_module_args(self):
        """Test that attempting to write a config with per module args yields expected results."""
        configuration_path = os.path.join(self.callpath, "test/configurations/test_write.cfg")
        sys.argv = ["ec2rl", "run"]
        module_path = os.path.join(self.callpath, "test/modules/mod.d")
        modules = ec2rlcore.moduledir.ModuleDir(module_path)
        self.options = ec2rlcore.options.Options(subcommands=self.__subcommands)
        # Test for module present in modules list
        self.options.per_module_args["atop"] = {}
        self.options.per_module_args["atop"]["times"] = "1"
        # Test for module not present in modules list
        self.options.per_module_args["test"] = {}
        self.options.per_module_args["test"]["times"] = "1"
        # Test for module named "Global"
        self.options.per_module_args["Global"] = {}
        self.options.per_module_args["Global"]["times"] = "1"
        with mock.patch("{}.open".format(builtins_name), new_callable=mock.mock_open()) as open_mock:
            config = self.options.write_config(configuration_path, modules)

            self.assertIsInstance(config, configparser.ConfigParser)
            self.assertEqual(config["atop"]["times"], "1")
            self.assertEqual(config["test"]["times"], "1")
            self.assertFalse("times" in config["Global"])
            self.assertTrue(open_mock.called)

def test_csv_instances(mocker):
    mock_open = mocker.patch("%s.open" % builtin_mock, mock.mock_open(read_data=DATA))
    mock_csv_reader = mocker.patch("csv.DictReader")

    # for whatever reason mock_open is not sufficent since the DictReader will return nothing
    # so mocking the csv reader is necessary
    ret = []
    for line in DATA.split("\n"):
        ret.append(dict(zip(['name','address','random'], line.split("|"))))
    mock_csv_reader.return_value = ret

    csv_obj = CsvInventory('/tmp/dummy/path', ' name,address, random ', ' | ')
    instances = csv_obj.instances()
    print(instances)
    expected_instances = [Instance(name='devenv-pubsrv', address='13.14.15.16', source='csv'),
                          Instance(name='testenv-pubsrv', address='1.2.3.4', source='csv'),
                          Instance(name='devenv-pubsrv', address='9.10.11.12', source='csv'),
                          Instance(name='testenv-pubsrv', address='5.6.7.8', source='csv'),
                          Instance(name='testenv-formsvc', address='17.18.19.20', source='csv')]
    assert set(instances) == set(expected_instances)

def test_base_snap_config_uwsgi(self, mock_os, mock_utils,
                                    mock_renderer):
        '''Ensure wrapped binary of uwsgi called with correct arguments'''
        self.mock_snap_utils(mock_utils)
        snap = base.OpenStackSnap(os.path.join(TEST_DIR,
                                               'snap-openstack.yaml'))
        mock_os.path.exists.side_effect = self.mock_exists
        mock_os.environ = {}
        mock_os.path.basename.side_effect = 'keystone.conf'
        builtin = '__builtin__'
        if sys.version_info > (3, 0):
            builtin = 'builtins'
        with patch('{}.open'.format(builtin), mock_open(), create=True):
            snap.execute(['snap-openstack',
                          'keystone-uwsgi'])
        mock_os.execvpe.assert_called_with(
            '/snap/keystone/current/bin/uwsgi',
            ['/snap/keystone/current/bin/uwsgi', '--master',
             '--die-on-term', '-H', '/snap/keystone/current/usr',
             '--emperor', '/var/snap/keystone/common/etc/uwsgi/snap',
             '--logto', '/var/snap/keystone/common/log/uwsgi.log'],
            {},
        )

def test_param_broken_cfg(testdir, content):
    '''verifies pytest-github loads completed info from provided --github-cfg parameter'''

    # create github.yml config for testing

    with mock.patch('os.path.isfile', return_value=True) as mock_isfile:
        with mock.patch('pytest_github.plugin.open', mock.mock_open(read_data=content), create=True) as mock_open:
            with mock.patch('pytest_github.plugin.GitHubPytestPlugin') as mock_plugin:
                result = testdir.runpytest()

    # Assert py.test exit code
    assert result.ret == EXIT_NOTESTSCOLLECTED

    # Assert mock isfile called
    mock_isfile.assert_called_once_with('github.yml')

    # Assert mock open called on provided file
    mock_open.assert_called_once_with('github.yml', 'r')

    # Assert plugin initialized as expected
    mock_plugin.assert_called_once_with(None, None, completed_labels=[])

def test_restore_success_scenario(self, mock_open, mock_wait,
                                      mock_post, mock_gen_hmac, mock_warn):

        data = json.dumps({"refresh_conf_url": "aa"})
        mock_open.side_effect = [mock.mock_open(read_data=data).return_value,
                                 mock.mock_open(read_data=data).return_value,
                                 mock.mock_open(read_data=data).return_value]

        mock_gen_hmac.return_value = "headers"

        mock_post.side_effect = [
            requests.exceptions.RequestException,
            mock.Mock(status_code=500),
            mock.Mock(status_code=200)
        ]
        conf.restore(["hmac"], 50)
        mock_open.assert_has_calls(
            [mock.call(conf._RUNTIME_CONF_FILE % 50, "rw")] * 3)
        self.assertEqual(1, mock_warn.call_count)
        mock_post.assert_has_calls([
            mock.call("aa", headers="headers"),
            mock.call("aa", headers="headers"),
            mock.call("aa", headers="headers")
        ])
        mock_wait.assert_has_calls([mock.call(1), mock.call(1)])

def test_generate_uses_last_modified_if_created_is_missing():
    kinto_client = mock.MagicMock()
    data = ADDONS_DATA.copy()
    del data['details']['created']
    kinto_client.get_records.return_value = [data]
    collections = ['/buckets/blocklists/collections/addons']

    with mock.patch('amo2kinto.generator.os.makedirs'):
        f = mock.mock_open()
        with mock.patch('amo2kinto.generator.open', f, create=True):
            generate(kinto_client, collections, 'tmp', 'collection.tpl', 'record.tpl')

            assert f.return_value.write.call_count == 2

            # Present in index
            assert b'May 13, 2013' in f.return_value.write.call_args_list[0][0][0]

            # Present in the record file
            assert b'May 13, 2013' in f.return_value.write.call_args_list[1][0][0]

def test_program_flag_reads_hquery_program_from_file(capsys, mocker):
    expected_filename = 'filename.hq'
    mocked_open = mock_open(read_data='''
                                        //p
                                        ->
                                        $_/text()''')
    mocker.patch('hq.hq.docopt').return_value = simulate_args_dict(
        program=expected_filename)
    mocker.patch('sys.stdin.read').return_value = wrap_html_body('<p>foo</p>')
    mocker.patch('hq.hq.open', mocked_open, create=True)

    main()

    actual, _ = capture_console_output(capsys)
    mocked_open.assert_called_with(expected_filename)
    assert actual == 'foo'

def test_get_recovery_data(self, mock_yaml_safe_load, mock_isfile):
        import os
        recovery_file = os.path.join("/path/to/a/base_folder", "print_recovery_data.yaml")

        mock_isfile.return_value = True

        data = dict(path="some_path.gco",
                    origin="local",
                    pos=1234,
                    date=123456789)
        mock_yaml_safe_load.return_value = data

        with mock.patch("__builtin__.open", mock.mock_open(read_data=data), create=True) as m:
            result = self.file_manager.get_recovery_data()

            self.assertDictEqual(data, result)

            m.assert_called_with(recovery_file)

            mock_handle = m()
            mock_yaml_safe_load.assert_called_with(mock_handle)

def test_get_value_for_attribute_with_a_present_attribute(self, mock_meta_req):
        """Test get_value_for_attribute returns correctly.

        Setup:
            * Mock out a httplib.HTTPResponse .
            * Return that from _issue_http_request.

        Expected results:
            * A matching string.
        """
        mock_response = 'expected_response'

        with mock.patch('httplib.HTTPResponse',
                        mock.mock_open(read_data=mock_response)) as mock_http_resp:
            mock_http_resp.return_value.status = httplib.OK
            mock_meta_req.side_effect = mock_http_resp

            actual_response = metadata_server.get_value_for_attribute('')

        self.assertEqual(actual_response, mock_response)

def test_show_stops(self, monkeypatch):
        def mock_init(*args, **kwargs):
            return None

        def mock_func(*args):
            return True

        monkeypatch.setattr('focli.foline.FoliPrint.__init__', mock_init)
        monkeypatch.setattr('focli.foline.FoliPrint.print_lines', mock_func)
        monkeypatch.setattr('sys.argv', ['focli', '157'])
        assert focli.main() == 0
        fcontent = '{"123":"abc"}'
        monkeypatch.setattr('sys.argv', ['focli'])
        with mock.patch.object(builtins, 'open',
                               mock.mock_open(read_data=fcontent)):
            assert focli.main() == 0
        with mock.patch.object(builtins, 'open',
                               mock.mock_open(read_data="{}")):
            assert focli.main() == 1

def test__prepare_variables_configdrive_file(self, mem_req_mock):
        i_info = self.node.instance_info
        i_info['configdrive'] = 'fake-content'
        self.node.instance_info = i_info
        self.node.save()
        self.config(tempdir='/path/to/tmpfiles')
        expected = {"image": {"url": "http://image",
                              "validate_certs": "yes",
                              "source": "fake-image",
                              "mem_req": 2000,
                              "disk_format": "qcow2",
                              "checksum": "md5:checksum"},
                    'configdrive': {'type': 'file',
                                    'location': '/path/to/tmpfiles/%s.cndrive'
                                    % self.node.uuid}}
        with mock.patch.object(ansible_deploy, 'open', mock.mock_open(),
                               create=True) as open_mock:
            with task_manager.acquire(self.context, self.node.uuid) as task:
                self.assertEqual(expected,
                                 ansible_deploy._prepare_variables(task))
            open_mock.assert_has_calls((
                mock.call('/path/to/tmpfiles/%s.cndrive' % self.node.uuid,
                          'w'),
                mock.call().__enter__(),
                mock.call().write('fake-content'),
                mock.call().__exit__(None, None, None)))

def test__get_clean_steps(self, load_mock):
        steps = [{"interface": "deploy",
                  "name": "foo",
                  "args": {"spam": {"required": True, "value": "ham"}}},
                 {"name": "bar",
                  "interface": "deploy",
                  "priority": 100}]
        load_mock.return_value = steps
        expected = [{"interface": "deploy",
                     "step": "foo",
                     "priority": 10,
                     "abortable": False,
                     "argsinfo": {"spam": {"required": True}},
                     "args": {"spam": "ham"}},
                    {"interface": "deploy",
                     "step": "bar",
                     "priority": 100,
                     "abortable": False,
                     "argsinfo": {},
                     "args": {}}]
        d_info = self.node.driver_info
        d_info['ansible_clean_steps_config'] = 'custom_clean'
        self.node.driver_info = d_info
        self.node.save()
        self.config(group='ansible', playbooks_path='/path/to/playbooks')

        with mock.patch.object(ansible_deploy, 'open', mock.mock_open(),
                               create=True) as open_mock:
            self.assertEqual(
                expected,
                ansible_deploy._get_clean_steps(
                    self.node, interface="deploy",
                    override_priorities={"foo": 10}))
            open_mock.assert_has_calls((
                mock.call('/path/to/playbooks/custom_clean'),))
            load_mock.assert_called_once_with(
                open_mock().__enter__.return_value)

def test_read(self):
        fake_file = "fake_file_{}".format(self.build.timestamp())
        mock_read = mock_open()

        with patch('ardy.core.build.build.open', mock_read, create=False):
            self.build.read(fake_file)

        self.assertEqual(mock_read.call_count, 1)

def test_read():
    ''' Test fs.read() works. '''
    filename = 'somefile'
    data = 'somedata'

    with patch('__builtin__.open', mock_open(read_data=data)) as mock_file:
        assert fs.read(filename) == data
        mock_file.assert_called_with(filename, 'r')

def test_write():
    ''' Test fs.write() works. '''
    filename = 'somefile'
    data = 'somedata'
    m = mock_open()

    with patch('__builtin__.open', m) as mock_file:
        fs.write(filename, data)
        mock_file.assert_called_with(filename, 'w')
        m().write.assert_called_with(data)

def test_is_rotational(self, hwd):
        read_data = "1"
        with patch("srv.salt._modules.cephdisks.open", mock_open(read_data=read_data)) as mock_file:
            expect = read_data
            out = hwd._is_rotational('disk/in/question')
            assert expect == out

def test_is_rotational_not(self, hwd):
        read_data = "0"
        with patch("srv.salt._modules.cephdisks.open", mock_open(read_data=read_data)) as mock_file:
            expect = read_data
            out = hwd._is_rotational('disk/in/question')
            assert expect == out

def test_is_removable_not(self, hwd):
        read_data = "0"
        with patch("srv.salt._modules.cephdisks.open", mock_open(read_data=read_data)) as mock_file:
            expect = None
            out = hwd._is_removable('disk/in/question')
            assert expect == out

def test_is_removable(self, hwd):
        read_data = "1"
        with patch("srv.salt._modules.cephdisks.open", mock_open(read_data=read_data)) as mock_file:
            expect = True
            out = hwd._is_removable('disk/in/question')
            assert expect == out

def test_write_checksum(self, log_mock, cfg):
        fs.CreateFile("{}/{}".format(checksum_dir, 'rgw.conf'), contents="foo=bar") 
        cfg = cfg('rgw')
        m = mock_open()
        with patch('__builtin__.open', m, create=True):
            ret = cfg.write_checksum('0b0b0b0b0b0b0b0b0b0b0')
            log_mock.debug.assert_called()
            m.assert_called_once_with('/srv/salt/ceph/configuration/files/ceph.conf.checksum/rgw.conf', 'w')
            m().write.assert_called_once_with('0b0b0b0b0b0b0b0b0b0b0')
            fs.RemoveFile('/srv/salt/ceph/configuration/files/ceph.conf.checksum/rgw.conf')