Adapt champ tests to pytest

custom/champ/tests/__init__.py

@@ -1,63 +0,0 @@
-import os
-
-from django.test.utils import override_settings
-
-from unittest import mock
-import postgres_copy
-import sqlalchemy
-
-from corehq.apps.domain.models import Domain
-from corehq.apps.domain.shortcuts import create_domain
-from corehq.apps.userreports.models import StaticDataSourceConfiguration
-from corehq.apps.userreports.util import get_indicator_adapter, get_table_name
-from corehq.sql_db.connections import UCR_ENGINE_ID, connection_manager
-from corehq.util.test_utils import require_db_context
-
-
-@require_db_context
-def setUpModule():
-    _call_center_domain_mock = mock.patch(
-        'corehq.apps.callcenter.data_source.call_center_data_source_configuration_provider'
-    )
-    _call_center_domain_mock.start()
-
-    domain = create_domain('champ-cameroon')
-
-    try:
-        configs = StaticDataSourceConfiguration.by_domain(domain.name)
-        adapters = [get_indicator_adapter(config) for config in configs]
-
-        for adapter in adapters:
-            adapter.build_table()
-
-        engine = connection_manager.get_engine(UCR_ENGINE_ID)
-        metadata = sqlalchemy.MetaData(bind=engine)
-        metadata.reflect(bind=engine, extend_existing=True)
-        path = os.path.join(os.path.dirname(__file__), 'fixtures')
-        for file_name in os.listdir(path):
-            with open(os.path.join(path, file_name), encoding='utf-8') as f:
-                table_name = get_table_name(domain.name, file_name[:-4])
-                table = metadata.tables[table_name]
-                postgres_copy.copy_from(f, table, engine, format='csv', null='', header=True)
-    except Exception:
-        tearDownModule()
-        raise
-
-    _call_center_domain_mock.stop()
-
-
-@require_db_context
-def tearDownModule():
-    _call_center_domain_mock = mock.patch(
-        'corehq.apps.callcenter.data_source.call_center_data_source_configuration_provider'
-    )
-    _call_center_domain_mock.start()
-
-    configs = StaticDataSourceConfiguration.by_domain('champ-cameroon')
-    adapters = [get_indicator_adapter(config) for config in configs]
-
-    for adapter in adapters:
-        adapter.drop_table()
-
-    Domain.get_by_name('champ-cameroon').delete()
-    _call_center_domain_mock.stop()

custom/champ/tests/utils.py

@@ -1,21 +1,67 @@
 import json
-from django.test.testcases import TestCase
+import os
+from contextlib import ExitStack
+from unittest import mock
+
 from django.test.client import RequestFactory
-from django.test.testcases import SimpleTestCase
+from django.test.testcases import SimpleTestCase, TestCase
+
+import postgres_copy
+import sqlalchemy
 from fakecouch import FakeCouchDb
+from unmagic import fixture, get_fixture_value
 
 from corehq.apps.accounting.models import SoftwarePlanEdition
 from corehq.apps.accounting.tests.utils import DomainSubscriptionMixin
 from corehq.apps.accounting.utils import clear_plan_version_cache
-from corehq.apps.users.models import WebUser
-
-from corehq.apps.domain.models import Domain
+from corehq.apps.domain.shortcuts import create_domain
 from corehq.apps.userreports.expressions.factory import ExpressionFactory
 from corehq.apps.userreports.filters.factory import FilterFactory
-from corehq.apps.userreports.models import DataSourceConfiguration
+from corehq.apps.userreports.models import (
+    DataSourceConfiguration,
+    StaticDataSourceConfiguration,
+)
 from corehq.apps.userreports.specs import FactoryContext
-from corehq.apps.users.models import CommCareUser
-import os
+from corehq.apps.userreports.util import get_indicator_adapter, get_table_name
+from corehq.apps.users.models import CommCareUser, WebUser
+from corehq.sql_db.connections import UCR_ENGINE_ID, connection_manager
+
+
+@fixture(scope="package")
+def create_domain_with_ucr_fixtures():
+    # ucr_fixtures will be setup for the first test that uses it (a
+    # subclass of ChampTestCase) and torn down after all tests in the
+    # "champ.tests" package have been run.
+
+    def with_db(func):
+        with db_blocker.unblock():
+            func()
+
+    db_blocker = get_fixture_value("django_db_blocker")
+    _call_center_domain_mock = mock.patch(
+        'corehq.apps.callcenter.data_source.call_center_data_source_configuration_provider'
+    )
+    with _call_center_domain_mock, ExitStack() as on_exit:
+        domain = create_domain('champ-cameroon')
+        on_exit.callback(with_db, domain.delete)
+
+        configs = StaticDataSourceConfiguration.by_domain(domain.name)
+        adapters = [get_indicator_adapter(config) for config in configs]
+        for adapter in adapters:
+            adapter.build_table()
+            on_exit.callback(with_db, adapter.drop_table)
+
+        engine = connection_manager.get_engine(UCR_ENGINE_ID)
+        metadata = sqlalchemy.MetaData(bind=engine)
+        metadata.reflect(bind=engine, extend_existing=True)
+        path = os.path.join(os.path.dirname(__file__), 'fixtures')
+        for file_name in os.listdir(path):
+            with open(os.path.join(path, file_name), encoding='utf-8') as f:
+                table_name = get_table_name(domain.name, file_name[:-4])
+                table = metadata.tables[table_name]
+                postgres_copy.copy_from(f, table, engine, format='csv', null='', header=True)
+
+        yield domain
 
 
 class ChampTestCase(TestCase, DomainSubscriptionMixin):
@@ -24,8 +70,7 @@ class ChampTestCase(TestCase, DomainSubscriptionMixin):
     def setUpClass(cls):
         super().setUpClass()
         cls.factory = RequestFactory()
-        # gets created + removed in package level setup / teardown
-        domain = Domain.get_or_create_with_name('champ-cameroon')
+        domain = create_domain_with_ucr_fixtures.get_value()
         domain.is_active = True
         domain.save()
         cls.domain = domain