Python json.loads函数代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了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；未经允许，请勿转载。