feat: add streaming assertions (#1)

Author: othercodes

pyproject.toml

@@ -6,7 +6,7 @@ readme = "README.md"
 license = { text = "Apache-2.0" }
 requires-python = ">=3.11"
 authors = [
-    { name = "Unay Santisteban", email = "usantisteban@outlook.com" }
+    { name = "Unay Santisteban", email = "usantisteban@othercode.io" }
 ]
 keywords = ["django", "testing", "assertions", "fluent", "pytest"]
 classifiers = [
@@ -15,7 +15,6 @@ classifiers = [
     "License :: OSI Approved :: Apache Software License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
-
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
@@ -33,9 +32,9 @@ dependencies = [
 ]
 
 [project.urls]
-Homepage = "https://github.com/usantisteban/pyssertive"
-Repository = "https://github.com/usantisteban/pyssertive.git"
-Issues = "https://github.com/usantisteban/pyssertive/issues"
+Homepage = "https://github.com/othercodes/pyssertive"
+Repository = "https://github.com/othercodes/pyssertive.git"
+Issues = "https://github.com/othercodes/pyssertive/issues"
 
 [build-system]
 requires = ["setuptools>=70.0.0"]
@@ -77,7 +76,6 @@ addopts = "--cov=pyssertive --cov-report=term-missing --cov-fail-under=100"
 [tool.coverage.run]
 branch = true
 source = ["src/pyssertive"]
-omit = ["src/pyssertive/db.py"]
 
 [tool.coverage.report]
 exclude_lines = [

src/pyssertive/http/assertions.py

@@ -315,3 +315,105 @@ def assert_cookie_expired(self, name: str) -> Self:
         is_expired = max_age == 0 or max_age == "0" or cookie.value == ""
         assert is_expired, f"Cookie '{name}' is not expired (max-age={max_age}, value='{cookie.value}')"
         return self  # type: ignore[return-value]
+
+
+class StreamingAssertionsMixin:
+    _response: HttpResponse
+
+    def _get_streaming_content(self) -> str:
+        # Streaming content can only be consumed once, so cache it for reuse
+        from django.http import StreamingHttpResponse
+
+        if hasattr(self, "_cached_streaming_content"):
+            return self._cached_streaming_content  # type: ignore[return-value]
+
+        if isinstance(self._response, StreamingHttpResponse):
+            content = b"".join(self._response.streaming_content).decode("utf-8")
+        else:
+            content = self._response.content.decode("utf-8")
+
+        self._cached_streaming_content = content  # type: ignore[attr-defined]
+        return content
+
+    def _get_streaming_lines(self, strip_empty: bool = True) -> list[str]:
+        # Handles both \n and \r\n line endings
+        content = self._get_streaming_content()
+        content = content.replace("\r\n", "\n")
+        lines = content.split("\n")
+        if strip_empty:
+            lines = [line for line in lines if line.strip()]
+        return lines
+
+    def assert_streaming(self) -> Self:
+        from django.http import StreamingHttpResponse
+
+        assert isinstance(self._response, StreamingHttpResponse), (
+            f"Expected StreamingHttpResponse, got {type(self._response).__name__}"
+        )
+        return self  # type: ignore[return-value]
+
+    def assert_download(self, filename: str | None = None) -> Self:
+        # If filename provided, also asserts the filename matches.
+        disposition = self._response.get("Content-Disposition", "")
+        assert "attachment" in disposition, f"Expected Content-Disposition: attachment header, got '{disposition}'"
+        if filename is not None:
+            assert f'filename="{filename}"' in disposition or f"filename={filename}" in disposition, (
+                f"Expected filename '{filename}' in Content-Disposition, got '{disposition}'"
+            )
+        return self  # type: ignore[return-value]
+
+    def assert_streaming_contains(self, text: str) -> Self:
+        content = self._get_streaming_content()
+        assert text in content, f"Expected streaming content to contain '{text}'"
+        return self  # type: ignore[return-value]
+
+    def assert_streaming_not_contains(self, text: str) -> Self:
+        content = self._get_streaming_content()
+        assert text not in content, f"Expected streaming content to NOT contain '{text}'"
+        return self  # type: ignore[return-value]
+
+    def assert_streaming_matches(self, pattern: str) -> Self:
+        content = self._get_streaming_content()
+        assert re.search(pattern, content), f"Expected streaming content to match pattern '{pattern}'"
+        return self  # type: ignore[return-value]
+
+    def assert_streaming_line_count(
+        self, exact: int | None = None, *, min: int | None = None, max: int | None = None
+    ) -> Self:
+        # Counts non-empty lines only
+        lines = self._get_streaming_lines()
+        count = len(lines)
+
+        if exact is not None:
+            assert count == exact, f"Expected exactly {exact} lines, got {count}"
+        if min is not None:
+            assert count >= min, f"Expected at least {min} lines, got {count}"
+        if max is not None:
+            assert count <= max, f"Expected at most {max} lines, got {count}"
+
+        return self  # type: ignore[return-value]
+
+    def assert_streaming_empty(self) -> Self:
+        # Checks for no non-empty lines
+        lines = self._get_streaming_lines()
+        assert len(lines) == 0, f"Expected empty streaming content, got {len(lines)} lines"
+        return self  # type: ignore[return-value]
+
+    def assert_streaming_csv_header(self, columns: list[str] | str) -> Self:
+        # columns: list of names or comma-separated string
+        lines = self._get_streaming_lines(strip_empty=False)
+        has_content = len(lines) > 0 and lines[0].strip()
+        assert has_content, "Expected CSV content but response is empty"
+
+        header = lines[0]
+        expected = ",".join(columns) if isinstance(columns, list) else columns
+
+        assert header == expected, f"Expected CSV header '{expected}', got '{header}'"
+        return self  # type: ignore[return-value]  # type: ignore[return-value]
+
+    def assert_streaming_line(self, index: int, expected: str) -> Self:
+        lines = self._get_streaming_lines()
+        assert index < len(lines), f"Line index {index} out of range (only {len(lines)} lines)"
+        actual = lines[index]
+        assert actual == expected, f"Line {index}: expected '{expected}', got '{actual}'"
+        return self  # type: ignore[return-value]

src/pyssertive/http/client.py

@@ -11,6 +11,7 @@
     HTMLContentAssertionsMixin,
     HttpStatusAssertionsMixin,
     JsonContentAssertionsMixin,
+    StreamingAssertionsMixin,
 )
 from pyssertive.http.debug import DebugResponseMixin
 from pyssertive.http.django import (
@@ -24,6 +25,7 @@ class FluentResponse(
     DebugResponseMixin,
     SessionAssertionsMixin,
     CookieAssertionsMixin,
+    StreamingAssertionsMixin,
     TemplateContextAssertionsMixin,
     FormValidationAssertionsMixin,
     HTMLContentAssertionsMixin,

tests/app/urls.py

@@ -46,4 +46,10 @@
     path("cookie-set/", views.cookie_set_view),
     path("cookie-expire/", views.cookie_expire_view),
     path("cookie-detailed/", views.cookie_detailed_view),
+    # Streaming and download endpoints
+    path("streaming-csv/", views.streaming_csv_view),
+    path("streaming-text/", views.streaming_text_view),
+    path("streaming-empty/", views.streaming_empty_view),
+    path("download/", views.download_view),
+    path("inline/", views.inline_view),
 ]

tests/app/views.py

@@ -209,3 +209,47 @@ def cookie_detailed_view(request: HttpRequest) -> HttpResponse:
     response.set_cookie("detailed", "value", max_age=3600, path="/api/")
     response.set_cookie("maxage_only", "value", max_age=7200)
     return response
+
+
+def streaming_csv_view(request: HttpRequest) -> HttpResponse:
+    from django.http import StreamingHttpResponse
+
+    def generate_csv():
+        yield "Name,Email,Age\r\n"
+        yield "John,john@example.com,30\r\n"
+        yield "Jane,jane@example.com,25\r\n"
+        yield "Bob,bob@example.com,35\r\n"
+
+    response = StreamingHttpResponse(generate_csv(), content_type="text/csv")
+    response["Content-Disposition"] = 'attachment; filename="users.csv"'
+    return response
+
+
+def streaming_text_view(request: HttpRequest) -> HttpResponse:
+    from django.http import StreamingHttpResponse
+
+    def generate_text():
+        yield "Line 1\n"
+        yield "Line 2\n"
+        yield "Line 3\n"
+
+    return StreamingHttpResponse(generate_text(), content_type="text/plain")
+
+
+def streaming_empty_view(request: HttpRequest) -> HttpResponse:
+    from django.http import StreamingHttpResponse
+
+    def generate_empty():
+        yield ""
+
+    return StreamingHttpResponse(generate_empty(), content_type="text/plain")
+
+
+def download_view(request: HttpRequest) -> HttpResponse:
+    response = HttpResponse("file content here", content_type="application/octet-stream")
+    response["Content-Disposition"] = 'attachment; filename="report.txt"'
+    return response
+
+
+def inline_view(request: HttpRequest) -> HttpResponse:
+    return HttpResponse("inline content", content_type="text/plain")