Method test_workflow_stream

scripts/stress-test/sse_benchmark.py:400–554

Test workflow SSE streaming endpoint

(self)

    @task
    def test_workflow_stream(self) -> None:
        """Test workflow SSE streaming endpoint"""

        question = random.choice(self.questions)
        self.user_counter += 1

        headers = {
            "Authorization": f"Bearer {self.api_token}",
            "Content-Type": "application/json",
            "Accept": "text/event-stream",
            "Cache-Control": "no-cache",
        }

        data = WorkflowRequestData(
            inputs=WorkflowInputs(question=question),
            response_mode="streaming",
            user=f"user_{self.user_counter}",
        )

        start_time = time.time()
        first_event_time = None
        event_count = 0
        inter_event_times: list[float] = []
        last_event_time = None
        ttfe = 0
        request_success = False
        bytes_received = 0

        metrics.connection_started()

        # Use catch_response context manager directly
        with self.client.request(
            method="POST",
            url=WORKFLOW_PATH,
            headers=headers,
            json=data,
            stream=True,
            catch_response=True,
            timeout=(CONNECT_TIMEOUT, READ_TIMEOUT),
            name="/v1/workflows/run",  # Name for Locust stats
        ) as response:
            try:
                # Validate response
                if response.status_code >= 400:
                    error_type: ErrorType = "http_4xx" if response.status_code < 500 else "http_5xx"
                    metrics.record_error(error_type)
                    response.failure(f"HTTP {response.status_code}")
                    return

                content_type = response.headers.get("Content-Type", "")
                if "text/event-stream" not in content_type and "application/json" not in content_type:
                    logger.error(f"Expected text/event-stream, got: {content_type}")
                    metrics.record_error("invalid_response")
                    response.failure(f"Invalid content type: {content_type}")
                    return

                # Parse SSE events
                parser = SSEParser()