python类mock_open()的实例源码

def test_lattice_from_file( self, mock_lattice, mock_site ):
        cell_lengths = np.array( [ 2.0, 3.0, 4.0 ] )
        example_file = """1\n
                          site: 2
                          centre: 21.4669 -1.37 6.1334
                          vertices: 1 315 435 649
                          neighbours: 4 5 6
                          label: H"""
        mock_site.return_value = 'site'
        mock_lattice.return_value = 'lattice'
        with patch( 'builtins.open', mock_open( read_data=example_file), create=True) as m:
            lattice = init_lattice.lattice_from_sites_file( 'filename', cell_lengths )
        site_calls = mock_site.mock_calls[0][1]
        self.assertEqual( site_calls[0], 2 ) # site number
        np.testing.assert_array_equal( site_calls[1], np.array( [ 21.4669, -1.37, 6.1334 ] ) ) # r
        self.assertEqual( site_calls[2], [ 4, 5, 6 ] ) # neighbour list
        self.assertEqual( site_calls[3], 0.0 ) # ?
        self.assertEqual( site_calls[4], 'H' ) # site label 
        self.assertEqual( mock_lattice.mock_calls[0][1][0], [ 'site' ] )
        np.testing.assert_array_equal( mock_lattice.mock_calls[0][2]['cell_lengths'], cell_lengths )
        self.assertEqual( lattice, 'lattice' )

def test_lattice_from_file_with_energy( self, mock_lattice, mock_site ):
        cell_lengths = np.array( [ 2.0, 3.0, 4.0 ] )
        example_file = """1\n
                          site: 2
                          centre: 21.4669 -1.37 6.1334
                          vertices: 1 315 435 649
                          neighbours: 4 5 6
                          energy: -1.0
                          label: H"""
        mock_site.return_value = 'site'
        mock_lattice.return_value = 'lattice'
        with patch( 'builtins.open', mock_open( read_data=example_file), create=True) as m:
            lattice = init_lattice.lattice_from_sites_file( 'filename', cell_lengths )
        site_calls = mock_site.mock_calls[0][1]
        self.assertEqual( site_calls[0], 2 ) # site number
        np.testing.assert_array_equal( site_calls[1], np.array( [ 21.4669, -1.37, 6.1334 ] ) ) # r
        self.assertEqual( site_calls[2], [ 4, 5, 6 ] ) # neighbour list
        self.assertEqual( site_calls[3], -1.0 ) # site energy
        self.assertEqual( site_calls[4], 'H' ) # site label 
        self.assertEqual( mock_lattice.mock_calls[0][1][0], [ 'site' ] )
        np.testing.assert_array_equal( mock_lattice.mock_calls[0][2]['cell_lengths'], cell_lengths )
        self.assertEqual( lattice, 'lattice' )

def test_save_config():
    """Test Config object can save to update a file."""
    mock_config = mock.mock_open(read_data=CONFIG_DATA)
    with mock.patch('two1.commands.util.config.open', mock_config, create=True):
        c = config.Config('config_file')

    num_config_keys = len(c.state.keys())

    # Update an existing key and add a new one
    with mock.patch('two1.commands.util.config.open', mock_config, create=True):
        c.set('username', 'TEST_USERNAME', should_save=True)
        c.set('some_list_key', [123, 456, 789], should_save=True)

    # Import the newly saved configuration file
    new_config = json.loads(mock_config.return_value.write.call_args[0][0])

    mock_config.assert_called_with('config_file', mode='w')
    assert c.username == 'TEST_USERNAME'
    assert c.some_list_key == [123, 456, 789]
    assert new_config['username'] == 'TEST_USERNAME'
    assert new_config['some_list_key'] == [123, 456, 789]
    assert len(new_config.keys()) == num_config_keys + 1

def setup_method(self, method):
        self.patchers = []

        patcher = patch('os.mkdir')
        self.mkdir = patcher.start()
        self.patchers.append(patcher)

        patcher = patch('omnic.worker.subprocessmanager.subprocess')
        self.subprocess = patcher.start()
        class FakeResult:
            stdout = tree_object.encode('utf8')
        self.subprocess.run.return_value = FakeResult
        self.patchers.append(patcher)

        self.open = mock_open()
        patcher = patch('builtins.open', self.open)
        patcher.start()
        self.patchers.append(patcher)
        singletons.settings.use_settings(Settings)
        self.rgraph = resolvergraph.ResolverGraph()  # Create resolver graph

        from omnic.builtin.resolvers.git import GitUpdater
        GitUpdater.reset_hash_cache()

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_passes_required_arguments_in_payload_as_json_when_image_is_file(self, post_mock):
        post_mock.return_value.status_code = 200

        m = mock.mock_open(read_data=str.encode('test'))
        with mock.patch('builtins.open', m, create=True):
            kairos_face.enroll_face('sub_id', 'gallery', file='/a/image/file.jpg')

        _, kwargs = post_mock.call_args
        expected_payload = {
            'image': 'dGVzdA==',
            'subject_id': 'sub_id',
            'gallery_name': 'gallery',
            'multiple_faces': False
        }
        self.assertTrue('json' in kwargs)
        self.assertEqual(expected_payload, kwargs['json'])

def test_passes_multiple_faces_argument_in_payload_when_flag_is_set(self, post_mock):
        post_mock.return_value.status_code = 200

        m = mock.mock_open(read_data=str.encode('test'))
        with mock.patch('builtins.open', m, create=True):
            kairos_face.enroll_face('sub_id', 'gallery', file='/a/image/file.jpg', multiple_faces=True)

        _, kwargs = post_mock.call_args
        expected_payload = {
            'image': 'dGVzdA==',
            'subject_id': 'sub_id',
            'gallery_name': 'gallery',
            'multiple_faces': True
        }
        self.assertTrue('json' in kwargs)
        self.assertEqual(expected_payload, kwargs['json'])

def test_read_config_file_with_valid_data(self):
        """
        Test the read_config_file function with valid data.
        """

        # Check handling of storage_handler.
        data = {
            'storage_handlers': [
                {'name': 'commissaire.storage.etcd'},
            ],
        }
        with mock.patch('builtins.open',
                mock.mock_open(read_data=json.dumps(data))) as _open:
            conf = config.read_config_file()
            self.assertIsInstance(conf, dict)
            data['authentication_plugins'] = {}
            self.assertEquals(data, conf)

def test_read_config_file_with_storge_handler_as_dict(self):
        """
        Verify the read_config_file function turns storage_handlers into a list.
        """
        data = {
            'storage_handlers': {
                'name': 'commissaire.storage.etcd',
            }
        }
        with mock.patch('builtins.open',
                mock.mock_open(read_data=json.dumps(data))) as _open:
            conf = config.read_config_file()
            self.assertIsInstance(conf, dict)
            data['storage_handlers'] = [data['storage_handlers']]
            data['authentication_plugins'] = {}
            self.assertEquals(data, conf)

def test_read_config_file_with_valid_authentication_plugin(self):
        """
        Verify the read_config_file function parses valid
        authentication_plugin directives.
        """
        plugin_name = 'commissaire_htp.authentication.httpbasicauth'
        data = {
            'authentication_plugins': [{
                'name': plugin_name,
                'users': {},
            }]
        }
        with mock.patch('builtins.open',
                mock.mock_open(read_data=json.dumps(data))) as _open:
            conf = config.read_config_file()
            self.assertIsInstance(conf, dict)
            self.assertTrue(
                plugin_name in conf['authentication_plugins'].keys())
            self.assertEquals(
                data['authentication_plugins'][0]['users'],
                conf['authentication_plugins'][plugin_name]['users'])

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_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_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_get():
    """
    Ensure a successful get results in the expected file getting written on
    the local file system with the expected content.
    """
    mock_serial = mock.MagicMock()
    with mock.patch('microfs.execute', return_value=(b'hello', b'')) as exe:
        mo = mock.mock_open()
        with mock.patch('microfs.open', mo, create=True):
            assert microfs.get('hello.txt', 'local.txt', mock_serial)
            commands = [
                "from microbit import uart",
                "f = open('{}', 'rb')".format('hello.txt'),
                "r = f.read",
                "result = True",
                "while result:\n    result = r(32)\n    if result:\n        "
                "uart.write(result)\n",
                "f.close()",
            ]
            exe.assert_called_once_with(commands, mock_serial)
            mo.assert_called_once_with('local.txt', 'wb')
            handle = mo()
            handle.write.assert_called_once_with(b'hello')

def test_get_no_target():
    """
    Ensure a successful get results in the expected file getting written on
    the local file system with the expected content. In this case, since no
    target is provided, use the name of the remote file.
    """
    with mock.patch('microfs.execute', return_value=(b'hello', b'')) as exe:
        mo = mock.mock_open()
        with mock.patch('microfs.open', mo, create=True):
            assert microfs.get('hello.txt')
            commands = [
                "from microbit import uart",
                "f = open('{}', 'rb')".format('hello.txt'),
                "r = f.read",
                "result = True",
                "while result:\n    result = r(32)\n    if result:\n        "
                "uart.write(result)\n",
                "f.close()",
            ]
            exe.assert_called_once_with(commands, None)
            mo.assert_called_once_with('hello.txt', 'wb')
            handle = mo()
            handle.write.assert_called_once_with(b'hello')

def test_readline_data(self):
        # Check that readline will return all the lines from the fake file
        mock = mock_open(read_data='foo\nbar\nbaz\n')
        with patch('%s.open' % __name__, mock, create=True):
            h = open('bar')
            line1 = h.readline()
            line2 = h.readline()
            line3 = h.readline()
        self.assertEqual(line1, 'foo\n')
        self.assertEqual(line2, 'bar\n')
        self.assertEqual(line3, 'baz\n')

        # Check that we properly emulate a file that doesn't end in a newline
        mock = mock_open(read_data='foo')
        with patch('%s.open' % __name__, mock, create=True):
            h = open('bar')
            result = h.readline()
        self.assertEqual(result, 'foo')

def test_interleaved_reads(self):
        # Test that calling read, readline, and readlines pulls data
        # sequentially from the data we preload with
        mock = mock_open(read_data='foo\nbar\nbaz\n')
        with patch('%s.open' % __name__, mock, create=True):
            h = open('bar')
            line1 = h.readline()
            rest = h.readlines()
        self.assertEqual(line1, 'foo\n')
        self.assertEqual(rest, ['bar\n', 'baz\n'])

        mock = mock_open(read_data='foo\nbar\nbaz\n')
        with patch('%s.open' % __name__, mock, create=True):
            h = open('bar')
            line1 = h.readline()
            rest = h.read()
        self.assertEqual(line1, 'foo\n')
        self.assertEqual(rest, 'bar\nbaz\n')

def test_passes_required_arguments_in_payload_as_json_when_image_is_file(self, post_mock):
        post_mock.return_value.status_code = 200

        m = mock.mock_open(read_data=str.encode('test'))
        with mock.patch('builtins.open', m, create=True):
            kairos_face.enroll_face('sub_id', 'gallery', file='/a/image/file.jpg')

        _, kwargs = post_mock.call_args
        expected_payload = {
            'image': 'dGVzdA==',
            'subject_id': 'sub_id',
            'gallery_name': 'gallery',
            'multiple_faces': False
        }
        self.assertTrue('json' in kwargs)
        self.assertEqual(expected_payload, kwargs['json'])

def test_count(self):
        with patch('acscore.metric.nesting_loops.open', mock_open(read_data=self.files[0])):
            result1 = self.nesting_loops.count('')
        self.assertEqual({'2': 1, '3': 1}, result1)

        with patch('acscore.metric.nesting_loops.open', mock_open(read_data=self.files[1])):
            result2 = self.nesting_loops.count('')
        self.assertEqual({'2': 1, '3': 1}, result2)

        with patch('acscore.metric.nesting_loops.open', mock_open(read_data=self.files[2])):
            result3 = self.nesting_loops.count('')
        self.assertEqual({'2': 1}, result3)

        with patch('acscore.metric.nesting_loops.open', mock_open(read_data=self.files[3])):
            result4 = self.nesting_loops.count('')
        self.assertEqual({'1': 3}, result4)

        with patch('acscore.metric.nesting_loops.open', mock_open(read_data=self.files[4])):
            result5 = self.nesting_loops.count('')
        self.assertEqual({}, result5)

def setUp(self):
        self.gettempdirmocked = self.patch(
            'onmydesk.core.outputs.tempfile.gettempdir', return_value='/tmp')

        uuid4_mocked = mock.MagicMock()
        uuid4_mocked.hex = 'asjkdlajksdlakjdlakjsdljalksdjla'
        self.uuid4_mocked = self.patch(
            'onmydesk.core.outputs.uuid4', return_value=uuid4_mocked)

        self.open_result_mocked = mock.MagicMock()
        self.open_mocked = mock.mock_open()
        self.open_mocked.return_value = self.open_result_mocked
        self.patch('onmydesk.core.outputs.open', create=True, new=self.open_mocked)

        self.writer = mock.MagicMock()
        self.patch('onmydesk.core.outputs.csv.writer', return_value=self.writer)

        self.iterable_object = [
            ('Alisson', 38),
            ('Joao', 13),
        ]

def setUp(self):
        self.gettempdirmocked = self.patch(
            'onmydesk.core.outputs.tempfile.gettempdir', return_value='/tmp')

        self.iterable_object = [
            ('Alisson', 38),
            ('Joao', 13),
        ]

        self.open_result_mocked = mock.MagicMock()
        self.open_mocked = mock.mock_open()
        self.open_mocked.return_value = self.open_result_mocked
        self.patch('onmydesk.core.outputs.open', create=True, new=self.open_mocked)

        self.writer_mocked = mock.MagicMock()
        self.patch('onmydesk.core.outputs.csv.writer',
                   return_value=self.writer_mocked)

        uuid4_mocked = mock.MagicMock()
        uuid4_mocked.hex = 'asjkdlajksdlakjdlakjsdljalksdjla'
        self.uuid4_mocked = self.patch(
            'onmydesk.core.outputs.uuid4', return_value=uuid4_mocked)

def test_not_remove_lib_if_not_upgrade(self, mock_os, mock_shutil):
        """Test removal of library with no upgrade."""
        ha_version = __version__

        mock_os.path.isdir = mock.Mock(return_value=True)

        mock_open = mock.mock_open()
        with mock.patch('homeassistant.config.open', mock_open, create=True):
            opened_file = mock_open.return_value
            opened_file.readline.return_value = ha_version

            self.hass.config.path = mock.Mock()

            config_util.process_ha_config_upgrade(self.hass)

            assert mock_os.path.isdir.call_count == 0
            assert mock_shutil.rmtree.call_count == 0

def test_assemble_config_sample(self):
        mock_open = mock.mock_open()
        sauna_instance = Sauna()
        with mock.patch('builtins.open', mock_open):
            sauna_instance.assemble_config_sample('/foo')
        mock_open.assert_called_once_with('/foo/sauna-sample.yml', 'w')
        f = mock_open()
        generated_yaml_string = f.write.call_args[0][0]
        # Will raise a yaml error if generated content is not valid yaml
        yaml.safe_load(generated_yaml_string)

def setup_method(self, test_method):
        self.mopen = mock_open()
        self.patcher = patch('services.csv_service.csv', spec=True)
        self.mcsv = self.patcher.start()

def test_result(self):
        video_id = 'avoijfsdfa'
        download = Download()
        m = mock_open(read_data=FIXTURE_WEBVTT)

        with patch('ytcc.download.open', m, create=True):
            with patch('ytcc.storage.Storage.remove_file', Mock()):
                download.get_result = Mock(return_value=0)
                self.assertEqual(FIXTURE_WEBVTT_STRIPPED,
                                 download.get_captions(video_id))

def get_arp_table():
    with open(INPUT_DIR + '/arp.txt') as farp:
        file_content = mock_open(read_data = farp.read())
        file_content.return_value.__iter__ = lambda self : iter(self.readline, '')
        # file_content = MagicMock(name = 'open', spec = open)
        # file_content.return_value = iter(farp.readlines())
        with patch('builtins.open', file_content):
            return _get_arp_table()

# Replace the function with mocked one

def test__load_method_called_with_the_right_path(self):
        dummy_path = '/dummy/path'
        with mock.patch(builtin_module + '.open') as mock_open:
            file_handler._load(dummy_path)
            mock_open.assert_called_with(dummy_path, 'r')

def test__load_method_returns_the_loaded_file_content_as_a_list_of_string(self):
        mock_open = mock.mock_open(read_data='line 1\nline 2')
        expected = ['line 1', 'line 2']
        with mock.patch(builtin_module + '.open', mock_open):
            result = file_handler._load('...')
            self.assertEqual(expected, result)

def test__config_file_is_opened_from_the_right_path(self, mock_path, mock_yaml):
        dummy_config_path = '/config/path'
        mock_path.return_value = dummy_config_path
        with mock.patch(open_mock_string) as mock_open:
            config._load_config()
            mock_open.assert_called_with(dummy_config_path, 'r')

def test__file_not_found__raises_error(self, mock_yaml):
        error_message = 'file not found'
        mock_open = mock.MagicMock(side_effect = IOError(error_message))
        with self.assertRaises(Exception) as cm:
            with mock.patch(open_mock_string, mock_open):
                config._load_config()
        assert_exception(self, cm, IOError, error_message)