# 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 pytest
from unittest.mock import patch, MagicMock

from rhodecode.apps.repository.views.repo_settings_advanced import RepoSettingsAdvancedView
from rhodecode.bootstrap import TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS
from rhodecode.lib import auth
from rhodecode.lib.str_utils import safe_str
from rhodecode.model.db import Repository, User
from rhodecode.model.meta import Session
from rhodecode.model.repo import RepoModel, ForksAction
from rhodecode.tests import HG_REPO, GIT_REPO, assert_session_flash, no_newline_id_generator, login_user_session
from rhodecode.tests.fixtures.rc_fixture import Fixture
from rhodecode.tests.utils import repo_on_filesystem
from rhodecode.tests.routes import route_path

fixture = Fixture()


@pytest.mark.usefixtures("autologin_user", "app")
class TestAdminRepoSettingsAdvanced(object):
    def test_set_repo_fork_has_no_self_id(self, autologin_user, backend):
        repo = backend.repo
        response = self.app.get(route_path("edit_repo_advanced", repo_name=backend.repo_name))
        opt = """<option value="%s">vcs_test_git</option>""" % repo.repo_id
        response.mustcontain(no=[opt])

    def test_set_fork_of_target_repo(self, autologin_user, backend, csrf_token):
        target_repo = "target_%s" % backend.alias
        fixture.create_repo(target_repo, repo_type=backend.alias)
        repo2 = Repository.get_by_repo_name(target_repo)
        response = self.app.post(
            route_path("edit_repo_advanced_fork", repo_name=backend.repo_name),
            params={"id_fork_of": repo2.repo_id, "csrf_token": csrf_token},
        )
        repo = Repository.get_by_repo_name(backend.repo_name)
        repo2 = Repository.get_by_repo_name(target_repo)
        assert_session_flash(response, "Marked repo %s as fork of %s" % (repo.repo_name, repo2.repo_name))

        assert repo.fork == repo2
        response = response.follow()
        # check if given repo is selected

        opt = 'This repository is a fork of <a href="%s">%s</a>' % (
            route_path("repo_summary", repo_name=repo2.repo_name),
            repo2.repo_name,
        )

        response.mustcontain(opt)

        fixture.destroy_repo(target_repo, forks_action=ForksAction.DETACH)

    @pytest.mark.backends("hg", "git")
    def test_set_fork_of_other_type_repo(self, autologin_user, backend, csrf_token):
        TARGET_REPO_MAP = {
            "git": {"type": "hg", "repo_name": HG_REPO},
            "hg": {"type": "git", "repo_name": GIT_REPO},
        }
        target_repo = TARGET_REPO_MAP[backend.alias]

        repo2 = Repository.get_by_repo_name(target_repo["repo_name"])
        response = self.app.post(
            route_path("edit_repo_advanced_fork", repo_name=backend.repo_name),
            params={"id_fork_of": repo2.repo_id, "csrf_token": csrf_token},
        )
        assert_session_flash(response, "Cannot set repository as fork of repository with other type")

    def test_set_fork_of_none(self, autologin_user, backend, csrf_token):
        # mark it as None
        response = self.app.post(
            route_path("edit_repo_advanced_fork", repo_name=backend.repo_name),
            params={"id_fork_of": None, "csrf_token": csrf_token},
        )
        assert_session_flash(response, "Marked repo %s as fork of %s" % (backend.repo_name, "Nothing"))
        assert backend.repo.fork is None

    def test_set_fork_of_same_repo(self, autologin_user, backend, csrf_token):
        repo = Repository.get_by_repo_name(backend.repo_name)
        response = self.app.post(
            route_path("edit_repo_advanced_fork", repo_name=backend.repo_name),
            params={"id_fork_of": repo.repo_id, "csrf_token": csrf_token},
        )
        assert_session_flash(response, "An error occurred during this operation")

    @pytest.mark.parametrize("suffix", ["", "ąęł", "123"], ids=no_newline_id_generator)
    def test_advanced_repo_delete(self, autologin_user, backend, suffix, csrf_token):
        repo = backend.create_repo(name_suffix=suffix, private_repo=True)
        repo.private = True
        Session().commit()

        repo_name = repo.repo_name
        repo_name_str = safe_str(repo.repo_name)

        response = self.app.post(
            route_path("edit_repo_advanced_delete", repo_name=repo_name_str), params={"csrf_token": csrf_token}
        )
        assert_session_flash(response, "Deleted repository `{}`".format(repo_name))
        response.follow()

        # check if repo was deleted from db
        assert RepoModel().get_by_repo_name(repo_name) is None
        assert not repo_on_filesystem(repo_name_str)

    @patch("rhodecode.model.quota_model.SettingsModel")
    def test_advanced_repo_delete_decrement_repo_quota(self, settings_model_mock, backend):
        test_quota_config = {
            "rhodecode_user_quotas_enabled": True,
            "rhodecode_global_user_repo_quotas": 1,
        }

        mock_instance = settings_model_mock.return_value
        mock_instance.get_all_settings.return_value = test_quota_config

        session = login_user_session(self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
        csrf_token = auth.get_csrf_token(session)

        user: User = User.get_by_username(TEST_USER_REGULAR_LOGIN)
        user.update_quotas(
            max_disk_space_mb_allowance=100,
            max_repository_count_allowance=1,
        )

        repo = backend.create_repo(name_suffix="quota")
        repo.user = user
        Session().commit()

        repo_name = repo.repo_name
        repo_name_str = safe_str(repo.repo_name)

        response = self.app.post(
            route_path("edit_repo_advanced_delete", repo_name=repo_name_str), params={"csrf_token": csrf_token}
        )
        assert_session_flash(response, "Deleted repository `{}`".format(repo_name))
        response.follow()

        user: User = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        # check if repo was deleted from db
        assert RepoModel().get_by_repo_name(repo_name) is None
        assert not repo_on_filesystem(repo_name_str)
        user_repos = Repository.get_all_repos(user.user_id)
        assert len(user_repos) == 0

    @pytest.mark.parametrize("suffix", ["", "ąęł", "123"], ids=no_newline_id_generator)
    def test_delete_broken_repo(self, autologin_user, backend, suffix, csrf_token):
        repo = backend.create_repo(name_suffix=suffix, private_repo=True)
        repo.private = True
        Session().commit()

        repo_name = repo.repo_name
        repo_name_str = safe_str(repo.repo_name)

        # delete from file system
        RepoModel()._delete_filesystem_repo(repo)
        assert not repo_on_filesystem(repo_name_str)

        response = self.app.post(
            route_path("edit_repo_advanced_delete", repo_name=repo_name_str), params={"csrf_token": csrf_token}
        )
        assert_session_flash(response, "Deleted repository `{}`".format(repo_name))
        response.follow()

        # check if repo was deleted from db
        assert RepoModel().get_by_repo_name(repo_name) is None
        assert not repo_on_filesystem(repo_name_str)

    @patch.object(RepoSettingsAdvancedView, "_maybe_needs_password_change")
    @patch.object(RepoSettingsAdvancedView, "_maybe_needs_2fa_configuration")
    @patch.object(RepoSettingsAdvancedView, "_maybe_needs_2fa_check")
    @patch("rhodecode.apps._base.ScmModel")
    @patch("rhodecode.apps._base.IssueTrackerSettingsModel")
    @patch("rhodecode.lib.helpers.route_path")
    def test_advanced_repo_delete__call_remote_cache_invalidation(self, *ignore):
        request = MagicMock()
        db_repo = MagicMock()
        scm_instance = MagicMock()

        db_repo.scm_instance.return_value = scm_instance
        request.db_repo = db_repo
        request.params = {}

        obj = RepoSettingsAdvancedView(MagicMock(), request)
        # bypass decorators
        unwrapped__edit_advanced_delete = RepoSettingsAdvancedView.__dict__[
            "edit_advanced_delete"
        ].__wrapped__.__wrapped__.__wrapped__

        try:
            unwrapped__edit_advanced_delete(obj)
        except Exception as ignore:
            # ignore error, because of mocks method will fail with exception which is not related to the current test
            pass

        db_repo.scm_instance.assert_called_once_with(cache=False)
        scm_instance.vcsserver_invalidate_cache.assert_called_once_with(delete=True)

    @pytest.mark.parametrize("suffix", ["", "ąęł", "123"], ids=no_newline_id_generator)
    def test_advanced_repo_archive(self, autologin_user, backend, suffix, csrf_token):
        repo = backend.create_repo(name_suffix=suffix)
        repo_name = repo.repo_name
        repo_name_str = safe_str(repo.repo_name)

        response = self.app.post(
            route_path("edit_repo_advanced_archive", repo_name=repo_name_str), params={"csrf_token": csrf_token}
        )

        assert_session_flash(response, "Archived repository `{}`".format(repo_name))

        response = self.app.get(route_path("repo_summary", repo_name=repo_name_str))
        response.mustcontain("This repository has been archived. It is now read-only.")

        # check if repo was deleted from db
        assert RepoModel().get_by_repo_name(repo_name).archived is True
