implement pytest plugin and test for backend module

.pre-commit-config.yaml

+# SPDX-FileCopyrightText: 2020-2023 Helmholtz-Zentrum hereon GmbH
+#
+# SPDX-License-Identifier: CC0-1.0
+
 # https://pre-commit.com/
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.0.1
+    rev: v4.4.0
     hooks:
       - id: trailing-whitespace
       - id: end-of-file-fixer
       - id: check-yaml
   # isort should run before black as black sometimes tweaks the isort output
   - repo: https://github.com/PyCQA/isort
-    rev: 5.9.3
+    rev: 5.12.0
     hooks:
       - id: isort
         args:
@@ -19,32 +23,35 @@ repos:
           - --filter-files
           - -skip-gitignore
           - --float-to-top
-          - -p
-          - .
   # https://github.com/python/black#version-control-integration
   - repo: https://github.com/psf/black
-    rev: 22.3.0
+    rev: 23.1.0
     hooks:
       - id: black
         args:
           - --line-length
           - "79"
           - --exclude
-          - migrations
-          - --exclude
           - venv
-          - dasf_broker
-          - testproject
   - repo: https://github.com/keewis/blackdoc
-    rev: v0.3.4
+    rev: v0.3.8
     hooks:
       - id: blackdoc
   - repo: https://github.com/pycqa/flake8
     rev: 6.0.0
     hooks:
       - id: flake8
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.4.1
+    hooks:
+      - id: mypy
+        additional_dependencies:
+          - types-PyYAML
+          - django-stubs==4.2.3
+          - django-environ
+          - "daphne"
+          - git+https://codebase.helmholtz.cloud/hcdc/django/clm-community/django-academic-community.git@django-cms-4
+          - djangocms-frontend
+          - types-requests
         args:
-          - --exclude
-          - migrations
-          - --exclude
-          - versioneer.py
+          - --ignore-missing-imports

conftest.py

@@ -20,3 +20,25 @@
 #
 # You should have received a copy of the EUPL-1.2 license along with this
 # program. If not, see https://www.eupl.eu/.
+
+from pathlib import Path
+from typing import Callable
+
+import pytest
+
+
+@pytest.fixture(scope="session")
+def base_test_modules_path() -> Path:
+    return Path(__file__).parent / "tests" / "test_modules"
+
+
+@pytest.fixture(scope="session")
+def get_test_module_path(
+    base_test_modules_path: Path,
+) -> Callable[[str], str]:
+    """Get the path to a test module"""
+
+    def get_path(mod: str) -> str:
+        return str(base_test_modules_path / (mod + ".py"))
+
+    return get_path

dasf_broker/app_settings.py

@@ -48,7 +48,6 @@ DASF_CREATE_TOPIC_ON_MESSAGE: bool = getattr(
 
 
 class StoreMessageOptions(str, Enum):
-
     DISABLED = "disabled"
     CACHE = "cache"
     CACHEALL = "cacheall"

dasf_broker/consumers.py

@@ -129,7 +129,6 @@ class PongConsumer(JsonWebsocketConsumer):
     """A consumer to handle pong messages of a topic."""
 
     def connect(self):
-
         from guardian.shortcuts import get_anonymous_user
 
         try:
@@ -142,7 +141,6 @@ class PongConsumer(JsonWebsocketConsumer):
         self.accept()
 
     def receive_json(self, content, **kwargs):
-
         from dasf_broker.models import BrokerTopic
 
         # HACK: daphne seems to not take the FORCE_SCRIPT_NAME into account.
@@ -189,7 +187,6 @@ class TopicConsumer(JsonWebsocketConsumer):
         return BrokerTopic.objects.filter(slug=self.dasf_topic_slug).first()
 
     def connect(self):
-
         from guardian.shortcuts import get_anonymous_user
 
         try:

dasf_broker/management/commands/dasf_topic.py

@@ -256,7 +256,6 @@ class Command(BaseCommand):
                     slug=topic_slug,
                 )
             except BrokerTopic.DoesNotExist:
-
                 raise ValueError(
                     f"A topic with the slug {topic_slug} does not exist. "
                     "If you want to create it, please use "

dasf_broker/tests/live_ws_server_helper.py

+"""A module for a live_server with support for django-channels.
+"""
+
+# SPDX-FileCopyrightText: 2024 Helmholtz-Zentrum hereon GmbH
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from functools import partial
+from typing import Any, Dict
+
+from channels.routing import get_default_application
+from daphne.testing import DaphneProcess
+from django.contrib.staticfiles.handlers import ASGIStaticFilesHandler
+from django.core.exceptions import ImproperlyConfigured
+from django.db import connections
+from django.test.utils import modify_settings
+
+
+def make_application(*, static_wrapper):
+    # Module-level function for pickle-ability
+    application = get_default_application()
+    if static_wrapper is not None:
+        application = static_wrapper(application)
+    return application
+
+
+class ChannelsLiveServer:
+    """
+    Does basically the same as TransactionTestCase but also launches a
+    live Daphne server in a separate process, so
+    that the tests may use another test framework, such as Selenium,
+    instead of the built-in dummy client.
+    """
+
+    host = "localhost"
+    ProtocolServerProcess = DaphneProcess
+    static_wrapper = ASGIStaticFilesHandler
+    serve_static = True
+
+    def __init__(self, addr: str, *, start: bool = True) -> None:
+        self.live_server_kwargs: Dict[str, Any] = {}
+        try:
+            host, port = addr.split(":")
+        except ValueError:
+            host = addr
+        else:
+            self.live_server_kwargs["port"] = int(port)
+
+        for connection in connections.all():
+            if self._is_in_memory_db(connection):
+                raise ImproperlyConfigured(
+                    "ChannelLiveServerTestCase can not be used with in memory databases"
+                )
+
+        self.host = host
+
+        self._live_server_modified_settings = modify_settings(
+            ALLOWED_HOSTS={"append": self.host}
+        )
+
+        self._live_server_modified_settings.enable()
+
+        get_application = partial(
+            make_application,
+            static_wrapper=self.static_wrapper if self.serve_static else None,
+        )
+        self._server_process = self.ProtocolServerProcess(
+            self.host, get_application, **self.live_server_kwargs
+        )
+
+        if start:
+            self._server_process.start()
+            self._server_process.ready.wait()
+        self._port = self._server_process.port.value
+
+    @property
+    def url(self):
+        return "http://%s:%s" % (self.host, self._port)
+
+    @property
+    def ws_url(self):
+        return "ws://%s:%s" % (self.host, self._port)
+
+    def stop(self):
+        self._live_server_modified_settings.disable()
+        self._server_process.terminate()
+        self._server_process.join()
+
+    def _is_in_memory_db(self, connection):
+        """
+        Check if DatabaseWrapper holds in memory database.
+        """
+        if connection.vendor == "sqlite":
+            return connection.is_in_memory_db()

dasf_broker/tests/plugin.py

+"""pytest plugin for dasf setups."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess as spr
+import sys
+import time
+import uuid
+from typing import TYPE_CHECKING, Callable, Dict, Generator, List, Union
+
+import pytest
+from pytest_django.lazy_django import skip_if_no_django
+
+if TYPE_CHECKING:
+    from .live_ws_server_helper import ChannelsLiveServer
+
+
+@pytest.fixture
+def live_ws_server(
+    transactional_db, request: pytest.FixtureRequest
+) -> Generator[ChannelsLiveServer, None, None]:
+    """Run a live Daphne server in the background during tests
+
+    The address the server is started from is taken from the
+    --liveserver command line option or if this is not provided from
+    the DJANGO_LIVE_TEST_SERVER_ADDRESS environment variable.  If
+    neither is provided ``localhost`` is used.  See the Django
+    documentation for its full syntax.
+
+    Notes
+    -----
+    - Static assets will be automatically served when
+      ``django.contrib.staticfiles`` is available in INSTALLED_APPS.
+    - this fixture is a combination of the ``live_server`` fixture of
+      pytest-django and channels ``ChannelsLiveServerTestCase``. Different
+      from pytest-djangos fixture, we need a function-scope however, as the
+      daphne process runs in a different process (and not in a different
+      thread as with pytest-django).
+    - You can't use an in-memory database for your live tests. Therefore
+      include a test database file name in your settings to tell Django to use
+      a file database if you use SQLite:
+
+      .. code-block:: python
+
+          DATABASES = {
+              "default": {
+                  "ENGINE": "django.db.backends.sqlite3",
+                  "NAME": os.path.join(BASE_DIR, "db.sqlite3"),
+                  "TEST": {
+                      "NAME": os.path.join(BASE_DIR, "db_test.sqlite3"),
+                  },
+              },
+          }
+    """
+    import dasf_broker.tests.live_ws_server_helper as live_ws_server_helper
+
+    skip_if_no_django()
+
+    server = live_ws_server_helper.ChannelsLiveServer("localhost")
+    yield server
+    server.stop()
+
+
+def create_brokertopic(slug: str):
+    from guardian.shortcuts import assign_perm, get_anonymous_user
+
+    from dasf_broker.models import BrokerTopic
+
+    topic = BrokerTopic.objects.create(slug=slug, is_public=True)
+    assign_perm("can_consume", get_anonymous_user(), topic)
+
+
+@pytest.fixture
+def random_topic(db) -> str:
+    """Generate a random topic for testing."""
+    topic_slug = "test_topic_" + uuid.uuid4().urn[9:]
+    create_brokertopic(topic_slug)
+    return topic_slug
+
+
+def get_test_module_args(
+    topic: str,
+    live_ws_server: ChannelsLiveServer,
+) -> List[str]:
+    """Generate a command for subprocess to launch the test module."""
+    from dasf_broker import app_settings
+
+    return [
+        "-t",
+        topic,
+        "--websocket-url",
+        "%s/%s"
+        % (live_ws_server.ws_url, app_settings.DASF_WEBSOCKET_URL_ROUTE),
+    ]
+
+
+@pytest.fixture
+def get_module_command(live_ws_server) -> Callable[[str, str], List[str]]:
+    """A factory to generate commands from a backend module."""
+
+    def factory(topic: str, path_or_mod: str) -> List[str]:
+        """Generate a command to connect a script backend module."""
+        if os.path.exists(path_or_mod):
+            base = [sys.executable, path_or_mod]
+        else:
+            base = [sys.executable, "-m", path_or_mod]
+        return base + get_test_module_args(topic, live_ws_server)
+
+    return factory
+
+
+@pytest.fixture
+def connect_module(
+    get_module_command: Callable[[str, str], List[str]],
+) -> Generator[Callable[[str, str], spr.Popen], None, None]:
+    """Get a factory that connects DASF backend module scripts"""
+    processes: List[spr.Popen] = []
+
+    def connect(topic: str, path_or_mod: str) -> spr.Popen:
+        """Connect a script to the live_server"""
+        command = get_module_command(topic, path_or_mod)
+        try:
+            process = spr.Popen(command + ["listen"])
+        except Exception:
+            raise
+        else:
+            processes.append(process)
+            time.sleep(1)
+            return process
+
+    yield connect
+
+    for process in processes:
+        process.terminate()
+
+
+@pytest.fixture
+def test_dasf_connect(
+    get_module_command: Callable[[str, str], List[str]]
+) -> Callable[[str, str], str]:
+    """Factory to test connecting a DASF module"""
+
+    def test_connect(topic: str, path_or_mod: str) -> str:
+        command = get_module_command(topic, path_or_mod)
+        output = spr.check_output(command + ["test-connect"])
+        return output.decode("utf-8")
+
+    return test_connect
+
+
+@pytest.fixture
+def test_dasf_request(
+    get_module_command: Callable[[str, str], List[str]], tmpdir
+) -> Callable[[str, str, Union[Dict, str]], str]:
+    """A factory to send requests via DASF.
+
+    Notes
+    -----
+    Sending a request requires a connected backend module! See the fixture
+    :func:`connect_module`.
+    """
+
+    def test_request(
+        topic: str, path_or_mod: str, request: Union[Dict, str]
+    ) -> str:
+        """Send a DASF request to a backend module."""
+        command = get_module_command(topic, path_or_mod)
+        request_path: str
+        if not isinstance(request, str):
+            request_path = str(tmpdir.mkdir(topic).join("request.json"))
+            with open(request_path, "w") as f:
+                json.dump(request, f)
+        else:
+            request_path = request
+
+        output = spr.check_output(command + ["send-request", request_path])
+        return output.decode("utf-8")
+
+    return test_request

dasf_broker/views.py

@@ -38,7 +38,6 @@ from dasf_broker import models  # noqa: F401
 
 
 class HttpResponseServiceUnavailable(HttpResponse):
-
     status_code = 503
 
 

setup.cfg

@@ -50,6 +50,9 @@ install_requires =
     django-guardian
     channels
 
+[options.entry_points]
+pytest11 =
+    dasf = dasf_broker.tests.plugin
 
 [options.package_data]
 * =
@@ -63,23 +66,26 @@ exclude =
     testproject
 
 [options.extras_require]
+pytest =
+    pytest-django
+
 testsite =
+    %(pytest)s
     tox
     requests
     types-requests
-    isort==5.9.3
-    black==22.3.0
-    blackdoc==0.3.4
+    isort==5.12.0
+    black==23.1.0
+    blackdoc==0.3.8
     flake8==6.0.0
     daphne
+    demessaging
     pre-commit
     mypy
     django-stubs
-    pytest-django
     pytest-cov
     djangorestframework
 
-
 dev =
     %(testsite)s
     Sphinx

testproject/settings.py

@@ -124,6 +124,9 @@ DATABASES = {
     "default": {
         "ENGINE": "django.db.backends.sqlite3",
         "NAME": BASE_DIR / "db.sqlite3",
+        "TEST": {
+            "NAME": BASE_DIR / "db_test.sqlite3",
+        },
     }
 }
 
@@ -142,7 +145,6 @@ if os.getenv("RUNNING_TESTS"):
     }
 
 else:
-
     CACHES = {
         "default": {
             "BACKEND": "django_redis.cache.RedisCache",

tests/test_module_connection.py

+"""Tests to connect a backend module."""
+
+import subprocess as spr
+from typing import Callable, Dict, Union
+
+
+def test_hello_world_connect(
+    test_dasf_connect: Callable[[str, str], str],
+    random_topic: str,
+    get_test_module_path: Callable[[str], str],
+):
+    """Test connecting the hello world backend module."""
+    modpath = get_test_module_path("hello_world")
+    test_dasf_connect(random_topic, modpath)
+
+
+def test_dasf_request(
+    connect_module: Callable[[str, str], spr.Popen],
+    test_dasf_request: Callable[[str, str, Union[Dict, str]], str],
+    random_topic: str,
+    get_test_module_path: Callable[[str], str],
+):
+    """Test sending a request to the backend module."""
+
+    modpath = get_test_module_path("hello_world")
+    connect_module(random_topic, modpath)
+    output = test_dasf_request(
+        random_topic, modpath, {"func_name": "hello_world"}
+    )
+
+    assert "Hello World!" in output

tests/test_modules/hello_world.py

+"""A simple backend module for DASF."""
+
+from demessaging import main
+
+__all__ = ["hello_world"]
+
+
+def hello_world() -> str:
+    """Simple function to be called via DASF"""
+    return "Hello World!"
+
+
+if __name__ == "__main__":
+    main()