# 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 . # # 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 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) assert 'pushed to' in call.call_args[0][1].title # specific commit was parsed and serialized assert 'change that fixes #41' in call.call_args[0][1].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)