I have this code in my main.py file of my FastAPI app.

@asynccontextmanager
async def lifespan(_application: FastAPI) -> AsyncGenerator:
    # Startup
    await redis_manager.connect()
    await database.connect()

    yield

    # Shutdown
    await database.disconnect()
    await redis_manager.disconnect()
    pulsar_manager.unsubscribe()

And because of the line pulsar_manager.unsubscribe() when I run pytest in Docker the tests are running twice each. I guess it is because this method is synchronous while lifespan is async. Could you please help me to understand: is lifespan shutdown called after evaluation of every router? Or is it shutdown when the whole app is shutdown? I need to unsubscribe when the whole appication is shutdown or restarted. ALso maybe you could help me rewrite my Pulsar Manager to work async? Here is the code for my PulsarManager:

import enum
import uuid
from datetime import datetime
from typing import Literal, Optional, Union
from uuid import UUID

from pydantic import BaseModel, Field, HttpUrl

import pulsar
from src.config import settings

__all__ = [
    "PulsarManager",
    "pulsar_manager",
    "PulsarSourceType",
    "PulsarActionType",
    "PulsarTwitterEvent",
    "PulsarDiscordEvent",
    "PulsarUserEvent",
    "PulsarUser",
    "PulsarDiscord",
    "PulsarTwitter",
    "PulsarAuth",
    "PulsarAuthEvent",
]


class PulsarUser(BaseModel):
    id: int
    uuid: UUID
    name: Optional[str]
    email: Optional[str]
    is_admin: bool
    apple_id: Optional[str]
    google_id: Optional[str]
    wallet_address: Optional[str]
    created_at: datetime
    updated_at: Optional[datetime]
    last_active_at: Optional[datetime]


class PulsarDiscord(BaseModel):
    uuid: UUID
    user_uuid: UUID
    id: str
    user_name: str


class PulsarTwitter(BaseModel):
    uuid: UUID
    user_uuid: UUID
    id: str
    name: Optional[str]
    user_name: str
    profile_image_url: Optional[HttpUrl]


class PulsarAuth(BaseModel):
    code: str
    state: str


@enum.unique
class PulsarSourceType(str, enum.Enum):
    TWITTER = "twitter"
    DISCORD = "discord"
    USER = "user"
    AUTH = "auth"


@enum.unique
class PulsarActionType(str, enum.Enum):
    CREATE = "create"
    DELETE = "delete"
    AUTH = "auth"


class PulsarTwitterEvent(BaseModel):
    source: Literal[PulsarSourceType.TWITTER] = PulsarSourceType.TWITTER
    action: PulsarActionType
    data: PulsarTwitter


class PulsarDiscordEvent(BaseModel):
    source: Literal[PulsarSourceType.DISCORD] = PulsarSourceType.DISCORD
    action: PulsarActionType
    data: PulsarDiscord


class PulsarUserEvent(BaseModel):
    source: Literal[PulsarSourceType.USER] = PulsarSourceType.USER
    action: PulsarActionType
    data: PulsarUser


class PulsarAuthEvent(BaseModel):
    source: Literal[PulsarSourceType.AUTH] = PulsarSourceType.AUTH
    action: PulsarActionType
    data: PulsarAuth


class PulsarEvent(BaseModel):
    payload: Union[
        PulsarTwitterEvent, PulsarDiscordEvent, PulsarUserEvent, PulsarAuthEvent
    ] = Field(discriminator="source")


class PulsarManager:
    def __init__(self) -> None:
        self._event_topic = f"persistent://{settings.PULSAR_TENANT_NAME}/{settings.PULSAR_NAMESPACE}/{settings.PULSAR_TOPIC_NAME}"
        self._auth_topic = f"persistent://{settings.PULSAR_TENANT_NAME}/{settings.PULSAR_NAMESPACE}/{settings.PULSAR_AUTH_TOPIC_NAME}"
        self._client = pulsar.Client(
            settings.PULSAR_SERVICE_URL,
            authentication=pulsar.AuthenticationToken(settings.PULSAR_TOKEN),
        )
        event_producer = self._client.create_producer(self._event_topic)
        auth_producer = self._client.create_producer(self._auth_topic)

        self._subscription_name = f"auth-subscription-{uuid.uuid4()}"

        self._source_to_producer_map = {
            PulsarSourceType.AUTH: auth_producer,
            PulsarSourceType.USER: event_producer,
            PulsarSourceType.DISCORD: event_producer,
            PulsarSourceType.TWITTER: event_producer,
        }

        self._auth_consumer = None

    def _get_producer(self, source_type: PulsarSourceType):
        return self._source_to_producer_map[source_type]

    def set_consumer_with_callback(self, listener):
        if self._auth_consumer is None:
            self._auth_consumer = self._client.subscribe(
                self._auth_topic,
                self._subscription_name,
                pulsar.ConsumerType.Exclusive,
                message_listener=listener,
            )

    def unsubscribe(self):
        if self._auth_consumer:
            self._auth_consumer.unsubscribe()

    def produce_event(
        self,
        event: Union[
            PulsarTwitterEvent, PulsarDiscordEvent, PulsarUserEvent, PulsarAuthEvent
        ],
    ):
        data = PulsarEvent(payload=event).model_dump_json()
        producer = self._get_producer(event.source)
        producer.send(data.encode("utf-8"))

    def consume_auth_event(self, msg) -> PulsarAuth:
        try:
            event = PulsarEvent.model_validate_json(msg.data().decode("utf-8"))
            self._auth_consumer.acknowledge(msg)
            return event.payload.data
        except Exception:
            self._auth_consumer.negative_acknowledge(msg)


pulsar_manager = PulsarManager()