unrhodecode/rhodecode/tests/integrations/test_slack.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 pytest
import mock
from mock import patch

from rhodecode import events
from rhodecode.integrations.types.handlers.slack import SlackDataHandler, SlackData
from rhodecode.model.db import Session, Integration
from rhodecode.integrations.types.slack import SlackIntegrationType
from rhodecode.tests import GIT_REPO


@pytest.fixture()
def base_slack_data():
    return {
        "pullrequest": {
            "url": "https://example.com/pr1",
            "pull_request_id": "1",
            "title": "started pr",
            "status": "new",
            "commits": ["1", "2"],
            "shadow_url": "http://shadow-url",
        },
        "actor": {"username": "foo-user"},
        "comment": {
            "comment_id": 1,
            "text": "test-comment",
            "status": "approved",
            "file": "text.py",
            "line": "1",
            "type": "note",
        },
        "push": {"branches": "", "commits": []},
        "repo": {
            "url": "https://example.com/repo1",
            "repo_name": GIT_REPO,
            "repo_type": "git",
        },
    }


@pytest.fixture()
def slack_settings():
    return {
        "service": "mock://slackintegration",
        "events": [
            "pullrequest-create",
            "repo-push",
        ],
        "channel": "#testing",
        "icon_emoji": ":recycle:",
        "username": "rhodecode-test",
    }


@pytest.fixture()
def slack_integration(request, app, slack_settings):
    integration = Integration()
    integration.name = "test slack integration"
    integration.enabled = True
    integration.integration_type = SlackIntegrationType.key
    integration.settings = slack_settings
    Session().add(integration)
    Session().commit()
    request.addfinalizer(lambda: Session().delete(integration))
    return integration


@pytest.fixture()
def slack_integration_empty(request, app, slack_settings):
    slack_settings["events"] = []
    integration = Integration()
    integration.name = "test slack integration"
    integration.enabled = True
    integration.integration_type = SlackIntegrationType.key
    integration.settings = slack_settings
    Session().add(integration)
    Session().commit()
    request.addfinalizer(lambda: Session().delete(integration))
    return integration


def test_slack_push(slack_integration, repo_push_event):
    with patch("rhodecode.integrations.types.slack.post_text_to_slack") as call:
        events.trigger(repo_push_event)
    # make sure we can handle both dicts and dataclass objects
    slack_data = call.call_args[0][1]
    if isinstance(slack_data, dict):
        slack_data = SlackData(**slack_data)

    assert "pushed to" in slack_data.title
    # specific commit was parsed and serialized
    assert "change that fixes #41" in slack_data.text


def test_slack_push_no_events(slack_integration_empty, repo_push_event):
    assert Integration.get(slack_integration_empty.integration_id).settings["events"] == []

    with patch("rhodecode.integrations.types.slack.post_text_to_slack") as call:
        events.trigger(repo_push_event)
    assert not call.call_args


def test_slack_data_handler_wrong_event():
    handler = SlackDataHandler()
    data = {"actor": {"username": "foo-user"}}
    with pytest.raises(ValueError):
        handler(events.RhodecodeEvent(), data)


@pytest.mark.parametrize(
    "event_type, args",
    [
        (
            events.PullRequestCommentEvent,
            (mock.MagicMock(name="pull-request"), mock.MagicMock(name="comment")),
        ),
        (
            events.PullRequestCommentEditEvent,
            (mock.MagicMock(name="pull-request"), mock.MagicMock(name="comment")),
        ),
        (
            events.PullRequestReviewEvent,
            (mock.MagicMock(name="pull-request"), mock.MagicMock(name="status")),
        ),
        (
            events.RepoPushEvent,
            (GIT_REPO, mock.MagicMock(name="pushed_commit_ids"), mock.MagicMock(name="extras")),
        ),
        (events.PullRequestEvent, (mock.MagicMock(),)),
        (events.RepoCreateEvent, (mock.MagicMock(),)),
    ],
)
def test_slack_data_handler(app, event_type: events.RhodecodeEvent, args, base_slack_data):
    handler = SlackDataHandler()
    handler(event_type(*args), base_slack_data)


def test_slack_pull_request_branch_info_display(app, base_slack_data):
    """Test that branch information is correctly displayed in slack notifications"""
    handler = SlackDataHandler()

    mock_pr = mock.MagicMock()
    mock_pr.source_ref_parts.name = "feature/login-improvements"
    mock_pr.target_ref_parts.name = "main"

    mock_event = mock.MagicMock()
    mock_event.__class__ = events.PullRequestCreateEvent
    mock_event.pullrequest = mock_pr

    slack_data = SlackData(title="default", text="default")

    result = handler.format_pull_request_event(mock_event, base_slack_data, slack_data)
    assert "`feature/login-improvements` → `main`" in result.text


@pytest.mark.parametrize(
    "expected_class",
    [
        events.PullRequestCreateEvent,
        events.PullRequestUpdateEvent,
        events.PullRequestMergeEvent,
        events.PullRequestCloseEvent,
    ],
)
def test_slack_pull_request_different_actions(app, base_slack_data, expected_class):
    handler = SlackDataHandler()

    mock_pr = mock.MagicMock()
    mock_pr.source_ref_parts.name = "bugfix/JIRA-456"
    mock_pr.target_ref_parts.name = "develop"

    slack_data = SlackData(title="default", text="default")
    branch_info = "`bugfix/JIRA-456` → `develop`"

    mock_event = mock.MagicMock()
    mock_event.__class__ = expected_class
    mock_event.pullrequest = mock_pr

    result = handler.format_pull_request_event(mock_event, base_slack_data, slack_data)
    assert branch_info in result.text, f"Branch info missing for {expected_class.__name__}"