Add assert_task_count() and an isolated-queue pytest recipe

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `django_ox.testing.assert_task_count()`, a read-only assertion that reads
+  the `OxTask` table directly, matching by backend alias, queue, task path,
+  and `args`/`kwargs` compared the same way `enqueue()` stores them. A
+  rolled-back enqueue is not counted. `docs/patterns.md` also documents an
+  isolated-queue `run_tasks()` pytest fixture recipe, with a repository
+  test, not a shipped fixture API.
 - `ox_retry` and `ox_discard` management commands for scriptable recovery
   of tasks. Both commands take one or more task UUIDs, deduplicate them, and
   delegate to `django_ox.actions.retry_many` or `discard_many`. Both support

docs/llms-full.txt

@@ -4654,6 +4654,86 @@ Use a real worker to test timeout enforcement and worker infrastructure. `run_ta
 
 Timeouts are inert. `deadline()` and `remaining()` return `None`. A configured timeout produces `run_tasks_timeout_inert` once per task path per call. A hanging body or callback hangs the test.
 
+### Assert on the queue directly
+
+`django_ox.testing.assert_task_count()` reads the `OxTask` table itself, so
+it works whether a task is still queued or has already run and left its
+row behind. It matches by backend alias, queue and task path, with `args`
+and `kwargs` compared the same way `enqueue()` stores them:
+
+```python
+from django_ox.testing import assert_task_count
+
+send_welcome_email.enqueue(user.pk)
+assert_task_count(
+    path="myapp.tasks.send_welcome_email", queue="default", args=[user.pk], count=1
+)
+```
+
+A rolled-back enqueue is not counted, since the row behind it never
+existed. `args` and `kwargs` are matched after the same JSON round trip
+`enqueue()` applies before storing them, so a tuple in `args` matches a
+stored list, and a raw `datetime`, `UUID` or `Decimal` is rejected with
+`TypeError`: supply that value's own stored JSON representation instead.
+
+### An isolated queue for one test
+
+`run_tasks()` claims whatever its backend's worker would, which includes
+rows other tests, fixtures or `setUpTestData` left behind on a shared
+queue. To run only a task under test, reserve a queue nothing else writes
+to, and check it is still empty before draining it. This is a documented
+recipe, not a shipped fixture or pytest plugin:
+
+```python
+import pytest
+from django_ox.models import OxTask
+from django_ox.testing import run_tasks
+
+# Must appear in this backend's own QUEUES list (a sibling of OPTIONS,
+# not inside it): TASKS["default"]["QUEUES"].
+ISOLATED_QUEUE = "isolated-recipe"
+
+
+@pytest.fixture
+def isolated_run_tasks():
+    # Checked once, here, before the test body enqueues anything: a row
+    # from setUpTestData or an earlier fixture is caught before it is ever
+    # claimed. The returned `run` does not repeat the check, since by then
+    # the only row it is meant to find is the one the test just enqueued.
+    if OxTask.objects.filter(
+        backend_name="default", queue_name=ISOLATED_QUEUE
+    ).exists():
+        raise ValueError(
+            f"queue {ISOLATED_QUEUE!r} already holds rows; reserve a queue "
+            "no other test or fixture writes to"
+        )
+
+    def run(**kwargs):
+        return run_tasks(queues=[ISOLATED_QUEUE], **kwargs)
+
+    return run
+```
+
+```python
+@task(queue_name=ISOLATED_QUEUE)
+def convert_video(path): ...
+
+
+def test_convert_video(isolated_run_tasks):
+    convert_video.enqueue("clip.mp4")
+    [result] = isolated_run_tasks()
+    assert result.return_value is not None
+```
+
+Fixture setup runs before the test body, so the emptiness check sees only
+what was there beforehand, never the task the test is about to enqueue.
+A queue that already holds work, including a row `setUpTestData` left
+there, fails fast with `ValueError` rather than draining someone else's
+task. Passing `queues=` to the wrapped `run_tasks()` call, not a model
+filter, is what keeps the drain from touching unrelated rows:
+`tests/test_testing_assertions.py` tests this recipe's isolation and
+failure behaviour directly.
+
 ## Optional coordination features
 
 [Unique tasks](features/unique-tasks.md), [batches](features/batches.md),
@@ -8524,11 +8604,15 @@ export names this page does not list; those names are not public.
   is always `null`. On `worker_claim_recovery_failed` and
   `worker_claim_recovery_expired`, it is always `"expired"`.
 - **The testing helpers** `django_ox.testing.ImmediateBackend`,
-  `django_ox.testing.DummyBackend` and `django_ox.testing.run_tasks`.
-  The backends accept policy declarations but do not enforce retries,
-  backoff or timeouts. `run_tasks()` drains due queued tasks through the
-  configured worker class, including retries and backoff. It does not
-  enforce timeouts. `run_tasks()` is public and provisional.
+  `django_ox.testing.DummyBackend`, `django_ox.testing.run_tasks` and
+  `django_ox.testing.assert_task_count`. The backends accept policy
+  declarations but do not enforce retries, backoff or timeouts.
+  `run_tasks()` drains due queued tasks through the configured worker
+  class, including retries and backoff. It does not enforce timeouts.
+  `run_tasks()` is public and provisional. `assert_task_count()` is
+  public and stable: its keyword-only signature
+  (`path`, `queue`, `count`, `backend`, `args`, `kwargs`) does not carry
+  `run_tasks()`'s provisional status.
 - **The database schema** of `OxTask` and `OxScheduleTick`, evolved only
   through shipped migrations.
 - **`django_ox.__version__`.**
@@ -9384,6 +9468,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `django_ox.testing.assert_task_count()`, a read-only assertion that reads
+  the `OxTask` table directly, matching by backend alias, queue, task path,
+  and `args`/`kwargs` compared the same way `enqueue()` stores them. A
+  rolled-back enqueue is not counted. `docs/patterns.md` also documents an
+  isolated-queue `run_tasks()` pytest fixture recipe, with a repository
+  test, not a shipped fixture API.
 - `ox_retry` and `ox_discard` management commands for scriptable recovery
   of tasks. Both commands take one or more task UUIDs, deduplicate them, and
   delegate to `django_ox.actions.retry_many` or `discard_many`. Both support

docs/patterns.md

@@ -447,6 +447,86 @@ Use a real worker to test timeout enforcement and worker infrastructure. `run_ta
 
 Timeouts are inert. `deadline()` and `remaining()` return `None`. A configured timeout produces `run_tasks_timeout_inert` once per task path per call. A hanging body or callback hangs the test.
 
+### Assert on the queue directly
+
+`django_ox.testing.assert_task_count()` reads the `OxTask` table itself, so
+it works whether a task is still queued or has already run and left its
+row behind. It matches by backend alias, queue and task path, with `args`
+and `kwargs` compared the same way `enqueue()` stores them:
+
+```python
+from django_ox.testing import assert_task_count
+
+send_welcome_email.enqueue(user.pk)
+assert_task_count(
+    path="myapp.tasks.send_welcome_email", queue="default", args=[user.pk], count=1
+)
+```
+
+A rolled-back enqueue is not counted, since the row behind it never
+existed. `args` and `kwargs` are matched after the same JSON round trip
+`enqueue()` applies before storing them, so a tuple in `args` matches a
+stored list, and a raw `datetime`, `UUID` or `Decimal` is rejected with
+`TypeError`: supply that value's own stored JSON representation instead.
+
+### An isolated queue for one test
+
+`run_tasks()` claims whatever its backend's worker would, which includes
+rows other tests, fixtures or `setUpTestData` left behind on a shared
+queue. To run only a task under test, reserve a queue nothing else writes
+to, and check it is still empty before draining it. This is a documented
+recipe, not a shipped fixture or pytest plugin:
+
+```python
+import pytest
+from django_ox.models import OxTask
+from django_ox.testing import run_tasks
+
+# Must appear in this backend's own QUEUES list (a sibling of OPTIONS,
+# not inside it): TASKS["default"]["QUEUES"].
+ISOLATED_QUEUE = "isolated-recipe"
+
+
+@pytest.fixture
+def isolated_run_tasks():
+    # Checked once, here, before the test body enqueues anything: a row
+    # from setUpTestData or an earlier fixture is caught before it is ever
+    # claimed. The returned `run` does not repeat the check, since by then
+    # the only row it is meant to find is the one the test just enqueued.
+    if OxTask.objects.filter(
+        backend_name="default", queue_name=ISOLATED_QUEUE
+    ).exists():
+        raise ValueError(
+            f"queue {ISOLATED_QUEUE!r} already holds rows; reserve a queue "
+            "no other test or fixture writes to"
+        )
+
+    def run(**kwargs):
+        return run_tasks(queues=[ISOLATED_QUEUE], **kwargs)
+
+    return run
+```
+
+```python
+@task(queue_name=ISOLATED_QUEUE)
+def convert_video(path): ...
+
+
+def test_convert_video(isolated_run_tasks):
+    convert_video.enqueue("clip.mp4")
+    [result] = isolated_run_tasks()
+    assert result.return_value is not None
+```
+
+Fixture setup runs before the test body, so the emptiness check sees only
+what was there beforehand, never the task the test is about to enqueue.
+A queue that already holds work, including a row `setUpTestData` left
+there, fails fast with `ValueError` rather than draining someone else's
+task. Passing `queues=` to the wrapped `run_tasks()` call, not a model
+filter, is what keeps the drain from touching unrelated rows:
+`tests/test_testing_assertions.py` tests this recipe's isolation and
+failure behaviour directly.
+
 ## Optional coordination features
 
 [Unique tasks](features/unique-tasks.md), [batches](features/batches.md),

docs/stability.md

@@ -112,11 +112,15 @@ export names this page does not list; those names are not public.
   is always `null`. On `worker_claim_recovery_failed` and
   `worker_claim_recovery_expired`, it is always `"expired"`.
 - **The testing helpers** `django_ox.testing.ImmediateBackend`,
-  `django_ox.testing.DummyBackend` and `django_ox.testing.run_tasks`.
-  The backends accept policy declarations but do not enforce retries,
-  backoff or timeouts. `run_tasks()` drains due queued tasks through the
-  configured worker class, including retries and backoff. It does not
-  enforce timeouts. `run_tasks()` is public and provisional.
+  `django_ox.testing.DummyBackend`, `django_ox.testing.run_tasks` and
+  `django_ox.testing.assert_task_count`. The backends accept policy
+  declarations but do not enforce retries, backoff or timeouts.
+  `run_tasks()` drains due queued tasks through the configured worker
+  class, including retries and backoff. It does not enforce timeouts.
+  `run_tasks()` is public and provisional. `assert_task_count()` is
+  public and stable: its keyword-only signature
+  (`path`, `queue`, `count`, `backend`, `args`, `kwargs`) does not carry
+  `run_tasks()`'s provisional status.
 - **The database schema** of `OxTask` and `OxScheduleTick`, evolved only
   through shipped migrations.
 - **`django_ox.__version__`.**

src/django_ox/testing.py

@@ -25,11 +25,17 @@
 enqueued the way a worker runs it, on the test's own thread and connection,
 so it works inside a TestCase's transaction; its docstring says what it
 does and does not reproduce.
+
+assert_task_count() is a read-only assertion against the OxTask table
+itself, for either kind of test: whether a task is still queued, or ran and
+left its row behind.
 """
 
 from __future__ import annotations
 
+import json
 import logging
+from collections.abc import Mapping, Sequence
 from threading import Lock
 from typing import Any
 
@@ -38,7 +44,7 @@
 from .compat import ImmediateBackend as _StockImmediateBackend
 from .tasks import PolicyTask, task_policy, validate_policy
 
-__all__ = ["DummyBackend", "ImmediateBackend", "run_tasks"]
+__all__ = ["DummyBackend", "ImmediateBackend", "assert_task_count", "run_tasks"]
 
 logger = logging.getLogger("django_ox")
 
@@ -232,3 +238,80 @@ def run_tasks(
         max_tasks=max_tasks,
         raise_failures=raise_failures,
     )
+
+
+def _json_round_trip(label: str, value: Any) -> Any:
+    try:
+        return json.loads(json.dumps(value))
+    except TypeError as exc:
+        raise TypeError(
+            f"assert_task_count() {label} must be values a standard JSON round "
+            f"trip preserves ({exc}); supply the task's stored JSON "
+            f"representation instead."
+        ) from exc
+
+
+def assert_task_count(
+    *,
+    path: str,
+    queue: str,
+    count: int,
+    backend: str = DEFAULT_TASK_BACKEND_ALIAS,
+    args: Sequence[Any] = (),
+    kwargs: Mapping[str, Any] | None = None,
+) -> None:
+    """
+    Assert that exactly `count` OxTask rows in `backend`'s `queue` have
+    `path` as their stored task path and match `args` and `kwargs`.
+
+    This reads database rows, not a separate enqueue record: a task whose
+    enqueue was rolled back is not counted, and the count includes a row
+    that has since run, retried or finished, since its path, args and
+    kwargs are unchanged by running. Narrow with `run_tasks(queues=...)` or
+    a status check first if only queued work should count.
+
+    `args` and `kwargs` are compared to the stored values after a JSON
+    round trip, the same lossy conversion `enqueue()` itself applies before
+    storing them: a tuple in `args` matches a stored list, but a type a
+    standard `json.dumps()` call cannot handle, such as a raw `datetime`,
+    `UUID` or `Decimal`, is rejected with `TypeError` rather than silently
+    miscompared. Pass that value's own stored JSON representation (a
+    string, an int or a plain list or dict) instead.
+
+    Raises `AssertionError`, naming `backend`, `queue`, `path`, the
+    normalised `args` and `kwargs`, and the expected and observed counts,
+    when the observed count differs.
+    """
+    if not isinstance(path, str):
+        raise TypeError(f"assert_task_count() path must be a string, got {path!r}.")
+    if not isinstance(queue, str):
+        raise TypeError(f"assert_task_count() queue must be a string, got {queue!r}.")
+    if not isinstance(backend, str):
+        raise TypeError(
+            f"assert_task_count() backend must be a string, got {backend!r}."
+        )
+    if isinstance(count, bool) or not isinstance(count, int):
+        raise TypeError(
+            f"assert_task_count() count must be a whole number, got {count!r}."
+        )
+    if count < 0:
+        raise ValueError(f"assert_task_count() count must be 0 or more, got {count!r}.")
+
+    want_args = _json_round_trip("args", list(args))
+    want_kwargs = _json_round_trip("kwargs", dict(kwargs) if kwargs else {})
+
+    # Imported here, not at the top: this module is named in TASKS settings,
+    # and the worker imports the models.
+    from .models import OxTask
+
+    rows = OxTask.objects.filter(backend_name=backend, queue_name=queue, task_path=path)
+    observed = sum(
+        1 for row in rows if row.args == want_args and row.kwargs == want_kwargs
+    )
+    if observed != count:
+        raise AssertionError(
+            f"expected {count} task(s) matching backend={backend!r} "
+            f"queue={queue!r} path={path!r} args={want_args!r} "
+            f"kwargs={want_kwargs!r}, found {observed} (of {rows.count()} row(s) "
+            f"stored for that path in that backend and queue)"
+        )

tests/settings.py

@@ -72,7 +72,9 @@
 TASKS: dict[str, dict[str, Any]] = {
     "default": {
         "BACKEND": "django_ox.backend.OxBackend",
-        "QUEUES": ["default", "emails"],
+        # "isolated-recipe" is reserved for the docs/patterns.md isolated-queue
+        # fixture recipe and its test; nothing else enqueues onto it.
+        "QUEUES": ["default", "emails", "isolated-recipe"],
         "OPTIONS": {"MAX_ATTEMPTS": 3},
     }
 }