refactor lifespan tests

Author: lesnik512

lite_bootstrap/bootstrappers/base.py

@@ -10,11 +10,13 @@
 
 
 class BaseBootstrapper(abc.ABC, typing.Generic[ApplicationT]):
+    SLOTS = "bootstrap_config", "instruments", "is_bootstrapped"
     instruments_types: typing.ClassVar[list[type[BaseInstrument]]]
     instruments: list[BaseInstrument]
     bootstrap_config: BaseConfig
 
     def __init__(self, bootstrap_config: BaseConfig) -> None:
+        self.is_bootstrapped = False
         if not self.is_ready():
             raise RuntimeError(self.not_ready_message)
 
@@ -38,10 +40,12 @@ def _prepare_application(self) -> ApplicationT: ...
     def is_ready(self) -> bool: ...
 
     def bootstrap(self) -> ApplicationT:
+        self.is_bootstrapped = True
         for one_instrument in self.instruments:
             one_instrument.bootstrap()
         return self._prepare_application()
 
     def teardown(self) -> None:
+        self.is_bootstrapped = False
         for one_instrument in self.instruments:
             one_instrument.teardown()

tests/test_fastapi_bootstrap.py

@@ -1,9 +1,3 @@
-import asyncio
-import contextlib
-import dataclasses
-import typing
-
-import fastapi
 import pytest
 import structlog
 from opentelemetry.sdk.trace.export import ConsoleSpanExporter
@@ -39,11 +33,10 @@ def fastapi_config() -> FastAPIConfig:
 def test_fastapi_bootstrap(fastapi_config: FastAPIConfig) -> None:
     bootstrapper = FastAPIBootstrapper(bootstrap_config=fastapi_config)
     application = bootstrapper.bootstrap()
-    test_client = TestClient(application)
-
+    assert bootstrapper.is_bootstrapped
     logger.info("testing logging", key="value")
 
-    try:
+    with TestClient(application) as test_client:
         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"}
@@ -59,8 +52,8 @@ def test_fastapi_bootstrap(fastapi_config: FastAPIConfig) -> None:
         response = test_client.get(str(application.redoc_url))
         assert response.status_code == status.HTTP_200_OK
         assert response.text
-    finally:
-        bootstrapper.teardown()
+
+    assert not bootstrapper.is_bootstrapped
 
 
 def test_fastapi_bootstrapper_not_ready() -> None:
@@ -82,21 +75,3 @@ def test_fastapi_bootstrapper_with_missing_instrument_dependency(
 ) -> None:
     with emulate_package_missing(package_name), pytest.warns(UserWarning, match=package_name):
         FastAPIBootstrapper(bootstrap_config=fastapi_config)
-
-
-def test_fastapi_bootstrap_lifespan(fastapi_config: FastAPIConfig) -> None:
-    @contextlib.asynccontextmanager
-    async def lifespan_manager(_: fastapi.FastAPI) -> typing.AsyncIterator[dict[str, typing.Any]]:
-        try:
-            yield {}
-        finally:
-            await asyncio.sleep(0)
-
-    fastapi_config = dataclasses.replace(fastapi_config, application=fastapi.FastAPI(lifespan=lifespan_manager))
-    bootstrapper = FastAPIBootstrapper(bootstrap_config=fastapi_config)
-    application = bootstrapper.bootstrap()
-
-    with TestClient(application) as test_client:
-        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"}

tests/test_faststream_bootstrap.py

@@ -45,17 +45,23 @@ 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)
-
+    assert bootstrapper.is_bootstrapped
     logger.info("testing logging", key="value")
 
-    async with TestRedisBroker(broker):
-        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"}
+    with TestClient(app=application) as test_client:
+        async with TestRedisBroker(broker):
+            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(bootstrap_config.prometheus_metrics_path)
+            assert response.status_code == status.HTTP_200_OK
 
-        response = test_client.get(bootstrap_config.prometheus_metrics_path)
-        assert response.status_code == status.HTTP_200_OK
+    assert not bootstrapper.is_bootstrapped
 
 
 async def test_faststream_bootstrap_health_check_wo_broker() -> None:

tests/test_litestar_bootstrap.py

@@ -33,32 +33,30 @@ def litestar_config() -> LitestarConfig:
 def test_litestar_bootstrap(litestar_config: LitestarConfig) -> None:
     bootstrapper = LitestarBootstrapper(bootstrap_config=litestar_config)
     application = bootstrapper.bootstrap()
-
-    try:
-        logger.info("testing logging", key="value")
-
-        assert application.cors_config
-        assert application.cors_config.allow_origins == litestar_config.cors_allowed_origins
-
-        with TestClient(app=application) as test_client:
-            response = test_client.get(litestar_config.health_checks_path)
-            assert response.status_code == status_codes.HTTP_200_OK
-            assert response.json() == {
-                "health_status": True,
-                "service_name": "microservice",
-                "service_version": "2.0.0",
-            }
-
-            response = test_client.get(litestar_config.prometheus_metrics_path)
-            assert response.status_code == status_codes.HTTP_200_OK
-            assert response.text
-
-            response = test_client.get(litestar_config.swagger_path)
-            assert response.status_code == status_codes.HTTP_200_OK
-            response = test_client.get(f"{litestar_config.service_static_path}/swagger-ui.css")
-            assert response.status_code == status_codes.HTTP_200_OK
-    finally:
-        bootstrapper.teardown()
+    assert bootstrapper.is_bootstrapped
+    logger.info("testing logging", key="value")
+    assert application.cors_config
+    assert application.cors_config.allow_origins == litestar_config.cors_allowed_origins
+
+    with TestClient(app=application) as test_client:
+        response = test_client.get(litestar_config.health_checks_path)
+        assert response.status_code == status_codes.HTTP_200_OK
+        assert response.json() == {
+            "health_status": True,
+            "service_name": "microservice",
+            "service_version": "2.0.0",
+        }
+
+        response = test_client.get(litestar_config.prometheus_metrics_path)
+        assert response.status_code == status_codes.HTTP_200_OK
+        assert response.text
+
+        response = test_client.get(litestar_config.swagger_path)
+        assert response.status_code == status_codes.HTTP_200_OK
+        response = test_client.get(f"{litestar_config.service_static_path}/swagger-ui.css")
+        assert response.status_code == status_codes.HTTP_200_OK
+
+    assert not bootstrapper.is_bootstrapped
 
 
 def test_litestar_bootstrapper_not_ready() -> None: