<commit_before>import calendar
import datetime

from unittest import mock

from django.test import TestCase

from corehq.apps.accounting.models import DomainUserHistory
from corehq.apps.analytics.models import (
    PartnerAnalyticsContact,
    PartnerAnalyticsReport,
    PartnerAnalyticsDataPoint,
)
from corehq.apps.analytics.utils.partner_analytics import (
    get_number_of_mobile_workers,
    generate_monthly_mobile_worker_statistics,
    track_partner_access,
    generate_monthly_web_user_statistics,
    generate_monthly_submissions_statistics,
    get_csv_details_for_partner,
    send_partner_emails,
    NUMBER_OF_MOBILE_WORKERS,
    NUMBER_OF_WEB_USERS,
    NUMBER_OF_SUBMISSIONS,
    ACCESS_ODATA,
)
from corehq.apps.domain.models import Domain
from corehq.apps.es.tests.utils import es_test
from dimagi.utils.dates import get_start_and_end_dates_of_month


def _get_fake_number_of_mobile_workers(domain, _year, _month):
    return {
        'test-001': 10,
        'test-002': 20,
        'test-003': 30,
    }[domain]


def _get_fake_number_of_web_users(domain, _year, _month):
    return {
        'test-001': 11,
        'test-002': 21,
        'test-003': 31,
    }[domain]


def _get_fake_number_of_submissions(domain, _year, _month):
    return {
        'test-001': 11,
        'test-002': 22,
        'test-003': 33,
    }[domain]


@es_test
class TestPartnerAnalyticsDataUtils(TestCase):

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        cls.date_start, cls.date_end = get_start_and_end_dates_of_month(2021, 11)
        cls.domain = Domain.get_or_create_with_name('test-partner-analytics', is_active=True)

        ## Data for Mobile Workers Tests
        # test month
        DomainUserHistory.objects.create(
            domain=cls.domain.name,
            record_date=cls.date_end,
            num_users=3
        )
        # previous month
        DomainUserHistory.objects.create(
            domain=cls.domain.name,
            record_date=cls.date_start - datetime.timedelta(days=1),
            num_users=6
        )
        # next month
        DomainUserHistory.objects.create(
            domain=cls.domain.name,
            record_date=cls.date_end + datetime.timedelta(days=1),
            num_users=9
        )

    @classmethod
    def tearDownClass(cls):
        DomainUserHistory.objects.all().delete()
        cls.domain.delete()
        super().tearDownClass()

    def test_get_number_of_mobile_workers(self):
        self.assertEqual(
            get_number_of_mobile_workers(
                self.domain.name, self.date_start.year, self.date_start.month
            ),
            3
        )


class TestPartnerAnalyticsReportUtils(TestCase):

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        today = datetime.datetime.utcnow()
        cls.month = today.month
        cls.year = today.year

        cls.contact = PartnerAnalyticsContact.objects.create(
            organization_name="<NAME>",
            emails=["<EMAIL>", "<EMAIL>"],
        )
        domains = ['test-001', 'test-002', 'test-003']

        PartnerAnalyticsReport.objects.create(
            contact=cls.contact,
            title="Number of Mobile Workers",<commit_msg>add remaining tests<commit_after>import calendar
import datetime

from unittest import mock

from django.test import TestCase

from corehq.apps.accounting.models import DomainUserHistory
from corehq.apps.analytics.models import (
    PartnerAnalyticsContact,
    PartnerAnalyticsReport,
    PartnerAnalyticsDataPoint,
)
from corehq.apps.analytics.utils.partner_analytics import (
    get_number_of_mobile_workers,
    get_number_of_web_users,
    get_number_of_submissions,
    generate_monthly_mobile_worker_statistics,
    track_partner_access,
    generate_monthly_web_user_statistics,
    generate_monthly_submissions_statistics,
    get_csv_details_for_partner,
    send_partner_emails,
    NUMBER_OF_MOBILE_WORKERS,
    NUMBER_OF_WEB_USERS,
    NUMBER_OF_SUBMISSIONS,
    ACCESS_ODATA,
)
from corehq.apps.domain.models import Domain
from corehq.apps.users.models import WebUser, Invitation
from corehq.elastic import get_es_new, send_to_elasticsearch
from corehq.form_processor.tests.utils import (
    FormProcessorTestUtils,
    create_form_for_test,
)
from corehq.pillows.mappings import USER_INDEX_INFO, XFORM_INDEX_INFO
from corehq.pillows.user import transform_user_for_elasticsearch
from corehq.pillows.xform import transform_xform_for_elasticsearch
from corehq.util.elastic import ensure_index_deleted
from dimagi.utils.dates import get_start_and_end_dates_of_month
from pillowtop.es_utils import initialize_index_and_mapping


def _get_fake_number_of_mobile_workers(domain, _year, _month):
    return {
        'test-001': 10,
        'test-002': 20,
        'test-003': 30,
    }[domain]


def _get_fake_number_of_web_users(domain, _year, _month):
    return {
        'test-001': 11,
        'test-002': 21,
        'test-003': 31,
    }[domain]


def _get_fake_number_of_submissions(domain, _year, _month):
    return {
        'test-001': 11,
        'test-002': 22,
        'test-003': 33,
    }[domain]


class TestPartnerAnalyticsDataUtils(TestCase):

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        cls.es = get_es_new()
        initialize_index_and_mapping(cls.es, USER_INDEX_INFO)
        initialize_index_and_mapping(cls.es, XFORM_INDEX_INFO)

        cls.date_start, cls.date_end = get_start_and_end_dates_of_month(2021, 11)
        cls.domain = Domain.get_or_create_with_name('test-partner-analytics', is_active=True)

        # Data for Mobile Workers Tests
        DomainUserHistory.objects.create(
            domain=cls.domain.name,
            record_date=cls.date_end,
            num_users=3
        )
        DomainUserHistory.objects.create(
            domain=cls.domain.name,
            record_date=cls.date_start - datetime.timedelta(days=1),
            num_users=6
        )
        DomainUserHistory.objects.create(
            domain=cls.domain.name,
            record_date=cls.date_end + datetime.timedelta(days=1),
            num_users=9
        )

        # Data for Web Users tests
        cls.users = [
            WebUser.create(
                cls.domain.name, '<EMAIL>', '<PASSWORD>', None, None,
                date=cls.date_start + datetime.timedelta(days=5)
            ),
            WebUser.create(
                cls.domain.name, '<EMAIL>', '<PASSWORD>', None, None,
                date=cls.date_start - datetime.timedelta(days=1)
            ),
            WebUser.create(
                cls.domain.name, '<EMAIL>', '<PASSWORD>', None, None,
                date=cls.date_end + datetime.timedelta(days=1)
            ),
            WebUser.create(
                cls.domain.name, '<EMAIL>', '<PASSWORD>', None, None,
                date=cls.date_start + datetime.timedelta(days=6)
            ),
        ]
        for user in cls.users:
            elastic_user = transform_user_for_elasticsearch(user.to_json())
            send_to_elasticsearch('users', elastic_user)

        invitations = [
            Invitation.objects.create(
                domain=cls.domain.name,
                email='<EMAIL>',
                invited_by='<EMAIL>',
                invited_on=cls.date_start + datetime.timedelta(days=5),
            ),
            Invitation.objects.create(
                domain=cls.domain.name,
                email='<EMAIL>',
                invited_by='<EMAIL>',
                invited_on=cls.date_start - datetime.timedelta(days=1),
            ),
            Invitation.objects.create(
                domain=cls.domain.name,
                email='<EMAIL>',
                invited_by='<EMAIL>',
                invited_on=cls.date_end + datetime.timedelta(days=1),
            ),
        ]
        for invitation in invitations:
            invitation.is_accepted = True
            invitation.save()

        # Data for forms tests
        forms = [
            create_form_for_test(
                cls.domain.name,
                received_on=cls.date_start - datetime.timedelta(days=1)
            ),
            create_form_for_test(
                cls.domain.name,
                received_on=cls.date_end + datetime.timedelta(days=1)
            ),
            create_form_for_test(
                cls.domain.name,
                received_on=cls.date_start + datetime.timedelta(days=3)
            ),
            create_form_for_test(
                cls.domain.name,
                received_on=cls.date_start + datetime.timedelta(days=12)
            ),
        ]
        for form in forms:
            elastic_form = transform_xform_for_elasticsearch(form.to_json())
            send_to_elasticsearch('forms', elastic_form)

        cls.es.indices.refresh(USER_INDEX_INFO.alias)
        cls.es.indices.refresh(XFORM_INDEX_INFO.alias)

    @classmethod
    def tearDownClass(cls):
        ensure_index_deleted(USER_INDEX_INFO.alias)
        ensure_index_deleted(XFORM_INDEX_INFO.alias)
        for user in cls.users:
            user.delete(cls.domain.name, None)
        FormProcessorTestUtils.delete_all_sql_forms(cls.domain.name)
        FormProcessorTestUtils.delete_all_sql_cases(cls.domain.name)
        Invitation.objects.all().delete()
        DomainUserHistory.objects.all().delete()
        cls.domain.delete()
        super().tearDownClass()

    def test_get_number_of_mobile_workers(self):
        self.assertEqual(
            get_number_of_mobile_workers(
                self.domain.name, self.date_start.year, self.date_start.month
            ),
            3
        )

    def test_get_number_of_web_users(self):
        self.assertEqual(
            get_number_of_web_users(
                self.domain.name, self.date_start.year, self.date_start.month
            ),
            4
        )

    def test_get_number_of_submissions(self):
        self.assertEqual(
            get_number_of_submissions(
                self.domain.name, self.date_start.year, self.date_start.month
            ),
            2
        )


class TestPartnerAnalyticsReportUtils(TestCase):

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        today = datetime.datetime.utcnow()
        cls.month = today.month
        cls.year = today.year

        cls.contact = PartnerAnalyticsContact.objects.create(
            organization_name="<NAME>",
            emails=["<EMAIL>", "<EMAIL>"],
        )
        domains = ['test-001', 'test-002', 'test-003']

        PartnerAnalyticsReport.objects.create(
            contact=cls.contact,
            title="Number of Mobile Workers",
{'fraction_non_alphanumeric': 0.09342560553633218, 'fraction_numerical': 0.013653792200505003, 'mean_word_length': 2.3273179838207843, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, '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': '1789656', 'n_tokens_mistral': 3410, 'n_tokens_neox': 3246, 'n_words': 492}