community.general/tests/unit/plugins/modules/test_slack.py

# Copyright (c) Ansible project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later

from __future__ import annotations

import json
from unittest.mock import Mock, mock_open, patch

import pytest
from ansible_collections.community.internal_test_tools.tests.unit.plugins.modules.utils import (
    AnsibleExitJson,
    AnsibleFailJson,
    ModuleTestCase,
    set_module_args,
)

from ansible_collections.community.general.plugins.modules import slack


class TestSlackModule(ModuleTestCase):
    def setUp(self):
        super().setUp()
        self.module = slack

    def tearDown(self):
        super().tearDown()

    @pytest.fixture
    def fetch_url_mock(self, mocker):
        return mocker.patch("ansible.module_utils.notification.slack.fetch_url")

    def test_without_required_parameters(self):
        """Failure must occurs when all parameters are missing"""
        with self.assertRaises(AnsibleFailJson):
            with set_module_args({}):
                self.module.main()

    def test_invalid_old_token(self):
        """Failure if there is an old style token"""
        with set_module_args(
            {
                "token": "test",
            }
        ):
            with self.assertRaises(AnsibleFailJson):
                self.module.main()

    def test_successful_message(self):
        """tests sending a message. This is example 1 from the docs"""
        with set_module_args({"token": "TXX/BYY/ZZZ", "msg": "test"}):
            with patch.object(slack, "fetch_url") as fetch_url_mock:
                fetch_url_mock.return_value = (None, {"status": 200})
                with self.assertRaises(AnsibleExitJson):
                    self.module.main()

        self.assertTrue(fetch_url_mock.call_count, 1)
        call_data = json.loads(fetch_url_mock.call_args[1]["data"])
        assert call_data["username"] == "Ansible"
        assert call_data["text"] == "test"
        assert fetch_url_mock.call_args[1]["url"] == "https://hooks.slack.com/services/TXX/BYY/ZZZ"

    def test_failed_message(self):
        """tests failing to send a message"""

        with set_module_args({"token": "TXX/BYY/ZZZ", "msg": "test"}):
            with patch.object(slack, "fetch_url") as fetch_url_mock:
                fetch_url_mock.return_value = (None, {"status": 404, "msg": "test"})
                with self.assertRaises(AnsibleFailJson):
                    self.module.main()

    def test_message_with_thread(self):
        """tests sending a message with a thread"""
        with set_module_args({"token": "TXX/BYY/ZZZ", "msg": "test", "thread_id": "100.00"}):
            with patch.object(slack, "fetch_url") as fetch_url_mock:
                fetch_url_mock.return_value = (None, {"status": 200})
                with self.assertRaises(AnsibleExitJson):
                    self.module.main()

        self.assertTrue(fetch_url_mock.call_count, 1)
        call_data = json.loads(fetch_url_mock.call_args[1]["data"])
        assert call_data["username"] == "Ansible"
        assert call_data["text"] == "test"
        assert call_data["thread_ts"] == "100.00"
        assert fetch_url_mock.call_args[1]["url"] == "https://hooks.slack.com/services/TXX/BYY/ZZZ"

    # https://github.com/ansible-collections/community.general/issues/1097
    def test_ts_in_message_does_not_cause_edit(self):
        with set_module_args({"token": "xoxa-123456789abcdef", "msg": "test with ts"}):
            with patch.object(slack, "fetch_url") as fetch_url_mock:
                mock_response = Mock()
                mock_response.read.return_value = '{"ok": true, "fake":"data"}'
                fetch_url_mock.return_value = (mock_response, {"status": 200})
                with self.assertRaises(AnsibleExitJson):
                    self.module.main()

        self.assertTrue(fetch_url_mock.call_count, 1)
        self.assertEqual(fetch_url_mock.call_args[1]["url"], "https://slack.com/api/chat.postMessage")

    def test_govslack_message(self):
        with set_module_args({"token": "xoxa-123456789abcdef", "domain": "slack-gov.com", "msg": "test with ts"}):
            with patch.object(slack, "fetch_url") as fetch_url_mock:
                mock_response = Mock()
                mock_response.read.return_value = '{"ok": true, "fake":"data"}'
                fetch_url_mock.return_value = (mock_response, {"status": 200})
                with self.assertRaises(AnsibleExitJson):
                    self.module.main()

        self.assertTrue(fetch_url_mock.call_count, 1)
        self.assertEqual(fetch_url_mock.call_args[1]["url"], "https://slack-gov.com/api/chat.postMessage")

    def test_edit_message(self):
        with set_module_args({"token": "xoxa-123456789abcdef", "msg": "test2", "message_id": "12345"}):
            with patch.object(slack, "fetch_url") as fetch_url_mock:
                mock_response = Mock()
                mock_response.read.return_value = '{"ok": true, "messages":[{"ts":"12345","msg":"test1"}]}'
                fetch_url_mock.side_effect = [
                    (mock_response, {"status": 200}),
                    (mock_response, {"status": 200}),
                ]
                with self.assertRaises(AnsibleExitJson):
                    self.module.main()

        self.assertTrue(fetch_url_mock.call_count, 2)
        self.assertEqual(fetch_url_mock.call_args[1]["url"], "https://slack.com/api/chat.update")
        call_data = json.loads(fetch_url_mock.call_args[1]["data"])
        self.assertEqual(call_data["ts"], "12345")

    def test_message_with_blocks(self):
        """tests sending a message with blocks"""
        with set_module_args(
            {
                "token": "TXX/BYY/ZZZ",
                "msg": "test",
                "blocks": [
                    {
                        "type": "section",
                        "text": {"type": "mrkdwn", "text": "*test*"},
                        "accessory": {
                            "type": "image",
                            "image_url": "https://docs.ansible.com/favicon.ico",
                            "alt_text": "test",
                        },
                    },
                    {"type": "section", "text": {"type": "plain_text", "text": "test", "emoji": True}},
                ],
            }
        ):
            with patch.object(slack, "fetch_url") as fetch_url_mock:
                fetch_url_mock.return_value = (None, {"status": 200})
                with self.assertRaises(AnsibleExitJson):
                    self.module.main()

        self.assertTrue(fetch_url_mock.call_count, 1)
        call_data = json.loads(fetch_url_mock.call_args[1]["data"])
        assert call_data["username"] == "Ansible"
        assert call_data["blocks"][1]["text"]["text"] == "test"
        assert fetch_url_mock.call_args[1]["url"] == "https://hooks.slack.com/services/TXX/BYY/ZZZ"

    def test_message_with_invalid_color(self):
        """tests sending invalid color value to module"""
        with set_module_args(
            {
                "token": "TXX/BYY/ZZZ",
                "msg": "test",
                "color": "aa",
            }
        ):
            with self.assertRaises(AnsibleFailJson) as exec_info:
                self.module.main()

        msg = (
            "Color value specified should be either one of"
            " ['normal', 'good', 'warning', 'danger'] or any valid"
            " hex value with length 3 or 6."
        )
        assert exec_info.exception.args[0]["msg"] == msg

    def test_upload_files_only(self):
        with set_module_args(
            {"token": "xoxb-12345", "channel": "C123", "files": [{"path": "/tmp/test.txt", "name": "hello.txt"}]}
        ):
            with patch.object(slack, "fetch_url") as fetch_url_mock:
                with patch("os.path.exists", return_value=True):
                    with patch("os.path.getsize", return_value=100):
                        with patch("builtins.open", mock_open(read_data=b"data")):
                            mock_resp = Mock()
                            mock_resp.read.side_effect = [
                                '{"ok": true, "upload_url": "https://upload", "file_id": "F1"}',
                                '{"ok": true}',
                            ]
                            fetch_url_mock.return_value = (mock_resp, {"status": 200})
                            with self.assertRaises(AnsibleExitJson) as result:
                                self.module.main()
                            self.assertTrue(result.exception.args[0]["changed"])


color_test = [
    ("#111111", True),
    ("#00aabb", True),
    ("#abc", True),
    ("#gghhjj", False),
    ("#ghj", False),
    ("#a", False),
    ("#aaaaaaaa", False),
    ("", False),
    ("aaaa", False),
    ("$00aabb", False),
    ("$00a", False),
]


@pytest.mark.parametrize("color_value, ret_status", color_test)
def test_is_valid_hex_color(color_value, ret_status):
    generated_value = slack.is_valid_hex_color(color_value)
    assert generated_value == ret_status