(self, tctx, pipeline)
| 19 | class TestServer: |
| 20 | @pytest.mark.parametrize("pipeline", ["pipeline", None]) |
| 21 | def test_simple(self, tctx, pipeline): |
| 22 | hdrs1 = Placeholder(RequestHeaders) |
| 23 | hdrs2 = Placeholder(RequestHeaders) |
| 24 | req2 = b"GET http://example.com/two HTTP/1.1\r\nHost: example.com\r\n\r\n" |
| 25 | playbook = Playbook(Http1Server(tctx)) |
| 26 | ( |
| 27 | playbook |
| 28 | >> DataReceived( |
| 29 | tctx.client, |
| 30 | b"POST http://example.com/one HTTP/1.1\r\n" |
| 31 | b"Content-Length: 3\r\n" |
| 32 | b"\r\n" |
| 33 | b"abc" + (req2 if pipeline else b""), |
| 34 | ) |
| 35 | << ReceiveHttp(hdrs1) |
| 36 | << ReceiveHttp(RequestData(1, b"abc")) |
| 37 | << ReceiveHttp(RequestEndOfMessage(1)) |
| 38 | >> ResponseHeaders(1, http.Response.make(200)) |
| 39 | << SendData(tctx.client, b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n") |
| 40 | >> ResponseEndOfMessage(1) |
| 41 | ) |
| 42 | if not pipeline: |
| 43 | playbook >> DataReceived(tctx.client, req2) |
| 44 | playbook << ReceiveHttp(hdrs2) |
| 45 | playbook << ReceiveHttp(RequestEndOfMessage(3)) |
| 46 | assert playbook |
| 47 | |
| 48 | @pytest.mark.parametrize("pipeline", ["pipeline", None]) |
| 49 | def test_connect(self, tctx, pipeline): |
nothing calls this directly
no test coverage detected