From 0816c97075ced2093d4fd00df6bc99346c69c6d7 Mon Sep 17 00:00:00 2001
From: Jonnas Figueiredo <125215987+JonnasFigueiredo@users.noreply.github.com>
Date: Wed, 19 Aug 2026 09:43:01 -0300
Subject: [PATCH 1/2] fix(toxiproxy): add Toxiproxy container module
Lets you simulate bad network conditions - latency, limited bandwidth,
dropped connections - against a service in integration tests. Toxiproxy
already ships as a module for the Java, Go and .NET bindings, so this just
catches Python up.
The container exposes the HTTP control API and a range of proxy ports.
create_proxy() puts a proxy in front of an upstream and add_toxic() adds the
failures; it talks to the control API over urllib so there are no extra
dependencies. Comes with an integration test, docs and an example.
---
docs/community/toxiproxy.rst | 2 +
docs/modules/toxiproxy.md | 26 ++++
docs/modules/toxiproxy_example.py | 31 ++++
mkdocs.yml | 1 +
pyproject.toml | 1 +
.../community/toxiproxy/__init__.py | 137 ++++++++++++++++++
tests/community/toxiproxy/test_toxiproxy.py | 41 ++++++
uv.lock | 2 +-
8 files changed, 240 insertions(+), 1 deletion(-)
create mode 100644 docs/community/toxiproxy.rst
create mode 100644 docs/modules/toxiproxy.md
create mode 100644 docs/modules/toxiproxy_example.py
create mode 100644 src/testcontainers/community/toxiproxy/__init__.py
create mode 100644 tests/community/toxiproxy/test_toxiproxy.py
diff --git a/docs/community/toxiproxy.rst b/docs/community/toxiproxy.rst
new file mode 100644
index 000000000..1399a516f
--- /dev/null
+++ b/docs/community/toxiproxy.rst
@@ -0,0 +1,2 @@
+.. autoclass:: testcontainers.community.toxiproxy.ToxiproxyContainer
+.. title:: testcontainers.community.toxiproxy.ToxiproxyContainer
diff --git a/docs/modules/toxiproxy.md b/docs/modules/toxiproxy.md
new file mode 100644
index 000000000..4354f667d
--- /dev/null
+++ b/docs/modules/toxiproxy.md
@@ -0,0 +1,26 @@
+# Toxiproxy
+
+Since testcontainers-python :material-tag: v4.16.0
+
+## Introduction
+
+The Testcontainers module for [Toxiproxy](https://github.com/Shopify/toxiproxy), a TCP proxy for
+simulating adverse network conditions. Put Toxiproxy in front of a dependency and inject latency,
+bandwidth limits, connection drops and other failures to test how your application behaves when its
+dependencies misbehave.
+
+## Adding this module to your project dependencies
+
+Please run the following command to add the Toxiproxy module to your python dependencies:
+
+```bash
+pip install testcontainers[toxiproxy]
+```
+
+## Usage example
+
+
+
+[Injecting latency with Toxiproxy](toxiproxy_example.py)
+
+
diff --git a/docs/modules/toxiproxy_example.py b/docs/modules/toxiproxy_example.py
new file mode 100644
index 000000000..3890d2daf
--- /dev/null
+++ b/docs/modules/toxiproxy_example.py
@@ -0,0 +1,31 @@
+import time
+
+import requests
+
+from testcontainers.community.nginx import NginxContainer
+from testcontainers.community.toxiproxy import ToxiproxyContainer
+from testcontainers.core.network import Network
+
+
+def latency_example():
+ with Network() as network:
+ # An upstream service and Toxiproxy share a network so Toxiproxy can reach it.
+ nginx = NginxContainer("nginx:alpine").with_network(network).with_network_aliases("nginx")
+ toxiproxy = ToxiproxyContainer().with_network(network)
+ with nginx, toxiproxy:
+ # Route traffic to the upstream "nginx:80" through Toxiproxy.
+ proxy = toxiproxy.create_proxy("nginx", "nginx:80")
+ url = f"http://{proxy.host}:{proxy.proxy_port}/"
+
+ print(f"Status without toxics: {requests.get(url).status_code}")
+
+ # Inject 1 second of downstream latency.
+ proxy.add_toxic("latency", {"latency": 1000})
+
+ start = time.monotonic()
+ requests.get(url)
+ print(f"Request took {time.monotonic() - start:.2f}s with the latency toxic")
+
+
+if __name__ == "__main__":
+ latency_example()
diff --git a/mkdocs.yml b/mkdocs.yml
index d918b4238..761670cd2 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -96,6 +96,7 @@ nav:
- modules/selenium.md
- modules/sftp.md
- modules/test_module_import.md
+ - modules/toxiproxy.md
- modules/vault.md
- System Requirements:
- system_requirements/index.md
diff --git a/pyproject.toml b/pyproject.toml
index 67ad180a8..4ba4209ed 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -105,6 +105,7 @@ registry = ["bcrypt>=5"]
selenium = ["selenium>=4"]
scylla = ["cassandra-driver>=3; python_version < '3.14'"]
sftp = ["cryptography"]
+toxiproxy = []
valkey = []
vault = []
weaviate = ["weaviate-client>=4"]
diff --git a/src/testcontainers/community/toxiproxy/__init__.py b/src/testcontainers/community/toxiproxy/__init__.py
new file mode 100644
index 000000000..bfa2996d7
--- /dev/null
+++ b/src/testcontainers/community/toxiproxy/__init__.py
@@ -0,0 +1,137 @@
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+import json
+from typing import Any, Optional
+from urllib.request import Request, urlopen
+
+from typing_extensions import Self
+
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.wait_strategies import HttpWaitStrategy
+
+
+def _control_request(url: str, method: str = "GET", body: Optional[dict[str, Any]] = None) -> dict[str, Any]:
+ data = json.dumps(body).encode() if body is not None else None
+ request = Request(url, data=data, method=method, headers={"Content-Type": "application/json"})
+ with urlopen(request, timeout=5) as response:
+ raw = response.read().decode()
+ return json.loads(raw) if raw else {}
+
+
+class ToxiproxyProxy:
+ """A proxy created on a running :class:`ToxiproxyContainer`.
+
+ Connect to the upstream *through* the proxy using :attr:`host` and
+ :attr:`proxy_port`, then inject failures with :meth:`add_toxic`.
+ """
+
+ def __init__(self, name: str, host: str, proxy_port: int, control_url: str) -> None:
+ self.name = name
+ self.host = host
+ self.proxy_port = proxy_port
+ self._control_url = control_url
+
+ def add_toxic(
+ self,
+ toxic_type: str,
+ attributes: dict[str, Any],
+ stream: str = "downstream",
+ toxicity: float = 1.0,
+ name: Optional[str] = None,
+ ) -> dict[str, Any]:
+ """Add a toxic (e.g. ``latency``, ``bandwidth``, ``timeout``) to the proxy.
+
+ See https://github.com/Shopify/toxiproxy#toxics for the available types
+ and their attributes.
+ """
+ payload = {
+ "name": name or f"{self.name}_{toxic_type}_{stream}",
+ "type": toxic_type,
+ "stream": stream,
+ "toxicity": toxicity,
+ "attributes": attributes,
+ }
+ return _control_request(f"{self._control_url}/proxies/{self.name}/toxics", "POST", payload)
+
+
+class ToxiproxyContainer(DockerContainer):
+ """Toxiproxy TCP proxy for simulating adverse network conditions in tests.
+
+ Toxiproxy sits in front of another service and lets tests inject latency,
+ bandwidth limits, connection drops and other failures to verify resilience.
+ See https://github.com/Shopify/toxiproxy.
+
+ Example:
+
+ .. doctest::
+
+ >>> from testcontainers.core.network import Network
+ >>> from testcontainers.community.nginx import NginxContainer
+ >>> from testcontainers.community.toxiproxy import ToxiproxyContainer
+
+ >>> with Network() as network:
+ ... nginx = NginxContainer("nginx:alpine").with_network(network).with_network_aliases("nginx")
+ ... toxiproxy = ToxiproxyContainer().with_network(network)
+ ... with nginx, toxiproxy:
+ ... proxy = toxiproxy.create_proxy("nginx", "nginx:80")
+ ... proxy.add_toxic("latency", {"latency": 1000})
+ """
+
+ CONTROL_PORT = 8474
+ FIRST_PROXY_PORT = 8666
+ LAST_PROXY_PORT = 8697
+
+ def __init__(self, image: str = "ghcr.io/shopify/toxiproxy:2.11.0", **kwargs: object) -> None:
+ super().__init__(image, **kwargs)
+ proxy_ports = range(self.FIRST_PROXY_PORT, self.LAST_PROXY_PORT + 1)
+ self.with_exposed_ports(self.CONTROL_PORT, *proxy_ports)
+ self.waiting_for(HttpWaitStrategy(self.CONTROL_PORT, "/version"))
+ self._next_proxy_port = self.FIRST_PROXY_PORT
+
+ def get_control_port(self) -> int:
+ """Host port mapped to the Toxiproxy HTTP control API."""
+ return self.get_exposed_port(self.CONTROL_PORT)
+
+ def get_control_url(self) -> str:
+ """Base URL of the Toxiproxy HTTP control API, reachable from the host."""
+ return f"http://{self.get_container_host_ip()}:{self.get_control_port()}"
+
+ def create_proxy(self, name: str, upstream: str) -> ToxiproxyProxy:
+ """Create a proxy in front of ``upstream``.
+
+ ``upstream`` must be reachable from the Toxiproxy container itself, e.g.
+ ``"host:port"`` of another container that shares a network (via
+ :meth:`~testcontainers.core.container.DockerContainer.with_network_aliases`).
+ Returns a :class:`ToxiproxyProxy` whose ``host``/``proxy_port`` you
+ connect through instead of talking to the upstream directly.
+ """
+ if self._next_proxy_port > self.LAST_PROXY_PORT:
+ max_proxies = self.LAST_PROXY_PORT - self.FIRST_PROXY_PORT + 1
+ raise RuntimeError(f"No free proxy ports left (at most {max_proxies} proxies are supported).")
+ listen_port = self._next_proxy_port
+ self._next_proxy_port += 1
+ _control_request(
+ f"{self.get_control_url()}/proxies",
+ "POST",
+ {"name": name, "listen": f"0.0.0.0:{listen_port}", "upstream": upstream, "enabled": True},
+ )
+ return ToxiproxyProxy(
+ name=name,
+ host=self.get_container_host_ip(),
+ proxy_port=self.get_exposed_port(listen_port),
+ control_url=self.get_control_url(),
+ )
+
+ def start(self) -> Self:
+ super().start()
+ return self
diff --git a/tests/community/toxiproxy/test_toxiproxy.py b/tests/community/toxiproxy/test_toxiproxy.py
new file mode 100644
index 000000000..5c006bcac
--- /dev/null
+++ b/tests/community/toxiproxy/test_toxiproxy.py
@@ -0,0 +1,41 @@
+import time
+
+import requests
+
+from testcontainers.community.nginx import NginxContainer
+from testcontainers.community.toxiproxy import ToxiproxyContainer
+from testcontainers.core.network import Network
+
+
+def test_toxiproxy_proxies_traffic_and_injects_latency():
+ with Network() as network:
+ nginx = NginxContainer("nginx:alpine").with_network(network).with_network_aliases("nginx")
+ toxiproxy = ToxiproxyContainer().with_network(network)
+ with nginx, toxiproxy:
+ proxy = toxiproxy.create_proxy("nginx", "nginx:80")
+ url = f"http://{proxy.host}:{proxy.proxy_port}/"
+
+ # Traffic flows through the proxy to the upstream nginx.
+ response = requests.get(url, timeout=10)
+ assert response.status_code == 200
+ assert "nginx" in response.text.lower()
+
+ # Injecting 1s of downstream latency slows the response down.
+ proxy.add_toxic("latency", {"latency": 1000})
+ start = time.monotonic()
+ assert requests.get(url, timeout=10).status_code == 200
+ assert time.monotonic() - start >= 1.0
+
+
+def test_create_proxy_runs_out_of_ports():
+ toxiproxy = ToxiproxyContainer()
+ with toxiproxy:
+ # Exhaust the available proxy port range without a real upstream; the
+ # control API happily registers proxies pointing at an unused address.
+ for i in range(ToxiproxyContainer.FIRST_PROXY_PORT, ToxiproxyContainer.LAST_PROXY_PORT + 1):
+ toxiproxy.create_proxy(f"p{i}", "example:1234")
+ try:
+ toxiproxy.create_proxy("one-too-many", "example:1234")
+ raise AssertionError("expected RuntimeError when out of proxy ports")
+ except RuntimeError:
+ pass
diff --git a/uv.lock b/uv.lock
index 3ad0e8d5b..e6c8190a5 100644
--- a/uv.lock
+++ b/uv.lock
@@ -5711,7 +5711,7 @@ requires-dist = [
{ name = "weaviate-client", marker = "extra == 'weaviate'", specifier = ">=4" },
{ name = "wrapt" },
]
-provides-extras = ["arangodb", "aws", "azurite", "cassandra", "clickhouse", "cosmosdb", "cockroachdb", "cratedb", "db2", "elasticsearch", "generic", "test-module-import", "google", "influxdb", "k3s", "kafka", "keycloak", "localstack", "mailpit", "memcached", "minio", "milvus", "mongodb", "mqtt", "mssql", "mysql", "nats", "neo4j", "nginx", "openfga", "opensearch", "ollama", "oracle", "oracle-free", "postgres", "qdrant", "rabbitmq", "redis", "registry", "selenium", "scylla", "sftp", "valkey", "vault", "weaviate", "chroma", "trino"]
+provides-extras = ["arangodb", "aws", "azurite", "cassandra", "clickhouse", "cosmosdb", "cockroachdb", "cratedb", "db2", "elasticsearch", "generic", "test-module-import", "google", "influxdb", "k3s", "kafka", "keycloak", "localstack", "mailpit", "memcached", "minio", "milvus", "mongodb", "mqtt", "mssql", "mysql", "nats", "neo4j", "nginx", "openfga", "opensearch", "ollama", "oracle", "oracle-free", "postgres", "qdrant", "rabbitmq", "redis", "registry", "selenium", "scylla", "sftp", "toxiproxy", "valkey", "vault", "weaviate", "chroma", "trino"]
[package.metadata.requires-dev]
dev = [
From 9eedff63e7aa12fe42b6240e53b4911035e7fd39 Mon Sep 17 00:00:00 2001
From: Dave Ankin
Date: Tue, 25 Aug 2026 23:26:01 -0400
Subject: [PATCH 2/2] fix
---
src/testcontainers/community/toxiproxy/__init__.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/src/testcontainers/community/toxiproxy/__init__.py b/src/testcontainers/community/toxiproxy/__init__.py
index bfa2996d7..8253bbb6d 100644
--- a/src/testcontainers/community/toxiproxy/__init__.py
+++ b/src/testcontainers/community/toxiproxy/__init__.py
@@ -84,7 +84,7 @@ class ToxiproxyContainer(DockerContainer):
... toxiproxy = ToxiproxyContainer().with_network(network)
... with nginx, toxiproxy:
... proxy = toxiproxy.create_proxy("nginx", "nginx:80")
- ... proxy.add_toxic("latency", {"latency": 1000})
+ ... _ = proxy.add_toxic("latency", {"latency": 1000})
"""
CONTROL_PORT = 8474