testing extras

lite_bootstrap/bootstrappers/fastapi_bootstrapper.py

@@ -123,6 +123,12 @@ class FastAPIBootstrapper(BaseBootstrapper["fastapi.FastAPI"]):
     bootstrap_config: FastAPIConfig
     not_ready_message = "fastapi is not installed"
 
+    def __init__(self, bootstrap_config: FastAPIConfig) -> None:
+        super().__init__(bootstrap_config)
+        self.bootstrap_config.application.title = bootstrap_config.service_name
+        self.bootstrap_config.application.debug = bootstrap_config.service_debug
+        self.bootstrap_config.application.version = bootstrap_config.service_version
+
     def is_ready(self) -> bool:
         return import_checker.is_fastapi_installed
 

lite_bootstrap/bootstrappers/litestar_bootstrapper.py

@@ -128,8 +128,5 @@ class LitestarBootstrapper(BaseBootstrapper["litestar.Litestar"]):
     def is_ready(self) -> bool:
         return import_checker.is_litestar_installed
 
-    def __init__(self, bootstrap_config: LitestarConfig) -> None:
-        super().__init__(bootstrap_config)
-
     def _prepare_application(self) -> "litestar.Litestar":
         return litestar.Litestar.from_config(self.bootstrap_config.application_config)

tests/conftest.py

@@ -1,9 +1,14 @@
+import contextlib
+import sys
 import typing
+from importlib import reload
 from unittest.mock import Mock
 
 import pytest
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor  # type: ignore[attr-defined]
 
+from lite_bootstrap import import_checker
+
 
 class CustomInstrumentor(BaseInstrumentor):  # type: ignore[misc]
     def instrumentation_dependencies(self) -> typing.Collection[str]:
@@ -16,3 +21,15 @@ def _uninstrument(self, **kwargs: typing.Mapping[str, typing.Any]) -> None:
 @pytest.fixture(autouse=True)
 def mock_sentry_init(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr("sentry_sdk.init", Mock)
+
+
+@contextlib.contextmanager
+def emulate_package_missing(package_name: str) -> typing.Iterator[None]:
+    old_module = sys.modules[package_name]
+    sys.modules[package_name] = None  # type: ignore[assignment]
+    reload(import_checker)
+    try:
+        yield
+    finally:
+        sys.modules[package_name] = old_module
+        reload(import_checker)

tests/test_fastapi_bootstrap.py

@@ -4,52 +4,65 @@
 from starlette import status
 from starlette.testclient import TestClient
 
-from lite_bootstrap import FastAPIBootstrapper, FastAPIConfig, import_checker
-from tests.conftest import CustomInstrumentor
+from lite_bootstrap import FastAPIBootstrapper, FastAPIConfig
+from tests.conftest import CustomInstrumentor, emulate_package_missing
 
 
 logger = structlog.getLogger(__name__)
 
 
-def test_fastapi_bootstrap() -> None:
-    health_checks_path = "/custom-health/"
-    prometheus_metrics_path = "/custom-metrics/"
-    bootstrapper = FastAPIBootstrapper(
-        bootstrap_config=FastAPIConfig(
-            service_name="microservice",
-            service_version="2.0.0",
-            service_environment="test",
-            service_debug=False,
-            opentelemetry_endpoint="otl",
-            opentelemetry_instrumentors=[CustomInstrumentor()],
-            opentelemetry_span_exporter=ConsoleSpanExporter(),
-            prometheus_metrics_path=prometheus_metrics_path,
-            sentry_dsn="https://testdsn@localhost/1",
-            health_checks_path=health_checks_path,
-            logging_buffer_capacity=0,
-        ),
+@pytest.fixture
+def fastapi_config() -> FastAPIConfig:
+    return FastAPIConfig(
+        service_name="microservice",
+        service_version="2.0.0",
+        service_environment="test",
+        service_debug=False,
+        opentelemetry_endpoint="otl",
+        opentelemetry_instrumentors=[CustomInstrumentor()],
+        opentelemetry_span_exporter=ConsoleSpanExporter(),
+        prometheus_metrics_path="/custom-metrics/",
+        sentry_dsn="https://testdsn@localhost/1",
+        health_checks_path="/custom-health/",
+        logging_buffer_capacity=0,
     )
+
+
+def test_fastapi_bootstrap(fastapi_config: FastAPIConfig) -> None:
+    bootstrapper = FastAPIBootstrapper(bootstrap_config=fastapi_config)
     application = bootstrapper.bootstrap()
     test_client = TestClient(application)
 
     logger.info("testing logging", key="value")
 
     try:
-        response = test_client.get(health_checks_path)
+        response = test_client.get(fastapi_config.health_checks_path)
         assert response.status_code == status.HTTP_200_OK
         assert response.json() == {"health_status": True, "service_name": "microservice", "service_version": "2.0.0"}
 
-        response = test_client.get(prometheus_metrics_path)
+        response = test_client.get(fastapi_config.prometheus_metrics_path)
         assert response.status_code == status.HTTP_200_OK
         assert response.text
     finally:
         bootstrapper.teardown()
 
 
 def test_fastapi_bootstrapper_not_ready() -> None:
-    import_checker.is_fastapi_installed = False
-    try:
-        with pytest.raises(RuntimeError, match="fastapi is not installed"):
-            FastAPIBootstrapper(bootstrap_config=FastAPIConfig())
-    finally:
-        import_checker.is_fastapi_installed = True
+    with emulate_package_missing("fastapi"), pytest.raises(RuntimeError, match="fastapi is not installed"):
+        FastAPIBootstrapper(bootstrap_config=FastAPIConfig())
+
+
+@pytest.mark.parametrize(
+    "package_name",
+    [
+        "opentelemetry",
+        "sentry_sdk",
+        "structlog",
+        "prometheus_fastapi_instrumentator",
+    ],
+)
+def test_fastapi_bootstrapper_with_missing_instrument_dependency(
+    fastapi_config: FastAPIConfig, package_name: str
+) -> None:
+    with emulate_package_missing(package_name), pytest.warns(UserWarning, match=package_name):
+        FastAPIBootstrapper(bootstrap_config=fastapi_config)

tests/test_faststream_bootstrap.py

@@ -1,14 +1,17 @@
+import typing
+
 import pytest
 import structlog
+from faststream.broker.core.usecase import BrokerUsecase
 from faststream.redis import RedisBroker, TestRedisBroker
 from faststream.redis.opentelemetry import RedisTelemetryMiddleware
 from faststream.redis.prometheus import RedisPrometheusMiddleware
 from opentelemetry.sdk.trace.export import ConsoleSpanExporter
 from starlette import status
 from starlette.testclient import TestClient
 
-from lite_bootstrap import FastStreamBootstrapper, FastStreamConfig, import_checker
-from tests.conftest import CustomInstrumentor
+from lite_bootstrap import FastStreamBootstrapper, FastStreamConfig
+from tests.conftest import CustomInstrumentor, emulate_package_missing
 
 
 logger = structlog.getLogger(__name__)
@@ -19,73 +22,68 @@ def broker() -> RedisBroker:
     return RedisBroker()
 
 
-async def test_faststream_bootstrap(broker: RedisBroker) -> None:
-    prometheus_metrics_path = "/custom-metrics/"
-    health_check_path = "/custom-health/"
-    bootstrapper = FastStreamBootstrapper(
-        bootstrap_config=FastStreamConfig(
-            broker=broker,
-            service_name="microservice",
-            service_version="2.0.0",
-            service_environment="test",
-            service_debug=False,
-            opentelemetry_endpoint="otl",
-            opentelemetry_instrumentors=[CustomInstrumentor()],
-            opentelemetry_span_exporter=ConsoleSpanExporter(),
-            opentelemetry_middleware_cls=RedisTelemetryMiddleware,
-            prometheus_metrics_path=prometheus_metrics_path,
-            prometheus_middleware_cls=RedisPrometheusMiddleware,
-            sentry_dsn="https://testdsn@localhost/1",
-            health_checks_path=health_check_path,
-            logging_buffer_capacity=0,
-        ),
+def build_faststream_config(broker: BrokerUsecase[typing.Any, typing.Any] | None = None) -> FastStreamConfig:
+    return FastStreamConfig(
+        service_name="microservice",
+        service_version="2.0.0",
+        service_environment="test",
+        service_debug=False,
+        opentelemetry_endpoint="otl",
+        opentelemetry_instrumentors=[CustomInstrumentor()],
+        opentelemetry_span_exporter=ConsoleSpanExporter(),
+        opentelemetry_middleware_cls=RedisTelemetryMiddleware,
+        prometheus_metrics_path="/custom-metrics/",
+        prometheus_middleware_cls=RedisPrometheusMiddleware,
+        sentry_dsn="https://testdsn@localhost/1",
+        health_checks_path="/custom-health/",
+        logging_buffer_capacity=0,
+        broker=broker,
     )
+
+
+async def test_faststream_bootstrap(broker: RedisBroker) -> None:
+    bootstrap_config = build_faststream_config(broker=broker)
+    bootstrapper = FastStreamBootstrapper(bootstrap_config=bootstrap_config)
     application = bootstrapper.bootstrap()
     test_client = TestClient(app=application)
 
     logger.info("testing logging", key="value")
 
     async with TestRedisBroker(broker):
-        response = test_client.get(health_check_path)
+        response = test_client.get(bootstrap_config.health_checks_path)
         assert response.status_code == status.HTTP_200_OK
         assert response.json() == {"health_status": True, "service_name": "microservice", "service_version": "2.0.0"}
 
-        response = test_client.get(prometheus_metrics_path)
+        response = test_client.get(bootstrap_config.prometheus_metrics_path)
         assert response.status_code == status.HTTP_200_OK
 
 
 async def test_faststream_bootstrap_health_check_wo_broker() -> None:
-    health_check_path = "/custom-health-check-path"
-    bootstrapper = FastStreamBootstrapper(
-        bootstrap_config=FastStreamConfig(
-            service_name="microservice",
-            service_version="2.0.0",
-            service_environment="test",
-            service_debug=False,
-            opentelemetry_endpoint="otl",
-            opentelemetry_instrumentors=[CustomInstrumentor()],
-            opentelemetry_span_exporter=ConsoleSpanExporter(),
-            opentelemetry_middleware_cls=RedisTelemetryMiddleware,
-            prometheus_middleware_cls=RedisPrometheusMiddleware,
-            sentry_dsn="https://testdsn@localhost/1",
-            health_checks_path=health_check_path,
-            logging_buffer_capacity=0,
-        ),
-    )
+    bootstrap_config = build_faststream_config()
+    bootstrapper = FastStreamBootstrapper(bootstrap_config=bootstrap_config)
     application = bootstrapper.bootstrap()
     test_client = TestClient(app=application)
 
-    response = test_client.get(health_check_path)
+    response = test_client.get(bootstrap_config.health_checks_path)
     assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
     assert response.text == "Service is unhealthy"
 
 
 def test_faststream_bootstrapper_not_ready() -> None:
-    import_checker.is_faststream_installed = False
-    try:
-        with pytest.raises(RuntimeError, match="faststream is not installed"):
-            FastStreamBootstrapper(
-                bootstrap_config=FastStreamConfig(),
-            )
-    finally:
-        import_checker.is_faststream_installed = True
+    with emulate_package_missing("faststream"), pytest.raises(RuntimeError, match="faststream is not installed"):
+        FastStreamBootstrapper(bootstrap_config=FastStreamConfig())
+
+
+@pytest.mark.parametrize(
+    "package_name",
+    [
+        "opentelemetry",
+        "sentry_sdk",
+        "structlog",
+        "prometheus_client",
+    ],
+)
+def test_faststream_bootstrapper_with_missing_instrument_dependency(broker: RedisBroker, package_name: str) -> None:
+    bootstrap_config = build_faststream_config(broker=broker)
+    with emulate_package_missing(package_name), pytest.warns(UserWarning, match=package_name):
+        FastStreamBootstrapper(bootstrap_config=bootstrap_config)

tests/test_free_bootstrap.py

@@ -3,23 +3,26 @@
 from opentelemetry.sdk.trace.export import ConsoleSpanExporter
 
 from lite_bootstrap import FreeBootstrapper, FreeBootstrapperConfig
-from tests.conftest import CustomInstrumentor
+from tests.conftest import CustomInstrumentor, emulate_package_missing
 
 
 logger = structlog.getLogger(__name__)
 
 
-def test_free_bootstrap() -> None:
-    bootstrapper = FreeBootstrapper(
-        bootstrap_config=FreeBootstrapperConfig(
-            service_debug=False,
-            opentelemetry_endpoint="otl",
-            opentelemetry_instrumentors=[CustomInstrumentor()],
-            opentelemetry_span_exporter=ConsoleSpanExporter(),
-            sentry_dsn="https://testdsn@localhost/1",
-            logging_buffer_capacity=0,
-        ),
+@pytest.fixture
+def free_bootstrapper_config() -> FreeBootstrapperConfig:
+    return FreeBootstrapperConfig(
+        service_debug=False,
+        opentelemetry_endpoint="otl",
+        opentelemetry_instrumentors=[CustomInstrumentor()],
+        opentelemetry_span_exporter=ConsoleSpanExporter(),
+        sentry_dsn="https://testdsn@localhost/1",
+        logging_buffer_capacity=0,
     )
+
+
+def test_free_bootstrap(free_bootstrapper_config: FreeBootstrapperConfig) -> None:
+    bootstrapper = FreeBootstrapper(bootstrap_config=free_bootstrapper_config)
     bootstrapper.bootstrap()
     try:
         logger.info("testing logging", key="value")
@@ -36,3 +39,18 @@ def test_free_bootstrap_logging_not_ready() -> None:
                 sentry_dsn="https://testdsn@localhost/1",
             ),
         )
+
+
+@pytest.mark.parametrize(
+    "package_name",
+    [
+        "opentelemetry",
+        "sentry_sdk",
+        "structlog",
+    ],
+)
+def test_free_bootstrapper_with_missing_instrument_dependency(
+    free_bootstrapper_config: FreeBootstrapperConfig, package_name: str
+) -> None:
+    with emulate_package_missing(package_name), pytest.warns(UserWarning, match=package_name):
+        FreeBootstrapper(bootstrap_config=free_bootstrapper_config)