feat(n8n): add n8n workflow automation container module

Closes #901

Author: demetere

modules/n8n/README.rst

@@ -0,0 +1,2 @@
+.. autoclass:: testcontainers.n8n.N8nContainer
+.. title:: testcontainers.n8n.N8nContainer

modules/n8n/testcontainers/n8n/__init__.py

@@ -0,0 +1,74 @@
+#
+#    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.
+
+from http.client import HTTPException
+from typing import Optional
+from urllib.error import URLError
+from urllib.request import urlopen
+
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.waiting_utils import wait_container_is_ready
+
+
+class N8nContainer(DockerContainer):
+    """
+    n8n workflow automation container.
+
+    Example:
+
+        .. doctest::
+
+            >>> from testcontainers.n8n import N8nContainer
+
+            >>> with N8nContainer() as n8n:
+            ...     url = n8n.get_url()
+            ...     assert url.startswith("http")
+    """
+
+    def __init__(
+        self,
+        image: str = "docker.n8n.io/n8nio/n8n:latest",
+        port: int = 5678,
+        encryption_key: Optional[str] = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(image, **kwargs)
+        self.port = port
+        self.encryption_key = encryption_key
+        self.with_exposed_ports(self.port)
+
+    def _configure(self) -> None:
+        self.with_env("N8N_PORT", str(self.port))
+        self.with_env("N8N_DIAGNOSTICS_ENABLED", "false")
+        if self.encryption_key:
+            self.with_env("N8N_ENCRYPTION_KEY", self.encryption_key)
+
+    def get_url(self) -> str:
+        host = self.get_container_host_ip()
+        port = self.get_exposed_port(self.port)
+        return f"http://{host}:{port}"
+
+    def get_webhook_url(self) -> str:
+        return f"{self.get_url()}/webhook-test"
+
+    @wait_container_is_ready(HTTPException, URLError, ConnectionError)
+    def _healthcheck(self) -> None:
+        url = f"{self.get_url()}/healthz"
+        with urlopen(url, timeout=5) as res:
+            if res.status > 299:
+                raise HTTPException()
+
+    def start(self) -> "N8nContainer":
+        super().start()
+        self._healthcheck()
+        return self

modules/n8n/tests/test_n8n.py

@@ -0,0 +1,19 @@
+import json
+from urllib.request import urlopen
+
+from testcontainers.n8n import N8nContainer
+
+
+def test_docker_run_n8n():
+    with N8nContainer("docker.n8n.io/n8nio/n8n:latest") as n8n:
+        url = n8n.get_url()
+        with urlopen(f"{url}/healthz") as response:
+            assert response.status == 200
+            data = json.loads(response.read().decode())
+            assert data["status"] == "ok"
+
+
+def test_n8n_get_webhook_url():
+    with N8nContainer() as n8n:
+        webhook_url = n8n.get_webhook_url()
+        assert "/webhook-test" in webhook_url

pyproject.toml

@@ -84,6 +84,7 @@ mongodb = ["pymongo>=4"]
 mqtt = []
 mssql = ["sqlalchemy>=2", "pymssql>=2"]
 mysql = ["sqlalchemy>=2", "pymysql[rsa]>=1"]
+n8n = []
 nats = ["nats-py>=2"]
 neo4j = ["neo4j>=6"]
 nginx = []
@@ -201,6 +202,7 @@ packages = [
   "modules/mqtt/testcontainers",
   "modules/mssql/testcontainers",
   "modules/mysql/testcontainers",
+  "modules/n8n/testcontainers",
   "modules/nats/testcontainers",
   "modules/neo4j/testcontainers",
   "modules/nginx/testcontainers",
@@ -250,6 +252,7 @@ dev-mode-dirs = [
   "modules/mqtt",
   "modules/mssql",
   "modules/mysql",
+  "modules/n8n",
   "modules/nats",
   "modules/neo4j",
   "modules/nginx",