from __future__ import absolute_import
from __future__ import print_function

import filecmp
import os
import ujson

from django.core import mail
from django.http import HttpResponse
from django.test import override_settings
from mock import patch
from typing import Any, Dict, List

from zerver.lib.actions import do_change_stream_invite_only
from zerver.models import get_realm, get_stream, get_user_profile_by_email, \
    Realm, Stream, UserProfile
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import (
    avatar_disk_path, get_test_image_file, tornado_redirected_to_list,
)

class BotTest(ZulipTestCase):
    def assert_num_bots_equal(self, count):
        # type: (int) -> None
        result = self.client_get("/json/bots")
        self.assert_json_success(result)
        json = ujson.loads(result.content)
        self.assertEqual(count, len(json['bots']))

    def create_bot(self, **extras):
        # type: (**Any) -> Dict[str, Any]
        bot_info = {
            'full_name': 'The Bot of Hamlet',
            'short_name': 'hambot',
        }
        bot_info.update(extras)
        result = self.client_post("/json/bots", bot_info)
        self.assert_json_success(result)
        return ujson.loads(result.content)

    def test_bot_domain(self):
        # type: () -> None
        self.login("hamlet@zulip.com")
        self.create_bot()
        self.assertTrue(UserProfile.objects.filter(email='hambot-bot@zulip.testserver').exists())
        # The other cases are hard to test directly, since we don't allow creating bots from
        # the wrong subdomain, and because 'testserver.example.com' is not a valid domain for the bot's email.
        # So we just test the Raelm.get_bot_domain function.
        realm = get_realm('zulip')
        with self.settings(REALMS_HAVE_SUBDOMAINS=True):
            self.assertEqual(realm.get_bot_domain(), 'zulip.testserver')
        Realm.objects.exclude(string_id='zulip').update(deactivated=True)
        self.assertEqual(realm.get_bot_domain(), 'testserver')

    def deactivate_bot(self):
        # type: () -> None
        result = self.client_delete("/json/bots/hambot-bot@zulip.testserver")
        self.assert_json_success(result)

    def test_add_bot_with_bad_username(self):
        # type: () -> None
        self.login("hamlet@zulip.com")
        self.assert_num_bots_equal(0)
        bot_info = dict(
            full_name='',
            short_name='',
        )
        result = self.client_post("/json/bots", bot_info)
        self.assert_json_error(result, 'Bad name or username')
        self.assert_num_bots_equal(0)

    def test_add_bot(self):
        # type: () -> None
        self.login("hamlet@zulip.com")
        self.assert_num_bots_equal(0)
        events = [] # type: List[Dict[str, Any]]
        with tornado_redirected_to_list(events):
            result = self.create_bot()
        self.assert_num_bots_equal(1)

        bot = get_user_profile_by_email('hambot-bot@zulip.testserver')

        event = [e for e in events if e['event']['type'] == 'realm_bot'][0]
        self.assertEqual(
            dict(
                type='realm_bot',
                op='add',
                bot=dict(email='hambot-bot@zulip.testserver',
                         user_id=bot.id,
                         full_name='The Bot of Hamlet',
                         is_active=True,
                         api_key=result['api_key'],
                         avatar_url=result['avatar_url'],
                         default_sending_stream=None,
                         default_events_register_stream=None,
                         default_all_public_streams=False,
                         owner='hamlet@zulip.com')
            ),
            event['event']
        )

        users_result = self.client_get('/json/users')
        members = ujson.loads(users_result.content)['members']
        bots = [m for m in members if m['email'] == 'hambot-bot@zulip.testserver']
        self.assertEqual(len(bots), 1)
        bot = bots[0]
        self.assertEqual(bot['bot_owner'], 'hamlet@zulip.com')
        self.assertEqual(bot['user_id'], get_user_profile_by_email('hambot-bot@zulip.testserver').id)

    def test_add_bot_with_username_in_use(self):
        # type: () -> None
        self.login("hamlet@zulip.com")
        self.assert_num_bots_equal(0)
        result = self.create_bot()
        self.assert_num_bots_equal(1)

        bot_info = dict(
            full_name='Duplicate',
            short_name='hambot',
        )
        result = self.client_post("/json/bots", bot_info)
        self.assert_json_error(result, 'Username already in use')

    def test_add_bot_with_user_avatar(self):
        # type: () -> None
        self.login("hamlet@zulip.com")
        self.assert_num_bots_equal(0)
        with get_test_image_file('img.png') as fp:
            self.create_bot(file=fp)
            profile = get_user_profile_by_email('hambot-bot@zulip.testserver')
            # Make sure that avatar image that we've uploaded is same with avatar image in the server
            self.assertTrue(filecmp.cmp(fp.name,
                                        os.path.splitext(avatar_disk_path(profile))[0] +
                                        ".original"))
        self.assert_num_bots_equal(1)

        self.assertEqual(profile.avatar_source, UserProfile.AVATAR_FROM_USER)
        self.assertTrue(os.path.exists(avatar_disk_path(profile)))

    def test_add_bot_with_too_many_files(self):
        # type: () -> None
        self.login("hamlet@zulip.com")
        self.assert_num_bots_equal(0)
        with get_test_image_file('img.png') as fp1, \
                get_test_image_file('img.gif') as fp2:
            bot_info = dict(
                full_name='whatever',
                short_name='whatever',
                file1=fp1,
                file2=fp2,
            )
            result = self.client_post("/json/bots", bot_info)
        self.assert_json_error(result, 'You may only upload one file at a time')
        self.assert_num_bots_equal(0)

    def test_add_bot_with_default_sending_stream(self):
        # type: () -> None
        self.login("hamlet@zulip.com")
        self.assert_num_bots_equal(0)
        result = self.create_bot(default_sending_stream='Denmark')
        self.assert_num_bots_equal(1)
        self.assertEqual(result['default_sending_stream'], 'Denmark')

        profile = get_user_profile_by_email('hambot-bot@zulip.testserver')
        self.assertEqual(profile.default_sending_stream.name, 'Denmark')

    def test_add_bot_with_default_sending_stream_not_subscribed(self):
        # type: () -> None
        self.login("hamlet@zulip.com")
        self.assert_num_bots_equal(0)
        result = self.create_bot(default_sending_stream='Rome')
        self.assert_num_bots_equal(1)
        self.assertEqual(result['default_sending_stream'], 'Rome')

        profile = get_user_profile_by_email('hambot-bot@zulip.testserver')
        self.assertEqual(profile.default_sending_stream.name, 'Rome')

    def test_bot_add_subscription(self):
        # type: () -> None
        """

        Calling POST /json/users/me/subscriptions should successfully add
        streams, and a stream to the
        list of subscriptions and confirm the right number of events
        are generated.
        When 'principals' has a bot, no notification message event or invitation email
        is sent when add_subscriptions_backend is c
alled in the above api call.
        """
        self.login("hamlet@zulip.com")

        # Normal user i.e. not a bot.
        request_data = {
            'principals': '["iago@zulip.com"]'
        }
        events = [] # type: List[Dict[str, Any]]
        with tornado_redirected_to_list(events):
            result = self.common_subscribe_to_streams("hamlet@zulip.com", ['Rome'], request_data)
            self.assert_json_success(result)

        msg_event = [e for e in events if e['event']['type'] == 'message']
        self.assert_length(msg_event, 1) # Notification message event is sent.

        # Create a bot.
        self.assert_num_bots_equal(0)
        result = self.create_bot()
        self.