(data: bytes, boundary: bytes | str)
| 830 | |
| 831 | |
| 832 | def parse_multipart_data(data: bytes, boundary: bytes | str) -> list[MultipartPart]: |
| 833 | parts: list[MultipartPart] = [] |
| 834 | done = False |
| 835 | |
| 836 | current_headers: dict[bytes, bytes] = {} |
| 837 | current_header_field: bytes = b"" |
| 838 | |
| 839 | def on_part_begin() -> None: |
| 840 | nonlocal current_headers |
| 841 | current_headers = {} |
| 842 | |
| 843 | def on_part_data(data: bytes, start: int, end: int) -> None: |
| 844 | parts.append(MultipartPart(current_headers, data[start:end])) |
| 845 | |
| 846 | def on_header_field(data: bytes, start: int, end: int) -> None: |
| 847 | nonlocal current_header_field |
| 848 | current_header_field = data[start:end] |
| 849 | |
| 850 | def on_header_value(data: bytes, start: int, end: int) -> None: |
| 851 | current_headers[current_header_field] = data[start:end] |
| 852 | |
| 853 | def on_end() -> None: |
| 854 | nonlocal done |
| 855 | done = True |
| 856 | |
| 857 | parser = MultipartParser( |
| 858 | boundary, |
| 859 | dict( |
| 860 | on_part_begin=on_part_begin, |
| 861 | on_part_data=on_part_data, |
| 862 | on_header_field=on_header_field, |
| 863 | on_header_value=on_header_value, |
| 864 | on_end=on_end, |
| 865 | ), |
| 866 | ) |
| 867 | parser.write(data) |
| 868 | parser.finalize() |
| 869 | assert done |
| 870 | return parts |
| 871 | |
| 872 | |
| 873 | def test_file_response_insert_ranges(file_response_client: TestClient) -> None: |
no test coverage detected