KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_streaming_response method in locust

Best Python code snippet using locust

test_client.py

Source:test_client.py

...26                self.assertRaises(exception, s.get, "/")27            except KeyError:28                self.fail("Invalid URL %s was not propagated" % url)29    30    def test_streaming_response(self):31        """32        Test a request to an endpoint that returns a streaming response33        """34        s = HttpSession("http://127.0.0.1:%i" % self.port)35        r = s.get("/streaming/30")36        37        # verify that the time reported includes the download time of the whole streamed response38        self.assertGreater(global_stats.get("/streaming/30", method="GET").avg_response_time, 250)39        global_stats.clear_all()40        41        # verify that response time does NOT include whole download time, when using stream=True42        r = s.get("/streaming/30", stream=True)43        self.assertGreater(global_stats.get("/streaming/30", method="GET").avg_response_time, 0)44        self.assertLess(global_stats.get("/streaming/30", method="GET").avg_response_time, 250)...

cancelling_streaming_responses.py

Source:cancelling_streaming_responses.py

1"""2start with:3uvicorn test_streaming_response:app --reload4"""5import asyncio6import time7import uvicorn8from fastapi import FastAPI9from fastapi.responses import StreamingResponse10app = FastAPI()11def infinite_generator():12    # not blocking, so doesn't need to be async13    # but if it was blocking, you could make this async and await it14    while True:15        yield b"some fake data "16def finite_generator():17    # not blocking, so doesn't need to be async18    # but if it was blocking, you could make this async and await it19    x = 020    while x < 3000:21        yield f"{x}"22        x += 123async def astreamer(generator):24    try:25        # if it was an async generator we'd do:26        # "async for data in generator:"27        # (there is no "yield from" for async generators)28        for i in generator:29            yield i30            await asyncio.sleep(0.001)31    except asyncio.CancelledError as e:32        print("cancelled", e)33def streamer(generator):34    try:35        # note: normally we would do "yield from generator"36        # but that won't work with next(generator) in the finally statement37        for i in generator:38            yield i39            time.sleep(0.001)40    except GeneratorExit:41        print("cancelled")42    finally:43        # showing that we can check here to see if all data was consumed44        # the except statement above effectively does the same thing45        try:46            next(generator)47            print("we didn't finish")48            return49        except StopIteration:50            print("we finished")51@app.get("/infinite")52async def infinite_stream():53    return StreamingResponse(streamer(infinite_generator()))54@app.get("/finite")55async def finite_stream():56    return StreamingResponse(streamer(finite_generator()))57@app.get("/ainfinite")58async def ainfinite_stream():59    return StreamingResponse(astreamer(infinite_generator()))60@app.get("/afinite")61async def afinite_stream():62    return StreamingResponse(astreamer(finite_generator()))63if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.