Testing LiveViews

djust ships LiveViewTestClient — a synchronous test client that lets you exercise LiveView behavior end-to-end without standing up a real WebSocket consumer. Inspired by Phoenix's LiveViewTest.

from djust.testing import LiveViewTestClient

def test_counter_increments():
    client = LiveViewTestClient(CounterView).mount()
    client.send_event("increment")
    client.assert_state(count=1)

The client instantiates the view, calls mount(), and invokes handlers directly — the same code path the WebSocket consumer runs, minus the transport layer.

Core methods

• mount(via_websocket=True, **params) — instantiate the view and run mount(). Returns self. The default simulates the WebSocket/SSE mount (the view gets the identity attributes a live connection sets); pass via_websocket=False to mount it the way the HTTP GET path does.
• send_event(name, raise_on_missing=True, **params) — call a handler by name. Returns a dict with state_before, state_after, duration_ms, success, error. An unknown handler raises NoHandlerFoundError unless you pass raise_on_missing=False. An exception raised inside the handler is not re-raised: it comes back as success=False with the message in error, so assert on result["success"].
• assert_state(**expected) — exact-match assertions on public view attrs.
• assert_state_contains(**expected) — checks expected in actual: a substring for strings, one element for lists, a key for dicts. It is not a subset match (items=["a"] fails against ["a", "b"]).
• get_state() — return a snapshot of public attrs.
• render(engine="rust") — render the view template and return the HTML.

Phoenix-parity assertions (v0.5.1)

Eight additional methods cover the rest of the production test surface.

assert_push_event(event_name, params=None)

Verify that some handler since mount() queued a matching client-bound push event. The test client doesn't clear the queue between send_event calls, so a push from an earlier event also satisfies the assertion.

class SaveView(LiveView):
    @event_handler
    def save(self, **kwargs):
        self.push_event("flash", {"type": "success", "message": "saved"})

client = LiveViewTestClient(SaveView).mount()
client.send_event("save")
client.assert_push_event("flash", {"type": "success"})
# Payload match is SUBSET — extra keys in the stored payload are OK.

assert_patch(path=None, params=None) / assert_redirect(path=None, params=None)

Verify the handler queued a navigation.

class FilterView(LiveView):
    @event_handler
    def filter(self, **kwargs):
        self.live_patch(params={"page": 2, "category": "books"}, path="/items/")

client = LiveViewTestClient(FilterView).mount()
client.send_event("filter")
client.assert_patch("/items/", {"page": 2})  # subset match on params

assert_redirect has the same signature and matches live_redirect calls.

render_async()

Drain pending start_async / assign_async tasks synchronously so tests can assert on their results.

class ReportView(LiveView):
    @event_handler
    def generate(self, **kwargs):
        self.generating = True
        self.start_async(self._build_report)

    def _build_report(self):
        try:
            self.report = build(...)
        except Exception as exc:
            # Always guard async callbacks — a crash that leaves `generating`
            # True will leave the client stuck in a loading state.
            self.error = str(exc)
        finally:
            self.generating = False

client = LiveViewTestClient(ReportView).mount()
client.send_event("generate")
client.assert_state(generating=True)      # pre-drain
client.render_async()
client.assert_state(generating=False)     # post-drain
assert client.view_instance.report is not None

render_async() runs exactly one batch of pending tasks; callbacks that queue more work require a second call. Matches production consumer semantics.

follow_redirect()

After a live_redirect, mount the destination view and return a new client rooted on it. The destination is mounted with the redirect's params plus the URL kwargs of the resolved path, and with the same user the original client was built with. Nothing else carries over: logging a user in inside a handler does not log the test client in.

class LoginView(LiveView):
    @event_handler
    def submit(self, email="", password="", **kwargs):
        if check_credentials(email, password):
            self.live_redirect("/dashboard/", params={"welcome": "1"})

class DashboardView(LiveView):  # routed at /dashboard/
    def mount(self, request, welcome="", **kwargs):
        self.welcome = welcome

login_client = LiveViewTestClient(LoginView).mount()
login_client.send_event("submit", email="a@b.com", password="x")
dashboard_client = login_client.follow_redirect()
dashboard_client.assert_state(welcome="1")

To test a destination that depends on request.user, build the first client with LiveViewTestClient(LoginView, user=some_user); follow_redirect() passes that user on.

assert_stream_insert(stream_name, item=None)

Verify a stream operation was queued.

class ChatView(LiveView):
    def mount(self, request, **kwargs):
        self.stream("messages", [])

    @event_handler
    def post(self, text="", **kwargs):
        from uuid import uuid4
        self.stream_insert("messages", {"id": uuid4().hex, "text": text})

client = LiveViewTestClient(ChatView).mount()
client.send_event("post", text="hello")
client.assert_stream_insert("messages", {"text": "hello"})

Dict items are matched by subset; other types by equality.

trigger_info(message)

Synthetically deliver a handle_info message (the hook pg_notify uses). Tests pubsub / database-notification handlers without real backend wiring. The test client doesn't need a real subscription.

class OrdersView(LiveView):
    def mount(self, request, **kwargs):
        self.listen("orders")
        self.count = 0

    def handle_info(self, message):
        if message.get("type") == "db_notify":
            self.count += 1

client = LiveViewTestClient(OrdersView).mount()
client.trigger_info({"type": "db_notify", "channel": "orders", "payload": {"event": "save"}})
client.assert_state(count=1)

assert_http_ws_djid_parity(**mount_kwargs)

Asserts that the HTTP-GET render path and the WebSocket-mount render path assign the same dj-id baseline for a view. In production the initial HTTP GET renders the DOM the browser holds and establishes a VDOM baseline, while a separate instance handles the WebSocket mount and establishes its own baseline; the first WS event ships patches keyed by dj-id. If the two baselines diverge, those patches miss dj-id resolution and fall back to path traversal — a getNodeByPath → null patch failure.

def test_dashboard_djid_parity():
    LiveViewTestClient(ProjectDashboardView).assert_http_ws_djid_parity(slug="acme")

Reach for this when a view renders differently on first paint vs. reconnect, or when you suspect a 1/N patches failed console warning traces to an HTTP-vs-WS dj-id mismatch. The harness builds two independent instances and compares the ordered dj-id sequence in the dj-root subtree; it returns that sequence so you can make further assertions.

Putting it together

A typical integration test combines several of these:

def test_create_item_flow():
    client = LiveViewTestClient(InventoryView).mount()
    client.send_event("create", name="widget")
    client.assert_stream_insert("items", {"name": "widget"})
    client.assert_push_event("flash", {"type": "success"})
    client.assert_patch("/items/", {"highlight": "widget"})

When to reach for each tool

• Unit test a handler → send_event + assert_state
• Test a push-event / flash → assert_push_event
• Test navigation → assert_patch / assert_redirect / follow_redirect
• Test async workflow → render_async
• Test streaming/append → assert_stream_insert
• Test pubsub / db_notify reaction → trigger_info
• Test rendered HTML → render() and assert on the returned string
• Snapshot test rendered HTML → inherit SnapshotTestMixin (separate)

For end-to-end browser tests, use Playwright against examples/demo_project (or your own Django app running djust).