Function TelemetryCompileApp

tests/integration/test_telemetry_compile.py:22–85

Reflex app exercising the features collected on the ``compile`` event. Includes a parent + child state hierarchy with Cookie / LocalStorage / SessionStorage to guard against the inherited-field double-count regression, a background event handler, a SharedState user subclass, and a d

(events_log_path: str = "")

def TelemetryCompileApp(events_log_path: str = ""):
    """Reflex app exercising the features collected on the ``compile`` event.

    Includes a parent + child state hierarchy with Cookie / LocalStorage /
    SessionStorage to guard against the inherited-field double-count
    regression, a background event handler, a SharedState user subclass,
    and a dynamic route.

    Args:
        events_log_path: Filesystem path to a JSONL file. Every
            ``telemetry.send`` invocation during compile is appended as
            one JSON object per line so the test can read it back.
    """
    import json
    import os
    from pathlib import Path

    import reflex as rx
    from reflex.istate.storage import Cookie, LocalStorage, SessionStorage
    from reflex.utils import telemetry

    # AppHarness force-disables telemetry. Re-enable it on the live config
    # singleton so the real _compile() path runs record_compile, and
    # redirect telemetry.send to disk so the test can read the payload.
    os.environ["REFLEX_TELEMETRY_ENABLED"] = "true"
    rx.config.get_config().telemetry_enabled = True

    sink = Path(events_log_path)
    sink.parent.mkdir(parents=True, exist_ok=True)

    def _capture(event, properties=None, **_kwargs):
        with sink.open("a") as fh:
            fh.write(
                json.dumps({"event": event, "properties": properties or {}}) + "\n"
            )
        return True

    telemetry.send = _capture

    class StorageRoot(rx.State):
        token: str = Cookie()
        local_pref: str = LocalStorage()

        @rx.event(background=True)
        async def heavy(self):
            """Background handler used to assert detection."""

    class StorageChild(StorageRoot):
        # ``token`` and ``local_pref`` are inherited here and must not be
        # re-counted on the child state.
        session: str = SessionStorage()

    class SharedThing(rx.SharedState):
        value: int = 0

    def index():
        return rx.box(rx.text("home"))