How to Mock a Socket Stream in Python
Simulate a streaming socket source with a generator to test stream-read and buffering logic without a real network.
Python code
23 linesimport socket
import threading
import time
def mock_socket_stream(data_chunks, delay=0.1):
"""Generator that simulates a streaming socket source."""
for chunk in data_chunks:
time.sleep(delay)
yield chunk
def read_stream_socket(stream_gen):
"""Reads from mock stream and prints received chunks."""
received = []
for chunk in stream_gen:
received.append(chunk)
print(f"Received: {chunk}")
return ''.join(received)
if __name__ == "__main__":
chunks = ["Hello, ", "world! ", "This ", "is ", "a ", "mock ", "socket ", "stream."]
mock_stream = mock_socket_stream(chunks, delay=0.2)
result = read_stream_socket(mock_stream)
print(f"\nFinal received message: {result}")Output
Received: Hello,
Received: world!
Received: This
Received: is
Received: a
Received: mock
Received: socket
Received: stream.
Final received message: Hello, world! This is a mock socket stream.How it works
The mock_socket_stream generator yields chunks with a small delay to mimic network latency, while read_stream_socket consumes the generator incrementally, simulating how you'd read from a real socket in a loop. Each chunk is appended to a list and joined at the end, replicating buffer accumulation. The time.sleep introduces a pause to make the simulation realistic without blocking a real I/O thread. This pattern lets you unit-test streaming logic deterministically without opening sockets or managing connections.
Common mistakes
- Using `return` instead of `yield` in the mock generator, which stops iteration early.
- Forgetting to call `time.sleep` in the mock, making the simulation unrealistically fast.
- Accumulating chunks with repeated string concatenation instead of a list and `join`, which is inefficient for many chunks.
Variations
- Use a `queue.Queue` with a background thread to feed chunks for a more realistic non-blocking producer.
- Use `yield from` with a list or file iterator to simplify the generator body.
Real-world use cases
- Test a Kafka consumer's incremental message processing without a live broker.
- Simulate a network feed of sensor data to validate buffering logic in a data pipeline.
- Mock a live streaming API for integration tests in a CI pipeline without network calls.
Sponsored
More from Big data & Spark
Keep learning
Related tutorials and quizzes for this topic.