Specific fixture scoping (GSI-270) (#59)

* Enable scope selection on test fixtures

* Update template files

* Bump version from 0.10.0 -> 0.10.1

* Move fixture creation completely to other module

* Rename fixture_factory.py to fixtures.py

* Implement basic Mongo Reset Function

* Implement S3 fixture reset function

* Update imports in test files

* Add empty_collections to Mongo fixture

* Replace s3 reset func with empty_buckets

* Add delete_topics() to kafka fixture

Fix type issue with Path parameter in temp_file_object

* Use set.update to get new buckets

* Change delete_topics so no topics is all topics

* Make fixtures.py source of fixture-related imports

* Remove duplicate code in mongo fixture

* Change function sig for S3 reset function

* Add comments to test

* Change S3 test parm to a list instead of a set

* Add a test for the mongo reset function

---------

Co-authored-by: TheByronHimes <TheByronHimes@gmail.com>

.devcontainer/dev_install

@@ -10,7 +10,7 @@ python -m pip install --upgrade pip
 pip install -e .[all]
 
 # install or upgrade dependencies for development and testing
-pip install --upgrade -r requirements-dev.txt
+pip install -r requirements-dev.txt
 
 # install pre-commit hooks to git
 pre-commit install

.static_files

@@ -15,6 +15,7 @@
 scripts/script_utils/__init__.py
 scripts/script_utils/cli.py
 
+scripts/__init__.py
 scripts/license_checker.py
 scripts/get_package_name.py
 scripts/update_config_docs.py

hexkit/__init__.py

@@ -15,4 +15,4 @@
 
 """A Toolkit for Building Microservices using the Hexagonal Architecture"""
 
-__version__ = "0.10.0"
+__version__ = "0.10.1"

hexkit/providers/akafka/testutils.py

@@ -23,10 +23,11 @@
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from functools import partial
-from typing import AsyncGenerator, Optional, Sequence
+from typing import AsyncGenerator, Optional, Sequence, Union
 
-import pytest_asyncio
 from aiokafka import AIOKafkaConsumer, TopicPartition
+from kafka import KafkaAdminClient
+from kafka.errors import KafkaError
 from testcontainers.kafka import KafkaContainer
 
 from hexkit.custom_types import Ascii, JsonObject
@@ -364,6 +365,31 @@ def record_events(self, *, in_topic: Ascii) -> EventRecorder:
 
         return EventRecorder(kafka_servers=self.kafka_servers, topic=in_topic)
 
+    def delete_topics(self, topics: Optional[Union[str, list[str]]] = None):
+        """
+        Delete given topic(s) from Kafka broker. When no topics are specified,
+        all existing topics will be deleted.
+        """
+
+        admin_client = KafkaAdminClient(bootstrap_servers=self.kafka_servers)
+        all_topics = admin_client.list_topics()
+        if topics is None:
+            topics = all_topics
+        elif isinstance(topics, str):
+            topics = [topics]
+        try:
+            existing_topics = set(all_topics)
+            for topic in topics:
+                if topic in existing_topics:
+                    try:
+                        admin_client.delete_topics([topic])
+                    except KafkaError as error:
+                        raise RuntimeError(
+                            f"Could not delete topic {topic} from Kafka"
+                        ) from error
+        finally:
+            admin_client.close()
+
     @asynccontextmanager
     async def expect_events(
         self, events: Sequence[ExpectedEvent], *, in_topic: Ascii
@@ -381,8 +407,7 @@ async def expect_events(
         )
 
 
-@pytest_asyncio.fixture
-async def kafka_fixture() -> AsyncGenerator[KafkaFixture, None]:
+async def kafka_fixture_function() -> AsyncGenerator[KafkaFixture, None]:
     """Pytest fixture for tests depending on the Kafka-base providers."""
 
     with KafkaContainer(image="confluentinc/cp-kafka:5.4.9-1-deb8") as kafka:

hexkit/providers/mongodb/testutils.py

@@ -21,9 +21,10 @@
 
 
 from dataclasses import dataclass
-from typing import Generator
+from typing import Generator, Optional, Union
 
-import pytest_asyncio
+from pymongo import MongoClient
+from pymongo.errors import ExecutionTimeout, OperationFailure
 from testcontainers.mongodb import MongoDbContainer
 
 from hexkit.providers.mongodb.provider import MongoDbConfig, MongoDbDaoFactory
@@ -33,9 +34,35 @@
 class MongoDbFixture:
     """Yielded by the `mongodb_fixture` function"""
 
+    client: MongoClient
     config: MongoDbConfig
     dao_factory: MongoDbDaoFactory
 
+    def empty_collections(
+        self,
+        exclude_collections: Optional[Union[str, list[str]]] = None,
+    ):
+        """Drop all mongodb collections in the database.
+
+        You can also specify collection(s) that should be excluded
+        from the operation, i.e. collections that should be kept.
+        """
+        db_name = self.config.db_name
+        if exclude_collections is None:
+            exclude_collections = []
+        if isinstance(exclude_collections, str):
+            exclude_collections = [exclude_collections]
+        excluded_collections = set(exclude_collections)
+        try:
+            collection_names = self.client[db_name].list_collection_names()
+            for collection_name in collection_names:
+                if collection_name not in excluded_collections:
+                    self.client[db_name].drop_collection(collection_name)
+        except (ExecutionTimeout, OperationFailure) as error:
+            raise RuntimeError(
+                f"Could not drop collection(s) of Mongo database {db_name}"
+            ) from error
+
 
 def config_from_mongodb_container(container: MongoDbContainer) -> MongoDbConfig:
     """Prepares a MongoDbConfig from an instance of a MongoDbContainer container."""
@@ -44,12 +71,21 @@ def config_from_mongodb_container(container: MongoDbContainer) -> MongoDbConfig:
     return MongoDbConfig(db_connection_str=db_connection_str, db_name="test")
 
 
-@pytest_asyncio.fixture
-def mongodb_fixture() -> Generator[MongoDbFixture, None, None]:
-    """Pytest fixture for tests depending on the MongoDbDaoFactory DAO."""
+def mongodb_fixture_function() -> Generator[MongoDbFixture, None, None]:
+    """
+    Pytest fixture for tests depending on the MongoDbDaoFactory DAO.
+    Obtained via get_fixture in hexkit.providers.testing.fixtures.get_fixture
+    """
 
     with MongoDbContainer(image="mongo:6.0.3") as mongodb:
         config = config_from_mongodb_container(mongodb)
         dao_factory = MongoDbDaoFactory(config=config)
+        client = mongodb.get_connection_client()
+
+        yield MongoDbFixture(
+            client=client,
+            config=config,
+            dao_factory=dao_factory,
+        )
 
-        yield MongoDbFixture(config=config, dao_factory=dao_factory)
+        client.close()

hexkit/providers/s3/testutils.py

@@ -24,10 +24,9 @@
 from contextlib import contextmanager
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import Generator, List
+from typing import Generator, List, Optional
 
 import pytest
-import pytest_asyncio
 import requests
 from pydantic import BaseModel, validator
 from testcontainers.localstack import LocalStackContainer
@@ -87,6 +86,23 @@ def __init__(self, config: S3Config, storage: S3ObjectStorage):
         """Initialize with config."""
         self.config = config
         self.storage = storage
+        self._buckets: set[str] = set()
+
+    async def empty_buckets(self, buckets_to_exclude: Optional[list[str]] = None):
+        """Clean the test artifacts or files from given bucket"""
+        if buckets_to_exclude is None:
+            buckets_to_exclude = []
+
+        for bucket in self._buckets:
+            if bucket in buckets_to_exclude:
+                continue
+
+            # Get list of all objects in the bucket
+            object_ids = await self.storage.list_all_object_ids(bucket_id=bucket)
+
+            # Delete all objects
+            for object_id in object_ids:
+                await self.storage.delete_object(bucket_id=bucket, object_id=object_id)
 
     async def populate_buckets(self, buckets: list[str]):
         """Populate the storage with buckets."""
@@ -95,6 +111,8 @@ async def populate_buckets(self, buckets: list[str]):
             self.storage, bucket_fixtures=buckets, object_fixtures=[]
         )
 
+        self._buckets.update(buckets)
+
     async def populate_file_objects(self, file_objects: list[FileObject]):
         """Populate the storage with file objects."""
 
@@ -103,8 +121,7 @@ async def populate_file_objects(self, file_objects: list[FileObject]):
         )
 
 
-@pytest_asyncio.fixture
-def s3_fixture() -> Generator[S3Fixture, None, None]:
+def s3_fixture_function() -> Generator[S3Fixture, None, None]:
     """Pytest fixture for tests depending on the S3ObjectStorage DAO."""
 
     with LocalStackContainer(image="localstack/localstack:0.14.5").with_services(
@@ -137,7 +154,7 @@ def temp_file_object(
         temp_file.flush()
 
         yield FileObject(
-            file_path=temp_file.name, bucket_id=bucket_id, object_id=object_id
+            file_path=Path(temp_file.name), bucket_id=bucket_id, object_id=object_id
         )
 
 

hexkit/providers/testing/fixtures.py

@@ -0,0 +1,119 @@
+# Copyright 2021 - 2023 Universität Tübingen, DKFZ, EMBL, and Universität zu Köln
+# for the German Human Genome-Phenome Archive (GHGA)
+#
+# 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.
+#
+"""Enable requesting test fixtures with chosen scope"""
+
+from typing import Callable, Type, Union
+
+import pytest_asyncio
+from pytest_asyncio.plugin import _ScopeName
+
+from hexkit.providers.akafka.testutils import (
+    EventBase,
+    EventRecorder,
+    ExpectedEvent,
+    KafkaFixture,
+    RecordedEvent,
+    ValidationError,
+    kafka_fixture_function,
+)
+from hexkit.providers.mongodb.testutils import (
+    MongoDbFixture,
+    config_from_mongodb_container,
+    mongodb_fixture_function,
+)
+from hexkit.providers.s3.testutils import (
+    MEBIBYTE,
+    TEST_FILE_DIR,
+    TEST_FILE_PATHS,
+    TIMEOUT,
+    FileObject,
+    S3Config,
+    S3Fixture,
+    S3ObjectStorage,
+    calc_md5,
+    check_part_size,
+    config_from_localstack_container,
+    download_and_check_test_file,
+    file_fixture,
+    get_initialized_upload,
+    multipart_upload_file,
+    populate_storage,
+    prepare_non_completed_upload,
+    s3_fixture_function,
+    temp_file_object,
+    typical_workflow,
+    upload_file,
+    upload_part,
+    upload_part_of_size,
+    upload_part_via_url,
+)
+
+__all__ = [
+    "EventBase",
+    "EventRecorder",
+    "ExpectedEvent",
+    "KafkaFixture",
+    "RecordedEvent",
+    "ValidationError",
+    "kafka_fixture_function",
+    "MongoDbFixture",
+    "config_from_mongodb_container",
+    "mongodb_fixture_function",
+    "MEBIBYTE",
+    "TEST_FILE_DIR",
+    "TEST_FILE_PATHS",
+    "TIMEOUT",
+    "FileObject",
+    "S3Config",
+    "S3Fixture",
+    "S3ObjectStorage",
+    "calc_md5",
+    "check_part_size",
+    "config_from_localstack_container",
+    "download_and_check_test_file",
+    "file_fixture",
+    "get_initialized_upload",
+    "multipart_upload_file",
+    "populate_storage",
+    "prepare_non_completed_upload",
+    "s3_fixture_function",
+    "temp_file_object",
+    "typical_workflow",
+    "upload_file",
+    "upload_part",
+    "upload_part_of_size",
+    "upload_part_via_url",
+]
+
+
+ProviderFixture = Union[Type[KafkaFixture], Type[MongoDbFixture], Type[S3Fixture]]
+
+fixture_type_to_function: dict[ProviderFixture, Callable] = {
+    KafkaFixture: kafka_fixture_function,
+    MongoDbFixture: mongodb_fixture_function,
+    S3Fixture: s3_fixture_function,
+}
+
+
+def get_fixture(fixture_type: ProviderFixture, scope: _ScopeName = "function"):
+    """Produce a test fixture with the desired scope"""
+    fixture_function = fixture_type_to_function[fixture_type]
+    return pytest_asyncio.fixture(scope=scope)(fixture_function)
+
+
+mongodb_fixture = get_fixture(MongoDbFixture)
+kafka_fixture = get_fixture(KafkaFixture)
+s3_fixture = get_fixture(S3Fixture)

scripts/__init__.py

@@ -0,0 +1,17 @@
+# Copyright 2021 - 2023 Universität Tübingen, DKFZ, EMBL, and Universität zu Köln
+# for the German Human Genome-Phenome Archive (GHGA)
+#
+# 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.
+#
+
+"""Scripts and utils used during development or in CI pipelines."""

scripts/update_template_files.py

@@ -33,7 +33,7 @@
 try:
     from script_utils.cli import echo_failure, echo_success, run
 except ImportError:
-    echo_failure = echo_success = print  # type: ignore
+    echo_failure = echo_success = print
 
     def run(main_fn):
         """Run main function without cli tools (typer)."""

tests/integration/test_akafka.py

@@ -26,7 +26,7 @@
     KafkaEventPublisher,
     KafkaEventSubscriber,
 )
-from hexkit.providers.akafka.testutils import (  # noqa: F401
+from hexkit.providers.testing.fixtures import (  # noqa: F401
     ExpectedEvent,
     KafkaFixture,
     kafka_fixture,