from werkzeug.datastructures import Headers from werkzeug.sansio.multipart import Data from werkzeug.sansio.multipart import Epilogue from werkzeug.sansio.multipart import Field from werkzeug.sansio.multipart import File from werkzeug.sansio.multipart import MultipartDecoder from werkzeug.sansio.multipart import MultipartEncoder from werkzeug.sansio.multipart import NeedData from werkzeug.sansio.multipart import Preamble def test_decoder_simple() -> None: boundary = b"---------------------------9704338192090380615194531385$" decoder = MultipartDecoder(boundary) data = """ -----------------------------9704338192090380615194531385$ Content-Disposition: form-data; name="fname" ß∑œß∂ƒå∂ -----------------------------9704338192090380615194531385$ Content-Disposition: form-data; name="lname"; filename="bob" asdasd -----------------------------9704338192090380615194531385$-- """.replace( "\n", "\r\n" ).encode( "utf-8" ) decoder.receive_data(data) decoder.receive_data(None) events = [decoder.next_event()] while not isinstance(events[-1], Epilogue) and len(events) < 6: events.append(decoder.next_event()) assert events == [ Preamble(data=b""), Field( name="fname", headers=Headers([("Content-Disposition", 'form-data; name="fname"')]), ), Data(data="ß∑œß∂ƒå∂".encode(), more_data=False), File( name="lname", filename="bob", headers=Headers( [("Content-Disposition", 'form-data; name="lname"; filename="bob"')] ), ), Data(data=b"asdasd", more_data=False), Epilogue(data=b" "), ] encoder = MultipartEncoder(boundary) result = b"" for event in events: result += encoder.send_event(event) assert data == result def test_chunked_boundaries() -> None: boundary = b"--boundary" decoder = MultipartDecoder(boundary) decoder.receive_data(b"--") assert isinstance(decoder.next_event(), NeedData) decoder.receive_data(b"--boundary\r\n") assert isinstance(decoder.next_event(), Preamble) decoder.receive_data(b"Content-Disposition: form-data;") assert isinstance(decoder.next_event(), NeedData) decoder.receive_data(b'name="fname"\r\n\r\n') assert isinstance(decoder.next_event(), Field) decoder.receive_data(b"longer than the boundary") assert isinstance(decoder.next_event(), Data) decoder.receive_data(b"also longer, but includes a linebreak\r\n--") assert isinstance(decoder.next_event(), Data) assert isinstance(decoder.next_event(), NeedData) decoder.receive_data(b"--boundary--\r\n") event = decoder.next_event() assert isinstance(event, Data) assert not event.more_data decoder.receive_data(None) assert isinstance(decoder.next_event(), Epilogue)