<commit_before>from __future__ import unicode_literals

import json

from mock import patch

from context_processors import GroupPermWrapper
from dateutil.relativedelta import relativedelta
from django.conf import settings
from django.contrib.auth.models import User, Group
from django.core import mail
from django.core.urlresolvers import reverse
from django.test.utils import override_settings
from django.utils import timezone
from redis_cache import get_redis_connection
from temba.campaigns.models import Campaign, CampaignEvent
from temba.contacts.models import ContactGroup, TEL_SCHEME, TWITTER_SCHEME, ExportContactsTask
from temba.orgs.models import Org, OrgCache, OrgEvent, OrgFolder, TopUp, Invitation, DAYFIRST, MONTHFIRST
from temba.channels.models import Channel, RECEIVE, SEND, TWILIO, TWITTER, PLIVO_AUTH_ID, PLIVO_AUTH_TOKEN, PLIVO_APP_ID
from temba.flows.models import Flow, ExportFlowResultsTask, ActionSet
from temba.msgs.models import Broadcast, Call, Label, Msg, Schedule, CALL_IN, INCOMING, ExportMessagesTask
from temba.tests import TembaTest, MockResponse
from temba.triggers.models import Trigger
from temba.utils import datetime_to_ms


class OrgContextProcessorTest(TembaTest):

    def test_group_perms_wrapper(self):
        administrators = Group.objects.get(name="Administrators")
        editors = Group.objects.get(name="Editors")
        viewers = Group.objects.get(name="Viewers")

        administrators_wrapper = GroupPermWrapper(administrators)
        self.assertTrue(administrators_wrapper['msgs']['msg_api'])
        self.assertTrue(administrators_wrapper["msgs"]["msg_inbox"])

        editors_wrapper = GroupPermWrapper(editors)
        self.assertFalse(editors_wrapper["msgs"]["org_plan"])
        self.assertTrue(editors_wrapper["msgs"]["msg_inbox"])

        viewers_wrapper = GroupPermWrapper(viewers)
        self.assertFalse(viewers_wrapper["msgs"]["msg_api"])
        self.assertTrue(viewers_wrapper["msgs"]["msg_inbox"])


class OrgTest(TembaTest):

    def test_edit(self):
        # use a manager now
        self.login(self.admin)

        # can we see the edit page
        response = self.client.get(reverse('orgs.org_edit'))
        self.assertEquals(200, response.status_code)

         # update the name and slug of the organization
        data = dict(name="Temba", timezone="Africa/Kigali", date_format=DAYFIRST, slug="nice temba")
        response = self.client.post(reverse('orgs.org_edit'), data)
        self.assertTrue('slug' in response.context['form'].errors)

        data = dict(name="Temba", timezone="Africa/Kigali", date_format=MONTHFIRST, slug="nice-temba")
        response = self.client.post(reverse('orgs.org_edit'), data)
        self.assertEquals(302, response.status_code)

        org = Org.objects.get(pk=self.org.pk)
        self.assertEquals("Temba", org.name)
        self.assertEquals("nice-temba", org.slug)

    def test_recommended_channel(self):
        self.org.timezone = 'Africa/Nairobi'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'africastalking')

        self.org.timezone = 'America/Phoenix'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'twilio')

        self.org.timezone = 'Asia/Jakarta'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'hub9')

        self.org.timezone = 'Africa/Mogadishu'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'shaqodoon')

        self.org.timezone = 'Europe/Amsterdam'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'nexmo')

        self.org.timezone = 'Africa/Kigali'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'android')

    def test_country(self):
        from temba.locations.models import AdminBoundary
        country_url = reverse('orgs.org_country')

        # can't see this page if not logged in
        self.assertLoginRedirect(self.client.get(country_url))

        # login as admin instead
        self.login(self.admin)
        response = self.client.get(country_url)
        self.assertEquals(200, response.status_code)

        # save with Rwanda as a country
        response = self.client.post(country_url, dict(country=AdminBoundary.objects.get(name='Rwanda').pk))

        # assert it has changed
        org = Org.objects.get(pk=self.org.pk)
        self.assertEquals("Rwanda", str(org.country))

        # clear it out
        response = self.client.post(country_url, dict(country=''))

        # assert it has been
        org = Org.objects.get(pk=self.org.pk)
        self.assertFalse(org.country)

    def test_plans(self):
        self.contact = self.create_contact("Joe", "+250788123123")

        self.create_msg(direction=INCOMING, contact=self.contact, text="Orange")

        # check start and end date for this plan
        self.assertEquals(timezone.now().date(), self.org.current_plan_start())
        self.assertEquals(timezone.now().date() + relativedelta(months=1), self.org.current_plan_end())

        # check our credits
        self.login(self.admin)
        response = self.client.get(reverse('orgs.org_home'))
        self.assertContains(response, "999")

        # view our topups
        response = self.client.get(reverse('orgs.topup_list'))

        # should say we have a 1,000 credits too
        self.assertContains(response, "999")

    def test_user_update(self):
        update_url = reverse('orgs.user_edit')
        login_url = reverse('users.user_login')

        # no access if anonymous
        response = self.client.get(update_url)
        self.assertRedirect(response, login_url)

        self.login(self.admin)

        # change the user language
        post_data = dict(language='pt-br', first_name='Admin', last_name='User', email='<EMAIL>', current_password='<PASSWORD>')
        response = self.client.post(update_url, post_data)
        self.assertRedirect(response, reverse('orgs.org_home'))

        # check that our user settings have changed
        settings = self.admin.get_settings()
        self.assertEquals('pt-br', settings.language)

    def test_webhook_headers(self):
        update_url = reverse('orgs.org_update', args=[self.org.pk])
        login_url = reverse('users.user_login')

        # no access if anonymous
        response = self.client.get(update_url)
        self.assertRedirect(response, login_url)

        self.login(self.admin)
        self.admin.set_org(self.org)

        response = self.client.get(update_url)
        self.assertEquals(200, response.status_code)

        # set a webhook with headers
        post_data = response.context['form'].initial
        post_data['webhook'] = 'http://webhooks.uniceflabs.org'
        post_data['header_1_key'] = 'Authorization'
        post_data['header_1_value'] = 'Authorization: Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ=='

        response = self.client.post(update_url, post_data)
        self.assertRedirect(response, reverse('orgs.org_home'))

        # check that our webhook settings have changed
        self.assertEquals('http://webhooks.uniceflabs.org', self.org.get_webhook_url())
        self.assertDictEquals({'Authorization': 'Authorization: Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ=='}, self.org.get_webhook_headers())

    def test_org_administration(self):
        manage_url = reverse('orgs.org_manage')
        update_url = reverse('orgs.org_update', args=[self.org.pk])
        login_url = reverse('users.user_login')

        # no access to anon
        response = self.client.get(manage_url)
        self.assertRedirect(response, login_url)

        response = self.client.get(update_url)<commit_msg>update webhook header test<commit_after>from __future__ import unicode_literals

import json

from mock import patch

from context_processors import GroupPermWrapper
from dateutil.relativedelta import relativedelta
from django.conf import settings
from django.contrib.auth.models import User, Group
from django.core import mail
from django.core.urlresolvers import reverse
from django.test.utils import override_settings
from django.utils import timezone
from redis_cache import get_redis_connection
from temba.campaigns.models import Campaign, CampaignEvent
from temba.contacts.models import ContactGroup, TEL_SCHEME, TWITTER_SCHEME, ExportContactsTask
from temba.orgs.models import Org, OrgCache, OrgEvent, OrgFolder, TopUp, Invitation, DAYFIRST, MONTHFIRST
from temba.channels.models import Channel, RECEIVE, SEND, TWILIO, TWITTER, PLIVO_AUTH_ID, PLIVO_AUTH_TOKEN, PLIVO_APP_ID
from temba.flows.models import Flow, ExportFlowResultsTask, ActionSet
from temba.msgs.models import Broadcast, Call, Label, Msg, Schedule, CALL_IN, INCOMING, ExportMessagesTask
from temba.tests import TembaTest, MockResponse
from temba.triggers.models import Trigger
from temba.utils import datetime_to_ms


class OrgContextProcessorTest(TembaTest):

    def test_group_perms_wrapper(self):
        administrators = Group.objects.get(name="Administrators")
        editors = Group.objects.get(name="Editors")
        viewers = Group.objects.get(name="Viewers")

        administrators_wrapper = GroupPermWrapper(administrators)
        self.assertTrue(administrators_wrapper['msgs']['msg_api'])
        self.assertTrue(administrators_wrapper["msgs"]["msg_inbox"])

        editors_wrapper = GroupPermWrapper(editors)
        self.assertFalse(editors_wrapper["msgs"]["org_plan"])
        self.assertTrue(editors_wrapper["msgs"]["msg_inbox"])

        viewers_wrapper = GroupPermWrapper(viewers)
        self.assertFalse(viewers_wrapper["msgs"]["msg_api"])
        self.assertTrue(viewers_wrapper["msgs"]["msg_inbox"])


class OrgTest(TembaTest):

    def test_edit(self):
        # use a manager now
        self.login(self.admin)

        # can we see the edit page
        response = self.client.get(reverse('orgs.org_edit'))
        self.assertEquals(200, response.status_code)

         # update the name and slug of the organization
        data = dict(name="Temba", timezone="Africa/Kigali", date_format=DAYFIRST, slug="nice temba")
        response = self.client.post(reverse('orgs.org_edit'), data)
        self.assertTrue('slug' in response.context['form'].errors)

        data = dict(name="Temba", timezone="Africa/Kigali", date_format=MONTHFIRST, slug="nice-temba")
        response = self.client.post(reverse('orgs.org_edit'), data)
        self.assertEquals(302, response.status_code)

        org = Org.objects.get(pk=self.org.pk)
        self.assertEquals("Temba", org.name)
        self.assertEquals("nice-temba", org.slug)

    def test_recommended_channel(self):
        self.org.timezone = 'Africa/Nairobi'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'africastalking')

        self.org.timezone = 'America/Phoenix'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'twilio')

        self.org.timezone = 'Asia/Jakarta'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'hub9')

        self.org.timezone = 'Africa/Mogadishu'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'shaqodoon')

        self.org.timezone = 'Europe/Amsterdam'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'nexmo')

        self.org.timezone = 'Africa/Kigali'
        self.org.save()
        self.assertEquals(self.org.get_recommended_channel(), 'android')

    def test_country(self):
        from temba.locations.models import AdminBoundary
        country_url = reverse('orgs.org_country')

        # can't see this page if not logged in
        self.assertLoginRedirect(self.client.get(country_url))

        # login as admin instead
        self.login(self.admin)
        response = self.client.get(country_url)
        self.assertEquals(200, response.status_code)

        # save with Rwanda as a country
        response = self.client.post(country_url, dict(country=AdminBoundary.objects.get(name='Rwanda').pk))

        # assert it has changed
        org = Org.objects.get(pk=self.org.pk)
        self.assertEquals("Rwanda", str(org.country))

        # clear it out
        response = self.client.post(country_url, dict(country=''))

        # assert it has been
        org = Org.objects.get(pk=self.org.pk)
        self.assertFalse(org.country)

    def test_plans(self):
        self.contact = self.create_contact("Joe", "+250788123123")

        self.create_msg(direction=INCOMING, contact=self.contact, text="Orange")

        # check start and end date for this plan
        self.assertEquals(timezone.now().date(), self.org.current_plan_start())
        self.assertEquals(timezone.now().date() + relativedelta(months=1), self.org.current_plan_end())

        # check our credits
        self.login(self.admin)
        response = self.client.get(reverse('orgs.org_home'))
        self.assertContains(response, "999")

        # view our topups
        response = self.client.get(reverse('orgs.topup_list'))

        # should say we have a 1,000 credits too
        self.assertContains(response, "999")

    def test_user_update(self):
        update_url = reverse('orgs.user_edit')
        login_url = reverse('users.user_login')

        # no access if anonymous
        response = self.client.get(update_url)
        self.assertRedirect(response, login_url)

        self.login(self.admin)

        # change the user language
        post_data = dict(language='pt-br', first_name='Admin', last_name='User', email='<EMAIL>', current_password='<PASSWORD>')
        response = self.client.post(update_url, post_data)
        self.assertRedirect(response, reverse('orgs.org_home'))

        # check that our user settings have changed
        settings = self.admin.get_settings()
        self.assertEquals('pt-br', settings.language)

    def test_webhook_headers(self):
        update_url = reverse('orgs.org_update', args=[self.org.pk])
        login_url = reverse('users.user_login')

        # no access if anonymous
        response = self.client.get(update_url)
        self.assertRedirect(response, login_url)

        self.login(self.admin)
        self.admin.set_org(self.org)

        response = self.client.get(update_url)

        # set a webhook with headers
        post_data = response.context['form'].initial
        post_data['webhook'] = 'http://webhooks.uniceflabs.org'
        post_data['header_1_key'] = 'Authorization'
        post_data['header_1_value'] = 'Authorization: Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ=='

        response = self.client.post(update_url, post_data)
        self.assertRedirect(response, reverse('orgs.org_home'))

        # check that our webhook settings have changed
        self.assertEquals('http://webhooks.uniceflabs.org', self.org.get_webhook_url())
        self.assertDictEquals({'Authorization': 'Authorization: Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ=='}, self.org.get_webhook_headers())

    def test_org_administration(self):
        manage_url = reverse('orgs.org_manage')
        update_url = reverse('orgs.org_update', args=[self.org.pk])
        login_url = reverse('users.user_login')

        # no access to anon
        response = self.client.get(manage_url)
        self.assertRedirect(response, login_url)

        response = self.client.get(update_url)
{'fraction_non_alphanumeric': 0.11238651102464332, 'fraction_numerical': 0.004993514915693904, 'mean_word_length': 3.441532258064516, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5858981', 'n_tokens_mistral': 4811, 'n_tokens_neox': 4635, 'n_words': 870}