def _parse_plugin()

in azurelinuxagent/common/protocol/extensions_goal_state_from_extensions_config.py [0:0]

• 26 lines of code
• 13 McCabe index (conditional complexity)

    def _parse_plugin(extension, plugin):
        """
        Sample config:

        <Plugins>
          <Plugin name="Microsoft.CPlat.Core.NullSeqB" version="2.0.1" location="https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/f72653efd9e349ed9842c8b99e4c1712/Microsoft.CPlat.Core_NullSeqB_useast2euap_manifest.xml" state="enabled" autoUpgrade="false" failoverlocation="https://zrdfepirv2cbz06prdstr01a.blob.core.windows.net/f72653efd9e349ed9842c8b99e4c1712/Microsoft.CPlat.Core_NullSeqB_useast2euap_manifest.xml" runAsStartupTask="false" isJson="true" useExactVersion="true">
          <Plugin name="Microsoft.Azure.Extensions.CustomScript" version="1.0" location="https://rdfecurrentuswestcache.blob.core.test-cint.azure-test.net/0e53c53ef0be4178bacb0a1fecf12a74/Microsoft.Azure.Extensions_CustomScript_usstagesc_manifest.xml" state="enabled" autoUpgrade="false" failoverlocation="https://rdfecurrentuswestcache2.blob.core.test-cint.azure-test.net/0e53c53ef0be4178bacb0a1fecf12a74/Microsoft.Azure.Extensions_CustomScript_usstagesc_manifest.xml" runAsStartupTask="false" isJson="true" useExactVersion="true">
            <additionalLocations>
              <additionalLocation>https://rdfecurrentuswestcache3.blob.core.test-cint.azure-test.net/0e53c53ef0be4178bacb0a1fecf12a74/Microsoft.Azure.Extensions_CustomScript_usstagesc_manifest.xml</additionalLocation>
              <additionalLocation>https://rdfecurrentuswestcache4.blob.core.test-cint.azure-test.net/0e53c53ef0be4178bacb0a1fecf12a74/Microsoft.Azure.Extensions_CustomScript_usstagesc_manifest.xml</additionalLocation>
            </additionalLocations>
          </Plugin>
          <Plugin name="Microsoft.CPlat.Core.NullSeqA" version="2.0.1" location="https://zrdfepirv2cbn04prdstr01a.blob.core.windows.net/f72653efd9e349ed9842c8b99e4c1712/Microsoft.CPlat.Core_NullSeqA_useast2euap_manifest.xml" state="enabled" autoUpgrade="false" failoverlocation="https://zrdfepirv2cbn06prdstr01a.blob.core.windows.net/f72653efd9e349ed9842c8b99e4c1712/Microsoft.CPlat.Core_NullSeqA_useast2euap_manifest.xml" runAsStartupTask="false" isJson="true" useExactVersion="true" encodedSignature="MIIn..." />
        </Plugins>


        Note that the `additionalLocations` subnode is populated with links
        generated by PIR for resiliency. In regions with this feature enabled,
        CRP will provide any extra links in the format above. If no extra links
        are provided, the subnode will not exist.
        """

        def _log_error_if_none(attr_name, value):
            # Plugin Name and Version are very essential fields, without them we wont be able to even report back to CRP
            # about that handler. For those cases we need to fail the GoalState completely but currently we dont support
            # reporting status at a GoalState level (we only report at a handler level).
            # Once that functionality is added to the GA, we would raise here rather than just report error in our logs.
            if value in (None, ""):
                add_event(op=WALAEventOperation.InvalidExtensionConfig,
                          message="{0} is None for ExtensionConfig, logging error".format(attr_name),
                          log_event=True, is_success=False)
            return value

        extension.name = _log_error_if_none("Extensions.Plugins.Plugin.name", getattrib(plugin, "name"))
        extension.version = _log_error_if_none("Extensions.Plugins.Plugin.version",
                                                            getattrib(plugin, "version"))
        extension.state = getattrib(plugin, "state")
        if extension.state in (None, ""):
            raise ExtensionsConfigError("Received empty Extensions.Plugins.Plugin.state, failing Handler")

        # extension.encoded_signature value should be None if the property does not exist for the plugin. getattrib
        # returns "" if an attribute does not exist in a node, so use hasattrib here to check if the attribute exists
        extension.encoded_signature = getattrib(plugin, "encodedSignature") if hasattrib(plugin, "encodedSignature") else None

        def getattrib_wrapped_in_list(node, attr_name):
            attr = getattrib(node, attr_name)
            return [attr] if attr not in (None, "") else []

        location = getattrib_wrapped_in_list(plugin, "location")
        failover_location = getattrib_wrapped_in_list(plugin, "failoverlocation")

        locations = location + failover_location

        additional_location_node = find(plugin, "additionalLocations")
        if additional_location_node is not None:
            nodes_list = findall(additional_location_node, "additionalLocation")
            locations += [gettext(node) for node in nodes_list]

        for uri in locations:
            extension.manifest_uris.append(uri)