<commit_before>import pytz
import uuid

from datetime import datetime, timedelta
from django.test import SimpleTestCase
from django.test.utils import override_settings

from corehq.elastic import get_es_new, send_to_elasticsearch
from corehq.form_processor.models import CommCareCaseSQL, CaseTransaction
from corehq.pillows.mappings.case_mapping import CASE_INDEX
from corehq.pillows.mappings.group_mapping import GROUP_INDEX
from corehq.pillows.mappings.user_mapping import USER_INDEX
from corehq.pillows.mappings.xform_mapping import XFORM_INDEX
from dimagi.utils.dates import DateSpan
from dimagi.utils.parsing import string_to_utc_datetime
from elasticsearch.exceptions import ConnectionError

from corehq.util.elastic import ensure_index_deleted
from corehq.apps.users.models import CommCareUser
from corehq.apps.groups.models import Group
from corehq.form_processor.utils import TestFormMetadata
from casexml.apps.case.models import CommCareCase, CommCareCaseAction
from casexml.apps.case.const import CASE_ACTION_CREATE
from corehq.pillows.xform import XFormPillow, get_sql_xform_to_elasticsearch_pillow
from corehq.pillows.user import UserPillow
from corehq.pillows.case import CasePillow, get_sql_case_to_elasticsearch_pillow
from corehq.apps.reports.analytics.esaccessors import (
    get_submission_counts_by_user,
    get_completed_counts_by_user,
    get_submission_counts_by_date,
    get_active_case_counts_by_owner,
    get_total_case_counts_by_owner,
    get_case_counts_closed_by_user,
    get_case_counts_opened_by_user,
    get_paged_forms_by_type,
    get_last_form_submissions_by_user,
    get_user_stubs,
    get_group_stubs,
    get_forms,
    get_form_counts_by_user_xmlns,
    get_form_duration_stats_by_user,
    get_form_duration_stats_for_users,
    get_last_form_submission_for_xmlns,
    guess_form_name_from_submissions_using_xmlns,
    get_active_case_count
)
from corehq.apps.es.aggregations import MISSING_KEY
from corehq.util.test_utils import make_es_ready_form, trap_extra_setup
from pillowtop.feed.interface import Change


class BaseESAccessorsTest(SimpleTestCase):
    pillow_class = None
    es_index = None

    def setUp(self):
        with trap_extra_setup(ConnectionError):
            ensure_index_deleted(self.es_index)
            self.domain = 'esdomain'
            self.pillow = self.get_pillow()

    def get_pillow(self):
        return self.pillow_class()

    def tearDown(self):
        ensure_index_deleted(self.es_index)


class TestFormESAccessors(BaseESAccessorsTest):

    pillow_class = XFormPillow
    es_index = XFORM_INDEX

    def _send_form_to_es(self, domain=None, completion_time=None, received_on=None, **metadata_kwargs):
        metadata = TestFormMetadata(
            domain=domain or self.domain,
            time_end=completion_time or datetime.utcnow(),
            received_on=received_on or datetime.utcnow(),
        )
        for attr, value in metadata_kwargs.items():
            setattr(metadata, attr, value)

        form_pair = make_es_ready_form(metadata)
        self._pillow_process_form(form_pair)
        es = get_es_new()
        es.indices.refresh(XFORM_INDEX)
        return form_pair

    def _pillow_process_form(self, form_pair):
        self.pillow.change_transport(form_pair.json_form)

    def test_get_forms(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        xmlns = 'http://a.b.org'
        app_id = '1234'
        user_id = 'abc'

        self._send_form_to_es(
            app_id=app_id,
            xmlns=xmlns,
            received_on=datetime(2013, 7, 2),
            user_id=user_id,
        )

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[user_id],
            app_ids=app_id,
            xmlnss=xmlns,
        )
        self.assertEqual(paged_result.total, 1)
        self.assertEqual(paged_result.hits[0]['xmlns'], xmlns)
        self.assertEqual(paged_result.hits[0]['form']['meta']['userID'], user_id)
        self.assertEqual(paged_result.hits[0]['received_on'], '2013-07-02T00:00:00.000000Z')

    def test_get_forms_multiple_apps_xmlnss(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        xmlns1, xmlns2 = 'http://a.b.org', 'http://b.c.org'
        app_id1, app_id2 = '1234', '4567'
        user_id = 'abc'

        self._send_form_to_es(
            app_id=app_id1,
            xmlns=xmlns1,
            received_on=datetime(2013, 7, 2),
            user_id=user_id,
        )
        self._send_form_to_es(
            app_id=app_id2,
            xmlns=xmlns2,
            received_on=datetime(2013, 7, 2),
            user_id=user_id,
        )
        self._send_form_to_es(
            app_id=app_id1,
            xmlns=xmlns1,
            received_on=datetime(2013, 7, 2),
            user_id=None,
        )

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[user_id],
            app_ids=[app_id1, app_id2],
            xmlnss=[xmlns1, xmlns2],
        )
        self.assertEqual(paged_result.total, 2)

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[user_id],
            app_ids=[app_id1, app_id2],
            xmlnss=[xmlns1],
        )
        self.assertEqual(paged_result.total, 1)

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[user_id],
            app_ids=[app_id1],
            xmlnss=[xmlns2],
        )
        self.assertEqual(paged_result.total, 0)

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[None],
            app_ids=[app_id1],
            xmlnss=[xmlns1],
        )
        self.assertEqual(paged_result.total, 1)

    def test_basic_completed_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        self._send_form_to_es(completion_time=datetime(2013, 7, 2))

        results = get_completed_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_completed_out_of_range_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        self._send_form_to_es(completion_time=datetime(2013, 8, 2))
        self._send_form_to_es(completion_time=datetime(2013, 7, 2))

        results = get_completed_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_completed_different_domain_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        self._send_form_to_es(completion_time=datetime(2013, 7, 3), domain='not-in-my-backyard')
        self._send_form_to_es(completion_time=datetime(2013, 7, 2))

        results = get_completed_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_basic_submission_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        received_on = datetime(2013, 7, 15)

        self._send_form_to_es(received_on=received_on)

        results = get_submission_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_get_last_form_submission_by_xmlns(self):
        xmlns = 'http://a.b.org'
        kwargs = {
            'user_id': 'u1',
            'app_id': '1234',
            'domain': self.domain,
        }

        first = datetime(2013, 7, 15, 0, 0, 0)
        second = datetime(2013, 7, 16, 0, 0, 0)
        third = datetime(2013, 7, 17, 0, 0, 0)

        self._send_form_to_es(received_on=second, xmlns=xmlns, **kwargs)
        self._send_form_to_es(received_on=third, xmlns=xmlns, **kwargs)
        self._send_form_to_es(received_on=first, xmlns=xmlns, **kwargs)

        form = get_last_form_submission_for_xmlns(self.domain, xmlns)
        self.assertEqual(string_to_utc_datetime(form['received_on']), third)

        form = get_last_form_submission_for_xmlns(self.domain, 'missing')
        self.assertIsNone(form)

    def test_guess_form_name_from_xmlns_not_found(self):
        self.assertEqual(None, guess_form_name_from_submissions_using_xmlns('missing', 'missing'))

    def test_guess_form_name_from_xmlns(self):
        form_name = 'my cool form'
        xmlns = 'http://a.b.org'
        self._send_form_to_es(
            xmlns=xmlns,
            form_name=form_name,
        )
        self.assertEqual(form_name, guess_form_name_from_submissions_using_xmlns(self.domain, xmlns))

    def test_submission_out_of_range_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        self._send_form_to_es(received_on=datetime(2013, 8, 15))

        self._send_form_to_es(received_on=datetime(2013, 7, 15))

        results = get_submission_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_submission_different_domain_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        received_on = datetime(2013, 7, 15)

        self._send_form_to_es(received_on=received_on)
        self._send_form_to_es(received_on=received_on, domain='not-in-my-backyard')

        results = get_submission_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_basic_submission_by_date(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        received_on = datetime(2013, 7, 15)

        self._send_form_to_es(received_on=received_on)

        results = get_submission_counts_by_date(
            self.domain,
            ['cruella_deville'],
            DateSpan(start, end),
            pytz.utc
        )
        self.assertEquals(results['2013-07-15'], 1)

    def test_get_paged_forms_by_type(self):
        self._send_form_to_es()
        self._send_form_to_es()

        paged_result = get_paged_forms_by_type(self.domain, ['xforminstance'], size=1)
        self.assertEqual(len(paged_result.hits), 1)
        self.assertEqual(paged_result.total, 2)

    def test_timezone_differences(self):
        """
        Our received_on dates are always in UTC, so if we submit a form right at midnight UTC, then the report
        should show that form being submitted the day before if viewing from an earlier timezone like New York
        """
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        received_on = datetime(2013, 7, 15, 0, 0, 0)
        timezone = pytz.timezone('America/New_York')

        self._send_form_to_es(received_on=received_on)

        results = get_submission_counts_by_date(
            self.domain,
            ['cruella_deville'],
            DateSpan(start, end),
            timezone
        )
        self.assertEquals(results['2013-07-14'], 1)

    def test_get_last_form_submission_for_user_for_app(self):
        kwargs_u1 = {
            'user_id': 'u1',
            'app_id': '1234',
            'domain': self.domain,
        }
        kwargs_u2 = {
            'user_id': 'u2',
            'app_id': '1234',
            'domain': self.domain,
        }
        kwargs_u3 = {
            'user_id': None,
            'app_id': '1234',
            'domain': self.domain,
        }

        first = datetime(2013, 7, 15, 0, 0, 0)
        second = datetime(2013, 7, 16, 0, 0, 0)
        third = datetime(2013, 7, 17, 0, 0, 0)

        self._send_form_to_es(received_on=second, xmlns='second', **kwargs_u1)
        self._send_form_to_es(received_on=third, xmlns='third', **kwargs_u1)
        self._send_form_to_es(received_on=first, xmlns='first', **kwargs_u1)

        self._send_form_to_es(received_on=second, xmlns='second', **kwargs_u2)
        self._send_form_to_es(received_on=third, xmlns='third', **kwargs_u2)
        self._send_form_to_es(received_on=first, xmlns='first', **kwargs_u2)

        self._send_form_to_es(received_on=second, xmlns='second', **kwargs_u3)
        self._send_form_to_es(received_on=third, xmlns='third', **kwargs_u3)
        self._send_form_to_es(received_on=first, xmlns='first', **kwargs_u3)

        result = get_last_form_submissions_by_user(self.domain, ['u1', 'u2', 'missing'])
        self.assertEqual(result['u1'][0]['xmlns'], 'third')
        self.assertEqual(result['u2'][0]['xmlns'], 'third')

        result = get_last_form_submissions_by_user(self.domain, ['u1'], '1234')
        self.assertEqual(result['u1'][0]['xmlns'], 'third')

        result = get_last_form_submissions_by_user(self.domain, ['u1', None], '1234')
        self.assertEqual(result['u1'][0]['xmlns'], 'third')
        self.assertEqual(result[None][0]['xmlns'], 'third')

    def test_get_form_counts_by_user_xmlns(self):
        user1, user2 = 'u1', 'u2'
        app1, app2 = '123', '567'
        xmlns1, xmlns2 = 'abc', 'efg'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        received_on = datetime(2013, 7, 15, 0, 0, 0)
        received_on_out = datetime(2013, 6, 15, 0, 0, 0)
        self._send_form_to_es(received_on=received_on_out, completion_time=received_on, user_id=user1, app_id=app1, xmlns=xmlns1)
        self._send_form_to_es(received_on=received_on, user_id=user1, app_id=app1, xmlns=xmlns1)
        self._send_form_to_es(received_on=received_on, user_id=user1, app_id=app1, xmlns=xmlns1)
        self._send_form_to_es(received_on=received_on, user_id=user1, app_id=app2, xmlns=xmlns2)
        self._send_form_to_es(received_on=received_on, user_id=user2, app_id=app2, xmlns=xmlns2)
        self._send_form_to_es(received_on=received_on, user_id=None, app_id=app2, xmlns=xmlns2)

        counts = get_form_counts_by_user_xmlns(self.domain, start, end)
        self.assertEqual(counts, {
            (user1, app1, xmlns1): 2,
            (user1, app2, xmlns2): 1,
            (user2, app2, xmlns2): 1,
        })

        counts_user1 = get_form_counts_by_user_xmlns(self.domain, start, end, user_ids=[user1])
        self.assertEqual(counts_user1, {
            (user1, app1, xmlns1): 2,
            (user1, app2, xmlns2): 1,
        })

        counts_xmlns2 = get_form_counts_by_user_xmlns(self.domain, start, end, xmlnss=[xmlns2])
        self.assertEqual(counts_xmlns2, {
            (user1, app2, xmlns2): 1,
            (user2, app2, xmlns2): 1,
        })

        by_completion = get_form_counts_by_user_xmlns(self.domain, start, end, by_submission_time=False)
        self.assertEqual(by_completion, {
            (user1, app1, xmlns1): 1
        })

        counts_missing_user = get_form_counts_by_user_xmlns(self.domain, start, end, user_ids=[None])
        self.assertEqual(counts_missing_user, {
            (None, app2, xmlns2): 1,
        })

    def test_get_form_duration_stats_by_user(self):
        """
        Tests the get_form_duration_stats_by_user basic ability to get duration stats
        grouped by user
        """
        user1, user2 = 'u1', 'u2'
        app1 = '123'
        xmlns1 = 'abc'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        time_start = datetime(2013, 6, 15, 0, 0, 0)
        completion_time = datetime(2013, 7, 15, 0, 0, 0)

        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user2,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=None,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )

        results = get_form_duration_stats_by_user(
            self.domain,
            app1,
            xmlns1,
            [user1, user2, None],
            start,
            end,
            by_submission_time=False
        )

        self.assertEqual(results[user1]['count'], 1)
        self.assertEqual(timedelta(milliseconds=results[user1]['max']), completion_time - time_start)
        self.assertEqual(results[user2]['count'], 1)
        self.assertEqual(timedelta(milliseconds=results[user2]['max']), completion_time - time_start)
        self.assertEqual(results[MISSING_KEY]['count'], 1)
        self.assertEqual(timedelta(milliseconds=results[MISSING_KEY]['max']), completion_time - time_start)

    def test_get_form_duration_stats_by_user_decoys(self):
        """
        Tests the get_form_duration_stats_by_user ability to filter out forms that
        do not fit within the filters specified
        """
        user1, user2 = 'u1', 'u2'
        app1, app2 = '123', '456'
        xmlns1, xmlns2 = 'abc', 'def'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        time_start = datetime(2013, 7, 2, 0, 0, 0)
        completion_time = datetime(2013, 7, 15, 0, 0, 0)
        received_on = datetime(2013, 7, 20, 0, 0, 0)
        received_on_late = datetime(2013, 7, 20, 0, 0, 0)

        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on,
        )

        # different app
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app2,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on,
        )

        # different xmlns
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns2,
            time_start=time_start,
            received_on=received_on,
        )

        # out of time range
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user2,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on_late,
        )

        results = get_form_duration_stats_by_user(
            self.domain,
            app1,
            xmlns1,
            [user1, user2],
            start,
            end,
            by_submission_time=True
        )

        self.assertEqual(results[user1]['count'], 1)
        self.assertEqual(timedelta(milliseconds=results[user1]['max']), completion_time - time_start)
        self.assertIsNone(results.get('user2'))

    def test_get_form_duration_stats_for_users(self):
        """
        Tests the get_form_duration_stats_for_users basic ability to get duration stats
        """
        user1, user2 = 'u1', 'u2'
        app1 = '123'
        xmlns1 = 'abc'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        time_start = datetime(2013, 6, 15, 0, 0, 0)
        completion_time = datetime(2013, 7, 15, 0, 0, 0)

        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user2,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=None,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )

        results = get_form_duration_stats_for_users(
            self.domain,
            app1,
            xmlns1,
            [user1, user2, None],
            start,
            end,
            by_submission_time=False
        )

        self.assertEqual(results['count'], 3)
        self.assertEqual(timedelta(milliseconds=results['max']), completion_time - time_start)

    def test_get_form_duration_stats_for_users_decoys(self):
        """
        Tests the get_form_duration_stats_for_users ability to filter out forms that
        do not fit within the filters specified
        """
        user1, user2 = 'u1', 'u2'
        app1, app2 = '123', '456'
        xmlns1, xmlns2 = 'abc', 'def'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        time_start = datetime(2013, 7, 2, 0, 0, 0)
        completion_time = datetime(2013, 7, 15, 0, 0, 0)
        received_on = datetime(2013, 7, 20, 0, 0, 0)
        received_on_late = datetime(2013, 8, 20, 0, 0, 0)

        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on,
        )

        # different app
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app2,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on,
        )

        # different xmlns
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns2,
            time_start=time_start,
            received_on=received_on,
        )

        # out of time range
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user2,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on_late,
        )

        results = get_form_duration_stats_for_users(
            self.domain,
            app1,
            xmlns1,
            [user1, user2],
            start,
            end,
            by_submission_time=True
        )

        self.assertEqual(results['count'], 1)
        self.assertEqual(timedelta(milliseconds=results['max']), completion_time - time_start)


@override_settings(TESTS_SHOULD_USE_SQL_BACKEND=True)
class TestFormESAccessorsSQL(TestFormESAccessors):
    def get_pillow(self):
        XFormPillow()  # initialize index
        return get_sql_xform_to_elasticsearch_pillow()

    def _pillow_process_form(self, form_pair):
        change = Change(
            id=form_pair.json_form['form_id'],
            sequence_id='123',
            document=form_pair.json_form,
        )
        self.pillow.processor(change, do_set_checkpoint=False)


class TestUserESAccessors(BaseESAccessorsTest):

    pillow_class = UserPillow
    es_index = USER_INDEX

    def setUp(self):
        super(TestUserESAccessors, self).setUp()
        self.username = 'superman'
        self.first_name = 'clark'
        self.last_name = 'kent'
        self.doc_type = 'CommCareUser'

    def _send_user_to_es(self, _id=None, is_active=True):
        user = CommCareUser(
            domain=self.domain,
            username=self.username,
            _id=_id or uuid.uuid4().hex,
            is_active=is_active,
            first_name=self.first_name,
            last_name=self.last_name,
        )
        self.pillow.change_transport(user.to_json())
        self.pillow.get_es_new().indices.refresh(self.pillow.es_index)
        return user

    def test_active_user_query(self):
        self._send_user_to_es('123')
        results = get_user_stubs(['123'])

        self.assertEquals(len(results), 1)<commit_msg>remove test dependency on UserPillow<commit_after>import pytz
import uuid

from datetime import datetime, timedelta
from django.test import SimpleTestCase
from django.test.utils import override_settings

from corehq.elastic import get_es_new, send_to_elasticsearch
from corehq.form_processor.models import CommCareCaseSQL, CaseTransaction
from corehq.pillows.mappings.case_mapping import CASE_INDEX
from corehq.pillows.mappings.group_mapping import GROUP_INDEX
from corehq.pillows.mappings.user_mapping import USER_INDEX, USER_INDEX_INFO
from corehq.pillows.mappings.xform_mapping import XFORM_INDEX
from dimagi.utils.dates import DateSpan
from dimagi.utils.parsing import string_to_utc_datetime
from elasticsearch.exceptions import ConnectionError

from corehq.util.elastic import ensure_index_deleted
from corehq.apps.users.models import CommCareUser
from corehq.apps.groups.models import Group
from corehq.form_processor.utils import TestFormMetadata
from casexml.apps.case.models import CommCareCase, CommCareCaseAction
from casexml.apps.case.const import CASE_ACTION_CREATE
from corehq.pillows.xform import XFormPillow, get_sql_xform_to_elasticsearch_pillow
from corehq.pillows.case import CasePillow, get_sql_case_to_elasticsearch_pillow
from corehq.apps.reports.analytics.esaccessors import (
    get_submission_counts_by_user,
    get_completed_counts_by_user,
    get_submission_counts_by_date,
    get_active_case_counts_by_owner,
    get_total_case_counts_by_owner,
    get_case_counts_closed_by_user,
    get_case_counts_opened_by_user,
    get_paged_forms_by_type,
    get_last_form_submissions_by_user,
    get_user_stubs,
    get_group_stubs,
    get_forms,
    get_form_counts_by_user_xmlns,
    get_form_duration_stats_by_user,
    get_form_duration_stats_for_users,
    get_last_form_submission_for_xmlns,
    guess_form_name_from_submissions_using_xmlns,
    get_active_case_count
)
from corehq.apps.es.aggregations import MISSING_KEY
from corehq.util.test_utils import make_es_ready_form, trap_extra_setup
from pillowtop.es_utils import initialize_index_and_mapping
from pillowtop.feed.interface import Change


class BaseESAccessorsTest(SimpleTestCase):
    pillow_class = None
    es_index = None

    def setUp(self):
        with trap_extra_setup(ConnectionError):
            ensure_index_deleted(self.es_index)
            self.domain = 'esdomain'
            self.pillow = self.get_pillow()

    def get_pillow(self):
        return self.pillow_class()

    def tearDown(self):
        ensure_index_deleted(self.es_index)


class TestFormESAccessors(BaseESAccessorsTest):

    pillow_class = XFormPillow
    es_index = XFORM_INDEX

    def _send_form_to_es(self, domain=None, completion_time=None, received_on=None, **metadata_kwargs):
        metadata = TestFormMetadata(
            domain=domain or self.domain,
            time_end=completion_time or datetime.utcnow(),
            received_on=received_on or datetime.utcnow(),
        )
        for attr, value in metadata_kwargs.items():
            setattr(metadata, attr, value)

        form_pair = make_es_ready_form(metadata)
        self._pillow_process_form(form_pair)
        es = get_es_new()
        es.indices.refresh(XFORM_INDEX)
        return form_pair

    def _pillow_process_form(self, form_pair):
        self.pillow.change_transport(form_pair.json_form)

    def test_get_forms(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        xmlns = 'http://a.b.org'
        app_id = '1234'
        user_id = 'abc'

        self._send_form_to_es(
            app_id=app_id,
            xmlns=xmlns,
            received_on=datetime(2013, 7, 2),
            user_id=user_id,
        )

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[user_id],
            app_ids=app_id,
            xmlnss=xmlns,
        )
        self.assertEqual(paged_result.total, 1)
        self.assertEqual(paged_result.hits[0]['xmlns'], xmlns)
        self.assertEqual(paged_result.hits[0]['form']['meta']['userID'], user_id)
        self.assertEqual(paged_result.hits[0]['received_on'], '2013-07-02T00:00:00.000000Z')

    def test_get_forms_multiple_apps_xmlnss(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        xmlns1, xmlns2 = 'http://a.b.org', 'http://b.c.org'
        app_id1, app_id2 = '1234', '4567'
        user_id = 'abc'

        self._send_form_to_es(
            app_id=app_id1,
            xmlns=xmlns1,
            received_on=datetime(2013, 7, 2),
            user_id=user_id,
        )
        self._send_form_to_es(
            app_id=app_id2,
            xmlns=xmlns2,
            received_on=datetime(2013, 7, 2),
            user_id=user_id,
        )
        self._send_form_to_es(
            app_id=app_id1,
            xmlns=xmlns1,
            received_on=datetime(2013, 7, 2),
            user_id=None,
        )

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[user_id],
            app_ids=[app_id1, app_id2],
            xmlnss=[xmlns1, xmlns2],
        )
        self.assertEqual(paged_result.total, 2)

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[user_id],
            app_ids=[app_id1, app_id2],
            xmlnss=[xmlns1],
        )
        self.assertEqual(paged_result.total, 1)

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[user_id],
            app_ids=[app_id1],
            xmlnss=[xmlns2],
        )
        self.assertEqual(paged_result.total, 0)

        paged_result = get_forms(
            self.domain,
            start,
            end,
            user_ids=[None],
            app_ids=[app_id1],
            xmlnss=[xmlns1],
        )
        self.assertEqual(paged_result.total, 1)

    def test_basic_completed_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        self._send_form_to_es(completion_time=datetime(2013, 7, 2))

        results = get_completed_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_completed_out_of_range_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        self._send_form_to_es(completion_time=datetime(2013, 8, 2))
        self._send_form_to_es(completion_time=datetime(2013, 7, 2))

        results = get_completed_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_completed_different_domain_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        self._send_form_to_es(completion_time=datetime(2013, 7, 3), domain='not-in-my-backyard')
        self._send_form_to_es(completion_time=datetime(2013, 7, 2))

        results = get_completed_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_basic_submission_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        received_on = datetime(2013, 7, 15)

        self._send_form_to_es(received_on=received_on)

        results = get_submission_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_get_last_form_submission_by_xmlns(self):
        xmlns = 'http://a.b.org'
        kwargs = {
            'user_id': 'u1',
            'app_id': '1234',
            'domain': self.domain,
        }

        first = datetime(2013, 7, 15, 0, 0, 0)
        second = datetime(2013, 7, 16, 0, 0, 0)
        third = datetime(2013, 7, 17, 0, 0, 0)

        self._send_form_to_es(received_on=second, xmlns=xmlns, **kwargs)
        self._send_form_to_es(received_on=third, xmlns=xmlns, **kwargs)
        self._send_form_to_es(received_on=first, xmlns=xmlns, **kwargs)

        form = get_last_form_submission_for_xmlns(self.domain, xmlns)
        self.assertEqual(string_to_utc_datetime(form['received_on']), third)

        form = get_last_form_submission_for_xmlns(self.domain, 'missing')
        self.assertIsNone(form)

    def test_guess_form_name_from_xmlns_not_found(self):
        self.assertEqual(None, guess_form_name_from_submissions_using_xmlns('missing', 'missing'))

    def test_guess_form_name_from_xmlns(self):
        form_name = 'my cool form'
        xmlns = 'http://a.b.org'
        self._send_form_to_es(
            xmlns=xmlns,
            form_name=form_name,
        )
        self.assertEqual(form_name, guess_form_name_from_submissions_using_xmlns(self.domain, xmlns))

    def test_submission_out_of_range_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        self._send_form_to_es(received_on=datetime(2013, 8, 15))

        self._send_form_to_es(received_on=datetime(2013, 7, 15))

        results = get_submission_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_submission_different_domain_by_user(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        received_on = datetime(2013, 7, 15)

        self._send_form_to_es(received_on=received_on)
        self._send_form_to_es(received_on=received_on, domain='not-in-my-backyard')

        results = get_submission_counts_by_user(self.domain, DateSpan(start, end))
        self.assertEquals(results['cruella_deville'], 1)

    def test_basic_submission_by_date(self):
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        received_on = datetime(2013, 7, 15)

        self._send_form_to_es(received_on=received_on)

        results = get_submission_counts_by_date(
            self.domain,
            ['cruella_deville'],
            DateSpan(start, end),
            pytz.utc
        )
        self.assertEquals(results['2013-07-15'], 1)

    def test_get_paged_forms_by_type(self):
        self._send_form_to_es()
        self._send_form_to_es()

        paged_result = get_paged_forms_by_type(self.domain, ['xforminstance'], size=1)
        self.assertEqual(len(paged_result.hits), 1)
        self.assertEqual(paged_result.total, 2)

    def test_timezone_differences(self):
        """
        Our received_on dates are always in UTC, so if we submit a form right at midnight UTC, then the report
        should show that form being submitted the day before if viewing from an earlier timezone like New York
        """
        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)
        received_on = datetime(2013, 7, 15, 0, 0, 0)
        timezone = pytz.timezone('America/New_York')

        self._send_form_to_es(received_on=received_on)

        results = get_submission_counts_by_date(
            self.domain,
            ['cruella_deville'],
            DateSpan(start, end),
            timezone
        )
        self.assertEquals(results['2013-07-14'], 1)

    def test_get_last_form_submission_for_user_for_app(self):
        kwargs_u1 = {
            'user_id': 'u1',
            'app_id': '1234',
            'domain': self.domain,
        }
        kwargs_u2 = {
            'user_id': 'u2',
            'app_id': '1234',
            'domain': self.domain,
        }
        kwargs_u3 = {
            'user_id': None,
            'app_id': '1234',
            'domain': self.domain,
        }

        first = datetime(2013, 7, 15, 0, 0, 0)
        second = datetime(2013, 7, 16, 0, 0, 0)
        third = datetime(2013, 7, 17, 0, 0, 0)

        self._send_form_to_es(received_on=second, xmlns='second', **kwargs_u1)
        self._send_form_to_es(received_on=third, xmlns='third', **kwargs_u1)
        self._send_form_to_es(received_on=first, xmlns='first', **kwargs_u1)

        self._send_form_to_es(received_on=second, xmlns='second', **kwargs_u2)
        self._send_form_to_es(received_on=third, xmlns='third', **kwargs_u2)
        self._send_form_to_es(received_on=first, xmlns='first', **kwargs_u2)

        self._send_form_to_es(received_on=second, xmlns='second', **kwargs_u3)
        self._send_form_to_es(received_on=third, xmlns='third', **kwargs_u3)
        self._send_form_to_es(received_on=first, xmlns='first', **kwargs_u3)

        result = get_last_form_submissions_by_user(self.domain, ['u1', 'u2', 'missing'])
        self.assertEqual(result['u1'][0]['xmlns'], 'third')
        self.assertEqual(result['u2'][0]['xmlns'], 'third')

        result = get_last_form_submissions_by_user(self.domain, ['u1'], '1234')
        self.assertEqual(result['u1'][0]['xmlns'], 'third')

        result = get_last_form_submissions_by_user(self.domain, ['u1', None], '1234')
        self.assertEqual(result['u1'][0]['xmlns'], 'third')
        self.assertEqual(result[None][0]['xmlns'], 'third')

    def test_get_form_counts_by_user_xmlns(self):
        user1, user2 = 'u1', 'u2'
        app1, app2 = '123', '567'
        xmlns1, xmlns2 = 'abc', 'efg'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        received_on = datetime(2013, 7, 15, 0, 0, 0)
        received_on_out = datetime(2013, 6, 15, 0, 0, 0)
        self._send_form_to_es(received_on=received_on_out, completion_time=received_on, user_id=user1, app_id=app1, xmlns=xmlns1)
        self._send_form_to_es(received_on=received_on, user_id=user1, app_id=app1, xmlns=xmlns1)
        self._send_form_to_es(received_on=received_on, user_id=user1, app_id=app1, xmlns=xmlns1)
        self._send_form_to_es(received_on=received_on, user_id=user1, app_id=app2, xmlns=xmlns2)
        self._send_form_to_es(received_on=received_on, user_id=user2, app_id=app2, xmlns=xmlns2)
        self._send_form_to_es(received_on=received_on, user_id=None, app_id=app2, xmlns=xmlns2)

        counts = get_form_counts_by_user_xmlns(self.domain, start, end)
        self.assertEqual(counts, {
            (user1, app1, xmlns1): 2,
            (user1, app2, xmlns2): 1,
            (user2, app2, xmlns2): 1,
        })

        counts_user1 = get_form_counts_by_user_xmlns(self.domain, start, end, user_ids=[user1])
        self.assertEqual(counts_user1, {
            (user1, app1, xmlns1): 2,
            (user1, app2, xmlns2): 1,
        })

        counts_xmlns2 = get_form_counts_by_user_xmlns(self.domain, start, end, xmlnss=[xmlns2])
        self.assertEqual(counts_xmlns2, {
            (user1, app2, xmlns2): 1,
            (user2, app2, xmlns2): 1,
        })

        by_completion = get_form_counts_by_user_xmlns(self.domain, start, end, by_submission_time=False)
        self.assertEqual(by_completion, {
            (user1, app1, xmlns1): 1
        })

        counts_missing_user = get_form_counts_by_user_xmlns(self.domain, start, end, user_ids=[None])
        self.assertEqual(counts_missing_user, {
            (None, app2, xmlns2): 1,
        })

    def test_get_form_duration_stats_by_user(self):
        """
        Tests the get_form_duration_stats_by_user basic ability to get duration stats
        grouped by user
        """
        user1, user2 = 'u1', 'u2'
        app1 = '123'
        xmlns1 = 'abc'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        time_start = datetime(2013, 6, 15, 0, 0, 0)
        completion_time = datetime(2013, 7, 15, 0, 0, 0)

        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user2,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=None,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )

        results = get_form_duration_stats_by_user(
            self.domain,
            app1,
            xmlns1,
            [user1, user2, None],
            start,
            end,
            by_submission_time=False
        )

        self.assertEqual(results[user1]['count'], 1)
        self.assertEqual(timedelta(milliseconds=results[user1]['max']), completion_time - time_start)
        self.assertEqual(results[user2]['count'], 1)
        self.assertEqual(timedelta(milliseconds=results[user2]['max']), completion_time - time_start)
        self.assertEqual(results[MISSING_KEY]['count'], 1)
        self.assertEqual(timedelta(milliseconds=results[MISSING_KEY]['max']), completion_time - time_start)

    def test_get_form_duration_stats_by_user_decoys(self):
        """
        Tests the get_form_duration_stats_by_user ability to filter out forms that
        do not fit within the filters specified
        """
        user1, user2 = 'u1', 'u2'
        app1, app2 = '123', '456'
        xmlns1, xmlns2 = 'abc', 'def'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        time_start = datetime(2013, 7, 2, 0, 0, 0)
        completion_time = datetime(2013, 7, 15, 0, 0, 0)
        received_on = datetime(2013, 7, 20, 0, 0, 0)
        received_on_late = datetime(2013, 7, 20, 0, 0, 0)

        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on,
        )

        # different app
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app2,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on,
        )

        # different xmlns
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns2,
            time_start=time_start,
            received_on=received_on,
        )

        # out of time range
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user2,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on_late,
        )

        results = get_form_duration_stats_by_user(
            self.domain,
            app1,
            xmlns1,
            [user1, user2],
            start,
            end,
            by_submission_time=True
        )

        self.assertEqual(results[user1]['count'], 1)
        self.assertEqual(timedelta(milliseconds=results[user1]['max']), completion_time - time_start)
        self.assertIsNone(results.get('user2'))

    def test_get_form_duration_stats_for_users(self):
        """
        Tests the get_form_duration_stats_for_users basic ability to get duration stats
        """
        user1, user2 = 'u1', 'u2'
        app1 = '123'
        xmlns1 = 'abc'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        time_start = datetime(2013, 6, 15, 0, 0, 0)
        completion_time = datetime(2013, 7, 15, 0, 0, 0)

        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user2,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=None,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
        )

        results = get_form_duration_stats_for_users(
            self.domain,
            app1,
            xmlns1,
            [user1, user2, None],
            start,
            end,
            by_submission_time=False
        )

        self.assertEqual(results['count'], 3)
        self.assertEqual(timedelta(milliseconds=results['max']), completion_time - time_start)

    def test_get_form_duration_stats_for_users_decoys(self):
        """
        Tests the get_form_duration_stats_for_users ability to filter out forms that
        do not fit within the filters specified
        """
        user1, user2 = 'u1', 'u2'
        app1, app2 = '123', '456'
        xmlns1, xmlns2 = 'abc', 'def'

        start = datetime(2013, 7, 1)
        end = datetime(2013, 7, 30)

        time_start = datetime(2013, 7, 2, 0, 0, 0)
        completion_time = datetime(2013, 7, 15, 0, 0, 0)
        received_on = datetime(2013, 7, 20, 0, 0, 0)
        received_on_late = datetime(2013, 8, 20, 0, 0, 0)

        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on,
        )

        # different app
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app2,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on,
        )

        # different xmlns
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user1,
            app_id=app1,
            xmlns=xmlns2,
            time_start=time_start,
            received_on=received_on,
        )

        # out of time range
        self._send_form_to_es(
            completion_time=completion_time,
            user_id=user2,
            app_id=app1,
            xmlns=xmlns1,
            time_start=time_start,
            received_on=received_on_late,
        )

        results = get_form_duration_stats_for_users(
            self.domain,
            app1,
            xmlns1,
            [user1, user2],
            start,
            end,
            by_submission_time=True
        )

        self.assertEqual(results['count'], 1)
        self.assertEqual(timedelta(milliseconds=results['max']), completion_time - time_start)


@override_settings(TESTS_SHOULD_USE_SQL_BACKEND=True)
class TestFormESAccessorsSQL(TestFormESAccessors):
    def get_pillow(self):
        XFormPillow()  # initialize index
        return get_sql_xform_to_elasticsearch_pillow()

    def _pillow_process_form(self, form_pair):
        change = Change(
            id=form_pair.json_form['form_id'],
            sequence_id='123',
            document=form_pair.json_form,
        )
        self.pillow.processor(change, do_set_checkpoint=False)


class TestUserESAccessors(SimpleTestCase):

    def setUp(self):
        self.username = 'superman'
        self.first_name = 'clark'
        self.last_name = 'kent'
        self.doc_type = 'CommCareUser'
        self.domain = 'user-esaccessors-test'
        self.es = get_es_new()
        ensure_index_deleted(USER_INDEX)
        initialize_index_and_mapping(self.es, USER_INDEX_INFO)

    @classmethod
    def tearDownClass(cls):
        ensure_index_deleted(USER_INDEX)

    def _send_user_to_es(self, _id=None, is_active=True):
        user = CommCareUser(
            domain=self.domain,
            username=self.username,
            _id=_id or uuid.uuid4().hex,
            is_active=is_active,
            first_name=self.first_name,
            last_name=self.last_name,
        )
        send_to_elasticsearch('users', user.to_json())
        self.es.indices.refresh(USER_INDEX)
        return user

    def test_active_user_query(self):
        self._send_user_to_es('123')
        results = get_user_stubs(['123'])

        self.assertEquals(len(results), 1)
{'fraction_non_alphanumeric': 0.09984195553682436, 'fraction_numerical': 0.034980507849541674, 'mean_word_length': 2.423459818208051, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 272}, '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': '4545893', 'n_tokens_mistral': 17021, 'n_tokens_neox': 15502, 'n_words': 2610}