rhodecode/tests/lib/test_base.py

# -*- coding: utf-8 -*-

# Copyright (C) 2010-2017 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 pytest
from mock import Mock, patch
from pylons import url

from rhodecode.lib import base
from rhodecode.lib.vcs.exceptions import RepositoryRequirementError
from rhodecode.model import db


@pytest.mark.parametrize('result_key, expected_value', [
    ('username', 'stub_username'),
    ('action', 'stub_action'),
    ('repository', 'stub_repo_name'),
    ('scm', 'stub_scm'),
    ('hooks', ['stub_hook']),
    ('config', 'stub_ini_filename'),
    ('ip', '1.2.3.4'),
    ('server_url', 'https://example.com'),
    ('user_agent', 'client-text-v1.1'),
    # TODO: johbo: Commpare locking parameters with `_get_rc_scm_extras`
    # in hooks_utils.
    ('make_lock', None),
    ('locked_by', [None, None, None]),
])
def test_vcs_operation_context_parameters(result_key, expected_value):
    result = call_vcs_operation_context()
    assert result[result_key] == expected_value


@patch('rhodecode.model.db.User.get_by_username', Mock())
@patch('rhodecode.model.db.Repository.get_by_repo_name')
def test_vcs_operation_context_checks_locking(mock_get_by_repo_name):
    mock_get_locking_state = mock_get_by_repo_name().get_locking_state
    mock_get_locking_state.return_value = (None, None, [None, None, None])
    call_vcs_operation_context(check_locking=True)
    assert mock_get_locking_state.called


@patch('rhodecode.model.db.Repository.get_locking_state')
def test_vcs_operation_context_skips_locking_checks_if_anonymouse(
        mock_get_locking_state):
    call_vcs_operation_context(
        username=db.User.DEFAULT_USER, check_locking=True)
    assert not mock_get_locking_state.called


@patch('rhodecode.model.db.Repository.get_locking_state')
def test_vcs_operation_context_can_skip_locking_check(mock_get_locking_state):
    call_vcs_operation_context(check_locking=False)
    assert not mock_get_locking_state.called


@patch.object(
    base, 'get_enabled_hook_classes', Mock(return_value=['stub_hook']))
@patch('rhodecode.lib.utils2.get_server_url',
       Mock(return_value='https://example.com'))
def call_vcs_operation_context(**kwargs_override):
    kwargs = {
        'repo_name': 'stub_repo_name',
        'username': 'stub_username',
        'action': 'stub_action',
        'scm': 'stub_scm',
        'check_locking': False,
    }
    kwargs.update(kwargs_override)
    config_file_patch = patch.dict(
        'rhodecode.CONFIG', {'__file__': 'stub_ini_filename'})
    settings_patch = patch.object(base, 'VcsSettingsModel')
    with config_file_patch, settings_patch as settings_mock:
        result = base.vcs_operation_context(
            environ={'HTTP_USER_AGENT': 'client-text-v1.1',
                     'REMOTE_ADDR': '1.2.3.4'}, **kwargs)
    settings_mock.assert_called_once_with(repo='stub_repo_name')
    return result


class TestBaseRepoController(object):
    def test_context_is_updated_when_update_global_counters_is_called(self):
        followers = 1
        forks = 2
        pull_requests = 3
        is_following = True
        scm_model = Mock(name="scm_model")
        db_repo = Mock(name="db_repo")
        scm_model.get_followers.return_value = followers
        scm_model.get_forks.return_value = forks
        scm_model.get_pull_requests.return_value = pull_requests
        scm_model.is_following_repo.return_value = is_following

        controller = base.BaseRepoController()
        with patch.object(base, 'c') as context_mock:
            controller._update_global_counters(scm_model, db_repo)

        scm_model.get_pull_requests.assert_called_once_with(db_repo)

        assert context_mock.repository_pull_requests == pull_requests


class TestBaseRepoControllerHandleMissingRequirements(object):
    def test_logs_error_and_sets_repo_to_none(self, app):
        controller = base.BaseRepoController()
        error_message = 'Some message'
        error = RepositoryRequirementError(error_message)
        context_patcher = patch.object(base, 'c')
        log_patcher = patch.object(base, 'log')
        request_patcher = patch.object(base, 'request')
        redirect_patcher = patch.object(base, 'redirect')
        controller.rhodecode_repo = 'something'

        with context_patcher as context_mock, log_patcher as log_mock, \
                request_patcher, redirect_patcher:
            context_mock.repo_name = 'abcde'
            controller._handle_missing_requirements(error)

        expected_log_message = (
            'Requirements are missing for repository %s: %s', 'abcde',
            error_message)
        log_mock.error.assert_called_once_with(*expected_log_message)

        assert controller.rhodecode_repo is None

    @pytest.mark.parametrize('path, should_redirect', [
        ('/abcde', False),
        ('/abcde/settings', False),
        ('/abcde/settings/vcs', False),
        ('/_admin/repos/abcde', False),  # Settings update
        ('/abcde/changelog', True),
        ('/abcde/files/tip', True),
        ('/abcde/settings/statistics', True),
    ])
    def test_redirects_if_not_summary_or_settings_page(
            self, app, path, should_redirect):
        repo_name = 'abcde'
        controller = base.BaseRepoController()
        error = RepositoryRequirementError('Some message')
        context_patcher = patch.object(base, 'c')
        controller.rhodecode_repo = repo_name
        request_patcher = patch.object(base, 'request')
        redirect_patcher = patch.object(base, 'redirect')

        with context_patcher as context_mock, \
                request_patcher as request_mock, \
                redirect_patcher as redirect_mock:
            request_mock.path = path
            context_mock.repo_name = repo_name
            controller._handle_missing_requirements(error)

        expected_url = url('summary_home', repo_name=repo_name)
        if should_redirect:
            redirect_mock.assert_called_once_with(expected_url)
        else:
            redirect_mock.call_count == 0


class TestBaseRepoControllerBefore(object):
    def test_flag_is_true_when_requirements_are_missing(self, before_mocks):
        controller = self._get_controller()

        handle_patcher = patch.object(
            controller, '_handle_missing_requirements')

        error = RepositoryRequirementError()
        before_mocks.repository.scm_instance.side_effect = error

        with handle_patcher as handle_mock:
            controller.__before__()

        handle_mock.assert_called_once_with(error)
        assert before_mocks['context'].repository_requirements_missing is True

    def test_flag_is_false_when_no_requirements_are_missing(
            self, before_mocks):
        controller = self._get_controller()

        handle_patcher = patch.object(
            controller, '_handle_missing_requirements')
        with handle_patcher as handle_mock:
            controller.__before__()
        handle_mock.call_count == 0
        assert before_mocks['context'].repository_requirements_missing is False

    def test_update_global_counters_is_called(self, before_mocks):
        controller = self._get_controller()

        update_counters_patcher = patch.object(
            controller, '_update_global_counters')

        with update_counters_patcher as update_counters_mock:
            controller.__before__()
        update_counters_mock.assert_called_once_with(
            controller.scm_model, before_mocks.repository)

    def _get_controller(self):
        controller = base.BaseRepoController()
        controller.scm_model = Mock()
        controller.rhodecode_repo = Mock()
        return controller


@pytest.fixture
def before_mocks(request):
    patcher = BeforePatcher()
    patcher.start()
    request.addfinalizer(patcher.stop)
    return patcher


class BeforePatcher(object):
    patchers = {}
    mocks = {}
    repository = None

    def __init__(self):
        self.repository = Mock()

    def start(self):
        self.patchers = {
            'request': patch.object(base, 'request'),
            'before': patch.object(base.BaseController, '__before__'),
            'context': patch.object(base, 'c'),
            'repo': patch.object(
                base.Repository, 'get_by_repo_name',
                return_value=self.repository)

        }
        self.mocks = {
            p: self.patchers[p].start() for p in self.patchers
        }

    def stop(self):
        for patcher in self.patchers.values():
            patcher.stop()

    def __getitem__(self, key):
        return self.mocks[key]