Add SDK tests for python sdk. (#975)

Make some tweaks to fully align with sdk tests.

Author: gazpachoking

sdk-test.py

@@ -0,0 +1,53 @@
+# /// script
+# requires-python = ">=3.13"
+# dependencies = [
+#     "datastar-py",
+#     "sanic",
+# ]
+# [tool.uv.sources]
+# datastar-py = { path = "." }
+# ///
+
+"""
+Runs a test server that the SDK tests can be run against.
+1. Start this server with `uv run sdk-test.py`
+2. Move to the sdk/tests folder.
+3. Run `test-all.sh http://127.0.0.1:8000` to run the tests.
+"""
+import re
+
+from sanic import Request, Sanic
+
+from datastar_py import ServerSentEventGenerator as SSE
+from datastar_py.sanic import DatastarResponse, read_signals
+from datastar_py.sse import DatastarEvent
+
+app = Sanic("datastar-sdk-test")
+
+
+@app.route("/test", methods=["GET", "POST"])
+async def test_route(request: Request) -> None:
+    signals = await read_signals(request)
+    events: list[dict] = signals["events"]
+
+    response = await request.respond(response=DatastarResponse())
+
+    for event in events:
+        await response.send(build_event(event))
+
+
+def build_event(input: dict) -> DatastarEvent:
+    event_type = input.pop("type")
+    signals_raw = input.pop("signals-raw", None)
+    kwargs = {camel_to_snake(k): v for k, v in input.items()}
+    if signals_raw:
+        kwargs["signals"] = signals_raw
+    return getattr(SSE, camel_to_snake(event_type))(**kwargs)
+
+
+def camel_to_snake(text: str) -> str:
+    return re.sub(r"(.)([A-Z])", r"\1_\2", text).lower()
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=8000)

src/datastar_py/sse.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
 import json
-from collections.abc import AsyncIterable, Iterable
+from collections.abc import AsyncIterable, Iterable, Mapping
 from itertools import chain
 from typing import Literal, Protocol, TypeAlias, Union, overload, runtime_checkable
 
 import datastar_py.consts as consts
+from datastar_py.attributes import _escape
 
 SSE_HEADERS: dict[str, str] = {
     "Cache-Control": "no-cache",
@@ -46,12 +47,11 @@ def _send(
         event_id: str | None = None,
         retry_duration: int | None = None,
     ) -> DatastarEvent:
-        prefix = []
+        prefix = [f"event: {event_type}"]
+
         if event_id:
             prefix.append(f"id: {event_id}")
 
-        prefix.append(f"event: {event_type}")
-
         if retry_duration and retry_duration != consts.DEFAULT_SSE_RETRY_DURATION:
             prefix.append(f"retry: {retry_duration}")
 
@@ -94,7 +94,7 @@ def patch_elements(
         if isinstance(elements, _HtmlProvider):
             elements = elements.__html__()
         data_lines = []
-        if mode:
+        if mode and mode != "outer":  # TODO: Should there be a constant for this?
             data_lines.append(f"{consts.MODE_DATALINE_LITERAL} {mode}")
         if selector:
             data_lines.append(f"{consts.SELECTOR_DATALINE_LITERAL} {selector}")
@@ -132,7 +132,7 @@ def remove_elements(
     @classmethod
     def patch_signals(
         cls,
-        signals: dict,
+        signals: dict | str,
         event_id: str | None = None,
         only_if_missing: bool | None = None,
         retry_duration: int | None = None,
@@ -146,7 +146,13 @@ def patch_signals(
                 f"{consts.ONLY_IF_MISSING_DATALINE_LITERAL} {_js_bool(only_if_missing)}"
             )
 
-        data_lines.append(f"{consts.SIGNALS_DATALINE_LITERAL} {json.dumps(signals)}")
+        signals_str = (
+            signals if isinstance(signals, str) else json.dumps(signals, separators=(",", ":"))
+        )
+        data_lines.extend(
+            f"{consts.SIGNALS_DATALINE_LITERAL} {line}"
+            for line in signals_str.splitlines()
+        )
 
         return ServerSentEventGenerator._send(
             consts.EventType.PATCH_SIGNALS, data_lines, event_id, retry_duration
@@ -157,15 +163,20 @@ def execute_script(
         cls,
         script: str,
         auto_remove: bool = True,
-        attributes: list[str] | None = None,
+        attributes: Mapping[str, str] | list[str] | None = None,
         event_id: str | None = None,
         retry_duration: int | None = None,
     ) -> DatastarEvent:
         attribute_string = ""
         if auto_remove:
-            attribute_string += " data-effect='el.remove()'"
+            attribute_string += ' data-effect="el.remove()"'
         if attributes:
-            attribute_string += " " + " ".join(attributes)
+            if isinstance(attributes, Mapping):
+                attribute_string += " " + " ".join(
+                    f'{_escape(k)}="{_escape(v)}"' for k, v in attributes.items()
+                )
+            else:
+                attribute_string += " " + " ".join(attributes)
         script_tag = f"<script{attribute_string}>{script}</script>"
 
         return ServerSentEventGenerator.patch_elements(