# Copyright (C) 2010-2024 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 mock
import pytest

from rhodecode.lib.str_utils import base64_to_str
from rhodecode.lib.utils2 import AttributeDict
from rhodecode.tests.fixtures.fixture_pyramid import ini_config
from rhodecode.tests.utils import CustomTestApp, AuthPluginManager

from rhodecode.lib.caching_query import FromCache
from rhodecode.lib.middleware import simplevcs
from rhodecode.lib.middleware.https_fixup import HttpsFixup
from rhodecode.lib.middleware.utils import scm_app_http
from rhodecode.model.db import User, _hash_key
from rhodecode.model.meta import Session, cache as db_cache
from rhodecode.tests import HG_REPO, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS
from rhodecode.tests.lib.middleware import mock_scm_app

from rhodecode.model.db import Permission, User
from rhodecode.model.meta import Session
from rhodecode.model.user import UserModel


@pytest.fixture()
def enable_auth_plugins(request, app):
    """
    Return a factory object that when called, allows to control which
    authentication plugins are enabled.
    """

    enabler = AuthPluginManager()
    request.addfinalizer(enabler.cleanup)

    return enabler


@pytest.fixture()
def test_user_factory(request, baseapp):
    def user_factory(username="test_user", password="qweqwe", first_name="John", last_name="Testing", **kwargs):
        usr = UserModel().create_or_update(
            username=username,
            password=password,
            email=f"{username}@rhodecode.org",
            firstname=first_name,
            lastname=last_name,
        )
        Session().commit()

        for k, v in kwargs.items():
            setattr(usr, k, v)
            Session().add(usr)

        new_usr = User.get_by_username(username)
        new_usr_id = new_usr.user_id
        assert new_usr == usr

        @request.addfinalizer
        def cleanup():
            if User.get(new_usr_id) is None:
                return

            perm = Permission.query().all()
            for p in perm:
                UserModel().revoke_perm(usr, p)

            UserModel().delete(new_usr_id)
            Session().commit()

        return usr

    return user_factory


class StubVCSController(simplevcs.SimpleVCS):
    SCM = "hg"
    stub_response_body = tuple()

    def __init__(self, *args, **kwargs):
        super(StubVCSController, self).__init__(*args, **kwargs)
        self._action = "pull"
        self._is_shadow_repo_dir = True
        self._name = HG_REPO
        self.set_repo_names(None)

    @property
    def is_shadow_repo_dir(self):
        return self._is_shadow_repo_dir

    def _get_repository_name(self, environ):
        return self._name

    def _get_action(self, environ):
        return self._action

    def _create_wsgi_app(self, repo_path, repo_name, config):
        def fake_app(environ, start_response):
            headers = [("Http-Accept", "application/mercurial")]
            start_response("200 OK", headers)
            return self.stub_response_body

        return fake_app

    def _create_config(self, extras, repo_name, scheme="http"):
        return None


@pytest.fixture()
def vcscontroller(baseapp, config_stub, request_stub):
    from rhodecode.config.middleware import ce_auth_resources

    config_stub.testing_securitypolicy()
    config_stub.include("rhodecode.authentication")

    for resource in ce_auth_resources:
        config_stub.include(resource)

    controller = StubVCSController(baseapp.config.get_settings(), request_stub.registry)
    app = HttpsFixup(controller, baseapp.config.get_settings())
    app = CustomTestApp(app)

    _remove_default_user_from_query_cache()

    # Sanity checks that things are set up correctly
    app.get("/" + HG_REPO, status=200)

    app.controller = controller
    return app


def _remove_default_user_from_query_cache():
    user = User.get_default_user(cache=True)
    query = Session().query(User).filter(User.username == user.username)
    query = query.options(FromCache("sql_cache_short", f"get_user_{_hash_key(user.username)}"))

    db_cache.invalidate(query, {}, FromCache("sql_cache_short", f"get_user_{_hash_key(user.username)}"))

    Session().expire(user)


def test_handles_exceptions_during_permissions_checks(
    vcscontroller, disable_anonymous_user, enable_auth_plugins, test_user_factory
):
    test_password = "qweqwe"
    test_user = test_user_factory(password=test_password, extern_type="headers", extern_name="headers")
    test_username = test_user.username

    enable_auth_plugins.enable(
        [
            "egg:rhodecode-enterprise-ce#headers",
            "egg:rhodecode-enterprise-ce#token",
            "egg:rhodecode-enterprise-ce#rhodecode",
        ],
        override={"egg:rhodecode-enterprise-ce#headers": {"auth_headers_header": "REMOTE_USER"}},
    )

    user_and_pass = f"{test_username}:{test_password}"
    auth_password = base64_to_str(user_and_pass)

    extra_environ = {
        "AUTH_TYPE": "Basic",
        "HTTP_AUTHORIZATION": f"Basic {auth_password}",
        "REMOTE_USER": test_username,
    }

    # Verify that things are hooked up correctly, we pass user with headers bound auth, and headers filled in
    vcscontroller.get("/", status=200, extra_environ=extra_environ)

    # Simulate trouble during permission checks
    with mock.patch(
        "rhodecode.model.db.User.get_by_username", side_effect=Exception("permission_error_test")
    ) as get_user:
        # Verify that a correct 500 is returned and check that the expected
        # code path was hit.
        vcscontroller.get("/", status=500, extra_environ=extra_environ)
        assert get_user.called


class StubFailVCSController(simplevcs.SimpleVCS):
    def _handle_request(self, environ, start_response):
        raise Exception("BOOM")


@pytest.fixture(scope="module")
def fail_controller(baseapp):
    controller = StubFailVCSController(baseapp.config.get_settings(), baseapp.config)
    controller = HttpsFixup(controller, baseapp.config.get_settings())
    controller = CustomTestApp(controller)
    return controller


def test_handles_exceptions_as_internal_server_error(fail_controller):
    fail_controller.get("/", status=500)


def test_provides_traceback_for_appenlight(fail_controller):
    response = fail_controller.get("/", status=500, extra_environ={"appenlight.client": "fake"})
    assert "appenlight.__traceback" in response.request.environ


def test_provides_utils_scm_app_as_scm_app_by_default(baseapp, request_stub):
    controller = StubVCSController(baseapp.config.get_settings(), request_stub.registry)
    assert controller.scm_app is scm_app_http


def test_allows_to_override_scm_app_via_config(baseapp, request_stub):
    config = baseapp.config.get_settings().copy()
    config["vcs.scm_app_implementation"] = "rhodecode.tests.lib.middleware.mock_scm_app"
    controller = StubVCSController(config, request_stub.registry)
    assert controller.scm_app is mock_scm_app


@pytest.mark.parametrize(
    "query_string, expected",
    [
        ("cmd=stub_command", True),
        ("cmd=listkeys", False),
    ],
)
def test_should_check_locking(query_string, expected):
    result = simplevcs._should_check_locking(query_string)
    assert result == expected


class TestShadowRepoRegularExpression(object):
    pr_segment = "pull-request"
    shadow_segment = "repository"

    @pytest.mark.parametrize(
        "url, expected",
        [
            # repo with/without groups
            ("My-Repo/{pr_segment}/1/{shadow_segment}", True),
            ("Group/My-Repo/{pr_segment}/2/{shadow_segment}", True),
            ("Group/Sub-Group/My-Repo/{pr_segment}/3/{shadow_segment}", True),
            ("Group/Sub-Group1/Sub-Group2/My-Repo/{pr_segment}/3/{shadow_segment}", True),
            # pull request ID
            ("MyRepo/{pr_segment}/1/{shadow_segment}", True),
            ("MyRepo/{pr_segment}/1234567890/{shadow_segment}", True),
            ("MyRepo/{pr_segment}/-1/{shadow_segment}", False),
            ("MyRepo/{pr_segment}/invalid/{shadow_segment}", False),
            # unicode
            ("Sp€çîál-Repö/{pr_segment}/1/{shadow_segment}", True),
            ("Sp€çîál-Gröüp/Sp€çîál-Repö/{pr_segment}/1/{shadow_segment}", True),
            # trailing/leading slash
            ("/My-Repo/{pr_segment}/1/{shadow_segment}", False),
            ("My-Repo/{pr_segment}/1/{shadow_segment}/", False),
            ("/My-Repo/{pr_segment}/1/{shadow_segment}/", False),
            # misc
            ("My-Repo/{pr_segment}/1/{shadow_segment}/extra", False),
            ("My-Repo/{pr_segment}/1/{shadow_segment}extra", False),
        ],
    )
    def test_shadow_repo_regular_expression(self, url, expected):
        from rhodecode.lib.middleware.simplevcs import SimpleVCS

        url = url.format(pr_segment=self.pr_segment, shadow_segment=self.shadow_segment)
        match_obj = SimpleVCS.shadow_repo_re.match(url)
        assert (match_obj is not None) == expected


@pytest.mark.backends("git", "hg")
class TestShadowRepoExposure(object):
    def test_pull_on_shadow_repo_propagates_to_wsgi_app(self, baseapp, request_stub):
        """
        Check that a pull action to a shadow repo is propagated to the
        underlying wsgi app.
        """
        controller = StubVCSController(baseapp.config.get_settings(), request_stub.registry)
        controller.is_shadow_repo = True
        controller._action = "pull"
        controller._is_shadow_repo_dir = True
        controller.stub_response_body = (b"dummy body value",)
        controller._get_default_cache_ttl = mock.Mock(return_value=(False, 0))

        environ_stub = {
            "HTTP_HOST": "test.example.com",
            "HTTP_ACCEPT": "application/mercurial",
            "REQUEST_METHOD": "GET",
            "wsgi.url_scheme": "http",
        }

        response = controller(environ_stub, mock.Mock())
        response_body = b"".join(response)

        # Assert that we got the response from the wsgi app.
        assert response_body == b"".join(controller.stub_response_body)

    def test_pull_on_shadow_repo_that_is_missing(self, baseapp, request_stub):
        """
        Check that a pull action to a shadow repo is propagated to the
        underlying wsgi app.
        """
        controller = StubVCSController(baseapp.config.get_settings(), request_stub.registry)
        controller.is_shadow_repo = True
        controller._action = "pull"
        controller._is_shadow_repo_dir = False
        controller.stub_response_body = (b"dummy body value",)
        environ_stub = {
            "HTTP_HOST": "test.example.com",
            "HTTP_ACCEPT": "application/mercurial",
            "REQUEST_METHOD": "GET",
            "wsgi.url_scheme": "http",
        }

        response = controller(environ_stub, mock.Mock())
        response_body = b"".join(response)

        # Assert that we got the response from the wsgi app.
        assert b"404 Not Found" in response_body

    def test_push_on_shadow_repo_raises(self, baseapp, request_stub):
        """
        Check that a push action to a shadow repo is aborted.
        """
        controller = StubVCSController(baseapp.config.get_settings(), request_stub.registry)
        controller.is_shadow_repo = True
        controller._action = "push"
        controller.stub_response_body = (b"dummy body value",)
        environ_stub = {
            "HTTP_HOST": "test.example.com",
            "HTTP_ACCEPT": "application/mercurial",
            "REQUEST_METHOD": "GET",
            "wsgi.url_scheme": "http",
        }

        response = controller(environ_stub, mock.Mock())
        response_body = b"".join(response)

        assert response_body != controller.stub_response_body
        # Assert that a 406 error is returned.
        assert b"406 Not Acceptable" in response_body

    def test_set_repo_names_no_shadow(self, baseapp, request_stub):
        """
        Check that the set_repo_names method sets all names to the one returned
        by the _get_repository_name method on a request to a non shadow repo.
        """
        environ_stub = {}
        controller = StubVCSController(baseapp.config.get_settings(), request_stub.registry)
        controller._name = "RepoGroup/MyRepo"
        controller.set_repo_names(environ_stub)
        assert not controller.is_shadow_repo
        assert (
            controller.url_repo_name
            == controller.acl_repo_name
            == controller.vcs_repo_name
            == controller._get_repository_name(environ_stub)
        )

    def test_set_repo_names_with_shadow(self, baseapp, pr_util, config_stub, request_stub):
        """
        Check that the set_repo_names method sets correct names on a request
        to a shadow repo.
        """
        from rhodecode.model.pull_request import PullRequestModel

        pull_request = pr_util.create_pull_request()
        shadow_url = "{target}/{pr_segment}/{pr_id}/{shadow_segment}".format(
            target=pull_request.target_repo.repo_name,
            pr_id=pull_request.pull_request_id,
            pr_segment=TestShadowRepoRegularExpression.pr_segment,
            shadow_segment=TestShadowRepoRegularExpression.shadow_segment,
        )
        controller = StubVCSController(baseapp.config.get_settings(), request_stub.registry)
        controller._name = shadow_url
        controller.set_repo_names({})

        # Get file system path to shadow repo for assertions.
        workspace_id = PullRequestModel()._workspace_id(pull_request)
        vcs_repo_name = pull_request.target_repo.get_shadow_repository_path(workspace_id)

        assert controller.vcs_repo_name == vcs_repo_name
        assert controller.url_repo_name == shadow_url
        assert controller.acl_repo_name == pull_request.target_repo.repo_name
        assert controller.is_shadow_repo

    def test_set_repo_names_with_shadow_but_missing_pr(self, baseapp, pr_util, config_stub, request_stub):
        """
        Checks that the set_repo_names method enforces matching target repos
        and pull request IDs.
        """
        pull_request = pr_util.create_pull_request()
        shadow_url = "{target}/{pr_segment}/{pr_id}/{shadow_segment}".format(
            target=pull_request.target_repo.repo_name,
            pr_id=999999999,
            pr_segment=TestShadowRepoRegularExpression.pr_segment,
            shadow_segment=TestShadowRepoRegularExpression.shadow_segment,
        )
        controller = StubVCSController(baseapp.config.get_settings(), request_stub.registry)
        controller._name = shadow_url
        controller.set_repo_names({})

        assert not controller.is_shadow_repo
        assert controller.url_repo_name == controller.acl_repo_name == controller.vcs_repo_name


class TestGenerateVcsResponse(object):
    def test_ensures_that_start_response_is_called_early_enough(self, baseapp):
        app_ini_config = baseapp.config.registry.settings["__file__"]
        self.call_controller_with_response_body(app_ini_config, iter(["a", "b"]))
        assert self.start_response.called

    def test_invalidates_cache_after_body_is_consumed(self, baseapp):
        app_ini_config = baseapp.config.registry.settings["__file__"]
        result = self.call_controller_with_response_body(app_ini_config, iter(["a", "b"]))
        assert not self.was_cache_invalidated()
        # Consume the result
        list(result)
        assert self.was_cache_invalidated()

    def test_raises_unknown_exceptions(self, baseapp):
        app_ini_config = baseapp.config.registry.settings["__file__"]
        result = self.call_controller_with_response_body(app_ini_config, self.raise_result_iter(vcs_kind="unknown"))
        with pytest.raises(Exception):
            list(result)

    def call_controller_with_response_body(self, ini_config, response_body):
        settings = {
            "vcs.hooks.protocol.v2": "celery",
            "vcs.hooks.direct_calls": False,
        }
        registry = AttributeDict()
        controller = StubVCSController(settings, registry)
        controller._invalidate_cache = mock.Mock()
        controller.stub_response_body = response_body
        self.start_response = mock.Mock()
        result = controller._generate_vcs_response(
            environ={},
            start_response=self.start_response,
            repo_path="fake_repo_path",
            extras={"config": ini_config},
            action="push",
        )
        self.controller = controller
        return result

    def raise_result_iter(self, vcs_kind="repo_locked"):
        """
        Simulates an exception due to a vcs raised exception if kind vcs_kind
        """
        raise self.vcs_exception(vcs_kind=vcs_kind)
        yield "never_reached"

    def vcs_exception(self, vcs_kind="repo_locked"):
        locked_exception = Exception("TEST_MESSAGE")
        locked_exception._vcs_kind = vcs_kind
        return locked_exception

    def was_cache_invalidated(self):
        return self.controller._invalidate_cache.called


class TestInitializeGenerator(object):
    def test_drains_first_element(self):
        gen = self.factory(["__init__", 1, 2])
        result = list(gen)
        assert result == [1, 2]

    @pytest.mark.parametrize(
        "values",
        [
            [],
            [1, 2],
        ],
    )
    def test_raises_value_error(self, values):
        with pytest.raises(ValueError):
            self.factory(values)

    @simplevcs.initialize_generator
    def factory(self, iterable):
        for elem in iterable:
            yield elem
