diff --git a/docs/howto/django/tests.md b/docs/howto/django/tests.md
index bb44601fe..bc829617b 100644
--- a/docs/howto/django/tests.md
+++ b/docs/howto/django/tests.md
@@ -101,9 +101,76 @@ don't leak connections at test database teardown.
 
 :::
 
-In addition, you can also run a worker in your integration tests. Whether you
-use `pytest-django` or Django's `TestCase` subclasses, this requires some
-additonal configuration.
+In addition, you can also run a worker in your integration tests.
+
+### DjangoTestingConnector and Pytest Plugin
+
+If you need to test jobs while being closer to reality—like modifying job scheduling via the ORM—you can use the `DjangoTestingConnector`. This connector leverages your test database but handles `listen/notify` in-memory.
+
+To make things easier, Procrastinate provides a pytest plugin (automatically enabled if both `pytest` and `django` are installed). It offers the `run_procrastinate_jobs` and `arun_procrastinate_jobs` fixtures, which act as a shortcut to replace the connector and run the worker.
+
+Here is an example:
+
+```python
+import pytest
+from procrastinate.contrib.django.models import ProcrastinateJob
+from mypackage.procrastinate import my_task
+
+@pytest.mark.django_db(transaction=True)
+def test_task(run_procrastinate_jobs):
+    # Run tasks
+    my_task.defer(a=1, b=2)
+
+    # You can interact with jobs using the ORM to mimic reality
+    # (e.g. changing scheduled_at or statuses)
+
+    # Process awaiting jobs
+    run_procrastinate_jobs()
+
+    # Check task has been executed
+    assert ProcrastinateJob.objects.filter(task_name="my_task").first().status == "succeeded"
+
+@pytest.mark.asyncio
+@pytest.mark.django_db(transaction=True)
+async def test_async_task(arun_procrastinate_jobs):
+    await my_task.defer_async(a=1, b=2)
+    await arun_procrastinate_jobs()
+```
+
+### Time traveling with freezegun
+
+You can also use tools like `freezegun` with `run_procrastinate_jobs` to test scheduled jobs by traveling through time. Note that if you intend to travel through time or modify jobs via ORM, you must have `transaction=True` on your `django_db` marker.
+
+```python
+import datetime
+import pytest
+import freezegun
+from procrastinate.contrib.django.models import ProcrastinateJob
+from mypackage.procrastinate import my_task
+
+@pytest.mark.django_db(transaction=True)
+def test_task_time_travel(run_procrastinate_jobs):
+    with freezegun.freeze_time("2025-01-01T00:00:00Z"):
+        my_task.defer(a=1, b=2)
+
+        # Modify the job via ORM to schedule it for tomorrow
+        ProcrastinateJob.objects.update(
+            scheduled_at=datetime.datetime(2025, 1, 2, 0, 0, 0, tzinfo=datetime.timezone.utc)
+        )
+
+        # Worker shouldn't pick it up yet
+        run_procrastinate_jobs()
+        assert ProcrastinateJob.objects.filter(status="todo").exists()
+
+    with freezegun.freeze_time("2025-01-02T01:00:00Z"):
+        # The job is now ready to be processed
+        run_procrastinate_jobs()
+        assert ProcrastinateJob.objects.filter(status="succeeded").exists()
+```
+
+### Manual configuration without the pytest plugin
+
+If you use `pytest-django` without using the fixtures, or Django's `TestCase` subclasses, running a worker requires some additional configuration.
 
 1. In order to run the worker, use the syntax outlined here: {doc}`scripts`.
 2. In order for Procrastinate to be able to use `SELECT FOR UPDATE`, use
@@ -121,6 +188,7 @@ additonal configuration.
 
 ```python
 from procrastinate.contrib.django import app
+from procrastinate.contrib.django.testing import DjangoTestingConnector
 from django.test import TransactionTestCase
 
 from mypackage.procrastinate import my_task
@@ -131,48 +199,11 @@ class TestingTaskClass(TransactionTestCase):
         my_task.defer(a=1, b=2)
 
         # Start worker
-        with app.replace_connector(app.connector.get_worker_connector()):
+        with app.replace_connector(DjangoTestingConnector()):
             app.run_worker(wait=False, install_signal_handlers=False, listen_notify=False)
 
         # Check task has been executed
-        assert ProcrastinateJob.objects.filter(task_name="my_task").status == "succeeded"
-```
-
-```python
-from procrastinate.contrib.django import app
-
-from mypackage.procrastinate import my_task
-
-@pytest.mark.django_db(transaction=True)
-def test_task():
-    # Run tasks
-    my_task.defer(a=1, b=2)
-
-    # Start worker
-    with app.replace_connector(app.connector.get_worker_connector()):
-        app.run_worker(wait=False, install_signal_handlers=False, listen_notify=False)
-
-    # Check task has been executed
-    assert ProcrastinateJob.objects.filter(task_name="my_task").status == "succeeded"
-
-# Or with a fixture
-@pytest.fixture
-def worker(transactional_db):
-    with app.replace_connector(app.connector.get_worker_connector()):
-        def f():
-            app.run_worker(wait=False, install_signal_handlers=False, listen_notify=False)
-            return app
-    yield f
-
-def test_task(worker):
-    # Run tasks
-    my_task.defer(a=1, b=2)
-
-    # Start worker
-    worker()
-
-    # Check task has been executed
-    assert ProcrastinateJob.objects.filter(task_name="my_task").status == "succeeded"
+        assert ProcrastinateJob.objects.filter(task_name="my_task").first().status == "succeeded"
 ```
 
 ## Making the models writable in tests
diff --git a/procrastinate/contrib/django/testing.py b/procrastinate/contrib/django/testing.py
new file mode 100644
index 000000000..cdbe98cb3
--- /dev/null
+++ b/procrastinate/contrib/django/testing.py
@@ -0,0 +1,186 @@
+from __future__ import annotations
+
+import datetime
+import json
+from collections.abc import Iterable
+from typing import Any
+
+from typing_extensions import LiteralString
+
+from procrastinate import connector, jobs, sql, utils
+from procrastinate.contrib.django import django_connector
+
+
+class DjangoTestingConnector(django_connector.DjangoConnector):
+    """
+    A testing connector for Django applications.
+
+    While `InMemoryConnector` is useful for fast unit tests without a database,
+    integration tests typically require hitting the database to test the full
+    application flow. However, running a standard `PsycopgConnector` worker
+    inside a Django test (which is often wrapped in a transaction) can cause
+    deadlocks or fail to see uncommitted data.
+
+    This connector exists to solve this problem. It uses the exact same database
+    connection as the Django test, allowing the Procrastinate worker to run inside
+    the same transaction. This means users can interact with the Procrastinate ORM
+    models (e.g. `ProcrastinateJob.objects.update(...)`) to manipulate job states
+    or schedules before the worker runs, providing an environment closer to reality.
+
+    It simulates the PostgreSQL `LISTEN/NOTIFY` system in-memory since Django's
+    transactional tests cannot rely on real database notifications.
+
+    This class is primarily used by developers writing integration tests for their
+    Django applications that depend on Procrastinate, often implicitly via the
+    provided pytest plugin fixtures.
+    """
+
+    def __init__(self, alias: str = "default") -> None:
+        super().__init__(alias=alias)
+        self.on_notification: connector.Notify | None = None
+        self.notify_channels: list[str] = []
+        self._time_override_value: datetime.datetime | None = None
+
+    async def listen_notify(
+        self, on_notification: connector.Notify, channels: Iterable[str]
+    ) -> None:
+        self.on_notification = on_notification
+        self.notify_channels = list(channels)
+
+    async def _notify(self, queue_name: str, notification: jobs.Notification) -> None:
+        if not self.on_notification:
+            return
+
+        destination_channels = {
+            "procrastinate_any_queue_v1",
+            f"procrastinate_queue_v1#{queue_name}",
+        }
+
+        for channel in set(self.notify_channels).intersection(destination_channels):
+            await self.on_notification(
+                channel=channel,
+                payload=json.dumps(notification),
+            )
+
+    def _override_time(self, time: datetime.datetime) -> None:
+        """
+        Sets the database time for the current connection to a static value.
+        """
+        if self._time_override_value is not None and self._time_override_value == time:
+            return
+
+        time_str = time.isoformat()
+        with self.connection.cursor() as cursor:
+            cursor.execute(
+                sql.queries["testing_override_time"].format(time_str=time_str)
+            )
+        self._time_override_value = time
+
+    def _unoverride_time(self) -> None:
+        """
+        Drops the time-overriding functions.
+        """
+        if self._time_override_value is None:
+            return
+
+        with self.connection.cursor() as cursor:
+            cursor.execute(sql.queries["testing_unoverride_time"])
+        self._time_override_value = None
+
+    def _check_and_apply_time_override(self) -> None:
+        now = datetime.datetime.now(datetime.timezone.utc)
+        is_time_frozen = type(now).__module__ != "datetime"
+
+        if is_time_frozen:
+            self._override_time(now)
+        else:
+            self._unoverride_time()
+
+    @django_connector.wrap_exceptions()
+    def execute_query_one(
+        self, query: LiteralString, **arguments: Any
+    ) -> dict[str, Any]:
+        self._check_and_apply_time_override()
+        result = super().execute_query_one(query, **arguments)
+
+        if query == sql.queries["defer_periodic_job"] and result.get("id"):
+            utils.async_to_sync(
+                self._notify,
+                queue_name=arguments["queue"],
+                notification={"type": "job_inserted", "job_id": result["id"]},
+            )
+        elif query == sql.queries["cancel_job"] and arguments.get("abort"):
+            with self.connection.cursor() as cursor:
+                cursor.execute(
+                    "SELECT queue_name FROM procrastinate_jobs WHERE id = %s",
+                    [arguments["job_id"]],
+                )
+                row = cursor.fetchone()
+                if row:
+                    utils.async_to_sync(
+                        self._notify,
+                        queue_name=row[0],
+                        notification={
+                            "type": "abort_job_requested",
+                            "job_id": arguments["job_id"],
+                        },
+                    )
+
+        return result
+
+    def _dictfetch(self, cursor: Any):
+        """
+        Return all rows from a cursor as a dict.
+
+        This method overrides the parent `_dictfetch` to handle manual JSON
+        deserialization. In a production environment, Procrastinate's async
+        connectors register custom `psycopg` adapters for composite types and JSONB.
+        However, the Django testing connection uses the default `psycopg` adapter
+        for the test database.
+
+        As a result, JSONB columns (like the `args` column returned by `fetch_job`)
+        are fetched as raw JSON strings rather than Python dictionaries. This override
+        intercepts the rows as they are fetched and manually deserializes the `args`
+        column.
+
+        This exists so that the Procrastinate worker, when running in a Django test
+        environment using this connector, receives properly formatted job parameters
+        and does not crash when attempting to access `job.task_kwargs.items()`.
+        """
+        columns = [col[0] for col in cursor.description]
+        return (
+            {
+                col: (
+                    json.loads(val) if col == "args" and isinstance(val, str) else val
+                )
+                for col, val in zip(columns, row)
+            }
+            for row in cursor.fetchall()
+        )
+
+    @django_connector.wrap_exceptions()
+    def execute_query(self, query: LiteralString, **arguments: Any) -> None:
+        self._check_and_apply_time_override()
+        super().execute_query(query, **arguments)
+
+    @django_connector.wrap_exceptions()
+    def execute_query_all(
+        self, query: LiteralString, **arguments: Any
+    ) -> list[dict[str, Any]]:
+        self._check_and_apply_time_override()
+        result = super().execute_query_all(query, **arguments)
+
+        if query == sql.queries["defer_jobs"]:
+            for i, row in enumerate(result):
+                job = arguments["jobs"][i]
+                if isinstance(job, dict):
+                    queue_name = job["queue_name"]
+                else:
+                    queue_name = getattr(job, "queue_name", "default")
+                utils.async_to_sync(
+                    self._notify,
+                    queue_name=queue_name,
+                    notification={"type": "job_inserted", "job_id": row["id"]},
+                )
+
+        return result
diff --git a/procrastinate/pytest_plugin.py b/procrastinate/pytest_plugin.py
new file mode 100644
index 000000000..4aeeb6bea
--- /dev/null
+++ b/procrastinate/pytest_plugin.py
@@ -0,0 +1,76 @@
+"""
+This module provides a Pytest plugin for Procrastinate, automatically registered
+via entry points if pytest is installed.
+
+It exists to provide convenient fixtures for running background jobs directly
+within a test suite, avoiding the boilerplate required to manually set up a
+worker and configure the `DjangoTestingConnector`.
+
+The plugin ensures a smooth developer experience for Procrastinate users who
+rely on pytest to test their applications, particularly those utilizing the
+Django integration.
+"""
+
+from __future__ import annotations
+
+import typing
+from collections.abc import Awaitable
+
+try:
+    import pytest
+except ImportError:
+    pytest = None  # type: ignore
+
+HAS_DJANGO = False
+try:
+    from procrastinate.contrib.django import app as django_app
+    from procrastinate.contrib.django import testing
+
+    HAS_DJANGO = True  # pyright: ignore[reportConstantRedefinition]
+except ImportError:
+    pass
+
+
+if pytest and HAS_DJANGO:
+
+    @pytest.fixture
+    def run_procrastinate_jobs() -> typing.Callable[..., None]:
+        """
+        Fixture that provides a synchronous function to execute all awaiting Procrastinate jobs.
+
+        In an integration test environment, you often defer jobs that need to be processed
+        before asserting the outcome. This fixture simplifies that process by injecting the
+        `DjangoTestingConnector` into the current app and running the worker.
+
+        It exists so developers can predictably execute jobs inline during their tests,
+        without spawning separate processes or managing background workers.
+        """
+
+        def f(**kwargs: typing.Any) -> None:
+            kwargs.setdefault("wait", False)
+            kwargs.setdefault("install_signal_handlers", False)
+            kwargs.setdefault("listen_notify", False)
+            with django_app.replace_connector(testing.DjangoTestingConnector()):  # pyright: ignore[reportPossiblyUnboundVariable]
+                django_app.run_worker(**kwargs)  # pyright: ignore[reportPossiblyUnboundVariable]
+
+        return f
+
+    @pytest.fixture
+    def arun_procrastinate_jobs() -> typing.Callable[..., Awaitable[None]]:
+        """
+        Fixture that provides an asynchronous function to execute all awaiting Procrastinate jobs.
+
+        Similar to `run_procrastinate_jobs`, but designed for use within `pytest.mark.asyncio`
+        tests. It allows developers to await the processing of background jobs within their
+        async test suites, replacing the app's connector with the `DjangoTestingConnector`
+        for the duration of the execution.
+        """
+
+        async def f(**kwargs: typing.Any) -> None:
+            kwargs.setdefault("wait", False)
+            kwargs.setdefault("install_signal_handlers", False)
+            kwargs.setdefault("listen_notify", False)
+            with django_app.replace_connector(testing.DjangoTestingConnector()):  # pyright: ignore[reportPossiblyUnboundVariable]
+                await django_app.run_worker_async(**kwargs)  # pyright: ignore[reportPossiblyUnboundVariable]
+
+        return f
diff --git a/procrastinate/sql/queries.sql b/procrastinate/sql/queries.sql
index f54a315fe..be9039f9d 100644
--- a/procrastinate/sql/queries.sql
+++ b/procrastinate/sql/queries.sql
@@ -235,3 +235,25 @@ SELECT procrastinate_update_heartbeat_v1(%(worker_id)s)
 -- prune_stalled_workers --
 -- Delete stalled workers that haven't sent a heartbeat in a while
 SELECT * FROM procrastinate_prune_stalled_workers_v1(%(seconds_since_heartbeat)s)
+
+-- testing_override_time --
+-- Create a schema for testing overrides, set search path, and override time functions
+CREATE SCHEMA IF NOT EXISTS _procrastinate_testing;
+SET search_path TO _procrastinate_testing, public, pg_catalog;
+CREATE OR REPLACE FUNCTION _procrastinate_testing.now() RETURNS timestamp with time zone AS $$
+    SELECT '{time_str}'::timestamp with time zone;
+$$ LANGUAGE sql;
+CREATE OR REPLACE FUNCTION _procrastinate_testing.transaction_timestamp() RETURNS timestamp with time zone AS $$
+    SELECT '{time_str}'::timestamp with time zone;
+$$ LANGUAGE sql;
+CREATE OR REPLACE FUNCTION _procrastinate_testing.statement_timestamp() RETURNS timestamp with time zone AS $$
+    SELECT '{time_str}'::timestamp with time zone;
+$$ LANGUAGE sql;
+CREATE OR REPLACE FUNCTION _procrastinate_testing.clock_timestamp() RETURNS timestamp with time zone AS $$
+    SELECT '{time_str}'::timestamp with time zone;
+$$ LANGUAGE sql;
+
+-- testing_unoverride_time --
+-- Drop the testing schema and reset search path
+DROP SCHEMA IF EXISTS _procrastinate_testing CASCADE;
+RESET search_path;
diff --git a/pyproject.toml b/pyproject.toml
index 9a8745ff9..f88151be3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -47,6 +47,9 @@ changelog = "https://github.com/procrastinate-org/procrastinate/releases"
 [project.scripts]
 procrastinate = 'procrastinate.cli:main'
 
+[project.entry-points.pytest11]
+procrastinate = "procrastinate.pytest_plugin"
+
 [dependency-groups]
 types = ["django-stubs"]
 release = ["dunamai"]
@@ -58,6 +61,7 @@ test = [
     "pytest-cov",
     "pytest-django",
     "pytest-mock",
+    "freezegun",
     "results",
 ]
 migration_test = ["nox", "httpx", "nox-uv"]
diff --git a/tests/integration/contrib/django/test_pytest_plugin.py b/tests/integration/contrib/django/test_pytest_plugin.py
new file mode 100644
index 000000000..30e506a00
--- /dev/null
+++ b/tests/integration/contrib/django/test_pytest_plugin.py
@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import datetime
+
+import freezegun
+import pytest
+
+from procrastinate.contrib.django import app
+from procrastinate.contrib.django.models import ProcrastinateJob
+
+
+@app.task(queue="pytest_plugin_queue")
+def my_test_task_plugin(a: int, b: int) -> int:
+    return a + b
+
+
+TASK_NAME = "tests.integration.contrib.django.test_pytest_plugin.my_test_task_plugin"
+
+
+@pytest.fixture(autouse=True)
+def clear_jobs(settings):
+    settings.PROCRASTINATE_READONLY_MODELS = False
+    ProcrastinateJob.objects.all().delete()
+
+
+@pytest.mark.django_db(transaction=True)
+def test_run_procrastinate_jobs(run_procrastinate_jobs):
+    my_test_task_plugin.defer(a=3, b=4)
+
+    assert ProcrastinateJob.objects.filter(task_name=TASK_NAME).count() == 1
+
+    run_procrastinate_jobs()
+
+    # Check the job succeeded
+    assert (
+        ProcrastinateJob.objects.filter(
+            task_name=TASK_NAME,
+            status="succeeded",
+        ).count()
+        == 1
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.django_db(transaction=True)
+async def test_arun_procrastinate_jobs(arun_procrastinate_jobs):
+    await my_test_task_plugin.defer_async(a=5, b=6)
+
+    assert await ProcrastinateJob.objects.filter(task_name=TASK_NAME).acount() == 1
+
+    await arun_procrastinate_jobs()
+
+    # Check the job succeeded
+    assert (
+        await ProcrastinateJob.objects.filter(
+            task_name=TASK_NAME,
+            status="succeeded",
+        ).acount()
+        == 1
+    )
+
+
+@pytest.mark.django_db(transaction=True)
+def test_run_procrastinate_jobs_time_travel(run_procrastinate_jobs):
+    with freezegun.freeze_time("2025-01-01T00:00:00Z"):
+        my_test_task_plugin.defer(a=3, b=4)
+        ProcrastinateJob.objects.update(
+            scheduled_at=datetime.datetime(
+                2025, 1, 2, 0, 0, 0, tzinfo=datetime.timezone.utc
+            )
+        )
+
+        assert (
+            ProcrastinateJob.objects.filter(
+                task_name=TASK_NAME,
+                status="todo",
+            ).count()
+            == 1
+        )
+
+        # worker shouldn't pick it up yet
+        run_procrastinate_jobs()
+        assert (
+            ProcrastinateJob.objects.filter(
+                task_name=TASK_NAME,
+                status="todo",
+            ).count()
+            == 1
+        )
+
+    with freezegun.freeze_time("2025-01-02T01:00:00Z"):
+        run_procrastinate_jobs()
+        assert (
+            ProcrastinateJob.objects.filter(
+                task_name=TASK_NAME,
+                status="succeeded",
+            ).count()
+            == 1
+        )
+
+
+@pytest.mark.django_db(transaction=True)
+def test_run_procrastinate_jobs_django_orm_modifications(run_procrastinate_jobs):
+    my_test_task_plugin.defer(a=3, b=4)
+    ProcrastinateJob.objects.update(
+        scheduled_at=datetime.datetime(
+            2100, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc
+        )
+    )
+
+    assert (
+        ProcrastinateJob.objects.filter(
+            task_name=TASK_NAME,
+            status="todo",
+        ).count()
+        == 1
+    )
+
+    # modify job scheduling using Django ORM
+    ProcrastinateJob.objects.filter(task_name=TASK_NAME).update(
+        scheduled_at=datetime.datetime(
+            2020, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc
+        )
+    )
+
+    run_procrastinate_jobs()
+
+    # it should be picked up now because the scheduled_at is in the past
+    assert (
+        ProcrastinateJob.objects.filter(
+            task_name=TASK_NAME,
+            status="succeeded",
+        ).count()
+        == 1
+    )
diff --git a/tests/integration/contrib/django/test_testing.py b/tests/integration/contrib/django/test_testing.py
new file mode 100644
index 000000000..bc101b250
--- /dev/null
+++ b/tests/integration/contrib/django/test_testing.py
@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+import asyncio
+import datetime
+import json
+
+import freezegun
+import pytest
+from asgiref.sync import async_to_sync
+
+from procrastinate import sql, types
+from procrastinate.contrib.django import testing
+
+
+@pytest.fixture
+def django_testing_connector(db):
+    return testing.DjangoTestingConnector(alias="default")
+
+
+async def test_defer_job_notifies(django_testing_connector):
+    event = asyncio.Event()
+    received = []
+
+    async def on_notification(*, channel: str, payload: str):
+        received.append((channel, payload))
+        event.set()
+
+    await django_testing_connector.listen_notify(
+        on_notification=on_notification, channels=["procrastinate_queue_v1#default"]
+    )
+
+    result = await django_testing_connector.execute_query_all_async(
+        query=sql.queries["defer_jobs"],
+        jobs=[
+            types.JobToDefer(
+                task_name="my_task",
+                queue_name="default",
+                lock=None,
+                queueing_lock=None,
+                args={},
+                scheduled_at=None,
+                priority=0,
+            )
+        ],
+    )
+
+    assert event.is_set()
+    assert len(received) == 1
+    channel, payload_str = received[0]
+    assert channel == "procrastinate_queue_v1#default"
+    payload = json.loads(payload_str)
+    assert payload["type"] == "job_inserted"
+    assert payload["job_id"] == result[0]["id"]
+
+
+def test_defer_job_sync_notifies(django_testing_connector):
+    received = []
+
+    async def on_notification(*, channel: str, payload: str):
+        received.append((channel, payload))
+
+    async_to_sync(django_testing_connector.listen_notify)(
+        on_notification=on_notification, channels=["procrastinate_queue_v1#default"]
+    )
+
+    result = django_testing_connector.execute_query_all(
+        query=sql.queries["defer_jobs"],
+        jobs=[
+            types.JobToDefer(
+                task_name="my_task",
+                queue_name="default",
+                lock=None,
+                queueing_lock=None,
+                args={},
+                scheduled_at=None,
+                priority=0,
+            )
+        ],
+    )
+
+    assert len(received) == 1
+    channel, payload_str = received[0]
+    assert channel == "procrastinate_queue_v1#default"
+    payload = json.loads(payload_str)
+    assert payload["type"] == "job_inserted"
+    assert payload["job_id"] == result[0]["id"]
+
+
+async def test_cancel_job_notifies(django_testing_connector):
+    event = asyncio.Event()
+    received = []
+
+    async def on_notification(*, channel: str, payload: str):
+        received.append((channel, payload))
+        event.set()
+
+    await django_testing_connector.listen_notify(
+        on_notification=on_notification, channels=["procrastinate_queue_v1#default"]
+    )
+
+    result = await django_testing_connector.execute_query_all_async(
+        query=sql.queries["defer_jobs"],
+        jobs=[
+            types.JobToDefer(
+                task_name="my_task",
+                queue_name="default",
+                lock=None,
+                queueing_lock=None,
+                args={},
+                scheduled_at=None,
+                priority=0,
+            )
+        ],
+    )
+
+    received.clear()
+    event.clear()
+
+    await django_testing_connector.execute_query_one_async(
+        query=sql.queries["cancel_job"],
+        job_id=result[0]["id"],
+        abort=True,
+        delete_job=False,
+    )
+
+    assert event.is_set()
+    assert len(received) == 1
+    channel, payload_str = received[0]
+    assert channel == "procrastinate_queue_v1#default"
+    payload = json.loads(payload_str)
+    assert payload["type"] == "abort_job_requested"
+    assert payload["job_id"] == result[0]["id"]
+
+
+def test_freezegun_mocking(django_testing_connector):
+    queries = [
+        "SELECT now()",
+        "SELECT transaction_timestamp()",
+        "SELECT statement_timestamp()",
+        "SELECT clock_timestamp()",
+    ]
+
+    # Without freezegun, time should not be overridden
+    for q in queries:
+        result = django_testing_connector.execute_query_one(query=q)
+        value = next(iter(result.values()))
+        assert value.year != 2000
+
+    # With freezegun, the override should be applied automatically
+    frozen_time = datetime.datetime(2000, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
+    with freezegun.freeze_time("2000-01-01T00:00:00Z"):
+        for q in queries:
+            result = django_testing_connector.execute_query_one(query=q)
+            value = next(iter(result.values()))
+            assert value == frozen_time
+
+    # After freezegun block, time should be back to normal
+    for q in queries:
+        result = django_testing_connector.execute_query_one(query=q)
+        value = next(iter(result.values()))
+        assert value.year != 2000
diff --git a/tests/unit/test_pytest_plugin.py b/tests/unit/test_pytest_plugin.py
new file mode 100644
index 000000000..004b871fe
--- /dev/null
+++ b/tests/unit/test_pytest_plugin.py
@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import importlib
+import sys
+
+
+def test_pytest_plugin_no_django(monkeypatch):
+    monkeypatch.setitem(sys.modules, "procrastinate.contrib.django", None)
+    monkeypatch.delitem(sys.modules, "procrastinate.pytest_plugin", raising=False)
+
+    pytest_plugin = importlib.import_module("procrastinate.pytest_plugin")
+
+    assert pytest_plugin.HAS_DJANGO is False
+    assert not hasattr(pytest_plugin, "run_procrastinate_jobs")
+
+
+def test_pytest_plugin_no_pytest(monkeypatch):
+    monkeypatch.setitem(sys.modules, "pytest", None)
+    monkeypatch.delitem(sys.modules, "procrastinate.pytest_plugin", raising=False)
+
+    pytest_plugin = importlib.import_module("procrastinate.pytest_plugin")
+
+    assert pytest_plugin.pytest is None
+    assert not hasattr(pytest_plugin, "run_procrastinate_jobs")
+
+
+def test_pytest_plugin_with_all_deps(monkeypatch):
+    monkeypatch.delitem(sys.modules, "procrastinate.pytest_plugin", raising=False)
+
+    pytest_plugin = importlib.import_module("procrastinate.pytest_plugin")
+
+    assert pytest_plugin.HAS_DJANGO is True
+    assert pytest_plugin.pytest is not None
+    assert hasattr(pytest_plugin, "run_procrastinate_jobs")
diff --git a/uv.lock b/uv.lock
index cf9a3d056..40c2a9662 100644
--- a/uv.lock
+++ b/uv.lock
@@ -572,6 +572,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/13/37/a065dc3bd6e49423a6532c642ca7378d3f467b1ef44c2800c937af7f9739/filelock-3.29.4-py3-none-any.whl", hash = "sha256:dac1648087d5115554850d113e7dd8c83ab2d38e3435dde2d4f163847e57b767", size = 42757, upload-time = "2026-06-13T16:11:59.582Z" },
 ]
 
+[[package]]
+name = "freezegun"
+version = "1.5.5"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "python-dateutil" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/95/dd/23e2f4e357f8fd3bdff613c1fe4466d21bfb00a6177f238079b17f7b1c84/freezegun-1.5.5.tar.gz", hash = "sha256:ac7742a6cc6c25a2c35e9292dfd554b897b517d2dec26891a2e8debf205cb94a", size = 35914, upload-time = "2025-08-09T10:39:08.338Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/5e/2e/b41d8a1a917d6581fc27a35d05561037b048e47df50f27f8ac9c7e27a710/freezegun-1.5.5-py3-none-any.whl", hash = "sha256:cd557f4a75cf074e84bc374249b9dd491eaeacd61376b9eb3c423282211619d2", size = 19266, upload-time = "2025-08-09T10:39:06.636Z" },
+]
+
 [[package]]
 name = "furo"
 version = "2025.12.19"
@@ -1073,6 +1085,7 @@ release = [
     { name = "dunamai" },
 ]
 test = [
+    { name = "freezegun" },
     { name = "pytest-asyncio" },
     { name = "pytest-benchmark" },
     { name = "pytest-cov" },
@@ -1135,6 +1148,7 @@ pg-implem = [
 ]
 release = [{ name = "dunamai" }]
 test = [
+    { name = "freezegun" },
     { name = "pytest-asyncio" },
     { name = "pytest-benchmark" },
     { name = "pytest-cov" },