#44 Features from pytest-itde and pytest-saas migrated to pytest-backend

pytest-backend/exasol/pytest_backend/__init__.py

@@ -16,12 +16,12 @@
     timestamp_name
 )
 import exasol.pytest_backend.project_short_tag as pst
-from exasol.pytest_backend.itde_config import (
+from exasol.pytest_backend.itde import (
     itde_pytest_addoption,
-    ONPREM_DB_OPTIONS, OnpremDBConfig,
-    ONPREM_BFS_OPTIONS, OnpremBfsConfig,
-    SSH_OPTIONS, SshConfig,
-    ITDE_OPTIONS, ItdeConfig
+    exasol_config,
+    bucketfs_config,
+    ssh_config,
+    itde_config
 )
 
 _BACKEND_OPTION = '--backend'
@@ -103,38 +103,6 @@ def use_saas(request) -> bool:
     return _is_backend_selected(request, _BACKEND_SAAS)
 
 
-@pytest.fixture(scope="session")
-def exasol_config(request) -> OnpremDBConfig:
-    """Returns the configuration settings of the exasol db for this session."""
-    cli_arguments = request.config.option
-    kwargs = ONPREM_DB_OPTIONS.kwargs(os.environ, cli_arguments)
-    return OnpremDBConfig(**kwargs)
-
-
-@pytest.fixture(scope="session")
-def bucketfs_config(request) -> OnpremBfsConfig:
-    """Returns the configuration settings of the bucketfs for this session."""
-    cli_arguments = request.config.option
-    kwargs = ONPREM_BFS_OPTIONS.kwargs(os.environ, cli_arguments)
-    return OnpremBfsConfig(**kwargs)
-
-
-@pytest.fixture(scope="session")
-def ssh_config(request) -> SshConfig:
-    """Returns the configuration settings for SSH access in this session."""
-    cli_arguments = request.config.option
-    kwargs = SSH_OPTIONS.kwargs(os.environ, cli_arguments)
-    return SshConfig(**kwargs)
-
-
-@pytest.fixture(scope="session")
-def itde_config(request) -> ItdeConfig:
-    """Returns the configuration settings of the ITDE for this session."""
-    cli_arguments = request.config.option
-    kwargs = ITDE_OPTIONS.kwargs(os.environ, cli_arguments)
-    return ItdeConfig(**kwargs)
-
-
 @pytest.fixture(scope="session")
 def backend_aware_onprem_database(request,
                                   use_onprem,

pytest-backend/exasol/pytest_backend/config.py

@@ -0,0 +1,104 @@
+from collections import ChainMap
+from dataclasses import dataclass
+from typing import (
+    Generic,
+    Optional,
+    TypeVar,
+)
+
+T = TypeVar("T")
+
+
+@dataclass(frozen=True)
+class Option(Generic[T]):
+    name: str
+    prefix: str
+    type: T
+    default: Optional[T] = None
+    help_text: str = ""
+
+    @property
+    def env(self):
+        """Environment variable name"""
+
+        def normalize(name):
+            name = name.replace("-", "_")
+            name = name.upper()
+            return name
+
+        return f"{normalize(self.prefix)}_{normalize(self.name)}"
+
+    @property
+    def cli(self):
+        """Cli argument name"""
+
+        def normalize(name):
+            name = name.replace("_", "-")
+            name = name.lower()
+            return name
+
+        return f"--{normalize(self.prefix)}-{normalize(self.name)}"
+
+    @property
+    def pytest(self):
+        """Pytest option name"""
+
+        def normalize(name):
+            name = name.replace("-", "_")
+            name = name.lower()
+            return name
+
+        return f"{normalize(self.prefix)}_{normalize(self.name)}"
+
+    @property
+    def help(self):
+        """Help text including information about default value."""
+        if not self.default:
+            return f"{self.help_text}."
+        return f"{self.help_text} (default: {self.default})."
+
+
+class OptionGroup:
+    """
+    Wraps a set of pytest options.
+    """
+
+    def __init__(self, prefix, options):
+        self._prefix = prefix
+        self._options = tuple(Option(prefix=prefix, **kwargs) for kwargs in options)
+        self._default = {o.name: o.default for o in self._options}
+        self._env = {}
+        self._cli = {}
+        self._kwargs = ChainMap(self._cli, self._env, self._default)
+
+    @property
+    def prefix(self):
+        """The option group prefix."""
+        return self._prefix
+
+    @property
+    def options(self):
+        """A tuple of all options which are part of this group."""
+        return self._options
+
+    def kwargs(self, environment, cli_arguments):
+        """
+        Given the default values, the passed environment and cli arguments it will
+        take care of the prioritization for the option values in regard of their
+        source(s) and return a kwargs dictionary with all options and their
+        appropriate value.
+        """
+        env = {
+            o.name: o.type(environment[o.env])
+            for o in self._options
+            if o.env in environment
+        }
+        cli = {
+            o.name: getattr(cli_arguments, o.pytest)
+            for o in self.options
+            if hasattr(cli_arguments, o.pytest)
+            and getattr(cli_arguments, o.pytest) is not None
+        }
+        self._env.update(env)
+        self._cli.update(cli)
+        return self._kwargs

pytest-backend/exasol/pytest_backend/itde.py

@@ -0,0 +1,168 @@
+import os
+from dataclasses import dataclass
+import pytest
+
+from exasol_integration_test_docker_environment.lib.test_environment.ports import Ports
+import exasol.pytest_backend.config as config
+
+
+@dataclass
+class OnpremDBConfig:
+    """Exasol database configuration"""
+
+    host: str
+    port: int
+    username: str
+    password: str
+
+
+@dataclass
+class OnpremBfsConfig:
+    """Bucketfs configuration"""
+
+    url: str
+    username: str
+    password: str
+
+
+@dataclass
+class SshConfig:
+    """SSH configuration"""
+
+    port: int
+
+
+@dataclass
+class ItdeConfig:
+    """Itde configuration settings"""
+
+    db_version: str
+
+
+_ONPREM_DB_OPTIONS = config.OptionGroup(
+    prefix="exasol",
+    options=(
+        {
+            "name": "host",
+            "type": str,
+            "default": "localhost",
+            "help_text": "Host to connect to",
+        },
+        {
+            "name": "port",
+            "type": int,
+            "default": Ports.forward.database,
+            "help_text": "Port on which the exasol db is listening",
+        },
+        {
+            "name": "username",
+            "type": str,
+            "default": "SYS",
+            "help_text": "Username used to authenticate against the exasol db",
+        },
+        {
+            "name": "password",
+            "type": str,
+            "default": "exasol",
+            "help_text": "Password used to authenticate against the exasol db",
+        },
+    ),
+)
+
+_ONPREM_BFS_OPTIONS = config.OptionGroup(
+    prefix="bucketfs",
+    options=(
+        {
+            "name": "url",
+            "type": str,
+            "default": f"http://127.0.0.1:{Ports.forward.bucketfs}",
+            "help_text": "Base url used to connect to the bucketfs service",
+        },
+        {
+            "name": "username",
+            "type": str,
+            "default": "w",
+            "help_text": "Username used to authenticate against the bucketfs service",
+        },
+        {
+            "name": "password",
+            "type": str,
+            "default": "write",
+            "help_text": "Password used to authenticate against the bucketfs service",
+        },
+    ),
+)
+
+_SSH_OPTIONS = config.OptionGroup(
+    prefix="ssh",
+    options=(
+        {
+            "name": "port",
+            "type": int,
+            "default": Ports.forward.ssh,
+            "help_text": "Port on which external processes can access the database via SSH protocol",
+        },
+    ),
+)
+
+
+_ITDE_OPTIONS = config.OptionGroup(
+    prefix="itde",
+    options=(
+        {
+            "name": "db_version",
+            "type": str,
+            "default": "8.18.1",
+            "help_text": "DB version to start, if value is 'external' an existing instance will be used",
+        },
+    ),
+)
+
+_ITDE_OPTION_GROUPS = (_ONPREM_DB_OPTIONS, _ONPREM_BFS_OPTIONS, _ITDE_OPTIONS, _SSH_OPTIONS)
+
+
+def _add_option_group(parser, group):
+    parser_group = parser.getgroup(group.prefix)
+    for option in group.options:
+        parser_group.addoption(
+            option.cli,
+            type=option.type,
+            help=option.help,
+        )
+
+
+def itde_pytest_addoption(parser):
+    for group in _ITDE_OPTION_GROUPS:
+        _add_option_group(parser, group)
+
+
+@pytest.fixture(scope="session")
+def exasol_config(request) -> OnpremDBConfig:
+    """Returns the configuration settings of the exasol db for this session."""
+    cli_arguments = request.config.option
+    kwargs = _ONPREM_DB_OPTIONS.kwargs(os.environ, cli_arguments)
+    return OnpremDBConfig(**kwargs)
+
+
+@pytest.fixture(scope="session")
+def bucketfs_config(request) -> OnpremBfsConfig:
+    """Returns the configuration settings of the bucketfs for this session."""
+    cli_arguments = request.config.option
+    kwargs = _ONPREM_BFS_OPTIONS.kwargs(os.environ, cli_arguments)
+    return OnpremBfsConfig(**kwargs)
+
+
+@pytest.fixture(scope="session")
+def ssh_config(request) -> SshConfig:
+    """Returns the configuration settings for SSH access in this session."""
+    cli_arguments = request.config.option
+    kwargs = _SSH_OPTIONS.kwargs(os.environ, cli_arguments)
+    return SshConfig(**kwargs)
+
+
+@pytest.fixture(scope="session")
+def itde_config(request) -> ItdeConfig:
+    """Returns the configuration settings of the ITDE for this session."""
+    cli_arguments = request.config.option
+    kwargs = _ITDE_OPTIONS.kwargs(os.environ, cli_arguments)
+    return ItdeConfig(**kwargs)