Add django application in tests that can be used to run django ORM in pytests tests seamlessly.

 Other drivers use contrived setups most that most likely stem from the django reusable apps testing strategy. The way we do it allows for total pytest compatability, we also added two fixtures to automatically delete any row added in test and if necessary, to drop the tables that the migrations created. The commands `makemigrations` and `migrate` are automatically called in every test run.

Author: surister

tests/conftest.py

@@ -0,0 +1,61 @@
+import logging
+
+import pytest
+import os, django
+from django.db import connection
+
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "tests.settings")
+django.setup()
+
+
+try:
+    from django.core.management import execute_from_command_line
+except ImportError as exc:
+    raise ImportError(
+        "Couldn't import Django. Are you sure it's installed and "
+        "available on your PYTHONPATH environment variable? Did you "
+        "forget to activate a virtual environment?"
+    ) from exc
+
+# Runs migrations at the beginning of every full test run, to check that migrations can run.
+execute_from_command_line(["manage.py", "makemigrations"])
+execute_from_command_line(["manage.py", "migrate"])
+
+logging.info("All migrations run successfully")
+
+
+@pytest.fixture(scope="function", autouse=True)
+def clean_database(request):
+    """After every test removes all the rows inserted in the model."""
+
+    yield
+
+    models = [
+        model
+        for model in request.module.__dict__.values()
+        if isinstance(model, django.db.models.base.ModelBase)
+    ]
+    for model in models:
+        with connection.cursor() as cursor:
+            cursor.execute(f"DELETE FROM {model._meta.db_table}")
+            cursor.execute(f"REFRESH TABLE {model._meta.db_table}")
+
+
+@pytest.fixture(scope="session", autouse=False)
+def cleanup_migrations():
+    """
+    Removes every table created from the migrations, not necessary in CI but very useful
+    in local while debugging migrations and model field creations, set autouse=True to use it.
+    :return:
+    """
+    yield
+
+    models = [
+        model
+        for model in request.module.__dict__.values()
+        if isinstance(model, django.db.models.base.ModelBase)
+    ]
+    for model in models:
+        with connection.cursor() as cursor:
+            cursor.execute(f"DROP TABLE IF EXISTS {model._meta.db_table}")

tests/settings.py

@@ -0,0 +1,51 @@
+# Default configuration, in the test we might change them to test logic.
+DATABASES = {
+    "default": {
+        "ENGINE": "cratedb_django",
+        "SERVERS": ["192.168.88.251:4200"],
+    }
+}
+
+
+# Below are all necessary settings to load typical django migrations,
+# which we use to test migrations
+
+SECRET_KEY = "fake-key"
+
+INSTALLED_APPS = [
+    "tests.test_app",
+    "django.contrib.admin",
+    "django.contrib.auth",
+    "django.contrib.contenttypes",
+    "django.contrib.sessions",
+    "django.contrib.messages",
+    "django.contrib.staticfiles",
+]
+
+MIDDLEWARE = [
+    "django.middleware.security.SecurityMiddleware",
+    "django.contrib.sessions.middleware.SessionMiddleware",
+    "django.middleware.common.CommonMiddleware",
+    "django.middleware.csrf.CsrfViewMiddleware",
+    "django.contrib.auth.middleware.AuthenticationMiddleware",
+    "django.contrib.messages.middleware.MessageMiddleware",
+    "django.middleware.clickjacking.XFrameOptionsMiddleware",
+]
+
+TEMPLATES = [
+    {
+        "BACKEND": "django.template.backends.django.DjangoTemplates",
+        "DIRS": [],
+        "APP_DIRS": True,
+        "OPTIONS": {
+            "context_processors": [
+                "django.template.context_processors.debug",
+                "django.template.context_processors.request",
+                "django.contrib.auth.context_processors.auth",
+                "django.contrib.messages.context_processors.messages",
+            ],
+        },
+    },
+]
+
+DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"

tests/test_app/__init__.py

@@ -0,0 +1,4 @@
+"""
+This module is a Django application that can contain models, views and urls, we only
+use it to register models and test the ORM.
+"""

tests/test_app/migrations/__init__.py

@@ -0,0 +1,50 @@
+import datetime, uuid
+
+from django.db import models
+
+from cratedb_django import CrateModel
+from cratedb_django.models import ObjectField
+
+
+class AllFieldsModel(CrateModel):
+    field_int = models.IntegerField(unique=False)
+    field_int_unique = models.IntegerField(unique=True)
+    field_int_not_indexed = models.IntegerField(db_index=False)
+    field_int_not_null = models.IntegerField(null=False)
+    field_int_null = models.IntegerField(null=True)
+    field_int_default = models.IntegerField(default=54321)
+    field_float = models.FloatField()
+    field_char = models.CharField(max_length=100)
+    field_bool = models.BooleanField()
+    # field_date = models.DateField()
+    # field_datetime = models.DateTimeField()
+    field_json = ObjectField(default=dict)
+    field_uuid = models.UUIDField()
+
+    @classmethod
+    def create_test(cls):
+        return cls.objects.create(
+            field_int=1,
+            field_int_unique=2,
+            field_int_not_indexed=3,
+            field_int_not_null=4,
+            field_int_null=5,
+            field_int_default=6,
+            field_float=0.1,
+            field_char="somechar",
+            field_bool=True,
+            field_date=datetime.datetime.today().date(),
+            field_datetime=datetime.datetime.today(),
+            field_json={"hello": "world"},
+            field_uuid=uuid.uuid4(),
+        )
+
+    class Meta:
+        app_label = "test_app"
+
+
+class SimpleModel(CrateModel):
+    field = models.TextField()
+
+    class Meta:
+        app_label = "test_app"

tests/test_app/models.py

@@ -1,59 +1,37 @@
-import datetime
-
 from django.forms.models import model_to_dict
-
-try:
-    from example.apps.simple_auth.models import MigrationTest
-except:
-    from apps.simple_auth.models import MigrationTest
-
-from tests import NoTransactionTestCase
-
-
-class MigrationTestTests(NoTransactionTestCase):
-    def tearDown(self):
-        # Clean up after each test
-        MigrationTest.objects.all().delete()
-        MigrationTest.refresh()
-        # pass
-
-    def test_simple_create(self):
-        """Test that an object is created and accounted for with all fields"""
-        assert MigrationTest.objects.count() == 0
-        expected = {
-            "id": 29147646,
-            "field_int": 1,
-            "field_int_unique": 2,
-            "field_int_not_indexed": 3,
-            "field_int_not_null": 4,
-            "field_int_null": 5,
-            "field_int_default": 6,
-            "field_float": 0.1,
-            "field_char": "somechar",
-            "field_bool": True,
-            "field_date": datetime.datetime(2025, 4, 23, 0, 0),
-            "field_datetime": datetime.datetime(2025, 4, 23, 14, 20, 15, 833000),
-            "field_json": {"hello": "world"},
-            "field_uuid": "00bde3702f844402b750c1b37d589084",
-        }
-        MigrationTest.objects.create(**expected)
-        MigrationTest.refresh()
-        assert MigrationTest.objects.count() == 1
-
-        obj = MigrationTest.objects.get()
-        assert model_to_dict(obj) == expected
-
-    def test_unique_duplicates(self):
-        # CrateDB does not support the unique constraint, creating two test objects which
-        # has static values should not error.
-        MigrationTest.create_test()
-        MigrationTest.create_test()
-        MigrationTest.refresh()
-        assert MigrationTest.objects.count() == 2
-
-    def test_random_id(self):
-        MigrationTest.create_test()
-        MigrationTest.refresh()
-        obj = MigrationTest.objects.get()
-        assert hasattr(obj, 'id')
-        assert isinstance(obj.id, int)
+from tests.test_app.models import AllFieldsModel, SimpleModel
+
+
+def test_insert_model_field():
+    """Test that we can insert a model and refresh it"""
+    assert SimpleModel.objects.count() == 0
+    SimpleModel.objects.create(field="text")
+    SimpleModel.refresh()
+    assert SimpleModel.objects.count() == 1
+
+
+def test_insert_all_fields():
+    """Test that an object is created and accounted for with all fields"""
+
+    expected = {
+        "id": 29147646,
+        "field_int": 1,
+        "field_int_unique": 2,
+        "field_int_not_indexed": 3,
+        "field_int_not_null": 4,
+        "field_int_null": 5,
+        "field_int_default": 6,
+        "field_float": 0.1,
+        "field_char": "somechar",
+        "field_bool": True,
+        # "field_date": datetime.datetime(2025, 4, 22, 0, 0, tzinfo=datetime.timezone.utc),
+        # "field_datetime": datetime.datetime(1, 1, 1, 1, 1, 1, 1),
+        "field_json": {"hello": "world"},
+        "field_uuid": "00bde3702f844402b750c1b37d589084",
+    }
+    AllFieldsModel.objects.create(**expected)
+    AllFieldsModel.refresh()
+    assert AllFieldsModel.objects.count() == 1
+
+    obj = AllFieldsModel.objects.get()
+    assert model_to_dict(obj) == expected