(
tctx: context.Context,
open_h3_server_conn: connection.Server,
stream: str,
)
| 552 | |
| 553 | @pytest.mark.parametrize("stream", ["stream", ""]) |
| 554 | def test_request_trailers( |
| 555 | tctx: context.Context, |
| 556 | open_h3_server_conn: connection.Server, |
| 557 | stream: str, |
| 558 | ): |
| 559 | playbook, cff = start_h3_proxy(tctx) |
| 560 | tctx.server = open_h3_server_conn |
| 561 | sff = FrameFactory(tctx.server, is_client=False) |
| 562 | |
| 563 | def enable_streaming(flow: HTTPFlow): |
| 564 | flow.request.stream = stream |
| 565 | |
| 566 | flow = tutils.Placeholder(HTTPFlow) |
| 567 | ( |
| 568 | playbook |
| 569 | # request client |
| 570 | >> cff.receive_headers(example_request_headers) |
| 571 | << (request_headers := http.HttpRequestHeadersHook(flow)) |
| 572 | << cff.send_decoder() # for receive_headers |
| 573 | >> cff.receive_data(b"Hello World!") |
| 574 | >> tutils.reply(to=request_headers, side_effect=enable_streaming) |
| 575 | ) |
| 576 | if not stream: |
| 577 | ( |
| 578 | playbook |
| 579 | >> cff.receive_headers(example_request_trailers, end_stream=True) |
| 580 | << (request := http.HttpRequestHook(flow)) |
| 581 | << cff.send_decoder() # for receive_headers |
| 582 | >> tutils.reply(to=request) |
| 583 | ) |
| 584 | ( |
| 585 | playbook |
| 586 | # request server |
| 587 | << sff.send_init() |
| 588 | << sff.send_headers(example_request_headers) |
| 589 | << sff.send_data(b"Hello World!") |
| 590 | ) |
| 591 | if not stream: |
| 592 | playbook << sff.send_headers(example_request_trailers, end_stream=True) |
| 593 | ( |
| 594 | playbook |
| 595 | >> sff.receive_init() |
| 596 | << sff.send_encoder() |
| 597 | >> sff.receive_encoder() |
| 598 | >> sff.receive_decoder() # for send_headers |
| 599 | ) |
| 600 | if stream: |
| 601 | ( |
| 602 | playbook |
| 603 | >> cff.receive_headers(example_request_trailers, end_stream=True) |
| 604 | << (request := http.HttpRequestHook(flow)) |
| 605 | << cff.send_decoder() # for receive_headers |
| 606 | >> tutils.reply(to=request) |
| 607 | << sff.send_headers(example_request_trailers, end_stream=True) |
| 608 | ) |
| 609 | assert playbook >> sff.receive_decoder() # for send_headers |
| 610 | |
| 611 | assert cff.is_done and sff.is_done |
nothing calls this directly
no test coverage detected
searching dependent graphs…