From 2d81a786d9480b38e25a7132256076cdc371f182 Mon Sep 17 00:00:00 2001
From: PhiLily <252857470+PhiLily@users.noreply.github.com>
Date: Wed, 28 Oct 2026 11:05:36 +0200
Subject: [PATCH] Add assert_task_count() and an isolated-queue pytest recipe

django_ox.testing gains assert_task_count(). It reads OxTask rows for
one backend alias, queue and task path and counts those whose stored
args and kwargs equal the caller's after the JSON round trip enqueue()
applies before storing them, so a tuple in args matches a stored list.
A value the standard library's json.dumps() cannot handle, such as a
raw datetime, UUID or Decimal, is rejected with TypeError naming the
offending field rather than silently mismatching. A count that differs
raises AssertionError naming the backend, queue, path, the args and
kwargs after that round trip, and the expected and observed counts. The
helper reads live rows, so a rolled-back enqueue is not counted.

docs/patterns.md has two new sections. "Assert on the queue directly"
documents the helper with a short example. "An isolated queue for one
test" documents a pytest fixture recipe that wraps run_tasks() to drain
a single reserved queue, and refuses with ValueError at fixture setup,
before the test body enqueues anything, if that queue is already
occupied by setUpTestData or an earlier fixture. The snippet says where
the reserved name belongs: in that backend's own QUEUES list in the
TASKS setting, a sibling of OPTIONS rather than a key inside it. The
recipe is documentation, not a shipped API: tests/settings.py reserves
the isolated-recipe queue in the test backend's QUEUES for it.

docs/stability.md lists django_ox.testing.assert_task_count among the
public testing helpers. It is public and stable: its keyword-only
signature (path, queue, count, backend, args, kwargs) does not carry
the provisional status run_tasks() has on that page.

tests/test_testing_assertions.py, with its tasks in
tests/testing_assertions_tasks.py, covers the helper and the recipe:
JSON matching and rejection, backend and queue scoping, a finished row
still counted, the count mismatch message, and the occupied-queue
refusal. The tests run under both TestCase and TransactionTestCase.
TestCase wraps every test in a transaction it rolls back, so the
TransactionTestCase tests are the ones that read a row enqueue() has
committed, and they also show that a nested atomic() which rolls back
is still excluded without that outer transaction. CHANGELOG.md gets an
Added entry under Unreleased and docs/llms-full.txt is regenerated.

Closes #129
---
 CHANGELOG.md                      |   6 +
 docs/llms-full.txt                | 100 +++++++++++++++-
 docs/patterns.md                  |  80 +++++++++++++
 docs/stability.md                 |  14 ++-
 src/django_ox/testing.py          |  85 +++++++++++++-
 tests/settings.py                 |   4 +-
 tests/test_testing_assertions.py  | 186 ++++++++++++++++++++++++++++++
 tests/testing_assertions_tasks.py |  18 +++
 8 files changed, 481 insertions(+), 12 deletions(-)
 create mode 100644 tests/test_testing_assertions.py
 create mode 100644 tests/testing_assertions_tasks.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b969539..0259672 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/docs/llms-full.txt b/docs/llms-full.txt
index 7749461..d472634 100644
--- a/docs/llms-full.txt
+++ b/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
diff --git a/docs/patterns.md b/docs/patterns.md
index e83599f..a0aa561 100644
--- a/docs/patterns.md
+++ b/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),
diff --git a/docs/stability.md b/docs/stability.md
index 2677677..78ec63a 100644
--- a/docs/stability.md
+++ b/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__`.**
diff --git a/src/django_ox/testing.py b/src/django_ox/testing.py
index fc42485..40ef58c 100644
--- a/src/django_ox/testing.py
+++ b/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)"
+        )
diff --git a/tests/settings.py b/tests/settings.py
index 5d0ca3b..55ff0a0 100644
--- a/tests/settings.py
+++ b/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},
     }
 }
diff --git a/tests/test_testing_assertions.py b/tests/test_testing_assertions.py
new file mode 100644
index 0000000..7cff76e
--- /dev/null
+++ b/tests/test_testing_assertions.py
@@ -0,0 +1,186 @@
+"""
+django_ox.testing.assert_task_count(): reading OxTask rows directly, with
+JSON-normalised matching on args and kwargs.
+"""
+
+import pytest
+from django.db import transaction
+from django.test import TestCase, TransactionTestCase
+
+from django_ox.models import OxTask
+from django_ox.testing import assert_task_count, run_tasks
+
+from . import testing_assertions_tasks as t
+
+pytestmark = pytest.mark.django_db
+
+PATH = "tests.testing_assertions_tasks.note"
+ISOLATED_QUEUE = "isolated-recipe"
+
+
+@pytest.fixture
+def isolated_run_tasks():
+    """docs/patterns.md's isolated-queue recipe, reproduced to test it."""
+    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
+
+
+class TestAssertTaskCount(TestCase):
+    def test_matches_an_enqueued_task_by_path_args_and_kwargs(self):
+        t.note.enqueue("hello")
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=1)
+
+    def test_zero_when_nothing_matches(self):
+        assert_task_count(path=PATH, queue="default", count=0)
+
+    def test_tuple_args_match_the_stored_list(self):
+        t.note.enqueue("hello")
+        assert_task_count(path=PATH, queue="default", args=("hello",), count=1)
+
+    def test_kwargs_are_matched_too(self):
+        t.greet.enqueue(name="Ada", loud=True)
+        assert_task_count(
+            path="tests.testing_assertions_tasks.greet",
+            queue="default",
+            kwargs={"name": "Ada", "loud": True},
+            count=1,
+        )
+
+    def test_wrong_args_do_not_match(self):
+        t.note.enqueue("hello")
+        assert_task_count(path=PATH, queue="default", args=["goodbye"], count=0)
+
+    def test_count_mismatch_reports_expected_and_observed(self):
+        t.note.enqueue("hello")
+        t.note.enqueue("hello")
+        with pytest.raises(AssertionError) as excinfo:
+            assert_task_count(path=PATH, queue="default", args=["hello"], count=1)
+        message = str(excinfo.value)
+        assert "expected 1" in message
+        assert "found 2" in message
+        assert PATH in message
+        assert "default" in message
+
+    def test_a_rolled_back_enqueue_is_not_counted(self):
+        with pytest.raises(ValueError, match="roll back"), transaction.atomic():
+            t.note.enqueue("hello")
+            raise ValueError("roll back")
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=0)
+
+    def test_scoped_to_its_own_queue(self):
+        t.note.using(queue_name="emails").enqueue("hello")
+        assert_task_count(path=PATH, queue="emails", args=["hello"], count=1)
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=0)
+
+    def test_scoped_to_its_own_backend(self):
+        t.note.enqueue("hello")
+        assert_task_count(
+            path=PATH, queue="default", args=["hello"], backend="other", count=0
+        )
+
+    def test_a_finished_rows_path_args_and_kwargs_still_count(self):
+        from django_ox.testing import run_tasks
+
+        t.note.enqueue("hello")
+        run_tasks()
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=1)
+
+
+class TestAssertTaskCountUnderATransactionTestCase(TransactionTestCase):
+    """
+    TestCase above wraps every test in a transaction it rolls back; a plain
+    enqueue() there never truly commits. TransactionTestCase wraps nothing,
+    so the same assertion has to read a row that is actually committed, and
+    still has to exclude one that a nested atomic() rolled back.
+    """
+
+    def test_a_committed_enqueue_is_counted(self):
+        t.note.enqueue("hello")
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=1)
+
+    def test_a_rolled_back_nested_atomic_is_not_counted(self):
+        with pytest.raises(ValueError, match="roll back"), transaction.atomic():
+            t.note.enqueue("hello")
+            raise ValueError("roll back")
+        assert_task_count(path=PATH, queue="default", args=["hello"], count=0)
+
+
+class TestAssertTaskCountArguments:
+    def test_rejects_a_non_string_path(self):
+        with pytest.raises(TypeError, match="path"):
+            assert_task_count(path=123, queue="default", count=0)
+
+    def test_rejects_a_non_string_queue(self):
+        with pytest.raises(TypeError, match="queue"):
+            assert_task_count(path=PATH, queue=123, count=0)
+
+    def test_rejects_a_non_integer_count(self):
+        with pytest.raises(TypeError, match="count"):
+            assert_task_count(path=PATH, queue="default", count="1")
+
+    def test_rejects_a_negative_count(self):
+        with pytest.raises(ValueError, match="count"):
+            assert_task_count(path=PATH, queue="default", count=-1)
+
+    def test_rejects_a_datetime_argument(self):
+        from datetime import datetime
+
+        with pytest.raises(TypeError, match="JSON round trip"):
+            assert_task_count(
+                path=PATH, queue="default", args=[datetime.now()], count=0
+            )
+
+    def test_rejects_a_uuid_keyword_argument(self):
+        import uuid
+
+        with pytest.raises(TypeError, match="JSON round trip"):
+            assert_task_count(
+                path=PATH, queue="default", kwargs={"id": uuid.uuid4()}, count=0
+            )
+
+    def test_rejects_a_decimal_argument(self):
+        from decimal import Decimal
+
+        with pytest.raises(TypeError, match="JSON round trip"):
+            assert_task_count(
+                path=PATH, queue="default", args=[Decimal("1.5")], count=0
+            )
+
+
+def test_isolated_run_tasks_runs_only_the_reserved_queue(isolated_run_tasks):
+    """docs/patterns.md's "An isolated queue for one test" recipe."""
+    t.isolated_task.enqueue("clip.mp4")
+    t.note.enqueue("unrelated")
+    [result] = isolated_run_tasks()
+    assert result.return_value == "clip.mp4"
+    assert_task_count(
+        path="tests.testing_assertions_tasks.note",
+        queue="default",
+        args=["unrelated"],
+        count=1,
+    )
+
+
+def test_isolated_run_tasks_refuses_a_queue_already_holding_a_row(request):
+    # Simulates what setUpTestData or an earlier fixture would leave behind:
+    # a row in the reserved queue before the recipe's own check ever runs.
+    t.isolated_task.enqueue("leftover")
+    with pytest.raises(ValueError, match="already holds"):
+        request.getfixturevalue("isolated_run_tasks")
+    # Refused before any claim: the leftover row is untouched.
+    assert_task_count(
+        path="tests.testing_assertions_tasks.isolated_task",
+        queue=ISOLATED_QUEUE,
+        args=["leftover"],
+        count=1,
+    )
diff --git a/tests/testing_assertions_tasks.py b/tests/testing_assertions_tasks.py
new file mode 100644
index 0000000..777892d
--- /dev/null
+++ b/tests/testing_assertions_tasks.py
@@ -0,0 +1,18 @@
+"""Module-level tasks for tests/test_testing_assertions.py."""
+
+from django_ox.compat import task
+
+
+@task
+def note(label):
+    pass
+
+
+@task
+def greet(name, *, loud=False):
+    pass
+
+
+@task(queue_name="isolated-recipe")
+def isolated_task(label):
+    return label