Source code for pubtools.pulplib._impl.fake.controller

import warnings

from .client import FakeClient
from .state import FakeState


class FakeController(object):
    """A controller for fake instances of :class:`~pubtools.pulplib.Client`, to
    be used within automated tests.

    This class provide clients which have the same public interface as
    :class:`~pubtools.pulplib.Client`. These clients can do most of the same
    actions as a real client, but use a simple in-memory implementation rather
    than issuing requests to a remote Pulp server. Client data may be
    inspected and modified via the FakeController instance.

    Example:

        .. code-block:: python

            # imagine we have a function like this
            def delete_old_repos(client, days):
                # code here using client to find/delete all repos older
                # than given number of days
                ...

            # then we can use a fake client to test it like this
            def test_delete_old_repos():
                controller = FakeController()

                # Insert repositories into fake data
                repo_now = YumRepository(id='current-repo', created=datetime.utcnow())
                repo_old = FileRepository(id='old-repo', created=datetime.min)

                controller.insert_repository(repo_now)
                controller.insert_repository(repo_old)

                # Run the code under test against the fake client

                # It should succeed without crashing
                delete_old_repos(controller.client, 5)

                # The set of repos should now contain only the recent repo
                assert controller.repositories == [repo_now]

    **Limitations:**

        While :class:`~pubtools.pulplib.Client` allows searching on any fields
        which exist within Pulp's database, the fake client only supports searching
        on fields known to the :ref:`schemas` used by this library.
    """

    def __init__(self):
        self._state = FakeState()

        self._shared_client = self.new_client()

    def new_client(self):
        """Returns a new fake Client attached to this controller.

        All clients created from the same controller will share the same underlying
        state; changes made via one client will be visible to all other clients
        attached to this controller.

        .. versionadded:: 2.32.0
        """
        return FakeClient(self._state)

    @property
    def client(self):
        """A shared client instance attached to this controller.

        This property may be used if only a single client is required.
        Otherwise, :meth:`new_client` can be used to obtain multiple clients.
        """
        return self._shared_client

    @property
    def repositories(self):
        """The list of existing repositories in the fake client."""
        return self._state.repositories[:]

    def insert_repository(self, repository):
        """Add a repository to the set of existing repositories in the fake client.

        Args:
            repository (:class:`~pubtools.pulplib.Repository`)
                A repository object to insert.
        """
        with self._state.lock:
            self._state.repositories.append(repository)

    def insert_units(self, repository, units):
        """Add units to the set of existing content for a repository.

        Args:
            repository (:class:`~pubtools.pulplib.Repository`)
                A repository object, or ``None`` to insert units as orphans.
            units (list[:class:`~pubtools.pulplib.Unit`])
                A list of units to be inserted.

        .. versionadded:: 1.5.0

        .. versionadded:: 2.21.0
            Added ability to insert orphan units.
        """
        repo_id = repository.id if repository else None
        with self._state.lock:
            self._state.insert_repo_units(repo_id, units)

    @property
    def content_type_ids(self):
        """The list of content type IDs the fake client will claim to support.

        .. versionadded:: 1.4.0
        """
        return self._state.type_ids[:]

    def set_content_type_ids(self, type_ids):
        """Set the list of content type IDs the fake client will claim to support.

        Args:
            type_ids (list[str])
                A list of content type IDs (e.g. "rpm", "erratum", ...)

        .. versionadded:: 1.4.0
        """
        with self._state.lock:
            self._state.type_ids = type_ids[:]

    @property
    def publish_history(self):
        """A list of repository publishes triggered via clients associated with
        this controller.

        Each element of this list is a named tuple with the following attributes,
        in order:

            ``repository``:
                :class:`~pubtools.pulplib.Repository` for which publish was triggered
            ``tasks``:
                list of :class:`~pubtools.pulplib.Task` generated as a result
                of this publish
        """
        return self._state.publish_history[:]

    @property
    def sync_history(self):
        """A list of repository syncs triggered via clients associated with this
        controller.

        Each element of this list is a named tuple with the following attributes,
        in order:

            ``repository``:
                :class:`~pubtools.pulplib.Repository` for which sync was triggered
            ``tasks``:
                list of :class:`~pubtools.pulplib.Task` generated as a result
                of this sync
            ``sync_config``:
                :class:`~pubtools.pulplib.SyncConfig` (of the appropriate subclass) used for this sync

        .. versionadded:: 2.5.0
        """
        return self._state.sync_history[:]

    @property
    def upload_history(self):
        # A list of upload tasks triggered via this client.
        #
        # Each element of this list is a named tuple with the following attributes,
        # in order:
        #
        #     ``repository``:
        #         :class:`~pubtools.pulplib.Repository` for which upload was triggered
        #     ``tasks``:
        #         list of :class:`~pubtools.pulplib.Task` generated as a result
        #         of this upload
        #     ``name`` (str):
        #         the remote path used
        #     ``sha256`` (str):
        #         checksum of the file uploaded
        #
        # Deprecated: structure was unintentionally specific to ISO units and is also
        # unnecessary as uploading will make unit(s) available in the relevant repo.
        #
        warnings.warn(
            "upload_history is deprecated, check repo units instead", DeprecationWarning
        )

        return self._state.upload_history[:]

    @property
    def tasks(self):
        """The list of existing tasks in the fake."""
        return self._state.tasks[:]

    def insert_task(self, task):
        """Add a task to the set of existing tasks in the fake.

        Args:
            task (:class:`~pubtools.pulplib.Task`)
                A task object to insert.
        """
        with self._state.lock:
            self._state.tasks.append(task)

    @property
    def repo_lock_history(self):
        """
        A list containing all lock actions carried out.
        Possible actions are: lock, unlock, multi-unlock
        """
        return self._state.repo_lock_history