import json
import os
import random
import time
from contextlib import contextmanager
from datetime import datetime, timedelta
from decimal import Decimal
from functools import partial
from urlparse import SplitResult, urlsplit, urlunsplit

from django import forms, test
from django.db import connections, transaction, DEFAULT_DB_ALIAS
from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core.cache import cache
from django.core.files.storage import default_storage as storage
from django.core.management import call_command
from django.core.urlresolvers import reverse
from django.test.client import Client, RequestFactory
from django.utils import translation
from django.utils.translation import trans_real

import elasticsearch
import mock
import tower
from dateutil.parser import parse as dateutil_parser
from django_browserid.tests import mock_browserid
from nose.exc import SkipTest
from nose.tools import eq_
from pyquery import PyQuery as pq
from waffle import cache_sample, cache_switch
from waffle.models import Flag, Sample, Switch

import mkt
from lib.es.management.commands import reindex
from lib.post_request_task import task as post_request_task
from mkt.access.acl import check_ownership
from mkt.access.models import Group, GroupUser
from mkt.constants import regions
from mkt.constants.payments import PROVIDER_REFERENCE
from mkt.files.helpers import copyfileobj
from mkt.prices.models import AddonPremium, Price, PriceCurrency
from mkt.search.indexers import BaseIndexer
from mkt.site.fixtures import fixture
from mkt.site.utils import app_factory, website_factory  # NOQA
from mkt.translations.hold import clean_translations
from mkt.translations.models import Translation
from mkt.users.models import UserProfile
from mkt.webapps.models import Webapp


# We might now have gettext available in jinja2.env.globals when running tests.
# It's only added to the globals when activating a language with tower (which
# is usually done in the middlewares). During tests, however, we might not be
# running middlewares, and thus not activating a language, and thus not
# installing gettext in the globals, and thus not have it in the context when
# rendering templates.
tower.activate('en-us')


class DynamicBoolFieldsTestMixin():

    def setUp(self):
        """
        Create an instance of the DynamicBoolFields model and call super
        on the inheriting setUp.
        (e.g. RatingDescriptors.objects.create(addon=self.app))
        """
        self.app = app_factory()
        self.model = None
        self.related_name = ''  # Related name of the bool table on the Webapp.

        self.BOOL_DICT = []
        self.flags = []  # Flag names.
        self.expected = []  # Translation names.

    def _get_related_bool_obj(self):
        return getattr(self.app, self.related_name)

    def _flag(self):
        """Flag app with a handful of flags for testing."""
        self._get_related_bool_obj().update(
            **dict(('has_%s' % f.lower(), True) for f in self.flags))

    def _check(self, obj=None):
        if not obj:
            obj = self._get_related_bool_obj()

        for bool_name in self.BOOL_DICT:
            field = 'has_%s' % bool_name.lower()
            value = bool_name in self.flags
            if isinstance(obj, dict):
                eq_(obj[field], value,
                    u'Unexpected value for field: %s' % field)
            else:
                eq_(getattr(obj, field), value,
                    u'Unexpected value for field: %s' % field)

    def to_unicode(self, items):
        """
        Force unicode evaluation of lazy items in the passed list, for set
        comparison to a list of already-evaluated unicode strings.
        """
        return [unicode(i) for i in items]

    def test_bools_set(self):
        self._flag()
        self._check()

    def test_to_dict(self):
        self._flag()
        self._check(self._get_related_bool_obj().to_dict())

    def test_default_false(self):
        obj = self.model(addon=self.app)
        eq_(getattr(obj, 'has_%s' % self.flags[0].lower()), False)


def formset(*args, **kw):
    """
    Build up a formset-happy POST.

    *args is a sequence of forms going into the formset.
    prefix and initial_count can be set in **kw.
    """
    prefix = kw.pop('prefix', 'form')
    total_count = kw.pop('total_count', len(args))
    initial_count = kw.pop('initial_count', len(args))
    data = {prefix + '-TOTAL_FORMS': total_count,
            prefix + '-INITIAL_FORMS': initial_count}
    for idx, d in enumerate(args):
        data.update(('%s-%s-%s' % (prefix, idx, k), v)
                    for k, v in d.items())
    data.update(kw)
    return data


def initial(form):
    """Gather initial data from the form into a dict."""
    data = {}
    for name, field in form.fields.items():
        if form.is_bound:
            data[name] = form[name].data
        else:
            data[name] = form.initial.get(name, field.initial)
        # The browser sends nothing for an unchecked checkbox.
        if isinstance(field, forms.BooleanField):
            val = field.to_python(data[name])
            if not val:
                del data[name]
    return data


def check_links(expected, elements, selected=None, verify=True):
    """Useful for comparing an `expected` list of links against PyQuery
    `elements`. Expected format of links is a list of tuples, like so:

    [
        ('Home', '/'),
        ('Extensions', reverse('browse.extensions')),
        ...
    ]

    If you'd like to check if a particular item in the list is selected,
    pass as `selected` the title of the link.

    Links are verified by default.

    """
    for idx, item in enumerate(expected):
        # List item could be `(text, link)`.
        if isinstance(item, tuple):
            text, link = item
        # Or list item could be `link`.
        elif isinstance(item, basestring):
            text, link = None, item

        e = elements.eq(idx)
        if text is not None:
            eq_(e.text(), text)
        if link is not None:
            # If we passed an <li>, try to find an <a>.
            if not e.filter('a'):
                e = e.find('a')
            eq_(e.attr('href'), link)
            if verify and link != '#':
                eq_(Client().head(link, follow=True).status_code, 200,
                    '%r is dead' % link)
        if text is not None and selected is not None:
            e = e.filter('.selected, .sel') or e.parents('.selected, .sel')
            eq_(bool(e.length), text == selected)


class _JSONifiedResponse(object):

    def __init__(self, response):
        self._orig_response = response

    def __getattr__(self, n):
        return getattr(self._orig_response, n)

    def __getitem__(self, n):
        return self._orig_response[n]

    def __iter__(self):
        return iter(self._orig_response)

    @property
    def json(self):
        """Will return parsed JSON on response if there is any."""
        if self.content and 'application/json' in self['Content-Type']:
            if not hasattr(self, '_content_json'):
                self._content_json = json.loads(self.content)
            return self._content_json


class JSONClient(Client):

    def _with_json(self, response):
        if hasattr(response, 'json'):
            return response
        else:
            return _JSONifiedResponse(response)

    def get(self, *args, **kw):
        return self._with_json(super(JSONClient, self).get(*args, **kw))

    def delete(self, *args, **kw):
        return self._with_json(super(JSONClient, self).delete(*args, **kw))

    def post(self, *args, **kw):
        return self._with_json(super(JSONClient, self).post(*args, **kw))

    def put(self, *args, **kw):
        return self._with_json(super(JSONClient, self).put(*args, **kw))

    def patch(self, *args, **kw):
        return self._with_json(super(JSONClient, self).patch(*args, **kw))

    def options(self, *args, **kw):
        return self._with_json(super(JSONClient, self).options(*args, **kw))


ES_patchers = [mock.patch('elasticsearch.Elasticsearch'),
               mock.patch('mkt.websites.indexers.WebsiteIndexer', spec=True),
               mock.patch('mkt.webapps.indexers.WebappIndexer', spec=True),
               mock.patch('mkt.search.indexers.index', spec=True),
               mock.patch('mkt.search.indexers.BaseIndexer.unindex'),
               mock.patch('mkt.search.indexers.Reindexing', spec=True,
                          side_effect=lambda i: [i]),
               ]


def start_es_mock():
    for patch in ES_patchers:
        patch.start()


def stop_es_mock():
    for patch in ES_patchers:
        patch.stop()

    # Reset cached Elasticsearch objects.
    BaseIndexer._es = {}


def days_ago(days):
    return datetime.now().replace(microsecond=0) - timedelta(days=days)


class MockEsMixin(object):
    mock_es = True

    @classmethod
    def setUpClass(cls):
        if cls.mock_es:
            start_es_mock()
        try:
            super(MockEsMixin, cls).setUpClass()
        except Exception:
            # We need to unpatch here because tearDownClass will not be
            # called.
            if cls.mock_es:
                stop_es_mock()
            raise

    @classmethod
    def tearDownClass(cls):
        try:
            super(MockEsMixin, cls).tearDownClass()
        finally:
            if cls.mock_es:
                stop_es_mock()


class MockBrowserIdMixin(object):

    def mock_browser_id(self):
        cache.clear()
        real_login = self.client.login

        def fake_login(email, password=None):
            with mock_browserid(email=email):
                return real_login(email=email, assertion='test',
                                  audience='test')

        self.client.login = fake_login

    def login(self, profile):
        email = getattr(profile, 'email', profile)
        if '@' not in email:
            email += '@mozilla.com'
        assert self.client.login(email=email, password='password')


JINJA_INSTRUMENTED = False


class ClassFixtureTestCase(test.TestCase):
    """ Based on the changes to TestCase (& TransactionTestCase) in Django1.8.
    Fixtures are loaded once per class, and a class setUpTestData method is
    added to be overridden by sublasses.  `transaction.atomic()` is used to
    achieve test isolation.
    See orginal code:
    https://github.com/django/django/blob/1.8b2/django/test/testcases.py
    #L747-990.
    A noteable difference is that this class assumes the database supports
    transactions.  This class will be obsolete on upgrade to 1.8.
    """
    fixtures = None

    @classmethod
    def _databases_names(cls, include_mirrors=True):
        # If the test case has a multi_db=True flag, act on all databases,
        # including mirrors or not. Otherwise, just on the default DB.
        if getattr(cls, 'multi_db', False):
            return [alias for alias in connections
                    if (include_mirrors or
                        connections[alias].settings_dict['TEST']['MIRROR'])]
        else:
            return [DEFAULT_DB_ALIAS]

    @classmethod
    def _enter_atomics(cls):
        """Helper method to open atomic blocks for multiple databases"""
        atomics = {}
        for db_name in cls._databases_names():
            atomics[db_name] = transaction.atomic(using=db_name)
            atomics[db_name].__enter__()
        return atomics

    @classmethod
    def _rollback_atomics(cls, atomics):
        """Rollback atomic blocks opened through the previous method"""
        for db_name in reversed(cls._databases_names()):
            transaction.set_rollback(True, using=db_name)
            atomics[db_name].__exit__(None, None, None)

    @classmethod
    def setUpClass(cls):
        super(ClassFixtureTestCase, cls).setUpClass()
        cls.cls_atomics = cls._enter_atomics()

        try:
            if cls.fixtures:
                for db_name in cls._databases_names(include_mirrors=False):
                    call_command('loaddata', *cls.fixtures, **{
                        'verbosity': 0,
                        'commit': False,
                        'database': db_name,
                    })

            cls.setUpTestData()
        except Exception:
            cls._rollback_atomics(cls.cls_atomics)
            raise

    @classmethod
    def tearDownClass(cls):
        cls._rollback_atomics(cls.cls_atomics)
        for conn in connections.all():
            conn.close()
        super(ClassFixtureTestCase, cls).tearDownClass()

    @classmethod
    def setUpTestData(cls):
        """Load initial data for the TestCase"""
        pass

    def _should_reload_connections(self):
        return False

    def _fixture_setup(self):
        assert not self.reset_sequences, (
            'reset_sequences cannot be used on TestCase instances')
        self.atomics = self._enter_atomics()

    def _fixture_teardown(self):
        self._rollback_atomics(self.atomics)

    def _post_teardown(self):
        """Patch _post_teardown so connections don't get closed.
        In django 1.6's _post_teardown connections are closed and we don't want
        that to happen after each test anymore.  This method isn't copied from
        Django 1.8 code.
        https://github.com/django/django/blob/1.6.10/django/test/testcases.py
        #L788
        """
        if not self._should_reload_connections():
            real_connections_all = connections.all
            connections.all = lambda: []
        super(ClassFixtureTestCase, self)._post_teardown()
        if not self._should_reload_connections():
            connections.all = real_connections_all


class TestCase(MockEsMixin, MockBrowserIdMixin, ClassFixtureTestCase):
    """Base class for all mkt tests."""
    client_class = Client

    def shortDescription(self):
        # Stop nose using the test docstring and instead the test method name.
        pass

    def _pre_setup(self):
        super(TestCase, self)._pre_setup()

        # XXX See if we can remove this when we switch to Django 1.8.
        # Some crud from the migration system gets stuck here and we have to
        # kick it loose manually.
        from mkt.webapps.models import AddonUser
        UserProfile.addonuser_set.related.model = AddonUser

        # Clean the slate.
        cache.clear()
        post_request_task._discard_tasks()

        trans_real.deactivate()
        trans_real._translations = {}  # Django fails to clear this cache.
        trans_real.activate(settings.LANGUAGE_CODE)

        self.mock_browser_id()

        global JINJA_INSTRUMENTED
        if not JINJA_INSTRUMENTED:
            import jinja2
            old_render = jinja2.Template.render

            def instrumented_render(self, *args, **kwargs):
                context = dict(*args, **kwargs)
                test.signals.template_rendered.send(sender=self, template=self,
                                                    context=context)
                return old_render(self, *args, **kwargs)

            jinja2.Template.render = instrumented_render
            JINJA_INSTRUMENTED = True

    def _post_teardown(self):
        mkt.set_user(None)
        clean_translations(None)  # Make sure queued translations are removed.
        super(TestCase, self)._post_teardown()

    @contextmanager
    def activate(self, locale=None):
        """Active a locale."""
        old_locale = translation.get_language()
        if locale:
            translation.activate(locale)
        yield
        translation.activate(old_locale)

    def assertNoFormErrors(self, response):
        """Asserts that no form in the context has errors.

        If you add this check before checking the status code of the response
        you'll see a more informative error.
        """
        # TODO(Kumar) liberate upstream to Django?
        if response.context is None:
            # It's probably a redirect.
            return
        if len(response.templates) == 1:
            tpl = [response.context]
        else:
            # There are multiple contexts so iter all of them.
            tpl = response.context
        for ctx in tpl:
            for k, v in ctx.iteritems():
                if isinstance(v, (forms.BaseForm, forms.formsets.BaseFormSet)):
                    if isinstance(v, forms.formsets.BaseFormSet):
                        # Concatenate errors from each form in the formset.
                        msg = '\n'.join(f.errors.as_text() for f in v.forms)
                    else:
                        # Otherwise, just return the errors for this form.
                        msg = v.errors.as_text()
                    msg = msg.strip()
                    if msg != '':
                        self.fail('form %r had the following error(s):\n%s'
                                  % (k, msg))