from requests.exceptions import (RequestException, MissingSchema, InvalidSchema, InvalidURL) import gevent from locust.clients import HttpSession from locust.stats import global_stats from .testcases import WebserverTestCase class TestHttpSession(WebserverTestCase): def test_get(self): s = HttpSession("http://127.0.0.1:%i" % self.port) r = s.get("/ultra_fast") self.assertEqual(200, r.status_code) def test_connection_error(self): s = HttpSession("http://localhost:1") r = s.get("/", timeout=0.1) self.assertEqual(r.status_code, 0) self.assertEqual(None, r.content) self.assertRaises(RequestException, r.raise_for_status) def test_wrong_url(self): for url, exception in ( (u"http://\x94", InvalidURL), ("telnet://127.0.0.1", InvalidSchema), ("127.0.0.1", MissingSchema), ): s = HttpSession(url) try: self.assertRaises(exception, s.get, "/") except KeyError: self.fail("Invalid URL %s was not propagated" % url) def test_streaming_response(self): """ Test a request to an endpoint that returns a streaming response """ s = HttpSession("http://127.0.0.1:%i" % self.port) r = s.get("/streaming/30") # verify that the time reported includes the download time of the whole streamed response self.assertGreater(global_stats.get("/streaming/30", method="GET").avg_response_time, 250) global_stats.clear_all() # verify that response time does NOT include whole download time, when using stream=True r = s.get("/streaming/30", stream=True) self.assertGreater(global_stats.get("/streaming/30", method="GET").avg_response_time, 0) self.assertLess(global_stats.get("/streaming/30", method="GET").avg_response_time, 250) # download the content of the streaming response (so we don't get an ugly exception in the log) _ = r.content def test_slow_redirect(self): s = HttpSession("http://127.0.0.1:%i" % self.port) url = "/redirect?url=/redirect?delay=0.5" r = s.get(url) stats = global_stats.get(url, method="GET") self.assertEqual(1, stats.num_requests) self.assertGreater(stats.avg_response_time, 500) def test_post_redirect(self): s = HttpSession("http://127.0.0.1:%i" % self.port) url = "/redirect" r = s.post(url) self.assertEqual(200, r.status_code) post_stats = global_stats.get(url, method="POST") get_stats = global_stats.get(url, method="GET") self.assertEqual(1, post_stats.num_requests) self.assertEqual(0, get_stats.num_requests)
# | Change | User | Description | Committed | |
---|---|---|---|---|---|
#1 | 21680 | Robert Cowham |
Populate -o //guest/robert_cowham/p4benchmark/main/locust/... //guest/robert_cowham/p4benchmark/github/locust/.... |
||
//guest/robert_cowham/p4benchmark/main/locust/locust/test/test_client.py | |||||
#1 | 19772 | Robert Cowham | Initial version of locust |