fix(nats): Client-Free(ish) NATS container

index.rst

@@ -31,6 +31,7 @@ testcontainers-python facilitates the use of Docker containers for functional an
  modules/mongodb/README
  modules/mssql/README
  modules/mysql/README
+ modules/nats/README
  modules/neo4j/README
  modules/nginx/README
  modules/opensearch/README

modules/nats/README.rst

@@ -0,0 +1,2 @@
+.. autoclass:: testcontainers.nats.NatsContainer
+.. title:: testcontainers.nats.NatsContainer

modules/nats/testcontainers/nats/__init__.py

@@ -0,0 +1,75 @@
+#
+# 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 testcontainers.core.container import DockerContainer
+from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
+
+
+class NatsContainer(DockerContainer):
+ """
+ Nats container.
+
+ Example:
+
+ .. doctest::
+
+ >>> import asyncio
+ >>> from nats import connect as nats_connect
+ >>> from testcontainers.nats import NatsContainer
+
+ >>> async def test_doctest_usage():
+ ... with NatsContainer() as nats_container:
+ ... client = await nats_connect(nats_container.nats_uri())
+ ... sub_tc = await client.subscribe("tc")
+ ... await client.publish("tc", b"Test-Containers")
+ ... next_message = await sub_tc.next_msg(timeout=5.0)
+ ... await client.close()
+ ... return next_message.data
+ >>> asyncio.run(test_doctest_usage())
+ b'Test-Containers'
+ """
+
+ def __init__(
+ self,
+ image: str = "nats:latest",
+ client_port: int = 4222,
+ management_port: int = 8222,
+ expected_ready_log: str = "Server is ready",
+ ready_timeout_secs: int = 120,
+ **kwargs,
+ ) -> None:
+ super().__init__(image, **kwargs)
+ self.client_port = client_port
+ self.management_port = management_port
+ self._expected_ready_log = expected_ready_log
+ self._ready_timeout_secs = max(ready_timeout_secs, 0)
+ self.with_exposed_ports(self.client_port, self.management_port)
+
+ @wait_container_is_ready()
+ def _healthcheck(self) -> None:
+ wait_for_logs(self, self._expected_ready_log, timeout=self._ready_timeout_secs)
+
+ def nats_uri(self) -> str:
+ return f"nats://{self.get_container_host_ip()}:{self.get_exposed_port(self.client_port)}"
+
+ def nats_host_and_port(self) -> tuple[str, int]:
+ return self.get_container_host_ip(), self.get_exposed_port(self.client_port)
+
+ def nats_management_uri(self) -> str:
+ return f"nats://{self.get_container_host_ip()}:{self.get_exposed_port(self.management_port)}"
+
+ def start(self) -> "NatsContainer":
+ super().start()
+ self._healthcheck()
+ return self

modules/nats/tests/test_nats.py

@@ -0,0 +1,87 @@
+from testcontainers.nats import NatsContainer
+from uuid import uuid4
+import pytest
+
+from nats import connect as nats_connect
+from nats.aio.client import Client as NATSClient
+
+
+async def get_client(container: NatsContainer) -> "NATSClient":
+ """
+ Get a nats client.
+
+ Returns:
+ client: Nats client to connect to the container.
+ """
+ conn_string = container.nats_uri()
+ client = await nats_connect(conn_string)
+ return client
+
+
+def test_basic_container_ops():
+ with NatsContainer() as container:
+ # Not sure how to get type information without doing this
+ container: NatsContainer = container
+ h, p = container.nats_host_and_port()
+ assert h == "localhost"
+ uri = container.nats_uri()
+ management_uri = container.nats_management_uri()
+
+ assert uri != management_uri
+
+
+@pytest.mark.asyncio
+async def test_pubsub(anyio_backend):
+ with NatsContainer() as container:
+ nc: NATSClient = await get_client(container)
+
+ topic = str(uuid4())
+
+ sub = await nc.subscribe(topic)
+ sent_message = b"Test-Containers"
+ await nc.publish(topic, b"Test-Containers")
+ received_msg = await sub.next_msg()
+ print("Received:", received_msg)
+ assert sent_message == received_msg.data
+ await nc.flush()
+ await nc.close()
+
+
+@pytest.mark.asyncio
+async def test_more_complex_example(anyio_backend):
+ with NatsContainer() as container:
+ nc: NATSClient = await get_client(container)
+
+ sub = await nc.subscribe("greet.*")
+ await nc.publish("greet.joe", b"hello")
+
+ try:
+ await sub.next_msg(timeout=0.1)
+ except TimeoutError:
+ pass
+
+ await nc.publish("greet.joe", b"hello.joe")
+ await nc.publish("greet.pam", b"hello.pam")
+
+ first = await sub.next_msg(timeout=0.1)
+ assert b"hello.joe" == first.data
+
+ second = await sub.next_msg(timeout=0.1)
+ assert b"hello.pam" == second.data
+
+ await nc.publish("greet.bob", b"hello")
+
+ await sub.unsubscribe()
+ await nc.drain()
+
+
+@pytest.mark.asyncio
+async def test_doctest_usage():
+ """simpler to run test to mirror what is in the doctest"""
+ with NatsContainer() as nats_container:
+ client = await nats_connect(nats_container.nats_uri())
+ sub_tc = await client.subscribe("tc")
+ await client.publish("tc", b"Test-Containers")
+ next_message = await sub_tc.next_msg(timeout=5.0)
+ await client.close()
+ assert next_message.data == b"Test-Containers"

pyproject.toml

@@ -44,6 +44,7 @@ packages = [
  { include = "testcontainers", from = "modules/mongodb" },
  { include = "testcontainers", from = "modules/mssql" },
  { include = "testcontainers", from = "modules/mysql" },
+ { include = "testcontainers", from = "modules/nats" },
  { include = "testcontainers", from = "modules/neo4j" },
  { include = "testcontainers", from = "modules/nginx" },
  { include = "testcontainers", from = "modules/opensearch" },
@@ -79,6 +80,7 @@ pyyaml = { version = "*", optional = true }
 python-keycloak = { version = "*", optional = true }
 boto3 = { version = "*", optional = true }
 minio = { version = "*", optional = true }
+nats-py = { version = "*", optional = true }
 pymongo = { version = "*", optional = true }
 sqlalchemy = { version = "*", optional = true }
 pymssql = { version = "*", optional = true }
@@ -109,6 +111,7 @@ minio = ["minio"]
 mongodb = ["pymongo"]
 mssql = ["sqlalchemy", "pymssql"]
 mysql = ["sqlalchemy", "pymysql"]
+nats = ["nats-py"]
 neo4j = ["neo4j"]
 nginx = []
 opensearch = ["opensearch-py"]
@@ -136,6 +139,7 @@ sqlalchemy = "*"
 psycopg = "*"
 kafka-python = "^2.0.2"
 cassandra-driver = "*"
+pytest-asyncio = "0.23.5"
 
 [[tool.poetry.source]]
 name = "PyPI"