LESSON 132 · DEEPER PYTHON30 min

Test Binary Streams with io.BytesIO

Read a length-prefixed binary record from BytesIO and reject incomplete payloads.

Example check: Worked and guided code samples are checked against CPython 3.13. Practice uses Pyodide 0.29.5 (Python 3.13) in your browser. How we check lessons

Video lesson: Test Binary Streams with io.BytesIO

The preview is stored on this site. YouTube loads only when you press Play. Watch on YouTube

Read the video transcript

Select a timestamp to watch that moment on YouTube.

  1. 00:00

    Text streams and binary streams solve different problems. io.BytesIO holds bytes in memory and supports read, write, seek, and tell like a binary file. It lets a parser be tested without creating a file or guessing how text decoding should work. BytesIO wraps a bytes value in an in- memory stream. The first run starts at position zero, reads two bytes, and shows position two. A parser that reuses a stream must remember that read has changed its cursor. A second call will not magically begin at the start unless the caller seeks back. A two-byte header announces how many bytes to read next. If a packet ends early, the parser must notice rather than treating a partial

  2. 00:43

    payload as a complete message. BytesIO gives us the exact failure without a file. Before the first run, make a concrete prediction about the terminal output. A prediction turns a passive demonstration into a test: we can compare the actual result with the rule we thought the code followed. Keep the input small enough to inspect every value. Our sample record starts with a two-byte unsigned big-endian length, followed by exactly that many payload bytes. A naive parser calls read(length) and returns whatever arrives, even when the stream ends early. That turns a corrupt or truncated record into apparently valid data.

  3. 01:26

    A binary stream is deliberately different from StringIO, which handles text. Returning b'ab' makes the type visible in the output. Keep bytes as bytes while parsing a binary protocol. If a payload later represents text, decode it only after the protocol states which encoding is expected. The naive parser prints b'cat' for a correct record, which can conceal the missing validation. Test a short record too. A successful happy path does not prove the parser enforces its own length field. Watch the output as evidence rather than jumping directly to a replacement snippet. The broken program may exit cleanly while still violating its contract.

  4. 02:09

    We will name the exact input, the observed result, and the result we wanted before changing code. Read the two-byte header and verify its exact size before converting it with int.from_bytes. Then read the indicated payload and verify its exact length. Raise ValueError for either truncation. Keep the payload as bytes; decoding belongs to a later layer that knows the expected encoding. A two-byte big-endian header stores an unsigned length. The examples 00 03 and 01 00 decode to three and 256 respectively. The chosen byte order is part of the format. Both writer and reader must agree; changing it changes the meaning of the

  5. 02:51

    same two bytes. The parser should insist on two header bytes. Empty input or a one-byte prefix is incomplete, even if int.from_bytes would return an integer for the shorter value. Convert the header only after checking its size. Then request the declared number of payload bytes. The diagnostic run moves a cursor and then decodes a header. The stream position and byte order are both format rules. An incomplete header must be rejected before int.from_bytes, even though that function can convert one byte. The intermediate run isolates one mechanism. It is not the entire solution, and that is intentional.

  6. 03:34

    A small diagnostic is easier to explain than a large corrected application because it shows which assumption failed and which Python behavior we can rely on. Implement read_record(stream) for an io.BytesIO input containing a two-byte big-endian length followed by exactly that many bytes. Return the payload as bytes. Raise ValueError if the header or payload is truncated. The prepared call must print b'cat'. Check len(header) == 2; then size = int.from_bytes(header, 'big'). Read size bytes and check len(payload) == size. The final example requests three bytes from a stream that only contains two.

  7. 04:17

    BytesIO returns b'ab' without raising an exception. If our record claims a three-byte payload, returning b'ab' would make a truncated record look successful. Compare the actual number of bytes with the declared length and raise ValueError on a mismatch. The fixed parser checks two lengths. It returns raw bytes and does not assume text. For a production format, add a maximum allowed payload length to prevent a hostile header from requesting excessive work. After the fix, compare the observed output with the original prediction. The code shown on screen is executed locally and the terminal pane replays that captured run.

  8. 04:59

    The linked lesson gives you a separate checked task, so you can apply the idea rather than only copy a slide. BytesIO has a cursor. After reading once, seek(0) before another pass. The checked task verifies a valid b'cat' record, a missing header byte, and a payload shorter than its declared length. Here one read per field is demonstrated with BytesIO. A general stream may return a short chunk before end of input, so a production reader would accumulate bytes until the declared length or a true end of stream. It should also limit maximum accepted length for untrusted input. A production parser should also cap acceptable lengths before allocating or

  9. 05:41

    reading large payloads from untrusted data. The classroom task uses BytesIO: one read per field is sufficient to expose a truncated in-memory record. A general stream may return short chunks before its true end, so accumulate reads until the declared length or end of stream. The task tests a missing header, a short payload, an empty payload, and non-UTF-8 payload bytes without touching the filesystem. A declared length is part of the format contract. A separate layer can decode only when it knows the expected encoding. The final run accepts an empty payload with a zero length and rejects a three-

  10. 06:22

    byte claim followed by only two bytes. Both cases matter: zero is valid under this format, while truncation is not. An edge case is more valuable than repeating the same happy path. The final run challenges a boundary or invalid input and explains the decision the function makes. If your own implementation disagrees, use the lesson console to narrow the smallest failing case.

Understand the concept

Text streams and binary streams solve different problems. io.BytesIO holds bytes in memory and supports read, write, seek, and tell like a binary file. It lets a parser be tested without creating a file or guessing how text decoding should work.

Our sample record starts with a two-byte unsigned big-endian length, followed by exactly that many payload bytes. A naive parser calls read(length) and returns whatever arrives, even when the stream ends early. That turns a corrupt or truncated record into apparently valid data.

Read the two-byte header and verify its exact size before converting it with int.from_bytes. Then read the indicated payload and verify its exact length. Raise ValueError for either truncation. Keep the payload as bytes; decoding belongs to a later layer that knows the expected encoding.

BytesIO has a cursor. After reading once, seek(0) before another pass. The checked task verifies a valid b'cat' record, a missing header byte, and a payload shorter than its declared length. Here one read per field is demonstrated with BytesIO. A general stream may return a short chunk before end of input, so a production reader would accumulate bytes until the declared length or a true end of stream. It should also limit maximum accepted length for untrusted input.

  • BytesIO is an in-memory binary stream
  • read(n) can return fewer than n bytes
  • Validate header and payload lengths
  • Decode bytes only with a known format

See it step by step

Read the code, predict the output, then compare it with the result.

01. Observe binary cursor movement

PYTHON
from io import BytesIO
s=BytesIO(b'abc')
print(s.tell())
print(s.read(2))
print(s.tell())
EXPECTED OUTPUT
0
b'ab'
2

Binary reads advance the same sort of cursor as file reads.

02. Decode a two-byte length

PYTHON
print(int.from_bytes(b'\x00\x03','big'))
print(int.from_bytes(b'\x01\x00','big'))
EXPECTED OUTPUT
3
256

Byte order changes how the header becomes an integer.

03. Expose a short read

PYTHON
from io import BytesIO
s=BytesIO(b'ab')
part=s.read(3)
print(part)
print(len(part)==3)
EXPECTED OUTPUT
b'ab'
False

A short read must be handled explicitly.

A closer look

Follow the reasoning, inspect each result, then try the suggested changes in the console below.

01 / 03

Watch a binary cursor advance

BytesIO wraps a bytes value in an in-memory stream. The first run starts at position zero, reads two bytes, and shows position two. A parser that reuses a stream must remember that read has changed its cursor. A second call will not magically begin at the start unless the caller seeks back.

A binary stream is deliberately different from StringIO, which handles text. Returning b'ab' makes the type visible in the output. Keep bytes as bytes while parsing a binary protocol. If a payload later represents text, decode it only after the protocol states which encoding is expected.

PYTHON
from io import BytesIO
s=BytesIO(b'abc')
print(s.tell())
print(s.read(2))
print(s.tell())
EXPECTED OUTPUT
0
b'ab'
2
Follow the reasoning
  1. Create BytesIO around a known byte sequence.
  2. Read two bytes and inspect tell().
  3. Use seek(0) before a second complete parse.
02 / 03

Read the declared payload length

A two-byte big-endian header stores an unsigned length. The examples 00 03 and 01 00 decode to three and 256 respectively. The chosen byte order is part of the format. Both writer and reader must agree; changing it changes the meaning of the same two bytes.

The parser should insist on two header bytes. Empty input or a one-byte prefix is incomplete, even if int.from_bytes would return an integer for the shorter value. Convert the header only after checking its size. Then request the declared number of payload bytes.

PYTHON
print(int.from_bytes(b'\x00\x03','big'))
print(int.from_bytes(b'\x01\x00','big'))
EXPECTED OUTPUT
3
256
Follow the reasoning
  1. Check len(header) before conversion.
  2. Use int.from_bytes(header, 'big').
  3. Read exactly the declared payload length.
03 / 03

Catch a short read instead of accepting it

The final example requests three bytes from a stream that only contains two. BytesIO returns b'ab' without raising an exception. If our record claims a three-byte payload, returning b'ab' would make a truncated record look successful. Compare the actual number of bytes with the declared length and raise ValueError on a mismatch.

A production parser should also cap acceptable lengths before allocating or reading large payloads from untrusted data. The classroom task uses BytesIO: one read per field is sufficient to expose a truncated in-memory record. A general stream may return short chunks before its true end, so accumulate reads until the declared length or end of stream. The task tests a missing header, a short payload, an empty payload, and non-UTF-8 payload bytes without touching the filesystem.

PYTHON
from io import BytesIO
s=BytesIO(b'ab')
part=s.read(3)
print(part)
print(len(part)==3)
EXPECTED OUTPUT
b'ab'
False
Follow the reasoning
  1. Request more bytes than remain.
  2. Compare requested and actual lengths.
  3. Reject truncation while preserving valid arbitrary byte payloads.

Try it in Python

Edit the example and run it. Python starts in your browser the first time you click Run.

Python console

Ready to run
Need input()? Add one value per line
Ctrl / ⌘ + Enter to run
OUTPUT
Your output appears here.
Your turnImplement read_record(stream) for an io.BytesIO input containing a two-byte big-endian length followed by exactly that many bytes. Return the payload as bytes. Raise ValueError if the header or payload is truncated. The prepared call must print b'cat'.
Need a hint?

Check len(header) == 2; then size = int.from_bytes(header, 'big'). Read size bytes and check len(payload) == size.

Complete the task and select Check task to verify your code.

    Quick quiz

    Three questions. You can change your answers and try again.

    01.What type does BytesIO.read return?

    02.Does read(n) always return n bytes?

    03.Why keep an arbitrary payload as bytes?

    Typical mistakes

    Everyone meets these errors. See what causes them and how to fix them.

    Accepting a short payload

    COMMON MISTAKE
    return stream.read(size)
    FIX
    Check len(payload) == size

    What happens: Truncated data looks like a complete record.
    A declared length is part of the format contract.

    Decoding arbitrary bytes

    COMMON MISTAKE
    stream.read(size).decode()
    FIX
    Return bytes to the caller

    What happens: A valid binary payload may trigger UnicodeDecodeError.
    A separate layer can decode only when it knows the expected encoding.

    Browse this lesson’s problem fixes

    Sources and further reading