From ce23b5b32121846604be15bc4cb71f0d154cc7ef 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] fix(toxiproxy): add Toxiproxy container module Adds a ToxiproxyContainer under testcontainers.community.toxiproxy for simulating adverse network conditions (latency, bandwidth limits, connection drops, ...) in integration tests, mirroring the Toxiproxy modules already available for testcontainers Java, Go and .NET. - ToxiproxyContainer exposes the HTTP control API and a range of proxy ports and provides create_proxy() to route traffic to an upstream. - ToxiproxyProxy.add_toxic() injects toxics via the control API. - No extra runtime dependencies (the control API is called over urllib). - Adds an integration test (proxying + latency toxic), a docs page and example, and the community extra in pyproject. --- 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 ++++++ 7 files changed, 239 insertions(+) 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