rhodecode/tests/vcs/conftest.py

# Copyright (C) 2010-2023 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/

import time
import shutil
import datetime

import pytest

from rhodecode.lib.str_utils import safe_bytes
from rhodecode.lib.vcs.backends import get_backend
from rhodecode.lib.vcs.backends.base import Config
from rhodecode.lib.vcs.nodes import FileNode
from rhodecode.tests import get_new_dir
from rhodecode.tests.utils import check_skip_backends, check_xfail_backends


@pytest.fixture()
def vcs_repository_support(
        request, backend_alias, baseapp, _vcs_repo_container):
    """
    Provide a test repository for the test run.

    Depending on the value of `recreate_repo_per_test` a new repo for each
    test will be created.

    The parameter `--backends` can be used to limit this fixture to specific
    backend implementations.
    """
    cls = request.cls

    check_skip_backends(request.node, backend_alias)
    check_xfail_backends(request.node, backend_alias)

    if _should_create_repo_per_test(cls):
        _vcs_repo_container = _create_vcs_repo_container(request)

    repo = _vcs_repo_container.get_repo(cls, backend_alias=backend_alias)

    # TODO: johbo: Supporting old test class api, think about removing this
    cls.repo = repo
    cls.repo_path = repo.path
    cls.default_branch = repo.DEFAULT_BRANCH_NAME
    cls.Backend = cls.backend_class = repo.__class__
    cls.imc = repo.in_memory_commit

    return backend_alias, repo


@pytest.fixture(scope='class')
def _vcs_repo_container(request):
    """
    Internal fixture intended to help support class based scoping on demand.
    """
    return _create_vcs_repo_container(request)


def _create_vcs_repo_container(request):
    repo_container = VcsRepoContainer()
    if not request.config.getoption('--keep-tmp-path'):
        request.addfinalizer(repo_container.cleanup)
    return repo_container


class VcsRepoContainer(object):

    def __init__(self):
        self._cleanup_paths = []
        self._repos = {}

    def get_repo(self, test_class, backend_alias):
        if backend_alias not in self._repos:
            repo = _create_empty_repository(test_class, backend_alias)

            self._cleanup_paths.append(repo.path)
            self._repos[backend_alias] = repo
        return self._repos[backend_alias]

    def cleanup(self):
        for repo_path in reversed(self._cleanup_paths):
            shutil.rmtree(repo_path)


def _should_create_repo_per_test(cls):
    return getattr(cls, 'recreate_repo_per_test', False)


def _create_empty_repository(cls, backend_alias=None):
    Backend = get_backend(backend_alias or cls.backend_alias)
    repo_path = get_new_dir(str(time.time()))
    repo = Backend(repo_path, create=True)
    if hasattr(cls, '_get_commits'):
        commits = cls._get_commits()
        cls.tip = _add_commits_to_repo(repo, commits)

    return repo


@pytest.fixture()
def config():
    """
    Instance of a repository config.

    The instance contains only one value:

    - Section: "section-a"
    - Key:     "a-1"
    - Value:   "value-a-1"

    The intended usage is for cases where a config instance is needed but no
    specific content is required.
    """
    config = Config()
    config.set('section-a', 'a-1', 'value-a-1')
    return config


def _add_commits_to_repo(repo, commits):
    imc = repo.in_memory_commit
    tip = None

    for commit in commits:
        for node in commit.get('added', []):
            if not isinstance(node, FileNode):
                node = FileNode(safe_bytes(node.path), content=node.content)
            imc.add(node)

        for node in commit.get('changed', []):
            if not isinstance(node, FileNode):
                node = FileNode(safe_bytes(node.path), content=node.content)
            imc.change(node)

        for node in commit.get('removed', []):
            imc.remove(FileNode(safe_bytes(node.path)))

        tip = imc.commit(
            message=str(commit['message']),
            author=str(commit['author']),
            date=commit['date'],
            branch=commit.get('branch')
        )

    return tip


@pytest.fixture()
def vcs_repo(request, backend_alias):
    Backend = get_backend(backend_alias)
    repo_path = get_new_dir(str(time.time()))
    repo = Backend(repo_path, create=True)

    @request.addfinalizer
    def cleanup():
        shutil.rmtree(repo_path)

    return repo


@pytest.fixture()
def generate_repo_with_commits(vcs_repo):
    """
    Creates a fabric to generate N commits with some file nodes on a randomly
    generated repository
    """

    def commit_generator(num):
        start_date = datetime.datetime(2010, 1, 1, 20)
        for x in range(num):
            yield {
                'message': 'Commit %d' % x,
                'author': 'Joe Doe <[e-mail]>',
                'date': start_date + datetime.timedelta(hours=12 * x),
                'added': [
                    FileNode(b'file_%d.txt' % x, content=b'Foobar %d' % x),
                ],
                'modified': [
                    FileNode(b'file_%d.txt' % x,
                             content=b'Foobar %d modified' % (x-1)),
                ]
            }

    def commit_maker(num=5):
        _add_commits_to_repo(vcs_repo, commit_generator(num))
        return vcs_repo

    return commit_maker


@pytest.fixture()
def hg_repo(request, vcs_repo):
    repo = vcs_repo

    commits = repo._get_commits()
    _add_commits_to_repo(repo, commits)

    return repo


@pytest.fixture()
def hg_commit(hg_repo):
    return hg_repo.get_commit()


class BackendTestMixin(object):
    """
    This is a backend independent test case class which should be created
    with ``type`` method.

    It is required to set following attributes at subclass:

    - ``backend_alias``: alias of used backend (see ``vcs.BACKENDS``)
    - ``repo_path``: path to the repository which would be created for set of
      tests
    - ``recreate_repo_per_test``: If set to ``False``, repo would NOT be
      created
      before every single test. Defaults to ``True``.
    """
    recreate_repo_per_test = True

    @classmethod
    def _get_commits(cls):
        commits = [
            {
                'message': 'Initial commit',
                'author': 'Joe Doe <[e-mail]>',
                'date': datetime.datetime(2010, 1, 1, 20),
                'added': [
                    FileNode(b'foobar', content=b'Foobar'),
                    FileNode(b'foobar2', content=b'Foobar II'),
                    FileNode(b'foo/bar/baz', content=b'baz here!'),
                ],
            },
            {
                'message': 'Changes...',
                'author': 'Jane Doe <[e-mail]>',
                'date': datetime.datetime(2010, 1, 1, 21),
                'added': [
                    FileNode(b'some/new.txt', content=b'news...'),
                ],
                'changed': [
                    FileNode(b'foobar', b'Foobar I'),
                ],
                'removed': [],
            },
        ]
        return commits