Adapt inddex tests to pytest

custom/inddex/example_data/data.py

@@ -30,13 +30,13 @@ def populate_inddex_domain(domain):
     user = _get_or_create_user(domain)
     _import_cases(domain, user)
     _import_fixtures(domain)
-    _rebuild_datasource(domain)
+    return _rebuild_datasource(domain)
 
 
-def _get_or_create_user(domain):
+def _get_or_create_user(domain, create=True):
     username = format_username('nick', domain)
     user = CommCareUser.get_by_username(username, strict=True)
-    if not user:
+    if not user and create:
         user = CommCareUser.create(domain, username, 'secret', None, None)
     return user
 
@@ -142,3 +142,4 @@ def _rebuild_datasource(domain):
     config_id = StaticDataSourceConfiguration.get_doc_id(
         domain, 'food_consumption_indicators')
     rebuild_indicators(config_id, source='populate_inddex_test_domain')
+    return config_id

custom/inddex/tests/test_master_report.py

@@ -1,26 +1,29 @@
 import csv
 import doctest
 import os
+from contextlib import ExitStack
 from datetime import date
 
 from django.test import SimpleTestCase, TestCase
 from django.utils.functional import cached_property
 
 from memoized import memoized
 from unittest.mock import patch
+from unmagic import autouse, fixture, get_fixture_value
 
 from dimagi.utils.dates import DateSpan
 
 import custom.inddex.reports.r4_nutrient_stats
-from corehq.apps.domain.models import Domain
 from corehq.apps.domain.shortcuts import create_domain
 from corehq.apps.fixtures.models import LookupTable
+from corehq.apps.userreports.util import get_indicator_adapter, get_ucr_datasource_config_by_id
 from corehq.form_processor.models import CommCareCase
-from corehq.util.test_utils import require_db_context
+from corehq.sql_db.util import get_db_aliases_for_partitioned_query
 
 from ..example_data.data import (
     FOOD_CASE_TYPE,
     FOODRECALL_CASE_TYPE,
+    _get_or_create_user,
     populate_inddex_domain,
 )
 from ..fixtures import FixtureAccessor
@@ -64,22 +67,37 @@ def _overwrite_report(filename, actual_report):
         writer.writerows(rows)
 
 
-@require_db_context
-def setUpModule():
-    create_domain(name=DOMAIN)
-    try:
-        with patch('corehq.apps.callcenter.data_source.get_call_center_domains', lambda: []):
-            populate_inddex_domain(DOMAIN)
-    except Exception:
-        tearDownModule()
-        raise
+@autouse(__file__)
+@fixture(scope='module')
+def inddex_domain():
+    def on_exit(callback):
+        cleanup.callback(with_db, callback)
 
+    def with_db(func):
+        with db_blocker.unblock():
+            func()
 
-@require_db_context
-def tearDownModule():
-    Domain.get_by_name(DOMAIN).delete()
-    get_food_data.reset_cache()
-    _get_case_ids_by_external_id.reset_cache()
+    db_blocker = get_fixture_value("django_db_blocker")
+    with ExitStack() as cleanup:
+        cleanup.callback(_get_case_ids_by_external_id.reset_cache)
+        cleanup.callback(get_food_data.reset_cache)
+
+        with db_blocker.unblock():
+            domain = create_domain(name=DOMAIN)
+            on_exit(domain.delete)
+
+            with patch('corehq.apps.callcenter.data_source.get_call_center_domains', lambda: []):
+                datasource_config_id = populate_inddex_domain(DOMAIN)
+                config = get_ucr_datasource_config_by_id(datasource_config_id)
+
+        on_exit(lambda: _get_or_create_user(domain.name, create=False).delete(None, None))
+        on_exit(LookupTable.objects.filter(domain=domain.name).delete)
+        on_exit(get_indicator_adapter(config).drop_table)
+
+        for dbname in get_db_aliases_for_partitioned_query():
+            on_exit(CommCareCase.objects.using(dbname).filter(domain=domain.name).delete)
+
+        yield
 
 
 @memoized