path: root/src/buildstream/_elementsources.py

#
#  Copyright (C) 2016-2018 Codethink Limited
#  Copyright (C) 2017-2020 Bloomberg Finance LP
#
#  This program is free software; you can redistribute it and/or
#  modify it under the terms of the GNU Lesser General Public
#  License as published by the Free Software Foundation; either
#  version 2 of the License, or (at your option) any later version.
#
#  This library is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
#  Lesser General Public License for more details.
#
#  You should have received a copy of the GNU Lesser General Public
#  License along with this library. If not, see <http://www.gnu.org/licenses/>.

from typing import TYPE_CHECKING, Iterator

from ._context import Context

from .storage._casbaseddirectory import CasBasedDirectory

if TYPE_CHECKING:
    from typing import List

    from .source import Source

# An ElementSources object represents the combined sources of an element.
class ElementSources:
    def __init__(self, context: Context):

        self._context = context
        self._sources = []  # type: List[Source]
        self.vdir = None  # Directory with staged sources
        self._sourcecache = context.sourcecache  # Source cache
        self._is_resolved = False  # Whether the source is fully resolved or not
        self._cached = None  # If the sources are known to be successfully cached in CAS

        # the index of the last source in this element that requires previous
        # sources for staging
        self._last_source_requires_previous_idx = None

    # add_source():
    #
    # Append source to this list of element sources.
    #
    # Args:
    #   source (Source): The source to add
    #
    def add_source(self, source):
        self._sources.append(source)

    # sources():
    #
    # A generator function to enumerate the element sources
    #
    # Yields:
    #   Source: The individual sources
    #
    def sources(self) -> Iterator["Source"]:
        for source in self._sources:
            yield source

    # track():
    #
    # Calls track() on the Element sources
    #
    # Raises:
    #    SourceError: If one of the element sources has an error
    #
    # Returns:
    #    (list): A list of Source object ids and their new references
    #
    def track(self, workspace):
        refs = []
        for index, source in enumerate(self._sources):
            old_ref = source.get_ref()
            new_ref = source._track(self._sources[0:index])
            refs.append((source._unique_id, new_ref))

            # Complimentary warning that the new ref will be unused.
            if old_ref != new_ref and workspace:
                detail = (
                    "This source has an open workspace.\n"
                    + "To start using the new reference, please close the existing workspace."
                )
                source.warn("Updated reference will be ignored as source has open workspace", detail=detail)

        return refs

    # stage():
    #
    # Stage the element sources to a directory
    #
    # Returns:
    #     (:class:`.storage.Directory`): A virtual directory object to stage sources into.
    #
    def stage(self):
        # Assert sources are cached
        assert self.cached()

        self.vdir = CasBasedDirectory(self._context.get_cascache())

        if self._sources:
            # find last required source
            last_required_previous_idx = self._last_source_requires_previous()

            for source in self._sources[last_required_previous_idx:]:
                source_dir = self._sourcecache.export(source)
                self.vdir.import_files(source_dir)

        return self.vdir

    # fetch_done()
    #
    # Indicates that fetching the sources for this element has been done.
    #
    # Args:
    #   fetched_original (bool): Whether the original sources had been asked (and fetched) or not
    #
    def fetch_done(self, fetched_original):
        self._cached = True

        for source in self._sources:
            source._fetch_done(fetched_original)

    # push()
    #
    # Push the element's sources.
    #
    # Returns:
    #   (bool): True if the remote was updated, False if it already existed
    #           and no updated was required
    #
    def push(self):
        pushed = False

        for source in self.sources():
            if self._sourcecache.push(source):
                pushed = True

        return pushed

    # init_workspace():
    #
    # Initialises a new workspace from the element sources.
    #
    # Args:
    #   directory (str): Path of the workspace to init
    #
    def init_workspace(self, directory: str):
        for source in self.sources():
            source._init_workspace(directory)

    # fetch():
    #
    # Fetch the element sources.
    #
    # Raises:
    #    SourceError: If one of the element sources has an error
    #
    def fetch(self, fetch_original=False):
        previous_sources = []
        fetch_needed = False

        if self._sources and not fetch_original:
            for source in self._sources:
                if self._sourcecache.contains(source):
                    continue

                # try and fetch from source cache
                if not source._is_cached() and self._sourcecache.has_fetch_remotes():
                    if self._sourcecache.pull(source):
                        continue

                fetch_needed = True

        # We need to fetch original sources
        if fetch_needed or fetch_original:
            for source in self.sources():
                if not source._is_cached():
                    source._fetch(previous_sources)
                previous_sources.append(source)

            self._cache_sources()

    # get_unique_key():
    #
    # Return something which uniquely identifies the combined sources of the
    # element.
    #
    # Returns:
    #    (str, list, dict): A string, list or dictionary as unique identifier
    #
    def get_unique_key(self):
        result = []

        for source in self._sources:
            result.append({"key": source._get_unique_key(), "name": source._get_source_name()})

        return result

    # cached():
    #
    # Check if the element sources are cached in CAS, generating the source
    # cache keys if needed.
    #
    # Returns:
    #    (bool): True if the element sources are cached
    #
    def cached(self):
        if self._cached is not None:
            return self._cached

        sourcecache = self._sourcecache

        # Go through sources we'll cache generating keys
        for ix, source in enumerate(self._sources):
            if not source._key:
                if source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
                    source._generate_key(self._sources[:ix])
                else:
                    source._generate_key([])

        # Check all sources are in source cache
        for source in self._sources:
            if not sourcecache.contains(source):
                return False

        self._cached = True
        return True

    # is_resolved():
    #
    # Get whether all sources of the element are resolved
    #
    # Returns:
    #    (bool): True if all element sources are resolved
    #
    def is_resolved(self):
        return self._is_resolved

    # cached_original():
    #
    # Get whether all the sources of the element have their own cached
    # copy of their sources.
    #
    # Returns:
    #    (bool): True if all element sources have the original sources cached
    #
    def cached_original(self):
        return all(source._is_cached() for source in self._sources)

    # update_resolved_state():
    #
    # Updates source's resolved state
    #
    # An element's source state must be resolved before it may compute
    # cache keys, because the source's ref, whether defined in yaml or
    # from the workspace, is a component of the element's cache keys.
    #
    def update_resolved_state(self):
        for source in self._sources:
            if not source.is_resolved():
                break
        else:
            self._is_resolved = True

    # preflight():
    #
    # A internal wrapper for calling the abstract preflight() method on
    # the element and its sources.
    #
    def preflight(self):
        # Ensure that the first source does not need access to previous sources
        if self._sources and self._sources[0]._requires_previous_sources():
            from .element import ElementError  # pylint: disable=cyclic-import

            raise ElementError(
                "{}: {} cannot be the first source of an element "
                "as it requires access to previous sources".format(self, self._sources[0])
            )

        # Preflight the sources
        for source in self.sources():
            source._preflight()

    # _cache_sources():
    #
    # Caches the sources into the local CAS
    #
    def _cache_sources(self):
        if self._sources and not self.cached():
            last_requires_previous = 0
            # commit all other sources by themselves
            for idx, source in enumerate(self._sources):
                if source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
                    self._sourcecache.commit(source, self._sources[last_requires_previous:idx])
                    last_requires_previous = idx
                else:
                    self._sourcecache.commit(source, [])

    # _last_source_requires_previous
    #
    # This is the last source that requires previous sources to be cached.
    # Sources listed after this will be cached separately.
    #
    # Returns:
    #    (int): index of last source that requires previous sources
    #
    def _last_source_requires_previous(self):
        if self._last_source_requires_previous_idx is None:
            last_requires_previous = 0
            for idx, source in enumerate(self._sources):
                if source.BST_REQUIRES_PREVIOUS_SOURCES_STAGE:
                    last_requires_previous = idx
            self._last_source_requires_previous_idx = last_requires_previous
        return self._last_source_requires_previous_idx