<commit_before>from __future__ import unicode_literals

import json
import redis

from dash.orgs.models import Org
from dash.utils import random_string
from django.contrib.auth.models import User
from django.http import JsonResponse
from django.test import TestCase


class DashTest(TestCase):
    """
    Base class for dashboard test cases
    """
    @classmethod
    def setUpTestData(cls):
        cls.clear_cache()
        cls.superuser = User.objects.create_superuser(
            username="root", email="<EMAIL>", password="<PASSWORD>")

    def setUp(self):
        super(DashTest, self).setUp()

        self.clear_cache()

    @classmethod
    def clear_cache(cls):
        # we are extra paranoid here and actually hardcode redis to 'localhost' and '10'
        # Redis 10 is our testing redis db
        r = redis.StrictRedis(host='localhost', db=10)
        r.flushdb()

    def create_org(self, name, timezone, subdomain):
        return Org.objects.create(
            name=name, timezone=timezone, subdomain=subdomain, api_token=random_string(32),
            created_by=self.superuser, modified_by=self.superuser)

    def login(self, user):
        result = self.client.login(username=user.username, password=<PASSWORD>)
        self.assertTrue(result, "Couldn't login as %(user)s / %(user)s" % dict(user=user.username))

    def url_get(self, subdomain, url, params=None):
        return self._url_request(subdomain, 'get', url, params)

    def url_post(self, subdomain, url, data=None):
        return self._url_request(subdomain, 'post', url, data)

    def assertLoginRedirect(self, response, subdomain, next_url):
        self.assertRedirects(response,
                             'http://%s.localhost/users/login/?next=%s' % (subdomain, next_url))

    def _url_request(self, subdomain, method, url, data):
        if data is None:
            data = {}
        extra = {}
        if subdomain:
            extra['HTTP_HOST'] = '%s.localhost' % subdomain
        func = getattr(self.client, method)
        response = func(url, data, **extra)
        if isinstance(response, JsonResponse):
            response.json = json.loads(response.content)
        return response<commit_msg>Added MockClientQuery for mocking APIv2 client methods<commit_after>from __future__ import unicode_literals

import json
import itertools
import redis
import six

from dash.orgs.models import Org
from dash.utils import random_string
from django.contrib.auth.models import User
from django.http import JsonResponse
from django.test import TestCase


class DashTest(TestCase):
    """
    Base class for dashboard test cases
    """
    @classmethod
    def setUpTestData(cls):
        cls.clear_cache()
        cls.superuser = User.objects.create_superuser(
            username="root", email="<EMAIL>", password="<PASSWORD>")

    def setUp(self):
        super(DashTest, self).setUp()

        self.clear_cache()

    @classmethod
    def clear_cache(cls):
        # we are extra paranoid here and actually hardcode redis to 'localhost' and '10'
        # Redis 10 is our testing redis db
        r = redis.StrictRedis(host='localhost', db=10)
        r.flushdb()

    def create_org(self, name, timezone, subdomain):
        return Org.objects.create(
            name=name, timezone=timezone, subdomain=subdomain, api_token=random_string(32),
            created_by=self.superuser, modified_by=self.superuser)

    def login(self, user):
        result = self.client.login(username=user.username, password=user.username)
        self.assertTrue(result, "Couldn't login as %(user)s / %(user)s" % dict(user=user.username))

    def url_get(self, subdomain, url, params=None):
        return self._url_request(subdomain, 'get', url, params)

    def url_post(self, subdomain, url, data=None):
        return self._url_request(subdomain, 'post', url, data)

    def assertLoginRedirect(self, response, subdomain, next_url):
        self.assertRedirects(response,
                             'http://%s.localhost/users/login/?next=%s' % (subdomain, next_url))

    def _url_request(self, subdomain, method, url, data):
        if data is None:
            data = {}
        extra = {}
        if subdomain:
            extra['HTTP_HOST'] = '%s.localhost' % subdomain
        func = getattr(self.client, method)
        response = func(url, data, **extra)
        if isinstance(response, JsonResponse):
            response.json = json.loads(response.content)
        return response


class MockClientQuery(six.Iterator):
    """
    Mock for APIv2 client get_xxxxx return values. Pass lists of temba objects to mock each fetch the client would make.

    For example:
        mock_get_contacts.return_value = MockClientQuery(
            [TembaContact.create(...), TembaContact.create(...), TembaContact.create(...)]
            [TembaContact.create(...)]
        )

    Will return the three contacts on the first call to iterfetches, and one on the second call.

    """
    def __init__(self, *fetches):
        self.fetches = list(fetches)

    def iterfetches(self, retry_on_rate_exceed=False):
        return self

    def all(self, retry_on_rate_exceed=False):
        return list(itertools.chain.from_iterable(self.fetches))

    def first(self, retry_on_rate_exceed=False):
        return self.fetches[0][0] if self.fetches[0] else None

    def __iter__(self):
        return self

    def __next__(self):
        if not self.fetches:
            raise StopIteration()

        return self.fetches.pop(0)
{'fraction_non_alphanumeric': 0.10700706137968495, 'fraction_numerical': 0.003983342386384212, 'mean_word_length': 2.912181303116147, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 8, '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': '447016', 'n_tokens_mistral': 1672, 'n_tokens_neox': 1612, 'n_words': 419}