<commit_before>from datetime import datetime, timedelta
import uuid
import os
import hashlib

from django.conf import settings
from django.template import Template, Context

from django.test import TestCase
import lxml
from django.core.files.uploadedfile import UploadedFile
from mock import patch

from casexml.apps.case.tests.util import TEST_DOMAIN_NAME
from casexml.apps.case.xml import V2
from casexml.apps.phone.models import SimplifiedSyncLog
from corehq.apps.receiverwrapper.util import submit_form_locally
from corehq.blobs import get_blob_db
from corehq.blobs.tests.util import TemporaryS3BlobDB
from corehq.form_processor.interfaces.dbaccessors import CaseAccessors, FormAccessors
from couchforms.models import XFormInstance
from dimagi.utils.parsing import json_format_datetime
from corehq.form_processor.interfaces.processor import FormProcessorInterface
from corehq.form_processor.tests.utils import FormProcessorTestUtils, use_sql_backend
from corehq.util.test_utils import TestFileMixin, trap_extra_setup, flag_enabled

TEST_CASE_ID = "EOL9FIAKIQWOFXFOH0QAMWU64"
CREATE_XFORM_ID = "6RGAZTETE3Z2QC0PE2DKM88MO"
media_path = os.path.join(os.path.dirname(__file__), "data", "attachments")

MEDIA_FILES = {
    "fruity_file": os.path.join(media_path, "fruity.jpg"),  # first
    "dimagi_logo_file": os.path.join(media_path, "dimagi_logo.jpg"),
    "commcare_logo_file": os.path.join(media_path, "commcare-logo.png"),
    "globe_file": os.path.join(media_path, "globe.pdf"),
    "house_file": os.path.join(media_path, "house.jpg"),
}


class BaseCaseMultimediaTest(TestCase, TestFileMixin):

    file_path = ('data', 'multimedia')
    root = os.path.dirname(__file__)

    def setUp(self):
        super(BaseCaseMultimediaTest, self).setUp()
        self.formdb = FormAccessors()
        FormProcessorTestUtils.delete_all_cases()
        FormProcessorTestUtils.delete_all_xforms()

    def _formatXForm(self, doc_id, raw_xml, attachment_block, date=None):
        if date is None:
            date = datetime.utcnow()
        final_xml = Template(raw_xml).render(Context({
            "attachments": attachment_block,
            "time_start": json_format_datetime(date - timedelta(minutes=4)),
            "time_end": json_format_datetime(date),
            "date_modified": json_format_datetime(date),
            "doc_id": doc_id
        }))
        return final_xml

    def _prepAttachments(self, new_attachments, removes=[]):
        """
        Returns:
            attachment_block - An XML representation of the attachment
            dict_attachments - A key-value dict where the key is the name and the value is a Stream of the
            attachment
        """
        attachment_block = ''.join([self._singleAttachBlock(x) for x in new_attachments] + [self._singleAttachRemoveBlock(x) for x in removes])
        dict_attachments = dict((MEDIA_FILES[attach_name], self._attachmentFileStream(attach_name)) for attach_name in new_attachments)
        return attachment_block, dict_attachments

    def _singleAttachBlock(self, key):
        return '<n0:%s src="%s" from="local"/>' % (key, MEDIA_FILES[key])

    def _singleAttachRemoveBlock(self, key):
        return '<n0:%s />' % key

    def _attachmentFileStream(self, key):
        attachment_path = MEDIA_FILES[key]
        attachment = open(attachment_path, 'rb')
        return NoClose(UploadedFile(attachment, key))

    def _calc_file_hash(self, key):
        with open(MEDIA_FILES[key], 'rb') as attach:
            return hashlib.md5(attach.read()).hexdigest()

    def _do_submit(self, xml_data, dict_attachments, sync_token=None, date=None):
        """
        RequestFactory submitter - simulates direct submission to server
        directly (no need to call process case after fact)
        """
        with flag_enabled('MM_CASE_PROPERTIES'):
            result = submit_form_locally(
                xml_data,
                TEST_DOMAIN_NAME,
                attachments=dict_attachments,
                last_sync_token=sync_token,
                received_on=date
            )
        attachments = result.xform.attachments
        self.assertEqual(set(dict_attachments.keys()),
                         set(attachments.keys()))
        self.assertEqual(result.case.case_id, TEST_CASE_ID)

        return result.response, result.xform, result.cases

    def _submit_and_verify(self, doc_id, xml_data, dict_attachments,
                           sync_token=None, date=None):
        response, form, [case] = self._do_submit(xml_data, dict_attachments, sync_token, date=date)

        attachments = form.attachments
        self.assertEqual(len(dict_attachments), len(attachments))
        for k, vstream in dict_attachments.items():
            fileback = form.get_attachment(k)
            # rewind the pointer before comparing
            orig_attachment = vstream
            orig_attachment.seek(0)
            self.assertEqual(hashlib.md5(fileback).hexdigest(), hashlib.md5(orig_attachment.read()).hexdigest())

        case = CaseAccessors(TEST_DOMAIN_NAME).get_case(case.case_id)  # re-fetch case
        return form, case

    def _doCreateCaseWithMultimedia(self, attachments=['fruity_file']):
        xml_data = self.get_xml('multimedia_create')
        attachment_block, dict_attachments = self._prepAttachments(attachments)
        final_xml = self._formatXForm(CREATE_XFORM_ID, xml_data, attachment_block)
        return self._submit_and_verify(CREATE_XFORM_ID, final_xml, dict_attachments)

    def _doSubmitUpdateWithMultimedia(self, new_attachments=None, removes=None,<commit_msg>refetch form from DB<commit_after>from datetime import datetime, timedelta
import uuid
import os
import hashlib

from django.conf import settings
from django.template import Template, Context

from django.test import TestCase
import lxml
from django.core.files.uploadedfile import UploadedFile
from mock import patch

from casexml.apps.case.tests.util import TEST_DOMAIN_NAME
from casexml.apps.case.xml import V2
from casexml.apps.phone.models import SimplifiedSyncLog
from corehq.apps.receiverwrapper.util import submit_form_locally
from corehq.blobs import get_blob_db
from corehq.blobs.tests.util import TemporaryS3BlobDB
from corehq.form_processor.interfaces.dbaccessors import CaseAccessors, FormAccessors
from couchforms.models import XFormInstance
from dimagi.utils.parsing import json_format_datetime
from corehq.form_processor.interfaces.processor import FormProcessorInterface
from corehq.form_processor.tests.utils import FormProcessorTestUtils, use_sql_backend
from corehq.util.test_utils import TestFileMixin, trap_extra_setup, flag_enabled

TEST_CASE_ID = "EOL9FIAKIQWOFXFOH0QAMWU64"
CREATE_XFORM_ID = "6RGAZTETE3Z2QC0PE2DKM88MO"
media_path = os.path.join(os.path.dirname(__file__), "data", "attachments")

MEDIA_FILES = {
    "fruity_file": os.path.join(media_path, "fruity.jpg"),  # first
    "dimagi_logo_file": os.path.join(media_path, "dimagi_logo.jpg"),
    "commcare_logo_file": os.path.join(media_path, "commcare-logo.png"),
    "globe_file": os.path.join(media_path, "globe.pdf"),
    "house_file": os.path.join(media_path, "house.jpg"),
}


class BaseCaseMultimediaTest(TestCase, TestFileMixin):

    file_path = ('data', 'multimedia')
    root = os.path.dirname(__file__)

    def setUp(self):
        super(BaseCaseMultimediaTest, self).setUp()
        self.formdb = FormAccessors()
        FormProcessorTestUtils.delete_all_cases()
        FormProcessorTestUtils.delete_all_xforms()

    def _formatXForm(self, doc_id, raw_xml, attachment_block, date=None):
        if date is None:
            date = datetime.utcnow()
        final_xml = Template(raw_xml).render(Context({
            "attachments": attachment_block,
            "time_start": json_format_datetime(date - timedelta(minutes=4)),
            "time_end": json_format_datetime(date),
            "date_modified": json_format_datetime(date),
            "doc_id": doc_id
        }))
        return final_xml

    def _prepAttachments(self, new_attachments, removes=[]):
        """
        Returns:
            attachment_block - An XML representation of the attachment
            dict_attachments - A key-value dict where the key is the name and the value is a Stream of the
            attachment
        """
        attachment_block = ''.join([self._singleAttachBlock(x) for x in new_attachments] + [self._singleAttachRemoveBlock(x) for x in removes])
        dict_attachments = dict((MEDIA_FILES[attach_name], self._attachmentFileStream(attach_name)) for attach_name in new_attachments)
        return attachment_block, dict_attachments

    def _singleAttachBlock(self, key):
        return '<n0:%s src="%s" from="local"/>' % (key, MEDIA_FILES[key])

    def _singleAttachRemoveBlock(self, key):
        return '<n0:%s />' % key

    def _attachmentFileStream(self, key):
        attachment_path = MEDIA_FILES[key]
        attachment = open(attachment_path, 'rb')
        return NoClose(UploadedFile(attachment, key))

    def _calc_file_hash(self, key):
        with open(MEDIA_FILES[key], 'rb') as attach:
            return hashlib.md5(attach.read()).hexdigest()

    def _do_submit(self, xml_data, dict_attachments, sync_token=None, date=None):
        """
        RequestFactory submitter - simulates direct submission to server
        directly (no need to call process case after fact)
        """
        with flag_enabled('MM_CASE_PROPERTIES'):
            result = submit_form_locally(
                xml_data,
                TEST_DOMAIN_NAME,
                attachments=dict_attachments,
                last_sync_token=sync_token,
                received_on=date
            )
        xform = result.xform
        attachments = xform.attachments
        self.assertEqual(set(dict_attachments.keys()),
                         set(attachments.keys()))
        self.assertEqual(result.case.case_id, TEST_CASE_ID)

        return result.response, self.formdb.get_form(xform.form_id), result.cases

    def _submit_and_verify(self, doc_id, xml_data, dict_attachments,
                           sync_token=None, date=None):
        response, form, [case] = self._do_submit(xml_data, dict_attachments, sync_token, date=date)

        attachments = form.attachments
        self.assertEqual(len(dict_attachments), len(attachments))
        for k, vstream in dict_attachments.items():
            fileback = form.get_attachment(k)
            # rewind the pointer before comparing
            orig_attachment = vstream
            orig_attachment.seek(0)
            self.assertEqual(hashlib.md5(fileback).hexdigest(), hashlib.md5(orig_attachment.read()).hexdigest())

        case = CaseAccessors(TEST_DOMAIN_NAME).get_case(case.case_id)  # re-fetch case
        return form, case

    def _doCreateCaseWithMultimedia(self, attachments=['fruity_file']):
        xml_data = self.get_xml('multimedia_create')
        attachment_block, dict_attachments = self._prepAttachments(attachments)
        final_xml = self._formatXForm(CREATE_XFORM_ID, xml_data, attachment_block)
        return self._submit_and_verify(CREATE_XFORM_ID, final_xml, dict_attachments)

    def _doSubmitUpdateWithMultimedia(self, new_attachments=None, removes=None,
{'fraction_non_alphanumeric': 0.0921474358974359, 'fraction_numerical': 0.0035612535612535613, 'mean_word_length': 3.502204408817635, 'pattern_counts': {'":': 20, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 36}, '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': '7058680', 'n_tokens_mistral': 3435, 'n_tokens_neox': 3344, 'n_words': 694}