How to Create a StatsD UDP Metric Mock Server in Python

import socket
import threading
import time


def start_mock_statsd_server(host="127.0.0.1", port=8125, timeout=3):
    """Run a mock StatsD UDP server that captures metrics for a short window."""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((host, port))
    sock.settimeout(timeout)
    metrics = []

    def collect():
        while True:
            try:
                data, _ = sock.recvfrom(4096)
                metrics.append(data.decode("utf-8").strip())
            except socket.timeout:
                break

    thread = threading.Thread(target=collect, daemon=True)
    thread.start()
    thread.join(timeout + 1)
    sock.close()
    return metrics


if __name__ == "__main__":
    # Simulate a client sending metrics while the mock server is listening
    mock_thread = threading.Thread(target=lambda: time.sleep(0.2))
    mock_thread.start()
    results = start_mock_statsd_server(timeout=1)

    client_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    client_sock.sendto(b"page.visits:5|c", ("127.0.0.1", 8125))
    client_sock.sendto(b"api.latency:124|ms", ("127.0.0.1", 8125))
    client_sock.sendto(b"errors:1|c", ("127.0.0.1", 8125))
    client_sock.close()
    mock_thread.join()

    print("Captured metrics:", results)