Introduce proper download simulation for the user fixture (#3689)

* check outbox messages (see #3686)

* allow user fixture to test file downloads

* introduce proper download simulation

* update docs

* code review

* use constant for HTTP 200

---------

Co-authored-by: Falko Schindler <falko@zauberzeug.com>

nicegui/testing/user.py

@@ -12,6 +12,7 @@ from nicegui import Client, ElementFilter, ui
 from nicegui.element import Element
 from nicegui.nicegui import _on_handshake
 
+from .user_download import UserDownload
 from .user_interaction import UserInteraction
 from .user_navigate import UserNavigate
 from .user_notify import UserNotify
@@ -33,14 +34,16 @@ class User:
         self.forward_history: List[str] = []
         self.navigate = UserNavigate(self)
         self.notify = UserNotify()
+        self.download = UserDownload(self)
 
     def __getattribute__(self, name: str) -> Any:
-        if name not in {'notify', 'navigate'}:  # NOTE: avoid infinite recursion
+        if name not in {'notify', 'navigate', 'download'}:  # NOTE: avoid infinite recursion
             ui.navigate = self.navigate
             ui.notify = self.notify
+            ui.download = self.download
         return super().__getattribute__(name)
 
-    async def open(self, path: str, *, clear_forward_history: bool = True) -> None:
+    async def open(self, path: str, *, clear_forward_history: bool = True) -> Client:
         """Open the given path."""
         response = await self.http_client.get(path, follow_redirects=True)
         assert response.status_code == 200, f'Expected status code 200, got {response.status_code}'
@@ -56,6 +59,7 @@ class User:
         self.back_history.append(path)
         if clear_forward_history:
             self.forward_history.clear()
+        return self.client
 
     @overload
     async def should_see(self,

nicegui/testing/user_download.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, List, Optional, Union
+
+import httpx
+
+from .. import background_tasks
+
+if TYPE_CHECKING:
+    from .user import User
+
+
+class UserDownload:
+
+    def __init__(self, user: User) -> None:
+        self.http_responses: List[httpx.Response] = []
+        self.user = user
+
+    def __call__(self, src: Union[str, Path, bytes], filename: Optional[str] = None, media_type: str = '') -> Any:
+        background_tasks.create(self._get(src))
+
+    async def _get(self,  src: Union[str, Path, bytes]) -> None:
+        if isinstance(src, bytes):
+            await asyncio.sleep(0)
+            response = httpx.Response(httpx.codes.OK, content=src)
+        else:
+            response = await self.user.http_client.get(str(src))
+        self.http_responses.append(response)
+
+    async def next(self, *, timeout: float = 1.0) -> httpx.Response:
+        """Wait for a new download to happen.
+
+        :param timeout: the maximum time to wait (default: 1.0)
+        :returns: the HTTP response
+        """
+        assert self.user.client
+        downloads = len(self.http_responses)
+        deadline = time.time() + timeout
+        while len(self.http_responses) < downloads + 1:
+            await asyncio.sleep(0.1)
+            if time.time() > deadline:
+                raise TimeoutError('Download did not happen')
+        return self.http_responses[-1]

tests/test_user_simulation.py

@@ -1,6 +1,6 @@
 import csv
 from io import BytesIO
-from typing import Callable, Dict, Type
+from typing import Callable, Dict, Type, Union
 
 import pytest
 from fastapi import UploadFile
@@ -377,3 +377,22 @@ async def test_upload_table(user: User) -> None:
         {'name': 'Alice', 'age': '30'},
         {'name': 'Bob', 'age': '28'},
     ]
+
+
+@pytest.mark.parametrize('data', ['/data', b'Hello'])
+async def test_download_file(user: User, data: Union[str, bytes]) -> None:
+    @app.get('/data')
+    def get_data() -> PlainTextResponse:
+        return PlainTextResponse('Hello')
+
+    @ui.page('/')
+    def page():
+        ui.button('Download', on_click=lambda: ui.download(data))
+
+    await user.open('/')
+    assert len(user.download.http_responses) == 0
+    user.find('Download').click()
+    response = await user.download.next()
+    assert len(user.download.http_responses) == 1
+    assert response.status_code == 200
+    assert response.text == 'Hello'

website/documentation/content/user_documentation.py

@@ -156,6 +156,39 @@ def upload_table():
             ''')
 
 
+doc.text('Test Downloads', '''
+    You can verify that a download was triggered by checking `user.downloads.http_responses`.
+    By awaiting `user.downloads.next()` you can get the next download response.
+''')
+
+
+@doc.ui
+def check_outbox():
+    with ui.row().classes('gap-4 items-stretch'):
+        with python_window(classes='w-[500px]', title='some UI code'):
+            ui.markdown('''
+                ```python
+                @ui.page('/')
+                def page():
+                    def download():
+                        ui.download(b'Hello', filename='hello.txt')
+
+                    ui.button('Download', on_click=download)
+                ```
+            ''')
+
+        with python_window(classes='w-[500px]', title='user assertions'):
+            ui.markdown('''
+                ```python
+                await user.open('/')
+                assert len(user.download.http_responses) == 0
+                user.find('Download').click()
+                response = await user.download.next()
+                assert response.text == 'Hello'
+                ```
+            ''')
+
+
 doc.text('Multiple Users', '''
     Sometimes it is not enough to just interact with the UI as a single user.
     Besides the `user` fixture, we also provide the `create_user` fixture which is a factory function to create users.