def mark_download_url()

in src/buildstream/source.py [0:0]

• 41 lines of code
• 14 McCabe index (conditional complexity)

    def mark_download_url(self, url: str, *, primary: bool = True) -> None:
        """Identifies the URL that this Source uses to download

        Args:
           url (str): The URL used to download
           primary (bool): Whether this is the primary URL for the source

        .. note::

           This must be called for every URL in the configuration during
           :func:`Plugin.configure() <buildstream.plugin.Plugin.configure>` if
           :func:`Source.translate_url() <buildstream.source.Source.translate_url>`
           is not called.
        """
        # Only mark the Source level aliases on the main instance, not in
        # a reinstantiated instance in mirroring.
        if not self.__alias_override:
            if primary:
                expected_alias = _extract_alias(url)

                assert (
                    self.__expected_alias is None or self.__expected_alias == expected_alias
                ), "Attempt to mark primary URL with {}, already marked with {}".format(
                    expected_alias, self.__expected_alias
                )

                self.__expected_alias = expected_alias

        # Enforce proper behaviour of plugins by ensuring that all
        # aliased URLs have been marked at Plugin.configure() time.
        #
        if self._get_configuring():
            # Record marked urls while configuring
            #
            self.__marked_urls.add(url)
        else:
            # If an unknown aliased URL is seen after configuring,
            # this is an error.
            #
            # It is still possible that a URL that was not mentioned
            # in the element configuration can be marked, this is
            # the case for git submodules which might be automatically
            # discovered.
            #
            assert url in self.__marked_urls or not _extract_alias(
                url
            ), "URL was not seen at configure time: {}".format(url)

        alias = _extract_alias(url)

        # Issue a (fatal-able) warning if the source used a URL without specifying an alias
        if not alias:
            self.warn(
                "{}: Use of unaliased source download URL: {}".format(self, url),
                warning_token=CoreWarnings.UNALIASED_URL,
            )

        # If there is an alias in use, ensure that it exists in the project
        if alias:
            project = self._get_project()
            if not project.alias_exists(alias, first_pass=self.__first_pass, source=self):
                raise SourceError(
                    "{}: Invalid alias '{}' specified in URL: {}".format(self, alias, url),
                    reason="invalid-source-alias",
                )
            if not project.get_alias_uris(alias, first_pass=self.__first_pass, tracking=False):
                raise SourceError(
                    "{}: No fetch URI found for alias '{}'".format(self, alias),
                    detail="Check fetch controls in your user configuration",
                    reason="missing-source-alias-target",
                )
            if not project.get_alias_uris(alias, first_pass=self.__first_pass, tracking=True):
                raise SourceError(
                    "{}: No tracking URI found for alias '{}'".format(self, alias),
                    detail="Check track controls in your user configuration",
                    reason="missing-source-alias-target",
                )