Add Support for Case Deletion

corehq/apps/hqcase/case_deletion_utils.py

@@ -0,0 +1,126 @@
+from attrs import define, field
+
+from casexml.apps.case.xform import get_case_updates
+
+from corehq.apps.hqwebapp.doc_info import get_case_url
+from corehq.form_processor.interfaces.processor import CaseUpdateMetadata
+
+
+@define
+class DeleteCase:
+    id = field()
+    name = field()
+    url = field()
+    is_primary = field(default=False)
+    delete_forms = field(factory=list)
+
+
+@define
+class DeleteForm:
+    id = field()
+    name = field()
+    url = field()
+    is_primary = field(default=False)
+    affected_cases = field(factory=list)
+
+
+@define
+class FormAffectedCases:
+    case_name = field(default=None)
+    is_current_case = field(default=False)
+    actions = field(factory=list)
+
+
+@define
+class AffectedCase:
+    id = field()
+    name = field()
+    url = field()
+    affected_forms = field(factory=list)
+
+
+def get_or_create_affected_case(domain, case, affected_cases_display, form_cache, case_block_cache):
+    """Note: affected_cases_display is additionally mutated by this function"""
+    for affected_case in affected_cases_display:
+        if affected_case.id == case.case_id:
+            return affected_case
+    if not case.name:
+        case.name = _get_deleted_case_name(case, form_cache, case_block_cache)
+    affected_case = AffectedCase(id=case.case_id, name=case.name, url=get_case_url(domain, case.case_id))
+    affected_cases_display.append(affected_case)
+    return affected_case
+
+
+@define
+class AffectedForm:
+    name = field()
+    url = field()
+    actions = field()
+    is_primary = field(default=False)
+
+
+@define
+class ReopenedCase:
+    name = field()
+    url = field()
+    closing_form_url = field()
+    closing_form_name = field()
+    closing_form_is_primary = field(default=False)
+
+
+def get_deduped_ordered_forms_for_case(case, form_cache):
+    """
+    Returns deduplicated and chronologically ordered case xforms, if not already that.
+    Returned forms are inclusive of forms from revoked CaseTransactions (necessary in order to include
+    archived forms in the deletion workflow), which the case.xform_ids method does not support.
+    """
+    revoked_inclusive_xform_ids = list({t.form_id for t in case.transactions if t.is_form_transaction})
+    xform_objs = form_cache.get_forms(revoked_inclusive_xform_ids)
+    return sorted(xform_objs, key=lambda form: form.received_on)
+
+
+def prepare_case_for_deletion(case, form_cache, case_block_cache):
+    if not case.is_deleted and case.deleted_on is None:
+        # Normal state - not archived nor deleted
+        return case
+    elif case.is_deleted and case.deleted_on is None:
+        # Create form was archived > create CaseTransaction revoked > case name unassigned
+        case.name = _get_deleted_case_name(case, form_cache, case_block_cache)
+        return case
+    elif case.deleted_on:
+        # Case was deleted through the proper deletion workflow, so there's no need to delete it again
+        return None
+
+
+def _get_deleted_case_name(case, form_cache, case_block_cache):
+    """When a case's create form is archived, its name is reset to '', so this process sets it again
+    to properly display on the case deletion page, but does not save it to the case object"""
+    for t in case.transactions:
+        if t.is_case_create:
+            form_list = form_cache.get_forms([t.form_id])
+            if form_list:
+                create_form = form_list[0]
+                break
+            continue
+    else:
+        return '[Unknown Case]'
+    case_blocks = case_block_cache.get_case_blocks(create_form)
+    for case_block in case_blocks:
+        if 'create' in case_block and case_block['@case_id'] == case.case_id:
+            return case_block['create']['case_name']
+
+
+def get_all_cases_from_form(form, case_cache, case_block_cache):
+    case_updates = get_case_updates(form, case_block_cache=case_block_cache)
+    update_ids = [update.id for update in case_updates]
+    all_cases = case_cache.get_cases(update_ids)
+    all_actions = [{action.action_type_slug for action in update.actions} for update in case_updates]
+
+    touched_cases = {}
+    for case, actions in zip(all_cases, all_actions):
+        case_update_meta = CaseUpdateMetadata(case, False, '', actions)
+        if case.case_id in touched_cases:
+            touched_cases[case.case_id] = touched_cases[case.case_id].merge(case_update_meta)
+        else:
+            touched_cases[case.case_id] = case_update_meta
+    return touched_cases

corehq/apps/hqcase/tests/test_utils.py

@@ -1,18 +1,30 @@
+import uuid
 from contextlib import contextmanager
 from datetime import datetime
 
 from django.test import TestCase
 
-from casexml.apps.case.mock import CaseFactory
+from casexml.apps.case.mock import CaseBlock, CaseFactory
+from casexml.apps.case.xform import TempCaseBlockCache
 
 from corehq.apps.export.const import DEID_DATE_TRANSFORM, DEID_ID_TRANSFORM
 from corehq.apps.hqcase.case_helper import CaseCopier
+from corehq.apps.hqcase.case_deletion_utils import (
+    get_all_cases_from_form,
+    _get_deleted_case_name,
+    get_deduped_ordered_forms_for_case,
+)
 from corehq.apps.hqcase.utils import (
     get_case_value,
     get_deidentified_data,
     is_copied_case,
+    submit_case_blocks,
 )
 from corehq.form_processor.tests.utils import create_case
+from corehq.apps.reports.tests.test_case_data import _delete_all_cases_and_forms
+from corehq.form_processor.models import CommCareCase
+from corehq.form_processor.models.forms import TempFormCache
+from corehq.form_processor.models.cases import TempCaseCache
 
 DOMAIN = 'test-domain'
 
@@ -104,6 +116,64 @@ def test_invalid_deid_transform_blanks_property(self):
         self.assertTrue(censored_props['captain'] == '')
 
 
+class TestCaseDeletionUtil(TestCase):
+
+    def make_case(self):
+        cases = {
+            'main_case_id': uuid.uuid4().hex,
+            'child_case_id': uuid.uuid4().hex,
+        }
+        xforms = {}
+        main_xform, _ = submit_case_blocks([
+            CaseBlock(cases['main_case_id'], case_name="main_case", create=True).as_text(),
+        ], DOMAIN)
+        xforms['main_xform'] = main_xform
+        child_xform, _ = submit_case_blocks([
+            CaseBlock(cases['main_case_id'], update={}).as_text(),
+            CaseBlock(cases['child_case_id'], case_name="child1", create=True).as_text(),
+        ], DOMAIN)
+        xforms['child_xform'] = child_xform
+        self.addCleanup(_delete_all_cases_and_forms, DOMAIN)
+
+        return cases, xforms
+
+    def test_get_xform_irrespective_of_archived_state(self):
+        cases, xforms = self.make_case()
+        xforms['child_xform'].archive()
+        main_case = CommCareCase.objects.get_case(cases['main_case_id'], DOMAIN)
+        ordered_xforms = get_deduped_ordered_forms_for_case(main_case, TempFormCache())
+
+        self.assertItemsEqual(ordered_xforms, list(xforms.values()))
+
+    def test_xform_list_is_ordered(self):
+        cases, xforms = self.make_case()
+        main_case = CommCareCase.objects.get_case(cases['main_case_id'], DOMAIN)
+        ordered_xforms = get_deduped_ordered_forms_for_case(main_case, TempFormCache())
+
+        self.assertEqual(ordered_xforms, sorted(list(xforms.values()), key=lambda f: f.received_on))
+
+    def test_xform_list_is_deduped(self):
+        cases, xforms = self.make_case()
+        main_case = CommCareCase.objects.get_case(cases['main_case_id'], DOMAIN)
+        ordered_xforms = get_deduped_ordered_forms_for_case(main_case, TempFormCache())
+
+        self.assertEqual(len({f.form_id for f in list(xforms.values())}), len(ordered_xforms))
+
+    def test_get_deleted_case_name(self):
+        cases, xforms = self.make_case()
+        xforms['main_xform'].archive()
+        case = CommCareCase.objects.get_case(cases['main_case_id'], DOMAIN)
+
+        self.assertEqual(_get_deleted_case_name(case, TempFormCache(), TempCaseBlockCache()), "main_case")
+
+    def test_get_cases_irrespective_of_deleted_state(self):
+        cases, xforms = self.make_case()
+        xforms['child_xform'].archive()
+        cases_from_form = get_all_cases_from_form(xforms['child_xform'], TempCaseCache(), TempCaseBlockCache())
+
+        self.assertItemsEqual(list(cases.values()), list(cases_from_form.keys()))
+
+
 @contextmanager
 def get_case(*args, **kwargs):
     factory = CaseFactory(DOMAIN)

corehq/apps/hqwebapp/doc_info.py

@@ -194,10 +194,7 @@ def form_docinfo(domain, doc_id, is_deleted):
         id=doc_id,
         type="XFormInstance",
         type_display=_('Form'),
-        link=reverse(
-            'render_form_data',
-            args=[domain, doc_id],
-        ),
+        link=get_form_url(domain, doc_id),
         is_deleted=is_deleted,
     )
     return doc_info
@@ -221,6 +218,13 @@ def get_case_url(domain, case_id):
     )
 
 
+def get_form_url(domain, form_id):
+    return reverse(
+        'render_form_data',
+        args=[domain, form_id],
+    )
+
+
 def get_commcareuser_url(domain, user_id):
     return reverse(
         'edit_commcare_user',

corehq/apps/reports/display.py

@@ -8,6 +8,7 @@
 from dimagi.utils.couch import IncompatibleDocument, get_cached_property
 from dimagi.utils.couch.safe_index import safe_index
 
+from corehq.apps.app_manager.dbaccessors import get_app
 from corehq.apps.hqcase.utils import SYSTEM_FORM_XMLNS_MAP
 from corehq.apps.users.models import CouchUser
 from corehq.const import USER_DATETIME_FORMAT_WITH_SEC
@@ -143,3 +144,26 @@ def append_form_name(self, name, separator):
 
 def xmlns_to_name(domain, xmlns, app_id, lang=None, separator=None, form_name=None):
     return _FormType(domain, xmlns, app_id, form_name).get_label(lang, separator)
+
+
+def xmlns_to_name_for_case_deletion(domain, form):
+    """
+    The difference between this function and the one above is that in the event that the form name
+    can't be found, this function will attempt to recreate the standard 3 part structure, rather than
+    defaulting to the xmlns url (form.xmlns). Currently only used in the case deletion workflow.
+
+    TODO: Confirm it returning the form.xmlns isn't necessary + confirm this is the format we want to display
+          for unknown form names in report tables and merge the two functions into one.
+    """
+    form_name = xmlns_to_name(domain, form.xmlns, form.app_id)
+    if form_name == form.xmlns:
+        app_name = None
+        if form.app_id:
+            app_name = get_app(domain, form.app_id).name
+        extracted_name = [
+            app_name or "[Unknown App]",
+            "[Unknown Module]",
+            form.name or "[Unknown Form]"
+        ]
+        form_name = ' > '.join(extracted_name)
+    return form_name

corehq/apps/reports/exceptions.py

@@ -20,3 +20,7 @@ class TableauAPIError(Exception):
     def __init__(self, message, code=None):
         self.code = int(code) if code else None
         super().__init__(message)
+
+
+class TooManyCases(ValueError):
+    pass