python-zulip-api/zulip/integrations/bridge_with_matrix/test_matrix.py

from matrix_bridge import (
    check_zulip_message_validity,
    zulip_to_matrix,
)

from unittest import TestCase, mock
from subprocess import Popen, PIPE
import os

import shutil

from contextlib import contextmanager
from tempfile import mkdtemp

script_file = "matrix_bridge.py"
script_dir = os.path.dirname(__file__)
script = os.path.join(script_dir, script_file)

from typing import List, Iterator

sample_config_path = "matrix_test.conf"

sample_config_text = """[matrix]
host = https://matrix.org
username = username
password = password
room_id = #zulip:matrix.org

[zulip]
email = glitch-bot@chat.zulip.org
api_key = aPiKeY
site = https://chat.zulip.org
stream = test here
topic = matrix

"""

@contextmanager
def new_temp_dir():
    # type: () -> Iterator[str]
    path = mkdtemp()
    yield path
    shutil.rmtree(path)

class MatrixBridgeScriptTests(TestCase):
    def output_from_script(self, options):
        # type: (List[str]) -> List[str]
        popen = Popen(["python", script] + options, stdin=PIPE, stdout=PIPE, universal_newlines=True)
        return popen.communicate()[0].strip().split("\n")

    def test_no_args(self):
        # type: () -> None
        output_lines = self.output_from_script([])
        expected_lines = [
            "Options required: -c or --config to run, OR --write-sample-config.",
            "usage: {} [-h]".format(script_file)
        ]
        for expected, output in zip(expected_lines, output_lines):
            self.assertIn(expected, output)

    def test_help_usage_and_description(self):
        # type: () -> None
        output_lines = self.output_from_script(["-h"])
        usage = "usage: {} [-h]".format(script_file)
        description = "Script to bridge"
        self.assertIn(usage, output_lines[0])
        blank_lines = [num for num, line in enumerate(output_lines) if line == '']
        # There should be blank lines in the output
        self.assertTrue(blank_lines)
        # There should be finite output
        self.assertTrue(len(output_lines) > blank_lines[0])
        # Minimal description should be in the first line of the 2nd "paragraph"
        self.assertIn(description, output_lines[blank_lines[0] + 1])

    def test_write_sample_config(self):
        # type: () -> None
        with new_temp_dir() as tempdir:
            path = os.path.join(tempdir, sample_config_path)
            output_lines = self.output_from_script(["--write-sample-config", path])
            self.assertEqual(output_lines, ["Wrote sample configuration to '{}'".format(path)])

            with open(path) as sample_file:
                self.assertEqual(sample_file.read(), sample_config_text)

    def test_write_sample_config_from_zuliprc(self):
        # type: () -> None
        zuliprc_template = ["[api]", "email={email}", "key={key}", "site={site}"]
        zulip_params = {'email': 'foo@bar',
                        'key': 'some_api_key',
                        'site': 'https://some.chat.serverplace'}
        with new_temp_dir() as tempdir:
            path = os.path.join(tempdir, sample_config_path)
            zuliprc_path = os.path.join(tempdir, "zuliprc")
            with open(zuliprc_path, "w") as zuliprc_file:
                zuliprc_file.write("\n".join(zuliprc_template).format(**zulip_params))
            output_lines = self.output_from_script(["--write-sample-config", path,
                                                    "--from-zuliprc", zuliprc_path])
            self.assertEqual(output_lines,
                             ["Wrote sample configuration to '{}' using zuliprc file '{}'"
                              .format(path, zuliprc_path)])

            with open(path) as sample_file:
                sample_lines = [line.strip() for line in sample_file.readlines()]
                expected_lines = sample_config_text.split("\n")
                expected_lines[7] = 'email = {}'.format(zulip_params['email'])
                expected_lines[8] = 'api_key = {}'.format(zulip_params['key'])
                expected_lines[9] = 'site = {}'.format(zulip_params['site'])
                self.assertEqual(sample_lines, expected_lines[:-1])

    def test_detect_zuliprc_does_not_exist(self):
        # type: () -> None
        with new_temp_dir() as tempdir:
            path = os.path.join(tempdir, sample_config_path)
            zuliprc_path = os.path.join(tempdir, "zuliprc")
            # No writing of zuliprc file here -> triggers check for zuliprc absence
            output_lines = self.output_from_script(["--write-sample-config", path,
                                                    "--from-zuliprc", zuliprc_path])
            self.assertEqual(output_lines,
                             ["Could not write sample config: Zuliprc file '{}' does not exist."
                              .format(zuliprc_path)])

class MatrixBridgeZulipToMatrixTests(TestCase):
    valid_zulip_config = dict(
        stream="some stream",
        topic="some topic",
        email="some@email"
    )
    valid_msg = dict(
        sender_email="John@Smith.smith",  # must not be equal to config:email
        type="stream",  # Can only mirror Zulip streams
        display_recipient=valid_zulip_config['stream'],
        subject=valid_zulip_config['topic']
    )

    def test_zulip_message_validity_success(self):
        # type: () -> None
        zulip_config = self.valid_zulip_config
        msg = self.valid_msg
        # Ensure the test inputs are valid for success
        assert msg['sender_email'] != zulip_config['email']

        self.assertTrue(check_zulip_message_validity(msg, zulip_config))

    def test_zulip_message_validity_failure(self):
        # type: () -> None
        zulip_config = self.valid_zulip_config

        msg_wrong_stream = dict(self.valid_msg, display_recipient='foo')
        self.assertFalse(check_zulip_message_validity(msg_wrong_stream, zulip_config))

        msg_wrong_topic = dict(self.valid_msg, subject='foo')
        self.assertFalse(check_zulip_message_validity(msg_wrong_topic, zulip_config))

        msg_not_stream = dict(self.valid_msg, type="private")
        self.assertFalse(check_zulip_message_validity(msg_not_stream, zulip_config))

        msg_from_bot = dict(self.valid_msg, sender_email=zulip_config['email'])
        self.assertFalse(check_zulip_message_validity(msg_from_bot, zulip_config))

    def test_zulip_to_matrix(self):
        # type: () -> None
        room = mock.MagicMock()
        zulip_config = self.valid_zulip_config
        send_msg = zulip_to_matrix(zulip_config, room)

        msg = dict(self.valid_msg, sender_full_name="John Smith")

        expected = {
            'hi': '{} hi',
            '*hi*': '{} *hi*',
            '**hi**': '{} **hi**',
        }

        for content in expected:
            send_msg(dict(msg, content=content))

        for (method, params, _), expect in zip(room.method_calls, expected.values()):
            self.assertEqual(method, 'send_text')
            self.assertEqual(params[0], expect.format('<JohnSmith>'))