test(geo): (WIP!) add e2e integration tests for geo features

tests/conftest.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 import pytest
 
 from odata_query.grammar import ODataLexer, ODataParser
@@ -11,3 +13,11 @@ def lexer():
 @pytest.fixture
 def parser():
     return ODataParser()
+
+
+@pytest.fixture(scope="session")
+def data_dir():
+    data_dir_path = Path(__file__).parent / "data"
+    data_dir_path.mkdir(exist_ok=True)
+
+    return data_dir_path

tests/integration/django/apps.py

@@ -5,3 +5,32 @@ class ODataQueryConfig(AppConfig):
     name = "tests.integration.django"
     verbose_name = "OData Query Django test app"
     default = True
+
+
+class DbRouter:
+    """
+    Ensure that GeoDjango models go to the SpatiaLite database, while other
+    models use the default SQLite database.
+    """
+
+    GEO_APP = "django_geo"
+
+    def db_for_read(self, model, **hints):
+        if model._meta.app_label == self.GEO_APP:
+            return "geo"
+        return None
+
+    def db_for_write(self, model, **hints):
+        if model._meta.app_label == self.GEO_APP:
+            return "geo"
+        return None
+
+    def allow_relation(self, obj1, obj2, **hints):
+        return obj1._meta.app_label == obj2._meta.app_label
+
+    def allow_migrate(self, db: str, app_label: str, model_name=None, **hints):
+        if app_label != self.GEO_APP and db == "default":
+            return True
+        if app_label == self.GEO_APP and db == "geo":
+            return True
+        return False

tests/integration/django/conftest.py

@@ -5,3 +5,4 @@
 @pytest.fixture(scope="session")
 def django_db():
     management.call_command("migrate", "--run-syncdb")
+    management.call_command("migrate", "--run-syncdb", "--database", "geo")

tests/integration/django/settings.py

@@ -2,7 +2,19 @@
     "default": {
         "ENGINE": "django.db.backends.sqlite3",
         "NAME": "odata-query",
-    }
+    },
+    "geo": {
+        "ENGINE": "django.contrib.gis.db.backends.spatialite",
+        "NAME": "odata-query-geo",
+    },
 }
+DATABASE_ROUTERS = ["tests.integration.django.apps.DbRouter"]
 DEBUG = True
-INSTALLED_APPS = ["tests.integration.django.apps.ODataQueryConfig"]
+INSTALLED_APPS = [
+    "tests.integration.django.apps.ODataQueryConfig",
+    # GEO:
+    "django.contrib.gis",
+    "tests.integration.django_geo.apps.ODataQueryConfig",
+]
+
+# GDAL_LIBRARY_PATH = "/opt/homebrew/lib/libgdal.dylib"

tests/integration/django_geo/apps.py

@@ -0,0 +1,7 @@
+from django.apps import AppConfig
+
+
+class ODataQueryConfig(AppConfig):
+    name = "tests.integration.django_geo"
+    verbose_name = "OData Query GeoDjango test app"
+    default = True

tests/integration/django_geo/conftest.py

@@ -0,0 +1,37 @@
+import urllib.request as req
+from pathlib import Path
+from zipfile import ZipFile
+
+import pytest
+from django.core import management
+
+
+@pytest.fixture(scope="session")
+def django_db():
+    management.call_command("migrate", "--run-syncdb")
+
+
+@pytest.fixture(scope="session")
+def world_borders_dataset(data_dir: Path):
+    target_dir = data_dir / "world_borders"
+
+    if target_dir.exists():
+        return target_dir
+
+    filename_zip = target_dir.with_suffix(".zip")
+
+    if not filename_zip.exists():
+        breakpoint()
+        opener = req.build_opener()
+        opener.addheaders = [("Accept", "application/zip")]
+        req.install_opener(opener)
+        req.urlretrieve(
+            "https://thematicmapping.org/downloads/TM_WORLD_BORDERS-0.3.zip",
+            filename_zip,
+        )
+        assert filename_zip.exists()
+
+    with ZipFile(filename_zip, "r") as z:
+        z.extractall(target_dir)
+
+    return target_dir

tests/integration/django_geo/models.py

@@ -0,0 +1,25 @@
+# https://docs.djangoproject.com/en/4.2/ref/contrib/gis/tutorial/#geographic-models
+
+from django.contrib.gis.db import models
+
+
+class WorldBorder(models.Model):
+    # Regular Django fields corresponding to the attributes in the
+    # world borders shapefile.
+    name = models.CharField(max_length=50)
+    area = models.IntegerField()
+    pop2005 = models.IntegerField("Population 2005")
+    fips = models.CharField("FIPS Code", max_length=2, null=True)
+    iso2 = models.CharField("2 Digit ISO", max_length=2)
+    iso3 = models.CharField("3 Digit ISO", max_length=3)
+    un = models.IntegerField("United Nations Code")
+    region = models.IntegerField("Region Code")
+    subregion = models.IntegerField("Sub-Region Code")
+    lon = models.FloatField()
+    lat = models.FloatField()
+
+    # GeoDjango-specific: a geometry field (MultiPolygonField)
+    mpoly = models.MultiPolygonField()
+
+    def __str__(self):
+        return self.name

tests/integration/django_geo/test_querying.py

@@ -0,0 +1,64 @@
+from pathlib import Path
+from typing import Type
+
+import pytest
+from django.contrib.gis.db import models
+from django.contrib.gis.utils import LayerMapping
+
+from odata_query.django import apply_odata_query
+
+from .models import WorldBorder
+
+# The default spatial reference system for geometry fields is WGS84
+# (meaning the SRID is 4326)
+SRID = "SRID=4326"
+
+
+@pytest.fixture(scope="session")
+def sample_data_sess(django_db, world_borders_dataset: Path):
+    world_mapping = {
+        "fips": "FIPS",
+        "iso2": "ISO2",
+        "iso3": "ISO3",
+        "un": "UN",
+        "name": "NAME",
+        "area": "AREA",
+        "pop2005": "POP2005",
+        "region": "REGION",
+        "subregion": "SUBREGION",
+        "lon": "LON",
+        "lat": "LAT",
+        "mpoly": "MULTIPOLYGON",
+    }
+
+    world_shp = world_borders_dataset / "TM_WORLD_BORDERS-0.3.shp"
+    lm = LayerMapping(WorldBorder, world_shp, world_mapping, transform=False)
+    lm.save(strict=True, verbose=True)
+    yield
+    WorldBorder.objects.all().delete()
+
+
+@pytest.mark.parametrize(
+    "model, query, exp_results",
+    [
+        (
+            WorldBorder,
+            "geo.length(mpoly) gt 1000000",
+            154,
+        ),
+        (
+            WorldBorder,
+            f"geo.intersects(mpoly, geography'{SRID};Point(-95.3385 29.7245)')",
+            1,
+        ),
+    ],
+)
+def test_query_with_odata(
+    model: Type[models.Model],
+    query: str,
+    exp_results: int,
+    sample_data_sess,
+):
+    q = apply_odata_query(model.objects, query)
+    results = q.all()
+    assert len(results) == exp_results

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = py37-django3, py{38,39,310}-django{3,4}, linting, docs
+envlist = py37-django3, py{38,39,310,311}-django{3,4}, linting, docs
 skip_missing_interpreters = True
 isolated_build = True
 
@@ -9,6 +9,7 @@ python =
     3.8: py38, linting, docs
     3.9: py39
     3.10: py310
+    3.11: py311
 
 [testenv:linting]
 basepython = python3.8