rhodecode/tests/models/test_pullrequest.py

# 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 os

import mock
import pytest
import textwrap

import rhodecode
from rhodecode.lib.vcs.backends import get_backend
from rhodecode.lib.vcs.backends.base import MergeResponse, MergeFailureReason, Reference
from rhodecode.lib.vcs.exceptions import RepositoryError
from rhodecode.lib.vcs.nodes import FileNode
from rhodecode.model.comment import CommentsModel
from rhodecode.model.db import PullRequest, Session
from rhodecode.model.pull_request import PullRequestModel
from rhodecode.model.user import UserModel
from rhodecode.tests import TEST_USER_ADMIN_LOGIN
from rhodecode.lib.str_utils import safe_str

pytestmark = [
    pytest.mark.backends("git", "hg"),
]


@pytest.mark.usefixtures("config_stub")
class TestPullRequestModel(object):
    @pytest.fixture()
    def pull_request(self, request, backend, pr_util):
        """
        A pull request combined with multiples patches.
        """
        BackendClass = get_backend(backend.alias)
        merge_resp = MergeResponse(False, False, None, MergeFailureReason.UNKNOWN, metadata={"exception": "MockError"})
        self.merge_patcher = mock.patch.object(BackendClass, "merge", return_value=merge_resp)
        self.workspace_remove_patcher = mock.patch.object(BackendClass, "cleanup_merge_workspace")

        self.workspace_remove_mock = self.workspace_remove_patcher.start()
        self.merge_mock = self.merge_patcher.start()
        self.comment_patcher = mock.patch("rhodecode.model.changeset_status.ChangesetStatusModel.set_status")
        self.comment_patcher.start()
        self.notification_patcher = mock.patch("rhodecode.model.notification.NotificationModel.create")
        self.notification_patcher.start()
        self.helper_patcher = mock.patch("rhodecode.lib.helpers.route_path")
        self.helper_patcher.start()

        self.hook_patcher = mock.patch.object(PullRequestModel, "trigger_pull_request_hook")
        self.hook_mock = self.hook_patcher.start()

        self.invalidation_patcher = mock.patch("rhodecode.model.pull_request.ScmModel.mark_for_invalidation")
        self.invalidation_mock = self.invalidation_patcher.start()

        self.pull_request = pr_util.create_pull_request(mergeable=True, name_suffix="ąć")
        self.source_commit = self.pull_request.source_ref_parts.commit_id
        self.target_commit = self.pull_request.target_ref_parts.commit_id
        self.workspace_id = f"pr-{self.pull_request.pull_request_id}"
        self.repo_id = self.pull_request.target_repo.repo_id

        @request.addfinalizer
        def cleanup_pull_request():
            calls = [mock.call(self.pull_request, self.pull_request.author, "create")]
            self.hook_mock.assert_has_calls(calls)

            self.workspace_remove_patcher.stop()
            self.merge_patcher.stop()
            self.comment_patcher.stop()
            self.notification_patcher.stop()
            self.helper_patcher.stop()
            self.hook_patcher.stop()
            self.invalidation_patcher.stop()

        return self.pull_request

    def test_get_all(self, pull_request):
        prs = PullRequestModel().get_all(pull_request.target_repo)
        assert isinstance(prs, list)
        assert len(prs) == 1

    def test_count_all(self, pull_request):
        pr_count = PullRequestModel().count_all(pull_request.target_repo)
        assert pr_count == 1

    def test_get_awaiting_review(self, pull_request):
        prs = PullRequestModel().get_awaiting_review(pull_request.target_repo)
        assert isinstance(prs, list)
        assert len(prs) == 1

    def test_count_awaiting_review(self, pull_request):
        pr_count = PullRequestModel().count_awaiting_review(pull_request.target_repo)
        assert pr_count == 1

    def test_get_awaiting_my_review(self, pull_request):
        PullRequestModel().update_reviewers(
            pull_request, [(pull_request.author, ["author"], False, "reviewer", [])], pull_request.author
        )
        Session().commit()

        prs = PullRequestModel().get_awaiting_my_review(
            pull_request.target_repo.repo_name, user_id=pull_request.author.user_id
        )
        assert isinstance(prs, list)
        assert len(prs) == 1

    def test_count_awaiting_my_review(self, pull_request):
        PullRequestModel().update_reviewers(
            pull_request, [(pull_request.author, ["author"], False, "reviewer", [])], pull_request.author
        )
        Session().commit()

        pr_count = PullRequestModel().count_awaiting_my_review(
            pull_request.target_repo.repo_name, user_id=pull_request.author.user_id
        )
        assert pr_count == 1

    def test_delete_calls_cleanup_merge(self, pull_request):
        repo_id = pull_request.target_repo.repo_id
        PullRequestModel().delete(pull_request, pull_request.author)
        Session().commit()

        self.workspace_remove_mock.assert_called_once_with(repo_id, self.workspace_id)

    def test_close_calls_cleanup_and_hook(self, pull_request):
        PullRequestModel().close_pull_request(pull_request, pull_request.author)
        Session().commit()

        repo_id = pull_request.target_repo.repo_id

        self.workspace_remove_mock.assert_called_once_with(repo_id, self.workspace_id)
        self.hook_mock.assert_called_with(self.pull_request, self.pull_request.author, "close")

    def test_merge_status(self, pull_request):
        self.merge_mock.return_value = MergeResponse(True, False, None, MergeFailureReason.NONE)

        assert pull_request._last_merge_source_rev is None
        assert pull_request._last_merge_target_rev is None
        assert pull_request.last_merge_status is None

        merge_response, status, msg = PullRequestModel().merge_status(pull_request)
        assert status is True
        assert msg == "This pull request can be automatically merged."
        self.merge_mock.assert_called_with(
            self.repo_id,
            self.workspace_id,
            pull_request.target_ref_parts,
            pull_request.source_repo.scm_instance(),
            pull_request.source_ref_parts,
            dry_run=True,
            use_rebase=False,
            close_branch=False,
        )

        assert pull_request._last_merge_source_rev == self.source_commit
        assert pull_request._last_merge_target_rev == self.target_commit
        assert pull_request.last_merge_status is MergeFailureReason.NONE

        self.merge_mock.reset_mock()
        merge_response, status, msg = PullRequestModel().merge_status(pull_request)
        assert status is True
        assert msg == "This pull request can be automatically merged."
        assert self.merge_mock.called is False

    def test_merge_status_known_failure(self, pull_request):
        self.merge_mock.return_value = MergeResponse(
            False, False, None, MergeFailureReason.MERGE_FAILED, metadata={"unresolved_files": "file1"}
        )

        assert pull_request._last_merge_source_rev is None
        assert pull_request._last_merge_target_rev is None
        assert pull_request.last_merge_status is None

        merge_response, status, msg = PullRequestModel().merge_status(pull_request)
        assert status is False
        assert msg == "This pull request cannot be merged because of merge conflicts. file1"
        self.merge_mock.assert_called_with(
            self.repo_id,
            self.workspace_id,
            pull_request.target_ref_parts,
            pull_request.source_repo.scm_instance(),
            pull_request.source_ref_parts,
            dry_run=True,
            use_rebase=False,
            close_branch=False,
        )

        assert pull_request._last_merge_source_rev == self.source_commit
        assert pull_request._last_merge_target_rev == self.target_commit
        assert pull_request.last_merge_status is MergeFailureReason.MERGE_FAILED

        self.merge_mock.reset_mock()
        merge_response, status, msg = PullRequestModel().merge_status(pull_request)
        assert status is False
        assert msg == "This pull request cannot be merged because of merge conflicts. file1"
        assert self.merge_mock.called is False

    def test_merge_status_unknown_failure(self, pull_request):
        self.merge_mock.return_value = MergeResponse(
            False, False, None, MergeFailureReason.UNKNOWN, metadata={"exception": "MockError"}
        )

        assert pull_request._last_merge_source_rev is None
        assert pull_request._last_merge_target_rev is None
        assert pull_request.last_merge_status is None

        merge_response, status, msg = PullRequestModel().merge_status(pull_request)
        assert status is False
        assert msg == "This pull request cannot be merged because of an unhandled exception. MockError"
        self.merge_mock.assert_called_with(
            self.repo_id,
            self.workspace_id,
            pull_request.target_ref_parts,
            pull_request.source_repo.scm_instance(),
            pull_request.source_ref_parts,
            dry_run=True,
            use_rebase=False,
            close_branch=False,
        )

        assert pull_request._last_merge_source_rev is None
        assert pull_request._last_merge_target_rev is None
        assert pull_request.last_merge_status is None

        self.merge_mock.reset_mock()
        merge_response, status, msg = PullRequestModel().merge_status(pull_request)
        assert status is False
        assert msg == "This pull request cannot be merged because of an unhandled exception. MockError"
        assert self.merge_mock.called is True

    def test_merge_status_when_target_is_locked(self, pull_request):
        pull_request.target_repo.locked = [1, "12345.50", "lock_web"]
        merge_response, status, msg = PullRequestModel().merge_status(pull_request)
        assert status is False
        assert msg == "This pull request cannot be merged because the target repository is locked by user:1."

    def test_merge_status_requirements_check_target(self, pull_request):
        def has_largefiles(self, repo):
            return repo == pull_request.source_repo

        patcher = mock.patch.object(PullRequestModel, "_has_largefiles", has_largefiles)
        with patcher:
            merge_response, status, msg = PullRequestModel().merge_status(pull_request)

        assert status is False
        assert msg == "Target repository large files support is disabled."

    def test_merge_status_requirements_check_source(self, pull_request):
        def has_largefiles(self, repo):
            return repo == pull_request.target_repo

        patcher = mock.patch.object(PullRequestModel, "_has_largefiles", has_largefiles)
        with patcher:
            merge_response, status, msg = PullRequestModel().merge_status(pull_request)

        assert status is False
        assert msg == "Source repository large files support is disabled."

    def test_merge(self, pull_request, merge_extras):
        user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
        merge_ref = Reference("type", "name", "6126b7bfcc82ad2d3deaee22af926b082ce54cc6")
        self.merge_mock.return_value = MergeResponse(True, True, merge_ref, MergeFailureReason.NONE)

        merge_extras["repository"] = pull_request.target_repo.repo_name
        PullRequestModel().merge_repo(pull_request, pull_request.author, extras=merge_extras)
        Session().commit()

        message = "Merge pull request !{pr_id} from {source_repo} {source_ref_name}" "\n\n {pr_title}".format(
            pr_id=pull_request.pull_request_id,
            source_repo=safe_str(pull_request.source_repo.scm_instance().name),
            source_ref_name=pull_request.source_ref_parts.name,
            pr_title=safe_str(pull_request.title),
        )
        self.merge_mock.assert_called_with(
            self.repo_id,
            self.workspace_id,
            pull_request.target_ref_parts,
            pull_request.source_repo.scm_instance(),
            pull_request.source_ref_parts,
            user_name=user.short_contact,
            user_email=user.email,
            message=message,
            use_rebase=False,
            close_branch=False,
        )
        self.invalidation_mock.assert_called_once_with(pull_request.target_repo.repo_name)

        self.hook_mock.assert_called_with(self.pull_request, self.pull_request.author, "merge")

        pull_request = PullRequest.get(pull_request.pull_request_id)
        assert pull_request.merge_rev == "6126b7bfcc82ad2d3deaee22af926b082ce54cc6"

    def test_merge_with_status_lock(self, pull_request, merge_extras):
        user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
        merge_ref = Reference("type", "name", "6126b7bfcc82ad2d3deaee22af926b082ce54cc6")
        self.merge_mock.return_value = MergeResponse(True, True, merge_ref, MergeFailureReason.NONE)

        merge_extras["repository"] = pull_request.target_repo.repo_name

        with pull_request.set_state(PullRequest.STATE_UPDATING):
            assert pull_request.pull_request_state == PullRequest.STATE_UPDATING
            PullRequestModel().merge_repo(pull_request, pull_request.author, extras=merge_extras)
            Session().commit()

        assert pull_request.pull_request_state == PullRequest.STATE_CREATED

        message = "Merge pull request !{pr_id} from {source_repo} {source_ref_name}" "\n\n {pr_title}".format(
            pr_id=pull_request.pull_request_id,
            source_repo=safe_str(pull_request.source_repo.scm_instance().name),
            source_ref_name=pull_request.source_ref_parts.name,
            pr_title=safe_str(pull_request.title),
        )
        self.merge_mock.assert_called_with(
            self.repo_id,
            self.workspace_id,
            pull_request.target_ref_parts,
            pull_request.source_repo.scm_instance(),
            pull_request.source_ref_parts,
            user_name=user.short_contact,
            user_email=user.email,
            message=message,
            use_rebase=False,
            close_branch=False,
        )
        self.invalidation_mock.assert_called_once_with(pull_request.target_repo.repo_name)

        self.hook_mock.assert_called_with(self.pull_request, self.pull_request.author, "merge")

        pull_request = PullRequest.get(pull_request.pull_request_id)
        assert pull_request.merge_rev == "6126b7bfcc82ad2d3deaee22af926b082ce54cc6"

    def test_merge_failed(self, pull_request, merge_extras):
        user = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)
        merge_ref = Reference("type", "name", "6126b7bfcc82ad2d3deaee22af926b082ce54cc6")
        self.merge_mock.return_value = MergeResponse(False, False, merge_ref, MergeFailureReason.MERGE_FAILED)

        merge_extras["repository"] = pull_request.target_repo.repo_name
        PullRequestModel().merge_repo(pull_request, pull_request.author, extras=merge_extras)
        Session().commit()

        message = "Merge pull request !{pr_id} from {source_repo} {source_ref_name}" "\n\n {pr_title}".format(
            pr_id=pull_request.pull_request_id,
            source_repo=safe_str(pull_request.source_repo.scm_instance().name),
            source_ref_name=pull_request.source_ref_parts.name,
            pr_title=safe_str(pull_request.title),
        )
        self.merge_mock.assert_called_with(
            self.repo_id,
            self.workspace_id,
            pull_request.target_ref_parts,
            pull_request.source_repo.scm_instance(),
            pull_request.source_ref_parts,
            user_name=user.short_contact,
            user_email=user.email,
            message=message,
            use_rebase=False,
            close_branch=False,
        )

        pull_request = PullRequest.get(pull_request.pull_request_id)
        assert self.invalidation_mock.called is False
        assert pull_request.merge_rev is None

    def test_get_commit_ids(self, pull_request):
        # The PR has been not merged yet, so expect an exception
        with pytest.raises(ValueError):
            PullRequestModel()._get_commit_ids(pull_request)

        # Merge revision is in the revisions list
        pull_request.merge_rev = pull_request.revisions[0]
        commit_ids = PullRequestModel()._get_commit_ids(pull_request)
        assert commit_ids == pull_request.revisions

        # Merge revision is not in the revisions list
        pull_request.merge_rev = "f000" * 10
        commit_ids = PullRequestModel()._get_commit_ids(pull_request)
        assert commit_ids == pull_request.revisions + [pull_request.merge_rev]

    def test_get_diff_from_pr_version(self, pull_request):
        source_repo = pull_request.source_repo
        source_ref_id = pull_request.source_ref_parts.commit_id
        target_ref_id = pull_request.target_ref_parts.commit_id
        diff = PullRequestModel()._get_diff_from_pr_or_version(
            source_repo, source_ref_id, target_ref_id, hide_whitespace_changes=False, diff_context=6
        )
        assert b"file_1" in diff.raw.tobytes()

    def test_generate_title_returns_unicode(self):
        title = PullRequestModel().generate_pullrequest_title(
            source="source-dummy",
            source_ref="source-ref-dummy",
            target="target-dummy",
        )
        assert type(title) == str

    @pytest.mark.parametrize(
        "title, has_wip",
        [
            ("hello", False),
            ("hello wip", False),
            ("hello wip: xxx", False),
            ("[wip] hello", True),
            ("[wip] hello", True),
            ("wip: hello", True),
            ("wip hello", True),
        ],
    )
    def test_wip_title_marker(self, pull_request, title, has_wip):
        pull_request.title = title
        assert pull_request.work_in_progress == has_wip


@pytest.mark.usefixtures("config_stub")
class TestIntegrationMerge(object):

    def test_merge_fails_if_target_is_locked(self, pr_util, user_regular, merge_extras):
        pull_request = pr_util.create_pull_request(approved=True, mergeable=True)
        locked_by = [user_regular.user_id + 1, 12345.50, "lock_web"]
        pull_request.target_repo.locked = locked_by
        # TODO: johbo: Check if this can work based on the database, currently
        # all data is pre-computed, that's why just updating the DB is not
        # enough.
        merge_extras["locked_by"] = locked_by
        merge_extras["repository"] = pull_request.target_repo.repo_name
        # TODO: johbo: Needed for sqlite, try to find an automatic way for it
        Session().commit()
        merge_status = PullRequestModel().merge_repo(pull_request, user_regular, extras=merge_extras)
        Session().commit()

        assert not merge_status.executed


@pytest.mark.parametrize(
    "use_outdated, inlines_count, outdated_count",
    [
        (False, 1, 0),
        (True, 0, 1),
    ],
)
def test_outdated_comments(pr_util, use_outdated, inlines_count, outdated_count, config_stub):
    pull_request = pr_util.create_pull_request()
    pr_util.create_inline_comment(file_path="not_in_updated_diff")

    with outdated_comments_patcher(use_outdated) as outdated_comment_mock:
        pr_util.add_one_commit()
        assert_inline_comments(pull_request, visible=inlines_count, outdated=outdated_count)
    outdated_comment_mock.assert_called_with(pull_request)


@pytest.mark.parametrize(
    "mr_type, expected_msg",
    [
        (MergeFailureReason.NONE, "This pull request can be automatically merged."),
        (MergeFailureReason.UNKNOWN, "This pull request cannot be merged because of an unhandled exception. CRASH"),
        (
            MergeFailureReason.MERGE_FAILED,
            "This pull request cannot be merged because of merge conflicts. CONFLICT_FILE",
        ),
        (
            MergeFailureReason.PUSH_FAILED,
            "This pull request could not be merged because push to target:`some-repo@merge_commit` failed.",
        ),
        (
            MergeFailureReason.TARGET_IS_NOT_HEAD,
            "This pull request cannot be merged because the target `ref_name` is not a head.",
        ),
        (
            MergeFailureReason.HG_SOURCE_HAS_MORE_BRANCHES,
            "This pull request cannot be merged because the source contains more branches than the target.",
        ),
        (
            MergeFailureReason.HG_TARGET_HAS_MULTIPLE_HEADS,
            "This pull request cannot be merged because the target `ref_name` has multiple heads: `a,b,c`.",
        ),
        (
            MergeFailureReason.TARGET_IS_LOCKED,
            "This pull request cannot be merged because the target repository is locked by user:123.",
        ),
        (
            MergeFailureReason.MISSING_TARGET_REF,
            "This pull request cannot be merged because the target reference `ref_name` is missing.",
        ),
        (
            MergeFailureReason.MISSING_SOURCE_REF,
            "This pull request cannot be merged because the source reference `ref_name` is missing.",
        ),
        (
            MergeFailureReason.SUBREPO_MERGE_FAILED,
            "This pull request cannot be merged because of conflicts related to sub repositories.",
        ),
    ],
)
def test_merge_response_message(mr_type, expected_msg):
    merge_ref = Reference("type", "ref_name", "6126b7bfcc82ad2d3deaee22af926b082ce54cc6")
    metadata = {
        "unresolved_files": "CONFLICT_FILE",
        "exception": "CRASH",
        "target": "some-repo",
        "merge_commit": "merge_commit",
        "target_ref": merge_ref,
        "source_ref": merge_ref,
        "heads": ",".join(["a", "b", "c"]),
        "locked_by": "user:123",
    }

    merge_response = MergeResponse(True, True, merge_ref, mr_type, metadata=metadata)
    assert merge_response.merge_status_message == expected_msg


@pytest.fixture()
def merge_extras(request, user_regular):
    """
    Context for the vcs operation when running a merge.
    """

    extras = {
        "ip": "127.0.0.1",
        "username": user_regular.username,
        "user_id": user_regular.user_id,
        "action": "push",
        "repository": "fake_target_repo_name",
        "scm": "git",
        "config": request.config.getini("pyramid_config"),
        "repo_store": "",
        "make_lock": None,
        "locked_by": [None, None, None],
        "server_url": "http://test.example.com:5000",
        "hooks": ["push", "pull"],
        "is_shadow_repo": False,
    }
    return extras


@pytest.mark.usefixtures("config_stub")
class TestUpdateCommentHandling(object):
    @pytest.fixture(autouse=True, scope="class")
    def enable_outdated_comments(self, request, baseapp):
        config_patch = mock.patch.dict("rhodecode.CONFIG", {"rhodecode_use_outdated_comments": True})
        config_patch.start()

        @request.addfinalizer
        def cleanup():
            config_patch.stop()

    def test_comment_stays_unflagged_on_unchanged_diff(self, pr_util):
        commits = [
            {"message": "a"},
            {"message": "b", "added": [FileNode(b"file_b", b"test_content\n")]},
            {"message": "c", "added": [FileNode(b"file_c", b"test_content\n")]},
        ]
        pull_request = pr_util.create_pull_request(commits=commits, target_head="a", source_head="b", revisions=["b"])
        pr_util.create_inline_comment(file_path="file_b")
        pr_util.add_one_commit(head="c")

        assert_inline_comments(pull_request, visible=1, outdated=0)

    def test_comment_stays_unflagged_on_change_above(self, pr_util):
        original_content = b"".join((b"line %d\n" % x for x in range(1, 11)))
        updated_content = b"new_line_at_top\n" + original_content
        commits = [
            {"message": "a"},
            {"message": "b", "added": [FileNode(b"file_b", original_content)]},
            {"message": "c", "changed": [FileNode(b"file_b", updated_content)]},
        ]
        pull_request = pr_util.create_pull_request(commits=commits, target_head="a", source_head="b", revisions=["b"])

        with outdated_comments_patcher():
            comment = pr_util.create_inline_comment(line_no="n8", file_path="file_b")
            pr_util.add_one_commit(head="c")

        assert_inline_comments(pull_request, visible=1, outdated=0)
        assert comment.line_no == "n9"

    def test_comment_stays_unflagged_on_change_below(self, pr_util):
        original_content = b"".join([b"line %d\n" % x for x in range(10)])
        updated_content = original_content + b"new_line_at_end\n"
        commits = [
            {"message": "a"},
            {"message": "b", "added": [FileNode(b"file_b", original_content)]},
            {"message": "c", "changed": [FileNode(b"file_b", updated_content)]},
        ]
        pull_request = pr_util.create_pull_request(commits=commits, target_head="a", source_head="b", revisions=["b"])
        pr_util.create_inline_comment(file_path="file_b")
        pr_util.add_one_commit(head="c")

        assert_inline_comments(pull_request, visible=1, outdated=0)

    @pytest.mark.parametrize("line_no", ["n4", "o4", "n10", "o9"])
    def test_comment_flagged_on_change_around_context(self, pr_util, line_no):
        base_lines = [b"line %d\n" % x for x in range(1, 13)]
        change_lines = list(base_lines)
        change_lines.insert(6, b"line 6a added\n")

        # Changes on the last line of sight
        update_lines = list(change_lines)
        update_lines[0] = b"line 1 changed\n"
        update_lines[-1] = b"line 12 changed\n"

        def file_b(lines):
            return FileNode(b"file_b", b"".join(lines))

        commits = [
            {"message": "a", "added": [file_b(base_lines)]},
            {"message": "b", "changed": [file_b(change_lines)]},
            {"message": "c", "changed": [file_b(update_lines)]},
        ]

        pull_request = pr_util.create_pull_request(commits=commits, target_head="a", source_head="b", revisions=["b"])
        pr_util.create_inline_comment(line_no=line_no, file_path="file_b")

        with outdated_comments_patcher():
            pr_util.add_one_commit(head="c")
            assert_inline_comments(pull_request, visible=0, outdated=1)

    @pytest.mark.parametrize(
        "change, content",
        [
            ("changed", b"changed\n"),
            ("removed", b""),
        ],
        ids=["changed", b"removed"],
    )
    def test_comment_flagged_on_change(self, pr_util, change, content):
        commits = [
            {"message": "a"},
            {"message": "b", "added": [FileNode(b"file_b", b"test_content\n")]},
            {"message": "c", change: [FileNode(b"file_b", content)]},
        ]
        pull_request = pr_util.create_pull_request(commits=commits, target_head="a", source_head="b", revisions=["b"])
        pr_util.create_inline_comment(file_path="file_b")

        with outdated_comments_patcher():
            pr_util.add_one_commit(head="c")
            assert_inline_comments(pull_request, visible=0, outdated=1)


@pytest.mark.usefixtures("config_stub")
class TestUpdateChangedFiles(object):
    def test_no_changes_on_unchanged_diff(self, pr_util):
        commits = [
            {"message": "a"},
            {"message": "b", "added": [FileNode(b"file_b", b"test_content b\n")]},
            {"message": "c", "added": [FileNode(b"file_c", b"test_content c\n")]},
        ]
        # open a PR from a to b, adding file_b
        pull_request = pr_util.create_pull_request(
            commits=commits, target_head="a", source_head="b", revisions=["b"], name_suffix="per-file-review"
        )

        # modify PR adding new file file_c
        pr_util.add_one_commit(head="c")

        assert_pr_file_changes(pull_request, added=["file_c"], modified=[], removed=[])

    def test_modify_and_undo_modification_diff(self, pr_util):
        commits = [
            {"message": "a"},
            {"message": "b", "added": [FileNode(b"file_b", b"test_content b\n")]},
            {"message": "c", "changed": [FileNode(b"file_b", b"test_content b modified\n")]},
            {"message": "d", "changed": [FileNode(b"file_b", b"test_content b\n")]},
        ]
        # open a PR from a to b, adding file_b
        pull_request = pr_util.create_pull_request(
            commits=commits, target_head="a", source_head="b", revisions=["b"], name_suffix="per-file-review"
        )

        # modify PR modifying file file_b
        pr_util.add_one_commit(head="c")

        assert_pr_file_changes(pull_request, added=[], modified=["file_b"], removed=[])

        # move the head again to d, which rollbacks change,
        # meaning we should indicate no changes
        pr_util.add_one_commit(head="d")

        assert_pr_file_changes(pull_request, added=[], modified=[], removed=[])

    def test_updated_all_files_in_pr(self, pr_util):
        commits = [
            {"message": "a"},
            {
                "message": "b",
                "added": [
                    FileNode(b"file_a", b"test_content a\n"),
                    FileNode(b"file_b", b"test_content b\n"),
                    FileNode(b"file_c", b"test_content c\n"),
                ],
            },
            {
                "message": "c",
                "changed": [
                    FileNode(b"file_a", b"test_content a changed\n"),
                    FileNode(b"file_b", b"test_content b changed\n"),
                    FileNode(b"file_c", b"test_content c changed\n"),
                ],
            },
        ]
        # open a PR from a to b, changing 3 files
        pull_request = pr_util.create_pull_request(
            commits=commits, target_head="a", source_head="b", revisions=["b"], name_suffix="per-file-review"
        )

        pr_util.add_one_commit(head="c")

        assert_pr_file_changes(pull_request, added=[], modified=["file_a", "file_b", "file_c"], removed=[])

    def test_updated_and_removed_all_files_in_pr(self, pr_util):
        commits = [
            {"message": "a"},
            {
                "message": "b",
                "added": [
                    FileNode(b"file_a", b"test_content a\n"),
                    FileNode(b"file_b", b"test_content b\n"),
                    FileNode(b"file_c", b"test_content c\n"),
                ],
            },
            {
                "message": "c",
                "removed": [
                    FileNode(b"file_a", b"test_content a changed\n"),
                    FileNode(b"file_b", b"test_content b changed\n"),
                    FileNode(b"file_c", b"test_content c changed\n"),
                ],
            },
        ]
        # open a PR from a to b, removing 3 files
        pull_request = pr_util.create_pull_request(
            commits=commits, target_head="a", source_head="b", revisions=["b"], name_suffix="per-file-review"
        )

        pr_util.add_one_commit(head="c")

        assert_pr_file_changes(pull_request, added=[], modified=[], removed=["file_a", "file_b", "file_c"])


def test_update_writes_snapshot_into_pull_request_version(pr_util, config_stub):
    model = PullRequestModel()
    pull_request = pr_util.create_pull_request()
    pr_util.update_source_repository()

    model.update_commits(pull_request, pull_request.author)

    # Expect that it has a version entry now
    assert len(model.get_versions(pull_request)) == 1


def test_update_skips_new_version_if_unchanged(pr_util, config_stub):
    pull_request = pr_util.create_pull_request()
    model = PullRequestModel()
    model.update_commits(pull_request, pull_request.author)

    # Expect that it still has no versions
    assert len(model.get_versions(pull_request)) == 0


def test_update_assigns_comments_to_the_new_version(pr_util, config_stub):
    model = PullRequestModel()
    pull_request = pr_util.create_pull_request()
    comment = pr_util.create_comment()
    pr_util.update_source_repository()

    model.update_commits(pull_request, pull_request.author)

    # Expect that the comment is linked to the pr version now
    assert comment.pull_request_version == model.get_versions(pull_request)[0]


def test_update_adds_a_comment_to_the_pull_request_about_the_change(pr_util, config_stub):
    model = PullRequestModel()
    pull_request = pr_util.create_pull_request()
    pr_util.update_source_repository()
    pr_util.update_source_repository()

    update_response = model.update_commits(pull_request, pull_request.author)

    commit_id = update_response.common_ancestor_id
    # Expect to find a new comment about the change
    expected_message = textwrap.dedent(
        """\
        Pull request updated. Auto status change to |under_review|

        .. role:: added
        .. role:: removed
        .. parsed-literal::

          Changed commits:
            * :added:`1 added`
            * :removed:`0 removed`

          Changed files:
            * `A file_2 <#a_c-{}-92ed3b5f07b4>`_

        .. |under_review| replace:: *"Under Review"*"""
    ).format(commit_id[:12])
    pull_request_comments = sorted(pull_request.comments, key=lambda c: c.modified_at)
    update_comment = pull_request_comments[-1]
    assert update_comment.text == expected_message


def test_create_version_from_snapshot_updates_attributes(pr_util, config_stub):
    pull_request = pr_util.create_pull_request()

    # Avoiding default values
    pull_request.status = PullRequest.STATUS_CLOSED
    pull_request._last_merge_source_rev = "0" * 40
    pull_request._last_merge_target_rev = "1" * 40
    pull_request.last_merge_status = 1
    pull_request.merge_rev = "2" * 40

    # Remember automatic values
    created_on = pull_request.created_on
    updated_on = pull_request.updated_on

    # Create a new version of the pull request
    version = PullRequestModel()._create_version_from_snapshot(pull_request)

    # Check attributes
    assert version.title == pr_util.create_parameters["title"]
    assert version.description == pr_util.create_parameters["description"]
    assert version.status == PullRequest.STATUS_CLOSED

    # versions get updated created_on
    assert version.created_on != created_on

    assert version.updated_on == updated_on
    assert version.user_id == pull_request.user_id
    assert version.revisions == pr_util.create_parameters["revisions"]
    assert version.source_repo == pr_util.source_repository
    assert version.source_ref == pr_util.create_parameters["source_ref"]
    assert version.target_repo == pr_util.target_repository
    assert version.target_ref == pr_util.create_parameters["target_ref"]
    assert version._last_merge_source_rev == pull_request._last_merge_source_rev
    assert version._last_merge_target_rev == pull_request._last_merge_target_rev
    assert version.last_merge_status == pull_request.last_merge_status
    assert version.merge_rev == pull_request.merge_rev
    assert version.pull_request == pull_request


def test_link_comments_to_version_only_updates_unlinked_comments(pr_util, config_stub):
    version1 = pr_util.create_version_of_pull_request()
    comment_linked = pr_util.create_comment(linked_to=version1)
    comment_unlinked = pr_util.create_comment()
    version2 = pr_util.create_version_of_pull_request()

    PullRequestModel()._link_comments_to_version(version2)
    Session().commit()

    # Expect that only the new comment is linked to version2
    assert comment_unlinked.pull_request_version_id == version2.pull_request_version_id
    assert comment_linked.pull_request_version_id == version1.pull_request_version_id
    assert comment_unlinked.pull_request_version_id != comment_linked.pull_request_version_id


def test_calculate_commits():
    old_ids = [1, 2, 3]
    new_ids = [1, 3, 4, 5]
    change = PullRequestModel()._calculate_commit_id_changes(old_ids, new_ids)
    assert change.added == [4, 5]
    assert change.common == [1, 3]
    assert change.removed == [2]
    assert change.total == [1, 3, 4, 5]


def assert_inline_comments(pull_request, visible=None, outdated=None):
    if visible is not None:
        inline_comments = CommentsModel().get_inline_comments(
            pull_request.target_repo.repo_id, pull_request=pull_request
        )
        inline_cnt = len(CommentsModel().get_inline_comments_as_list(inline_comments))
        assert inline_cnt == visible
    if outdated is not None:
        outdated_comments = CommentsModel().get_outdated_comments(pull_request.target_repo.repo_id, pull_request)
        assert len(outdated_comments) == outdated


def assert_pr_file_changes(pull_request, added=None, modified=None, removed=None):
    pr_versions = PullRequestModel().get_versions(pull_request)
    # always use first version, ie original PR to calculate changes
    pull_request_version = pr_versions[0]
    old_diff_data, new_diff_data = PullRequestModel()._generate_update_diffs(pull_request, pull_request_version)
    file_changes = PullRequestModel()._calculate_file_changes(old_diff_data, new_diff_data)

    assert added == file_changes.added, "expected added:%s vs value:%s" % (added, file_changes.added)
    assert modified == file_changes.modified, "expected modified:%s vs value:%s" % (modified, file_changes.modified)
    assert removed == file_changes.removed, "expected removed:%s vs value:%s" % (removed, file_changes.removed)


def outdated_comments_patcher(use_outdated=True):
    return mock.patch.object(CommentsModel, "use_outdated_comments", return_value=use_outdated)