本文整理汇总了Python中pulp.common.compat.json.loads函数的典型用法代码示例。如果您正苦于以下问题：Python loads函数的具体用法？Python loads怎么用？Python loads使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了loads函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: extract_metadata
def extract_metadata(filename, temp_dir, module=None):
"""
Pulls the module's metadata file out of the module's tarball and updates the
module instance with its contents. The module instance itself is updated
as part of this call. It is up to the caller to delete the temp_dir after
this executes.
:param filename: full path to the module file
:type filename: str
:param temp_dir: location the module's files should be extracted to;
must exist prior to this call
:type temp_dir: str
:param module: module instance with name, author, version to help find
the directory which contains metadata.json (optional)
:type module: Module
:raise InvalidTarball: if the module file cannot be opened
:raise MissingModuleFile: if the module's metadata file cannot be found
"""
if module is None:
metadata = _extract_non_standard_json(filename, temp_dir)
return json.loads(metadata)
# Attempt to load from the standard metadata file location. If it's not
# found, try the brute force approach. If it's still not found, that call
# will raise the appropriate MissingModuleFile exception.
try:
metadata = _extract_json(module, filename, temp_dir)
return json.loads(metadata)
except MissingModuleFile:
metadata = _extract_non_standard_json(filename, temp_dir)
return json.loads(metadata)
开发者ID:bmbouter，项目名称:pulp_puppet，代码行数:34，代码来源:metadata.py
示例2: test_run
def test_run(self):
# Setup
data = {
OPTION_GROUP_ID.keyword: "test-group",
OPTION_NAME.keyword: "Group",
OPTION_DESCRIPTION.keyword: "Description",
OPTION_NOTES.keyword: {"a": "a", "b": "b"},
}
self.server_mock.request.return_value = 201, {}
# Test
self.command.run(**data)
# Verify
self.assertEqual(1, self.server_mock.request.call_count)
self.assertEqual("POST", self.server_mock.request.call_args[0][0])
body = self.server_mock.request.call_args[0][2]
body = json.loads(body)
self.assertEqual(body["id"], "test-group")
self.assertEqual(body["display_name"], "Group")
self.assertEqual(body["description"], "Description")
self.assertEqual(body["notes"], {"a": "a", "b": "b"})
self.assertEqual(1, len(self.prompt.get_write_tags()))
self.assertEqual(TAG_SUCCESS, self.prompt.get_write_tags()[0])
开发者ID:pcreech，项目名称:pulp，代码行数:27，代码来源:test_commands_repo_group.py
示例3: test_run_no_group
def test_run_no_group(self):
# Setup
data = {OPTION_GROUP_ID.keyword: "test-group"}
self.server_mock.request.return_value = 200, []
# Test
self.command.run(**data)
# Verify
self.assertEqual(1, self.server_mock.request.call_count)
# Group lookup call
call_args = self.server_mock.request.call_args_list[0]
self.assertEqual("POST", call_args[0][0])
url = call_args[0][1]
self.assertTrue(url.endswith("/repo_groups/search/"))
body = json.loads(call_args[0][2])
self.assertEqual(body["criteria"]["filters"]["id"], "test-group")
# Output
self.assertEqual(2, len(self.prompt.get_write_tags()))
self.assertEqual(TAG_TITLE, self.prompt.get_write_tags()[0])
self.assertEqual("not-found", self.prompt.get_write_tags()[1])
开发者ID:pcreech，项目名称:pulp，代码行数:26，代码来源:test_commands_repo_group.py
示例4: test_run_through_cli
def test_run_through_cli(self):
"""
This test isn't complete, but is at minimum being included to verify
903262 (--only-newest being parsed correctly). Ideally this should be
flushed out to test all of the options.
The test above kicks in after okaara has parsed the user input,
simulating what okaara would pass to the command itself. That's fine,
but doesn't pick up on issues in the parsing itself. This test will
simulate the string entered by the user from the command line and
force okaara to do the parsing and hand those results to the command.
"""
# Setup
self.server_mock.request.return_value = 201, {}
# Test
command = repo.RpmRepoCreateCommand(self.context)
self.cli.add_command(command)
self.cli.run("create --repo-id r --only-newest true".split())
# Verify
self.assertEqual(1, self.server_mock.request.call_count)
body = self.server_mock.request.call_args[0][2]
body = json.loads(body)
self.assertEqual(body['id'], 'r')
self.assertEqual(body['importer_config']['newest'], True) # not the string "true"
开发者ID:lzap，项目名称:pulp_rpm，代码行数:29，代码来源:test_extensions_repo.py
示例5: test_unset_queries
def test_unset_queries(self):
# make sure an empty list gets sent as the new value for "queries",
# and definitely not None
data = {
options.OPTION_REPO_ID.keyword : 'test-repo',
options.OPTION_NAME.keyword : 'Test Name',
options.OPTION_DESCRIPTION.keyword : 'Test Description',
options.OPTION_NOTES.keyword : {'a' : 'a'},
cudl.OPTION_FEED.keyword : 'http://localhost',
cudl.OPTION_HTTP.keyword : 'true',
cudl.OPTION_HTTPS.keyword : 'true',
cudl.OPTION_QUERY.keyword : None,
cudl.OPTION_QUERIES_UPDATE.keyword : []
}
self.server_mock.request.return_value = 200, {}
# Test
self.command.run(**data)
body = self.server_mock.request.call_args[0][2]
body = json.loads(body)
expected_config = {
u'feed' : u'http://localhost',
u'queries' : [], # this is the key part of this test
}
self.assertEqual(expected_config, body['importer_config'])
开发者ID:abhaychrungoo，项目名称:pulp_puppet，代码行数:28，代码来源:test_extension_repo.py
示例6: test_pulp_manage_db_loads_types
def test_pulp_manage_db_loads_types(self, listdir_mock):
"""
Test calling pulp-manage-db imports types on a clean types database.
"""
manage.main()
all_collection_names = types_db.all_type_collection_names()
self.assertEqual(len(all_collection_names), 1)
self.assertEqual(['units_test_type_id'], all_collection_names)
# Let's make sure we loaded the type definitions correctly
db_type_definitions = types_db.all_type_definitions()
self.assertEquals(len(db_type_definitions), 1)
test_json = json.loads(_test_type_json)
for attribute in ['id', 'display_name', 'description', 'unit_key', 'search_indexes']:
self.assertEquals(test_json['types'][0][attribute], db_type_definitions[0][attribute])
# Now let's ensure that we have the correct indexes
collection = types_db.type_units_collection('test_type_id')
indexes = collection.index_information()
self.assertEqual(indexes['_id_']['key'], [(u'_id', 1)])
# Make sure we have the unique constraint on all three attributes
self.assertEqual(indexes['attribute_1_1_attribute_2_1_attribute_3_1']['unique'], True)
self.assertEqual(indexes['attribute_1_1_attribute_2_1_attribute_3_1']['dropDups'], False)
self.assertEqual(indexes['attribute_1_1_attribute_2_1_attribute_3_1']['key'],
[(u'attribute_1', 1), (u'attribute_2', 1), (u'attribute_3', 1)])
# Make sure we indexes attributes 1 and 3
self.assertEqual(indexes['attribute_1_1']['dropDups'], False)
self.assertEqual(indexes['attribute_1_1']['key'], [(u'attribute_1', 1)])
self.assertEqual(indexes['attribute_3_1']['dropDups'], False)
self.assertEqual(indexes['attribute_3_1']['key'], [(u'attribute_3', 1)])
# Make sure we only have the indexes that we've hand inspected here
self.assertEqual(indexes.keys(), [u'_id_', u'attribute_1_1_attribute_2_1_attribute_3_1',
u'attribute_1_1', u'attribute_3_1'])
开发者ID:ryanschneider，项目名称:pulp，代码行数:35，代码来源:test_pulp_manage_db.py
示例7: test_run
def test_run(self):
# Setup
repo_id = 'test-repo'
data = {
OPTION_REPO_ID.keyword : repo_id,
OPTION_NAME.keyword : 'Test Repository',
OPTION_DESCRIPTION.keyword : 'Repository Description',
OPTION_NOTES.keyword : {'a' : 'a', 'b' : 'b'},
}
self.server_mock.request.return_value = 200, {}
# Test
self.command.run(**data)
# Verify
self.assertEqual(1, self.server_mock.request.call_count)
self.assertEqual('PUT', self.server_mock.request.call_args[0][0])
url = self.server_mock.request.call_args[0][1]
self.assertTrue(url.endswith('/repositories/%s/' % repo_id))
body = self.server_mock.request.call_args[0][2]
body = json.loads(body)
self.assertEqual(body['delta']['display_name'], 'Test Repository')
self.assertEqual(body['delta']['description'], 'Repository Description')
self.assertEqual(body['delta']['notes'], {'a' : 'a', 'b' : 'b'})
self.assertEqual(1, len(self.prompt.get_write_tags()))
self.assertEqual(TAG_SUCCESS, self.prompt.get_write_tags()[0])
开发者ID:ashcrow，项目名称:pulp，代码行数:30，代码来源:test_commands_repo_cudl.py
示例8: test_run_no_group
def test_run_no_group(self):
# Setup
data = {
OPTION_GROUP_ID.keyword : 'test-group',
}
self.server_mock.request.return_value = 200, []
# Test
self.command.run(**data)
# Verify
self.assertEqual(1, self.server_mock.request.call_count)
# Group lookup call
call_args = self.server_mock.request.call_args_list[0]
self.assertEqual('POST', call_args[0][0])
url = call_args[0][1]
self.assertTrue(url.endswith('/repo_groups/search/'))
body = json.loads(call_args[0][2])
self.assertEqual(body['criteria']['filters']['id'], 'test-group')
# Output
self.assertEqual(2, len(self.prompt.get_write_tags()))
self.assertEqual(TAG_TITLE, self.prompt.get_write_tags()[0])
self.assertEqual('not-found', self.prompt.get_write_tags()[1])
开发者ID:AndreaGiardini，项目名称:pulp，代码行数:28，代码来源:test_commands_repo_group.py
示例9: test_queries_overrides_query
def test_queries_overrides_query(self):
# make sure --queries overrides --query, which is deprecated
data = {
options.OPTION_REPO_ID.keyword : 'test-repo',
options.OPTION_NAME.keyword : 'Test Name',
options.OPTION_DESCRIPTION.keyword : 'Test Description',
options.OPTION_NOTES.keyword : {'a' : 'a'},
cudl.OPTION_FEED.keyword : 'http://localhost',
cudl.OPTION_HTTP.keyword : 'true',
cudl.OPTION_HTTPS.keyword : 'true',
cudl.OPTION_QUERY.keyword : ['q1', 'q2'],
cudl.OPTION_QUERIES.keyword : ['x', 'y']
}
self.server_mock.request.return_value = 200, {}
# Test
self.command.run(**data)
body = self.server_mock.request.call_args[0][2]
body = json.loads(body)
expected_config = {
u'feed' : u'http://localhost',
u'queries' : [u'x', u'y'],
}
self.assertEqual(expected_config, body['importer_config'])
开发者ID:abhaychrungoo，项目名称:pulp_puppet，代码行数:27，代码来源:test_extension_repo.py
示例10: test_run
def test_run(self):
# Setup
data = {
'from-repo-id': 'from',
'to-repo-id': 'to'
}
self.server_mock.request.return_value = 202, self.task()
mock_poll = mock.MagicMock().poll
self.command.poll = mock_poll
# Test
self.command.run(**data)
# Verify
call_args = self.server_mock.request.call_args[0]
self.assertEqual('POST', call_args[0])
self.assertTrue(call_args[1].endswith('/to/actions/associate/'))
body = json.loads(call_args[2])
self.assertEqual(body['source_repo_id'], 'from')
self.assertEqual(body['criteria']['type_ids'], [constants.TYPE_PUPPET_MODULE])
self.assertEqual(1, mock_poll.call_count)
开发者ID:asmacdo，项目名称:pulp_puppet，代码行数:25，代码来源:test_copy_modules.py
示例11: test_run
def test_run(self):
# Setup
data = {
OPTION_GROUP_ID.keyword : 'test-group',
OPTION_NAME.keyword : 'Group',
OPTION_DESCRIPTION.keyword : 'Description',
OPTION_NOTES.keyword : {'a' : 'a', 'b' : 'b'},
}
self.server_mock.request.return_value = 200, {}
# Test
self.command.run(**data)
# Verify
self.assertEqual(1, self.server_mock.request.call_count)
self.assertEqual('PUT', self.server_mock.request.call_args[0][0])
url = self.server_mock.request.call_args[0][1]
self.assertTrue(url.endswith('/repo_groups/test-group/'))
body = self.server_mock.request.call_args[0][2]
delta = json.loads(body)
self.assertTrue('display-name' not in delta)
self.assertEqual(delta['display_name'], 'Group')
self.assertEqual(delta['description'], 'Description')
self.assertEqual(delta['notes'], {'a' : 'a', 'b' : 'b'})
开发者ID:AndreaGiardini，项目名称:pulp，代码行数:27，代码来源:test_commands_repo_group.py
示例12: test_run
def test_run(self):
# Setup
data = {
OPTION_REPO_ID.keyword: "test-repo",
OPTION_NAME.keyword: "Test Repository",
OPTION_DESCRIPTION.keyword: "Repository Description",
OPTION_NOTES.keyword: ["a=a", "b=b"],
}
self.command.default_notes = {"foo": "bar"}
self.server_mock.request.return_value = 201, {}
# Test
self.command.run(**data)
# Verify
self.assertEqual(1, self.server_mock.request.call_count)
self.assertEqual("POST", self.server_mock.request.call_args[0][0])
body = self.server_mock.request.call_args[0][2]
body = json.loads(body)
self.assertEqual(body["id"], "test-repo")
self.assertEqual(body["display_name"], "Test Repository")
self.assertEqual(body["description"], "Repository Description")
self.assertEqual(body["notes"], {"a": "a", "b": "b", "foo": "bar"})
self.assertEqual(body["distributors"], [])
self.assertEqual(body["importer_type_id"], None)
self.assertEqual(1, len(self.prompt.get_write_tags()))
self.assertEqual(TAG_SUCCESS, self.prompt.get_write_tags()[0])
开发者ID:maxamillion，项目名称:pulp，代码行数:30，代码来源:test_cudl.py
示例13: _do_request
def _do_request(self, request_type, uri, params, additional_headers, serialize_json=True):
"""
Override the base class controller to allow for less deterministic
responses due to integration with the dispatch package.
"""
# Use the default headers established at setup and override/add any
headers = dict(PulpWebserviceTests.HEADERS)
if additional_headers is not None:
headers.update(additional_headers)
# Serialize the parameters if any are specified
if params is None:
params = {}
if serialize_json:
params = json.dumps(params)
# Invoke the API
f = getattr(PulpWebserviceTests.TEST_APP, request_type)
response = f('http://localhost' + uri, params=params, headers=headers, expect_errors=True)
# Collect return information and deserialize it
status = response.status
try:
body = json.loads(response.body)
except ValueError:
body = None
return status, body
开发者ID:AndreaGiardini，项目名称:pulp，代码行数:30，代码来源:base.py
示例14: test_to_json
def test_to_json(self):
# Setup
metadata = RepositoryMetadata()
metadata.update_from_json(VALID_REPO_METADATA_JSON)
# Test
serialized = metadata.to_json()
# Verify
parsed = json.loads(serialized)
self.assertEqual(2, len(parsed))
sorted_modules = sorted(parsed, key=lambda x: x["name"])
self.assertEqual(4, len(sorted_modules[0]))
self.assertEqual(sorted_modules[0]["name"], "common")
self.assertEqual(sorted_modules[0]["author"], "lab42")
self.assertEqual(sorted_modules[0]["version"], "0.0.1")
self.assertEqual(sorted_modules[0]["tag_list"], [])
self.assertEqual(4, len(sorted_modules[1]))
self.assertEqual(sorted_modules[1]["name"], "postfix")
self.assertEqual(sorted_modules[1]["author"], "lab42")
self.assertEqual(sorted_modules[1]["version"], "0.0.2")
self.assertEqual(sorted_modules[1]["tag_list"], ["postfix", "applications"])
开发者ID:aeria，项目名称:pulp_puppet，代码行数:26，代码来源:test_models.py
示例15: _interpret_operation_report
def _interpret_operation_report(output, operation, full_name):
"""
Makes a best effort to locate and deserialize the JSON output from the
"puppet module" tool. The tool does not document exactly how this will
be included in the output, so this method returns an empty dictionary if
a JSON-serialized report is not found. It also logs a warning in that
case.
:param output: text output from the "puppet module" tool, which
presumably includes some JSON
:type output: str
:param operation: one of "install", "upgrade", "uninstall", used only
for logging.
:type operation: str
:param full_name: full name in form "author/title", used only for
logging
:return: deserialized JSON output from the "puppet module" tool
:rtype: dict
"""
try:
potential_json = output.split('\n')[-2]
operation_report = json.loads(potential_json)
# if there was any error trying to parse puppet's JSON output, make
# an empty report
except (IndexError, ValueError):
logger.warning('failed to parse JSON output from %s of %s' % (operation, full_name))
operation_report = {}
return operation_report
开发者ID:abhaychrungoo，项目名称:pulp_puppet，代码行数:29，代码来源:puppet.py
示例16: test_migrate
def test_migrate(self, start_logging_mock, listdir_mock, mock_plugin_definitions,
mock_drop_indices):
"""
Ensure that migrate() imports types on a clean types database.
"""
migration.migrate()
self.assertTrue(mock_drop_indices.called)
all_collection_names = types_db.all_type_collection_names()
self.assertEqual(len(all_collection_names), 1)
self.assertEqual(['units_test_type_id'], all_collection_names)
# Let's make sure we loaded the type definitions correctly
db_type_definitions = types_db.all_type_definitions()
self.assertEquals(len(db_type_definitions), 1)
test_json = json.loads(_test_type_json)
for attribute in ['id', 'display_name', 'description', 'unit_key', 'search_indexes']:
self.assertEquals(test_json['types'][0][attribute], db_type_definitions[0][attribute])
# Now let's ensure that we have the correct indexes
collection = types_db.type_units_collection('test_type_id')
indexes = collection.index_information()
self.assertEqual(indexes['_id_']['key'], [(u'_id', 1)])
# Make sure we have the unique constraint on all three attributes
self.assertEqual(indexes['attribute_1_1_attribute_2_1_attribute_3_1']['unique'], True)
self.assertEqual(indexes['attribute_1_1_attribute_2_1_attribute_3_1']['key'],
[(u'attribute_1', 1), (u'attribute_2', 1), (u'attribute_3', 1)])
# Make sure we indexed attributes 1 and 3
self.assertEqual(indexes['attribute_1_1']['key'], [(u'attribute_1', 1)])
self.assertEqual(indexes['attribute_3_1']['key'], [(u'attribute_3', 1)])
# Make sure we only have the indexes that we've hand inspected here
self.assertEqual(indexes.keys(), [u'_id_', u'attribute_1_1_attribute_2_1_attribute_3_1',
u'attribute_1_1', u'attribute_3_1'])
开发者ID:alexxa，项目名称:pulp，代码行数:34，代码来源:test_0015_load_content_types.py
示例17: test_to_json
def test_to_json(self):
# Setup
metadata = RepositoryMetadata()
metadata.update_from_json(VALID_REPO_METADATA_JSON)
# Test
serialized = metadata.to_json()
# Verify
parsed = json.loads(serialized)
self.assertEqual(2, len(parsed))
sorted_modules = sorted(parsed, key=lambda x : x['name'])
self.assertEqual(4, len(sorted_modules[0]))
self.assertEqual(sorted_modules[0]['name'], 'common')
self.assertEqual(sorted_modules[0]['author'], 'lab42')
self.assertEqual(sorted_modules[0]['version'], '0.0.1')
self.assertEqual(sorted_modules[0]['tag_list'], [])
self.assertEqual(4, len(sorted_modules[1]))
self.assertEqual(sorted_modules[1]['name'], 'postfix')
self.assertEqual(sorted_modules[1]['author'], 'lab42')
self.assertEqual(sorted_modules[1]['version'], '0.0.2')
self.assertEqual(sorted_modules[1]['tag_list'], ['postfix', 'applications'])
开发者ID:christoe，项目名称:pulp_puppet，代码行数:26，代码来源:test_common_model.py
示例18: load_plugin_config
def load_plugin_config(config_file_name):
"""
@type config_file_name: str
@rtype: dict
"""
_LOG.debug('Loading config file: %s' % config_file_name)
contents = read_content(config_file_name)
cfg = json.loads(contents)
return cfg
开发者ID:jessegonzalez，项目名称:pulp，代码行数:9，代码来源:loading.py
示例19: test_from_dict
def test_from_dict(self):
# Setup
data = json.loads(VALID_MODULE_METADATA_JSON)
# Test
module = Module.from_dict(data)
# Verify
self.assert_valid_module(module)
开发者ID:christoe，项目名称:pulp_puppet，代码行数:9，代码来源:test_common_model.py
示例20: test_adds_repo_to_search
def test_adds_repo_to_search(self):
data = {OPTION_GROUP_ID.keyword: "test-group", FLAG_ALL.keyword: False, OPTION_REPO_ID.keyword: ["repo1"]}
self.server_mock.request.return_value = 200, {}
self.command.run(**data)
criteria = json.loads(self.server_mock.request.call_args[0][2])["criteria"]
self.assertEqual(criteria["filters"]["id"]["$in"], ["repo1"])
开发者ID:pcreech，项目名称:pulp，代码行数:9，代码来源:test_commands_repo_group.py
注：本文中的pulp.common.compat.json.loads函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论