我们从Python开源项目中，提取了以下49个代码示例，用于说明如何使用unittest.mock.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_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)
def test__config_is_written_to_the_right_place(self, mock_yaml, mock_path): dummy_path = '/config/path' mock_path.return_value = dummy_path with mock.patch(open_mock_string) as mock_open: config._create_default_config() mock_open.assert_called_with(dummy_path, 'w+')
def test__yaml_file_is_written_with_the_full_configuration(self, mock_yaml): mock_open = mock.mock_open() with mock.patch(open_mock_string, mock_open): config._create_default_config() mock_yaml.dump.assert_called_with(config._default_config, mock_open.return_value, default_flow_style=False)
def test_prompts_update_needed(): """Test Config object will suggest update if needed.""" tmp_config_data = json.loads(CONFIG_DATA) # set an old timestamp tmp_config_data['last_update_check'] = 0.0 mock_config = mock.mock_open(read_data=json.dumps(tmp_config_data)) with mock.patch('two1.commands.util.config.Config.save', return_value=None): with mock.patch('two1.commands.util.config.open', mock_config, create=True): with capture_stderr( config.Config, 'config_file', check_update=True) as output: output = output.strip() assert(len(output) > 0) assert(output in uxstring.UxString.update_required)
def test_needs_update_legacy_last_update_check(): """Test for a legacy two1.json with an older last_update_check, and that it does not throw an error""" 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') c.set('last_update_check', "", should_save=True) try: c.check_update() except ValueError: pytest.fail("Error dealing with legacy timestamp")
def test_last_update_check_set(): """Asert last_update_check is set in a config with none.""" mock_config = mock.mock_open(read_data=CONFIG_DATA) assert 'last_update_check' not in CONFIG_DATA with mock.patch('two1.commands.util.config.open', mock_config, create=True): conf = config.Config('config_file', check_update=True) # last_update_check should now be set after # initalizing the config object. assert hasattr(conf, 'last_update_check') assert 'last_update_check' in json.loads( mock_config.return_value.write.call_args[0][0])
def test_needs_old_last_update_check_with_new_version(): """Test for a last_update_check more than 3600 seconds ago, but version is new and that it does not suggest an update""" 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', check_update=True) c.set('last_update_check', 000000.000, should_save=True) with capture_stdout(config.Config, 'config_file', check_update=True) as output: assert len(output.strip()) == 0
def test_invalid_config_file(): """Test that an invalid `two1.json` file cannot be imported.""" mock_config = mock.mock_open(mock=mock.Mock(side_effect=ValueError)) with mock.patch('two1.commands.util.config.open', mock_config, create=True), pytest.raises(exceptions.FileDecodeError): # nopep8 config.Config('config_file') mock_config.assert_called_with('config_file', mode='r')
def setup_method(self, method): singletons.settings.use_settings(MockConfig) mocked_file = mock.mock_open(read_data=self.JSON_DATA) def mocked_exists(path): return path == self.JSON_PATH self.open_patch = mock.patch('omnic.web.viewer.open', mocked_file) self.exists_patch = mock.patch('os.path.exists', mocked_exists) self.open_patch.start() self.exists_patch.start()
def test_should_load_data_from_file(self, pickle_mock): with patch('builtins.open', mock_open()) as file_mock: load('dir/filename/') file_mock.assert_called_with('dir/filename/', 'rb') pickle_mock.load.assert_called_with(file_mock())
def test_should_save_data_to_file_using_highest_protocol(self, pickle_mock): data = 'data' with patch('builtins.open', mock_open(read_data=data)) as file_mock: save('dir/filename/', data) file_mock.assert_called_with('dir/filename/', 'wb') pickle_mock.dump.assert_called_with(data, file_mock(), pickle.HIGHEST_PROTOCOL)
def test_should_save_data_to_file_using_default_protocol(self, pickle_mock): data = 'data' with patch('builtins.open', mock_open(read_data=data)) as file_mock: save('dir/filename/', data, pickle.DEFAULT_PROTOCOL) file_mock.assert_called_with('dir/filename/', 'wb') pickle_mock.dump.assert_called_with(data, file_mock(), pickle.DEFAULT_PROTOCOL)
def test_get_config(self): mock_open = mock.mock_open(read_data=self.config) with mock.patch('fleece.cli.run.run.open', mock_open, create=True): config = run.get_config('./wat') self.assertDictEqual(yaml.load(self.config), config)
def test_config_attribute_returns_list_of_given_file(baseline): """ Config attribute should return given config file as a list. """ config = baseline.split('\n') data = mock.mock_open(read_data=baseline) with mock.patch('diffios.config.open', data, create=True) as mock_open: assert config == diffios.Config(data).config
def test_config_is_grouped_correctly_with_file(baseline, baseline_blocks): """ Should return valid config as list of hierarchical blocks, from a config given in a file. """ with mock.patch('diffios.config.os.path.isfile') as mock_isfile: mock_isfile.return_value = True config_data = mock.mock_open(read_data=baseline) with mock.patch( 'diffios.config.open', config_data, create=True) as mock_open: actual = diffios.Config( 'baseline.conf', ignore_lines=[]).included() mock_open.assert_called_once_with('baseline.conf') assert baseline_blocks == actual
def test_ignore_lines_from_file(ignores_file): """ Should return the lines in the given ignores file as list of lowercase strings. """ config = ['hostname ROUTER'] expected = ignores_file.lower().split('\n') ignores_data = mock.mock_open(read_data=ignores_file) with mock.patch('diffios.config.os.path.isfile') as mock_isfile: mock_isfile.return_value = True with mock.patch( 'diffios.config.open', ignores_data, create=True) as mock_open: actual = diffios.Config( config, ignore_lines='ignores_file').ignore_lines mock_open.assert_called_once_with('ignores_file') assert expected == actual
def test_returns_main_paths_if_cra_is_not_running(self): open_mock = mock_open( read_data='''{ "main.js": "static/js/main.1234.js", "main.css": "static/css/main.1234.css" }''' ) with patch('builtins.open', open_mock): self.assertEqual(asset_manifest.generate_manifest(False, server_url, '.'), { 'main_js': 'js/main.1234.js', 'main_css': 'css/main.1234.css', })
def test_load_secret(): with patch('ssb.util.open', mock_open(read_data=CONFIG_FILE), create=True): secret = load_ssb_secret() priv_key = b'\xfd\xba\x83\x04\x8f\xef\x18\xb0\xf9\xab-\xc6\xc4\xcb \x1cX\x18"\xba\xd8\xd3\xc2_O5\x1a\t\x84\xfa\xc7A' assert secret['id'] == '@rsYpBIcXsxjQAf0JNes+MHqT2DL+EfopWKAp4rGeEPQ=.ed25519' assert bytes(secret['keypair']) == priv_key assert bytes(secret['keypair'].verify_key) == b64decode('rsYpBIcXsxjQAf0JNes+MHqT2DL+EfopWKAp4rGeEPQ=')
def test_load_exception(): with pytest.raises(ConfigException): with patch('ssb.util.open', mock_open(read_data=CONFIG_FILE_INVALID), create=True): load_ssb_secret()
def test_client_script_put(self): """ Verify use cases for the client_script put requests. """ sys.argv = [''] read_data = six.b('1234567890') with mock.patch('requests.Session.put') as _put, \ mock.patch('os.path.realpath') as _realpath, \ mock.patch( 'argparse.FileType.__call__', mock.mock_open(read_data=read_data), create=True) as _filetype: _realpath.return_value = self.conf for cmd in ( ['cluster', 'create'], ['cluster', 'create', '-n', 'default'], ['cluster', 'create', '-t', 'kubernetes'], ['cluster', 'create', '-t', 'kubernetes', '-n', 'default'], ['cluster', 'deploy', 'start'], ['cluster', 'restart', 'start'], ['cluster', 'upgrade', 'start'], ['container_manager', 'create'], ['container_manager', 'create', '-o', '{}'], ['container_manager', 'create'], ['host', 'create', '-c', 'honeynut', '1.2.3.4'], ['host', 'join', '1.2.3.4'], ['network', 'create', '-n', 'test']): mock_return = requests.Response() mock_return._content = six.b('{}') mock_return.status_code = 201 mock_return.request = mock.MagicMock(path_url='/fake/path') _put.return_value = mock_return sys.argv[1:] = cmd + ['test'] if cmd[1] == 'deploy': sys.argv.append('1') # arbitrary version print(sys.argv) client_script.main() self.assertEquals(1, _put.call_count) _put.reset_mock()
