(tctx: context.Context)
| 435 | |
| 436 | |
| 437 | def test_simple(tctx: context.Context): |
| 438 | playbook, cff = start_h3_proxy(tctx) |
| 439 | flow = tutils.Placeholder(HTTPFlow) |
| 440 | server = tutils.Placeholder(connection.Server) |
| 441 | sff = FrameFactory(server, is_client=False) |
| 442 | assert ( |
| 443 | playbook |
| 444 | # request client |
| 445 | >> cff.receive_headers(example_request_headers, end_stream=True) |
| 446 | << (request := http.HttpRequestHeadersHook(flow)) |
| 447 | << cff.send_decoder() # for receive_headers |
| 448 | >> tutils.reply(to=request) |
| 449 | << http.HttpRequestHook(flow) |
| 450 | >> tutils.reply() |
| 451 | # request server |
| 452 | << commands.OpenConnection(server) |
| 453 | >> tutils.reply(None, side_effect=make_h3) |
| 454 | << sff.send_init() |
| 455 | << sff.send_headers(example_request_headers, end_stream=True) |
| 456 | >> sff.receive_init() |
| 457 | << sff.send_encoder() |
| 458 | >> sff.receive_encoder() |
| 459 | >> sff.receive_decoder() # for send_headers |
| 460 | # response server |
| 461 | >> sff.receive_headers(example_response_headers) |
| 462 | << (response := http.HttpResponseHeadersHook(flow)) |
| 463 | << sff.send_decoder() # for receive_headers |
| 464 | >> tutils.reply(to=response) |
| 465 | >> sff.receive_data(b"Hello, World!", end_stream=True) |
| 466 | << http.HttpResponseHook(flow) |
| 467 | >> tutils.reply() |
| 468 | # response client |
| 469 | << cff.send_headers(example_response_headers) |
| 470 | << cff.send_data(b"Hello, World!") |
| 471 | << cff.send_data(b"", end_stream=True) |
| 472 | >> cff.receive_decoder() # for send_headers |
| 473 | ) |
| 474 | assert cff.is_done and sff.is_done |
| 475 | assert flow().request.url == "http://example.com/" |
| 476 | assert flow().response.text == "Hello, World!" |
| 477 | |
| 478 | |
| 479 | @pytest.mark.parametrize("stream", ["stream", ""]) |
nothing calls this directly
no test coverage detected
searching dependent graphs…