Adapt testcontainers to be testing framework agnostic

cratedb_toolkit/testing/testcontainers/cratedb.py

@@ -10,6 +10,10 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
+# This is for Python 3.7 and 3.8 to support generic types
+# like `dict` instead of `typing.Dict
+from __future__ import annotations
+
 import logging
 import os
 from typing import Optional
@@ -19,6 +23,7 @@
 from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
 
 from cratedb_toolkit.testing.testcontainers.util import KeepaliveContainer, asbool
+from cratedb_toolkit.util import DatabaseAdapter
 
 logger = logging.getLogger(__name__)
 
@@ -51,8 +56,13 @@
     CRATEDB_PASSWORD = os.environ.get("CRATEDB_PASSWORD", "")
     CRATEDB_DB = os.environ.get("CRATEDB_DB", "doc")
     KEEPALIVE = asbool(os.environ.get("CRATEDB_KEEPALIVE", os.environ.get("TC_KEEPALIVE", False)))
+    CMD_OPTS = {
+        "discovery.type": "single-node",
+        "cluster.routing.allocation.disk.threshold_enabled": False,
+        "node.attr.storage": "hot",
+        "path.repo": "/tmp/snapshots",
+    }
 
-    # TODO: Dual-port use with 4200+5432.
     def __init__(
         self,
         image: str = "crate/crate:nightly",
@@ -61,34 +71,58 @@
         password: Optional[str] = None,
         dbname: Optional[str] = None,
         dialect: str = "crate",
+        cmd_opts: Optional[dict] = None,
+        extra_ports: Optional[list] = None,
         **kwargs,
     ) -> None:
         super().__init__(image=image, **kwargs)
 
         self._name = "testcontainers-cratedb"  # -{os.getpid()}
-        self._command = "-Cdiscovery.type=single-node -Ccluster.routing.allocation.disk.threshold_enabled=false"
-        # TODO: Generalize by obtaining more_opts from caller.
-        self._command += " -Cnode.attr.storage=hot"
-        self._command += " -Cpath.repo=/tmp/snapshots"
+
+        cmd_opts = cmd_opts if cmd_opts else {}
+        self._command = self._build_cmd({**self.CMD_OPTS, **cmd_opts})
 
         self.CRATEDB_USER = user or self.CRATEDB_USER
         self.CRATEDB_PASSWORD = password or self.CRATEDB_PASSWORD
         self.CRATEDB_DB = dbname or self.CRATEDB_DB
 
         self.port_to_expose = port
+        self.extra_ports = extra_ports or []
         self.dialect = dialect
 
+    @staticmethod
+    def _build_cmd(opts: dict) -> str:
+        """
+        Return a string with command options concatenated and optimised for ES5 use
+        """
+        cmd = []
+        for key, val in opts.items():
+            if isinstance(val, bool):
+                val = str(val).lower()
+            cmd.append("-C{}={}".format(key, val))
+        return " ".join(cmd)
+
+    def _configure_ports(self) -> None:
+        """
+        Bind all the ports exposed inside the container to the same port on the host
+        """
+        ports = [*[self.port_to_expose], *self.extra_ports]
+        for port in ports:
+            # If random port is needed on the host, use host=None
+            # or invoke self.with_exposed_ports
+            self.with_bind_ports(container=port, host=port)
+
     def _configure(self) -> None:
-        self.with_exposed_ports(self.port_to_expose)
+        self._configure_ports()
         self.with_env("CRATEDB_USER", self.CRATEDB_USER)
         self.with_env("CRATEDB_PASSWORD", self.CRATEDB_PASSWORD)
         self.with_env("CRATEDB_DB", self.CRATEDB_DB)
 
-    def get_connection_url(self, host=None) -> str:
+    def get_connection_url(self, host=None, dialect=None) -> str:
         # TODO: When using `db_name=self.CRATEDB_DB`:
         #       Connection.__init__() got an unexpected keyword argument 'database'
         return super()._create_connection_url(
-            dialect=self.dialect,
+            dialect=dialect or self.dialect,
             username=self.CRATEDB_USER,
             password=self.CRATEDB_PASSWORD,
             host=host,
@@ -101,3 +135,65 @@
         #       In `testcontainers-java`, there is the `HttpWaitStrategy`.
         # TODO: Provide a client instance.
         wait_for_logs(self, predicate="o.e.n.Node.*started", timeout=MAX_TRIES)
+
+
+class TestDrive:
+    """
+    Use different schemas for storing the subsystem database tables, and the
+    test/example data, so that they do not accidentally touch the default `doc`
+    schema.
+    """
+
+    EXT_SCHEMA = "testdrive-ext"
+    DATA_SCHEMA = "testdrive-data"
+
+    RESET_TABLES = [
+        f'"{EXT_SCHEMA}"."retention_policy"',
+        f'"{DATA_SCHEMA}"."raw_metrics"',
+        f'"{DATA_SCHEMA}"."sensor_readings"',
+        f'"{DATA_SCHEMA}"."testdrive"',
+        f'"{DATA_SCHEMA}"."foobar"',
+        f'"{DATA_SCHEMA}"."foobar_unique_single"',
+        f'"{DATA_SCHEMA}"."foobar_unique_composite"',
+        # cratedb_toolkit.io.{influxdb,mongodb}
+        '"testdrive"."demo"',
+    ]
+
+
+class CrateDBFixture:
+    """
+    A little helper wrapping Testcontainer's `CrateDBContainer` and
+    CrateDB Toolkit's `DatabaseAdapter`, agnostic of the test framework.
+    """
+
+    def __init__(self, crate_version: str = "nightly", **kwargs):
+        self.cratedb: Optional[CrateDBContainer] = None
+        self.image: str = "crate/crate:{}".format(crate_version)
+        self.database: Optional[DatabaseAdapter] = None
+        self.setup(**kwargs)
+
+    def setup(self, **kwargs):
+        self.cratedb = CrateDBContainer(image=self.image, **kwargs)
+        self.cratedb.start()
+        self.database = DatabaseAdapter(dburi=self.get_connection_url())
+
+    def finalize(self):
+        if self.cratedb:
+            self.cratedb.stop()
+
+    def reset(self, tables: Optional[list] = TestDrive.RESET_TABLES):
+        if tables and self.database:
+            for reset_table in tables:
+                self.database.connection.exec_driver_sql(f"DROP TABLE IF EXISTS {reset_table};")
+
+    def get_connection_url(self, *args, **kwargs):
+        if self.cratedb:
+            return self.cratedb.get_connection_url(*args, **kwargs)
+        return None
+
+    @property
+    def http_url(self):
+        """
+        Return a URL for HTTP interface
+        """
+        return self.get_connection_url(dialect="http")

pyproject.toml

@@ -235,6 +235,8 @@ extend-ignore = [
   "RET504",
   # Unnecessary `elif` after `return` statement
   "RET505",
+  # Probable insecure usage of temporary file or directory
+  "S108",
 ]
 
 extend-exclude = [

tests/conftest.py

@@ -3,56 +3,11 @@
 import pytest
 import responses
 
-from cratedb_toolkit.testing.testcontainers.cratedb import CrateDBContainer
-from cratedb_toolkit.util import DatabaseAdapter
+from cratedb_toolkit.testing.testcontainers.cratedb import CrateDBFixture, TestDrive
 from cratedb_toolkit.util.common import setup_logging
 
-# Use different schemas for storing the subsystem database tables, and the
-# test/example data, so that they do not accidentally touch the default `doc`
-# schema.
-TESTDRIVE_EXT_SCHEMA = "testdrive-ext"
-TESTDRIVE_DATA_SCHEMA = "testdrive-data"
-
-RESET_TABLES = [
-    f'"{TESTDRIVE_EXT_SCHEMA}"."retention_policy"',
-    f'"{TESTDRIVE_DATA_SCHEMA}"."raw_metrics"',
-    f'"{TESTDRIVE_DATA_SCHEMA}"."sensor_readings"',
-    f'"{TESTDRIVE_DATA_SCHEMA}"."testdrive"',
-    f'"{TESTDRIVE_DATA_SCHEMA}"."foobar"',
-    f'"{TESTDRIVE_DATA_SCHEMA}"."foobar_unique_single"',
-    f'"{TESTDRIVE_DATA_SCHEMA}"."foobar_unique_composite"',
-    # cratedb_toolkit.io.{influxdb,mongodb}
-    '"testdrive"."demo"',
-]
-
-
-class CrateDBFixture:
-    """
-    A little helper wrapping Testcontainer's `CrateDBContainer` and
-    CrateDB Toolkit's `DatabaseAdapter`, agnostic of the test framework.
-    """
-
-    def __init__(self):
-        self.cratedb = None
-        self.database: DatabaseAdapter = None
-        self.setup()
-
-    def setup(self):
-        # TODO: Make image name configurable.
-        self.cratedb = CrateDBContainer("crate/crate:nightly")
-        self.cratedb.start()
-        self.database = DatabaseAdapter(dburi=self.get_connection_url())
-
-    def finalize(self):
-        self.cratedb.stop()
-
-    def reset(self):
-        # TODO: Make list of tables configurable.
-        for reset_table in RESET_TABLES:
-            self.database.connection.exec_driver_sql(f"DROP TABLE IF EXISTS {reset_table};")
-
-    def get_connection_url(self, *args, **kwargs):
-        return self.cratedb.get_connection_url(*args, **kwargs)
+TESTDRIVE_DATA_SCHEMA = TestDrive.DATA_SCHEMA
+TESTDRIVE_EXT_SCHEMA = TestDrive.EXT_SCHEMA
 
 
 @pytest.fixture(scope="session", autouse=True)
@@ -63,7 +18,7 @@ def configure_database_schema(session_mocker):
 
     If not configured otherwise, the test suite currently uses `testdrive-ext`.
     """
-    session_mocker.patch("os.environ", {"CRATEDB_EXT_SCHEMA": TESTDRIVE_EXT_SCHEMA})
+    session_mocker.patch("os.environ", {"CRATEDB_EXT_SCHEMA": TestDrive.EXT_SCHEMA})
 
 
 @pytest.fixture(scope="session")