test(utils): cover queue enqueue forwarding

Author: bolinocroustibat

tests/test_analysis/test_analysis_csv.py

@@ -171,6 +171,49 @@ async def test_analyse_csv_send_udata_webhook(
         assert webhook.get(f"analysis:parsing:{k}", False) is None
 
 
+async def test_analyse_csv_enqueues_export_jobs_on_low_queue(
+    mocker, setup_catalog, rmock, catalog_content, db, fake_check, produce_mock
+):
+    """Parquet export is scheduled on the low RQ queue (CSV geo export disabled)."""
+    import asyncio
+
+    recorded: list[tuple[object, str | None]] = []
+
+    async def tracking_parquet_export(*args, **kwargs):
+        pass
+
+    mocker.patch("udata_hydra.analysis.csv.export_parquet", tracking_parquet_export)
+
+    def capture_enqueue(fn, *args, **kwargs):
+        recorded.append((fn, kwargs.get("_priority")))
+        kwargs = dict(kwargs)
+        kwargs.pop("_priority", None)
+        kwargs.pop("_exception", None)
+        result = fn(*args, **kwargs)
+        if asyncio.iscoroutine(result):
+            loop = asyncio.get_running_loop()
+            return loop.run_until_complete(result)
+        return result
+
+    mocker.patch("udata_hydra.utils.queue.enqueue", capture_enqueue)
+
+    check = await fake_check()
+    url = check["url"]
+    rmock.get(url, status=200, body=catalog_content)
+    with (
+        patch("udata_hydra.config.DB_TO_PARQUET", True),
+        patch("udata_hydra.config.MIN_LINES_FOR_PARQUET", 1),
+        patch("udata_hydra.config.DB_TO_GEOJSON", False),
+    ):
+        await analyse_csv(check=check)
+
+    assert any(f is tracking_parquet_export and p == "low" for f, p in recorded)
+
+    from udata_hydra.analysis.exports import export_geojson_pmtiles as exp_geo
+
+    assert not any(f is exp_geo for f, _ in recorded)
+
+
 @pytest.mark.parametrize(
     "forced_analysis",
     (

tests/test_utils/test_queue.py

@@ -0,0 +1,18 @@
+from unittest.mock import MagicMock
+
+from udata_hydra import config
+from udata_hydra.utils.queue import enqueue
+
+
+def test_enqueue_forwards_priority_exception_and_failure_ttl(mocker) -> None:
+    """Check that enqueue passes priority, exception flag, and TTL to the RQ queue.
+    Hydra uses custom _priority/_exception kwargs; this test ensures they reach Redis/RQ."""
+    rq_queue = MagicMock()
+    context_queue = mocker.patch("udata_hydra.utils.queue.context.queue", return_value=rq_queue)
+
+    enqueue(lambda: None, key="val", _priority="low", _exception=True)
+
+    context_queue.assert_called_once_with("low", True)
+    rq_queue.enqueue.assert_called_once_with(
+        mocker.ANY, key="val", failure_ttl=config.RQ_DEFAULT_FAILURE_TTL
+    )