<commit_before>import json
import uuid
from datetime import datetime

from django.utils.http import urlencode

from casexml.apps.case.mock import CaseBlock
from casexml.apps.case.models import CommCareCase
from dimagi.utils.parsing import json_format_datetime

from corehq.apps.api.models import ESCase
from corehq.apps.api.resources import v0_3, v0_4
from corehq.apps.domain.models import Domain
from corehq.apps.hqcase.utils import submit_case_blocks
from corehq.apps.users.models import WebUser
from corehq.elastic import get_es_new, send_to_elasticsearch
from corehq.form_processor.tests.utils import run_with_all_backends
from corehq.pillows.case import transform_case_for_elasticsearch
from corehq.pillows.mappings.case_mapping import CASE_INDEX_INFO
from corehq.util.elastic import reset_es_index
from pillowtop.es_utils import initialize_index_and_mapping

from .utils import APIResourceTest, FakeXFormES


class TestCommCareCaseResource(APIResourceTest):
    resource = v0_4.CommCareCaseResource

    def setUp(self):
        self.es = get_es_new()
        reset_es_index(CASE_INDEX_INFO)
        initialize_index_and_mapping(self.es, CASE_INDEX_INFO)

    def _setup_case(self):

        modify_date = datetime.utcnow()

        backend_case = CommCareCase(server_modified_on=modify_date, domain=self.domain.name)
        backend_case.save()
        self.addCleanup(backend_case.delete)

        translated_doc = transform_case_for_elasticsearch(backend_case.to_json())

        send_to_elasticsearch('cases', translated_doc)
        self.es.indices.refresh(CASE_INDEX_INFO.index)
        return backend_case

    def test_get_list(self):
        """
        Any case in the appropriate domain should be in the list from the API.
        """
        backend_case = self._setup_case()
        response = self._assert_auth_get_resource(self.list_endpoint)
        self.assertEqual(response.status_code, 200)

        api_cases = json.loads(response.content)['objects']
        self.assertEqual(len(api_cases), 1)

        api_case = api_cases[0]
        self.assertEqual(api_case['server_date_modified'], json_format_datetime(backend_case.server_modified_on))

    def test_get_list_format(self):
        """
        Any case in the appropriate domain should be in the list from the API.
        """
        backend_case = self._setup_case()

        # Get XML response
        response = self._assert_auth_get_resource(self.list_endpoint, headers={'HTTP_ACCEPT': 'application/xml'})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode().split('\n', 1)[0]
        self.assertEqual(content, "<?xml version='1.0' encoding='utf-8'?>")

        # Force JSON response with `format=json` parameter
        response = self._assert_auth_get_resource(
            self.list_endpoint + '?format=json',
            headers={'HTTP_ACCEPT': 'application/xml'}
        )
        self.assertEqual(response.status_code, 200)

        # we should still get the same case even with the `format` param
        # https://dimagi-dev.atlassian.net/browse/HI-656
        api_cases = json.loads(response.content)['objects']
        self.assertEqual(len(api_cases), 1)

        api_case = api_cases[0]
        self.assertEqual(api_case['server_date_modified'], json_format_datetime(backend_case.server_modified_on))

    @run_with_all_backends
    def test_parent_and_child_cases(self):
        fake_case_es = FakeXFormES(ESCase)
        v0_3.MOCK_CASE_ES = fake_case_es

        # Create cases
        parent_case_id = uuid.uuid4().hex
        parent_type = 'parent_case_type'
        parent_case = submit_case_blocks(
            CaseBlock.deprecated_init(
                case_id=parent_case_id,
                create=True,
                case_type=parent_type,
            ).as_text(),
            self.domain.name
        )[1][0]
        child_case_id = uuid.uuid4().hex
        child_case = submit_case_blocks(
            CaseBlock.deprecated_init(
                case_id=child_case_id,
                create=True,
                index={'parent': (parent_type, parent_case_id)}
            ).as_text(),
            self.domain.name
        )[1][0]

        fake_case_es.add_doc(parent_case_id, transform_case_for_elasticsearch(parent_case.to_json()))
        fake_case_es.add_doc(child_case_id, transform_case_for_elasticsearch(child_case.to_json()))

        # Fetch the child case through the API

        response = self._assert_auth_get_resource(self.single_endpoint(child_case_id) + "?parent_cases__full=true")
        self.assertEqual(
            response.status_code,
            200,
            "Status code was not 200. Response content was {}".format(response.content)
        )
        parent_cases = list(json.loads(response.content)['parent_cases'].values())

        # Confirm that the case appears in the resource
        self.assertEqual(len(parent_cases), 1)
        self.assertEqual(parent_cases[0]['id'], parent_case_id)

        # Fetch the parent case through the API

        response = self._assert_auth_get_resource(self.single_endpoint(parent_case_id) + "?child_cases__full=true")
        self.assertEqual(
            response.status_code,
            200,
            "Status code was not 200. Response content was {}".format(response.content)
        )<commit_msg>dont fake es in another test<commit_after>import json
import uuid
from datetime import datetime

from django.utils.http import urlencode

from casexml.apps.case.mock import CaseBlock
from casexml.apps.case.models import CommCareCase
from dimagi.utils.parsing import json_format_datetime

from corehq.apps.api.resources import v0_3, v0_4
from corehq.apps.domain.models import Domain
from corehq.apps.hqcase.utils import submit_case_blocks
from corehq.apps.users.models import WebUser
from corehq.elastic import get_es_new, send_to_elasticsearch
from corehq.form_processor.tests.utils import run_with_all_backends
from corehq.pillows.case import transform_case_for_elasticsearch
from corehq.pillows.mappings.case_mapping import CASE_INDEX_INFO
from corehq.util.elastic import reset_es_index
from pillowtop.es_utils import initialize_index_and_mapping

from .utils import APIResourceTest, FakeXFormES


class TestCommCareCaseResource(APIResourceTest):
    resource = v0_4.CommCareCaseResource

    def setUp(self):
        self.es = get_es_new()
        reset_es_index(CASE_INDEX_INFO)
        initialize_index_and_mapping(self.es, CASE_INDEX_INFO)

    def _setup_case(self):

        modify_date = datetime.utcnow()

        backend_case = CommCareCase(server_modified_on=modify_date, domain=self.domain.name)
        backend_case.save()
        self.addCleanup(backend_case.delete)

        translated_doc = transform_case_for_elasticsearch(backend_case.to_json())

        send_to_elasticsearch('cases', translated_doc)
        self.es.indices.refresh(CASE_INDEX_INFO.index)
        return backend_case

    def test_get_list(self):
        """
        Any case in the appropriate domain should be in the list from the API.
        """
        backend_case = self._setup_case()
        response = self._assert_auth_get_resource(self.list_endpoint)
        self.assertEqual(response.status_code, 200)

        api_cases = json.loads(response.content)['objects']
        self.assertEqual(len(api_cases), 1)

        api_case = api_cases[0]
        self.assertEqual(api_case['server_date_modified'], json_format_datetime(backend_case.server_modified_on))

    def test_get_list_format(self):
        """
        Any case in the appropriate domain should be in the list from the API.
        """
        backend_case = self._setup_case()

        # Get XML response
        response = self._assert_auth_get_resource(self.list_endpoint, headers={'HTTP_ACCEPT': 'application/xml'})
        self.assertEqual(response.status_code, 200)
        content = response.content.decode().split('\n', 1)[0]
        self.assertEqual(content, "<?xml version='1.0' encoding='utf-8'?>")

        # Force JSON response with `format=json` parameter
        response = self._assert_auth_get_resource(
            self.list_endpoint + '?format=json',
            headers={'HTTP_ACCEPT': 'application/xml'}
        )
        self.assertEqual(response.status_code, 200)

        # we should still get the same case even with the `format` param
        # https://dimagi-dev.atlassian.net/browse/HI-656
        api_cases = json.loads(response.content)['objects']
        self.assertEqual(len(api_cases), 1)

        api_case = api_cases[0]
        self.assertEqual(api_case['server_date_modified'], json_format_datetime(backend_case.server_modified_on))

    @run_with_all_backends
    def test_parent_and_child_cases(self):
        # Create cases
        parent_case_id = uuid.uuid4().hex
        parent_type = 'parent_case_type'
        parent_case = submit_case_blocks(
            CaseBlock.deprecated_init(
                case_id=parent_case_id,
                create=True,
                case_type=parent_type,
            ).as_text(),
            self.domain.name
        )[1][0]
        child_case_id = uuid.uuid4().hex
        child_case = submit_case_blocks(
            CaseBlock.deprecated_init(
                case_id=child_case_id,
                create=True,
                index={'parent': (parent_type, parent_case_id)}
            ).as_text(),
            self.domain.name
        )[1][0]

        self.addCleanup(child_case.delete)
        self.addCleanup(parent_case.delete)
        send_to_elasticsearch('cases', transform_case_for_elasticsearch(parent_case.to_json()))
        send_to_elasticsearch('cases', transform_case_for_elasticsearch(child_case.to_json()))
        self.es.indices.refresh(CASE_INDEX_INFO.index)

        # Fetch the child case through the API

        response = self._assert_auth_get_resource(self.single_endpoint(child_case_id) + "?parent_cases__full=true")
        self.assertEqual(
            response.status_code,
            200,
            "Status code was not 200. Response content was {}".format(response.content)
        )
        parent_cases = list(json.loads(response.content)['parent_cases'].values())

        # Confirm that the case appears in the resource
        self.assertEqual(len(parent_cases), 1)
        self.assertEqual(parent_cases[0]['id'], parent_case_id)

        # Fetch the parent case through the API

        response = self._assert_auth_get_resource(self.single_endpoint(parent_case_id) + "?child_cases__full=true")
        self.assertEqual(
            response.status_code,
            200,
            "Status code was not 200. Response content was {}".format(response.content)
        )
{'fraction_non_alphanumeric': 0.08894298286677277, 'fraction_numerical': 0.008987922479168617, 'mean_word_length': 3.0864575363427695, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 2, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 10}, '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': '4318367', 'n_tokens_mistral': 3219, 'n_tokens_neox': 3125, 'n_words': 596}