Quick Start¶
This guide will help you get up and running with Dummynet quickly. Dummynet is a Python library for creating network emulation environments, allowing you to simulate various network conditions for testing purposes.
Installation¶
The fastest way to get started is to install Dummynet using pip:
python3 -m pip install dummynet
Prerequisites¶
Dummynet requires:
Python 3.11 or higher
Linux operating system (tested on Ubuntu 20.04+)
Root privileges or appropriate capabilities for network namespace manipulation
iproute2andiptablespackages installed
Basic Usage¶
DummyNet depends upon the creation of a root host shell to run its commands. To create it, you can use the following pattern:
import dummynet
import logging
import os
# Check if we need to run as sudo
sudo = os.getuid() != 0
# Create a process monitor
log = logging.getLogger("dummynet")
log.setLevel(logging.DEBUG)
process_monitor = dummynet.ProcessMonitor(log=log)
# Create the host shell
shell = dummynet.HostShell(log=log, sudo=sudo, process_monitor=process_monitor)
Context Manager (Recommended)¶
The recommended way to use Dummynet is with a context manager, which ensures cleanup steps always run on exit:
with dummynet.DummyNet(shell=shell) as net:
# Your network code here, e.g.
net.link_veth_pair("veth0", "veth1")
# Cleanup happens automatically
Manual Management¶
For more control, you can manage the lifecycle manually:
net = dummynet.DummyNet(shell=shell)
try:
# Your network code here, e.g.
net.link_veth_pair("veth0", "veth1")
finally:
net.cleanup()
Warning
Always ensure net.cleanup() is called when using manual management to avoid
leaving any network artifacts dangling on your system!
Interface Management¶
Dummynet uses a special naming convention for interfaces to support concurrent execution:
# Create a veth pair using dummynet
net.link_veth_pair("veth0", "veth1")
This creates a veth pair locally called d-XXXX-veth0 and d-XXXX-veth1,
where:
d-is the Dummynet prefixXXXXis the current Python process ID encoded as a base58 stringveth0is your specified interface name
This naming scheme allows multiple Dummynet instances to run concurrently inside
multiple Python instances without conflicts, making it particularly useful for
parallelized testing with tools like pytest-xdist.
When using strings, these will always point to these scoped variants.
example_netns = net.netns_add("example")
net.link_set("eth0", namespace=example_netns)
# This will fail! As it actually points to `d-XXXX-eth0`,
# which doesn't exist!
# The correct actions would be to first create the scoped interfaces
net.link_veth_pair("eth0", "eth1")
net.link_set("eth0", namespace=example_netns)
# this will now move `d-XXXX-eth0` without error.
It is therefore reccomended to use the scoped variables returned by dummynet to explicitly show this behaviour, instead of relying on the string-based API.
example_netns = net.netns_add("example")
veth0, veth1 = net.link_veth_pair("veth0", "veth1")
net.link_set(veth0, namespace=example_netns)
To actually point at system devices, some are listed as UNSCOPED_NAMES, like
the loopback interface lo, which will never be scoped. If you still need a
system interface, you can use the InterfaceScoped class directly, and set its
uid to 0, which acts as a magic variable to always keep the interface
unscoped.
example_netns = net.netns_add("example")
eth0 = dummynet.InterfaceScoped("eth0", uid=0)
net.link_set(eth0, namespace=example_netns)
# This will actually move `eth0`, be careful!
Complete Example¶
Here’s a complete example demonstrating DummyNet:
1import dummynet
2import logging
3import sys
4import argparse
5
6
7def run():
8 parser = argparse.ArgumentParser(description="Program with debugger option")
9 parser.add_argument("--debug", action="store_true", help="Enable log debugger")
10 args = parser.parse_args()
11
12 log = logging.getLogger("dummynet")
13 log.setLevel(logging.DEBUG)
14
15 if args.debug:
16 console_handler = logging.StreamHandler(sys.stdout)
17 console_handler.setLevel(logging.DEBUG)
18 log.addHandler(console_handler)
19
20 process_monitor = dummynet.ProcessMonitor(log=log)
21 shell = dummynet.HostShell(log=log, sudo=True, process_monitor=process_monitor)
22 net = dummynet.DummyNet(shell=shell)
23
24 try:
25 cgroup0 = net.add_cgroup(
26 name="test_cgroup0",
27 shell=shell,
28 log=log,
29 cpu_limit=0.5,
30 memory_limit=200000000,
31 )
32
33 cgroup1 = net.add_cgroup(
34 name="test_cgroup1",
35 shell=shell,
36 log=log,
37 cpu_limit=0.2,
38 memory_limit=100000000,
39 )
40
41 # Get a list of the current namespaces
42 namespaces = net.netns_list()
43 assert namespaces == []
44
45 # create two namespaces
46 demo0 = net.netns_add(name="demo0")
47 demo1 = net.netns_add(name="demo1")
48
49 net.link_veth_add(p1_name="demo0-0", p2_name="demo1-0")
50
51 # Move the interfaces to the namespaces
52 net.link_set(namespace="demo0", interface="demo0-0")
53 net.link_set(namespace="demo1", interface="demo1-0")
54
55 # Bind an IP-address to the two peers in the link.
56 demo0.addr_add(ip="10.0.0.1/24", interface="demo0-0")
57 demo1.addr_add(ip="10.0.0.2/24", interface="demo1-0")
58
59 # Activate the interfaces.
60 demo0.up(interface="demo0-0")
61 demo1.up(interface="demo1-0")
62 demo0.up(interface="lo")
63 demo1.up(interface="lo")
64
65 # Test will run until last non-daemon process is done.
66 proc0 = demo0.run_async(cmd="ping -c 10 10.0.0.2", daemon=True)
67 proc1 = demo1.run_async(cmd="ping -c 10 10.0.0.1")
68
69 # # Add the processes to the cgroup.
70 cgroup0.add_pid(proc0.pid)
71 cgroup1.add_pid(proc1.pid)
72
73 # Print output as we go (optional)
74 def _proc0_stdout(data):
75 print("proc0: {}".format(data))
76
77 def _proc1_stdout(data):
78 print("proc1: {}".format(data))
79
80 proc0.stdout_callback = _proc0_stdout
81 proc1.stdout_callback = _proc1_stdout
82
83 while process_monitor.keep_running():
84 pass
85
86 # Check that the ping succeeded.
87 proc1.match(stdout="10 packets transmitted*", stderr=None)
88
89 # Since proc0 is a daemon we automatically kill it when the last
90 # non-daemon process is done. However we can still see the output
91 # it generated.
92 print(f"proc0: {proc0.stdout}")
93
94 finally:
95 net.cleanup()
96
97
98if __name__ == "__main__":
99 run()