conftest.py

"""Test main config.

Copyright (c) 2024 MultiFactor
License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE
"""

import asyncio
import os
import uuid
from contextlib import suppress
from dataclasses import dataclass
from itertools import chain
from typing import AsyncGenerator, AsyncIterator, Generator, Iterator
from unittest.mock import AsyncMock, Mock

import aioldap3
import httpx
import pytest
import pytest_asyncio
import redis.asyncio as redis
import uvloop
from alembic import command
from alembic.config import Config as AlembicConfig
from dishka import (
    AsyncContainer,
    Provider,
    Scope,
    from_context,
    make_async_container,
    provide,
)
from dishka.integrations.fastapi import setup_dishka
from fastapi import FastAPI, Request, Response
from loguru import logger
from multidirectory import _create_basic_app
from sqlalchemy import schema, text
from sqlalchemy.ext.asyncio import (
    AsyncConnection,
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
)

from api import shadow_router
from api.audit.adapter import AuditPoliciesAdapter
from api.auth.adapters import (
    AuthFastAPIAdapter,
    MFAFastAPIAdapter,
    SessionFastAPIGateway,
)
from api.auth.utils import get_ip_from_request, get_user_agent_from_request
from api.dhcp.adapter import DHCPAdapter
from api.dns.adapter import DNSFastAPIAdapter
from api.ldap_schema.adapters.attribute_type import AttributeTypeFastAPIAdapter
from api.ldap_schema.adapters.entity_type import LDAPEntityTypeFastAPIAdapter
from api.ldap_schema.adapters.object_class import ObjectClassFastAPIAdapter
from api.main.adapters.kerberos import KerberosFastAPIAdapter
from api.network.adapters.network import NetworkPolicyFastAPIAdapter
from api.password_policy.adapter import (
    PasswordBanWordsFastAPIAdapter,
    PasswordPolicyFastAPIAdapter,
)
from api.shadow.adapter import ShadowAdapter
from authorization_provider_protocol import AuthorizationProviderProtocol
from config import Settings
from constants import ENTITY_TYPE_DTOS_V1, ENTITY_TYPE_DTOS_V2
from enums import AuthorizationRules
from ioc import AuditRedisClient, MFACredsProvider, SessionStorageClient
from ldap_protocol.auth import AuthManager, MFAManager
from ldap_protocol.auth.setup_gateway import SetupGateway
from ldap_protocol.auth.use_cases import SetupUseCase
from ldap_protocol.dhcp import AbstractDHCPManager, StubDHCPManager
from ldap_protocol.dialogue import LDAPSession
from ldap_protocol.dns import (
    AbstractDNSManager,
    DNSSettingsDTO,
    StubDNSManager,
)
from ldap_protocol.dns.dns_gateway import DNSStateGateway
from ldap_protocol.dns.use_cases import DNSUseCase
from ldap_protocol.identity import IdentityProvider
from ldap_protocol.identity.provider_gateway import IdentityProviderGateway
from ldap_protocol.kerberos import AbstractKadmin
from ldap_protocol.kerberos.ldap_structure import KRBLDAPStructureManager
from ldap_protocol.kerberos.service import KerberosService
from ldap_protocol.kerberos.template_render import KRBTemplateRenderer
from ldap_protocol.ldap_requests.bind import BindRequest
from ldap_protocol.ldap_requests.contexts import (
    LDAPAbandonRequestContext,
    LDAPAddRequestContext,
    LDAPBindRequestContext,
    LDAPDeleteRequestContext,
    LDAPExtendedRequestContext,
    LDAPModifyDNRequestContext,
    LDAPModifyRequestContext,
    LDAPSearchRequestContext,
    LDAPUnbindRequestContext,
)
from ldap_protocol.ldap_schema._legacy.attribute_type.attribute_type_dao import (  # noqa: E501
    AttributeTypeDAOLegacy,
)
from ldap_protocol.ldap_schema._legacy.attribute_type.attribute_type_use_case import (  # noqa: E501
    AttributeTypeUseCaseLegacy,
)
from ldap_protocol.ldap_schema._legacy.object_class.object_class_dao import (
    ObjectClassDAOLegacy,
)
from ldap_protocol.ldap_schema._legacy.object_class.object_class_use_case import (  # noqa: E501
    ObjectClassUseCaseLegacy,
)
from ldap_protocol.ldap_schema.attribute_dao import AttributeDAO
from ldap_protocol.ldap_schema.attribute_type.attribute_type_dao import (
    AttributeTypeDAO,
)
from ldap_protocol.ldap_schema.attribute_type.attribute_type_system_flags_use_case import (  # noqa: E501
    AttributeTypeSystemFlagsUseCase,
)
from ldap_protocol.ldap_schema.attribute_type.attribute_type_use_case import (
    AttributeTypeUseCase,
)
from ldap_protocol.ldap_schema.attribute_value_validator import (
    AttributeValueValidator,
)
from ldap_protocol.ldap_schema.directory_dao import DirectoryDAO
from ldap_protocol.ldap_schema.dto import AttributeTypeDTO
from ldap_protocol.ldap_schema.entity_type.entity_type_dao import EntityTypeDAO
from ldap_protocol.ldap_schema.entity_type.entity_type_use_case import (
    EntityTypeUseCase,
)
from ldap_protocol.ldap_schema.object_class.object_class_dao import (
    ObjectClassDAO,
)
from ldap_protocol.ldap_schema.object_class.object_class_use_case import (
    ObjectClassUseCase,
)
from ldap_protocol.ldap_schema.schema_create_use_case import (
    DirectoryCreateUseCase,
)
from ldap_protocol.master_check_use_case import (
    MasterCheckUseCase,
    MasterGatewayProtocol,
)
from ldap_protocol.multifactor import LDAPMultiFactorAPI, MultifactorAPI
from ldap_protocol.permissions_checker import AuthorizationProvider
from ldap_protocol.policies.audit.audit_use_case import AuditUseCase
from ldap_protocol.policies.audit.destination_dao import AuditDestinationDAO
from ldap_protocol.policies.audit.events.managers import (
    NormalizedAuditManager,
    RawAuditManager,
)
from ldap_protocol.policies.audit.monitor import (
    AuditMonitor,
    AuditMonitorUseCase,
)
from ldap_protocol.policies.audit.policies_dao import AuditPoliciesDAO
from ldap_protocol.policies.audit.service import AuditService
from ldap_protocol.policies.network import (
    NetworkPolicyGateway,
    NetworkPolicyUseCase,
    NetworkPolicyValidatorGateway,
    NetworkPolicyValidatorProtocol,
    NetworkPolicyValidatorUseCase,
)
from ldap_protocol.policies.password import (
    PasswordPolicyDAO,
    PasswordPolicyUseCases,
    PasswordPolicyValidator,
)
from ldap_protocol.policies.password.ban_word_repository import (
    PasswordBanWordRepository,
)
from ldap_protocol.policies.password.settings import PasswordValidatorSettings
from ldap_protocol.policies.password.use_cases import (
    PasswordBanWordUseCases,
    UserPasswordHistoryUseCases,
)
from ldap_protocol.roles.access_manager import AccessManager
from ldap_protocol.roles.ace_dao import AccessControlEntryDAO
from ldap_protocol.roles.dataclasses import RoleDTO
from ldap_protocol.roles.migrations_ace_dao import (
    AccessControlEntryAttributeTypeRemapDAO,
)
from ldap_protocol.roles.role_dao import RoleDAO
from ldap_protocol.roles.role_use_case import RoleUseCase
from ldap_protocol.rootdse.gateway import SADomainGateway
from ldap_protocol.rootdse.gw_protocol import DomainReadProtocol
from ldap_protocol.rootdse.reader import DCInfoReader, RootDSEReader
from ldap_protocol.server import PoolClientHandler
from ldap_protocol.session_storage import RedisSessionStorage, SessionStorage
from ldap_protocol.session_storage.repository import SessionRepository
from ldap_protocol.utils.queries import get_user
from password_utils import PasswordUtils
from repo.pg.master_gateway import PGMasterGateway
from tests.constants import (
    TEST_DATA,
    admin_user_data_dict,
    user_data_dict,
    user_with_login_perm_data_dict,
)


class TestProvider(Provider):
    """Test provider."""

    __test__ = False

    scope = Scope.RUNTIME
    settings = from_context(provides=Settings, scope=scope)
    _cached_session: AsyncSession | None = None
    _cached_kadmin: Mock | None = None
    _cached_audit_service: Mock | None = None
    _cached_dns_manager: Mock | None = None
    _cached_dhcp_manager: Mock | None = None
    _session_id: uuid.UUID | None = None

    @provide(scope=Scope.APP, provides=AbstractKadmin)
    async def get_kadmin(self) -> AsyncIterator[AsyncMock]:
        """Get mock kadmin."""
        kadmin = Mock()

        ok_response = Mock()
        ok_response.status_code = 200
        ok_response.aiter_bytes.return_value = map(bytes, zip(b"test_string"))

        kadmin.setup = AsyncMock()
        kadmin.ktadd = AsyncMock(return_value=ok_response)
        kadmin.get_status = AsyncMock(return_value=False)
        kadmin.add_principal = AsyncMock()
        kadmin.del_principal = AsyncMock()
        kadmin.modify_princ = AsyncMock()
        kadmin.rename_princ = AsyncMock()
        kadmin.create_or_update_principal_pw = AsyncMock()
        kadmin.change_principal_password = AsyncMock()
        kadmin.lock_principal = AsyncMock()
        kadmin.reset_setup = AsyncMock()

        if not self._cached_kadmin:
            self._cached_kadmin = kadmin

        yield self._cached_kadmin

        self._cached_kadmin = None

    @provide(scope=Scope.APP, provides=AbstractDHCPManager)
    async def get_dhcp_mngr(self) -> AsyncIterator[AsyncMock]:
        """Get mock DHCP manager."""
        dhcp_manager = AsyncMock(spec=StubDHCPManager)

        if not self._cached_dhcp_manager:
            self._cached_dhcp_manager = dhcp_manager

        yield self._cached_dhcp_manager

        self._cached_dhcp_manager = None

    @provide(scope=Scope.APP, provides=AbstractDNSManager)
    async def get_dns_mngr(self) -> AsyncIterator[AsyncMock]:
        """Get mock DNS manager."""
        dns_manager = AsyncMock(spec=StubDNSManager)

        dns_manager.get_records.return_value = [
            {
                "name": "example.com",
                "type": "A",
                "records": [
                    {
                        "content": "127.0.0.1",
                        "disabled": False,
                        "modified_at": None,
                    },
                ],
                "ttl": 3600,
            },
        ]
        dns_manager.get_forward_zones.return_value = [
            {
                "id": "forward1",
                "name": "forward1.",
                "rrsets": [],
                "kind": "Forwarded",
                "type": "zone",
                "servers": ["127.0.0.1"],
                "recursion_desired": False,
            },
        ]
        dns_manager.get_master_zones.return_value = [
            {
                "id": "zone1",
                "name": "example.com.",
                "rrsets": [
                    {
                        "name": "example.com",
                        "type": "A",
                        "records": [
                            {
                                "content": "127.0.0.1",
                                "disabled": False,
                                "modified_at": None,
                            },
                        ],
                        "ttl": 3600,
                    },
                ],
                "dnssec": False,
                "nameservers": ["ns1.example.com."],
                "kind": "Master",
                "type": "zone",
            },
        ]

        if not self._cached_dns_manager:
            self._cached_dns_manager = dns_manager

        yield self._cached_dns_manager

        self._cached_dns_manager = None

    @provide(scope=Scope.REQUEST, provides=DNSSettingsDTO, cache=False)
    async def get_dns_mngr_settings(
        self,
        dns_state_gateway: DNSStateGateway,
        settings: Settings,
        root_dse_gw: DomainReadProtocol,
    ) -> AsyncIterator["DNSSettingsDTO"]:
        """Get DNS manager's settings."""
        domain = await root_dse_gw.get_domain()
        yield await dns_state_gateway.get_dns_manager_settings(
            settings,
            domain.name,
        )

    directory_create_use_case = provide(
        DirectoryCreateUseCase,
        scope=Scope.REQUEST,
    )
    attribute_type_dao = provide(AttributeTypeDAO, scope=Scope.REQUEST)
    attribute_type_dao_legacy = provide(
        AttributeTypeDAOLegacy,
        scope=Scope.REQUEST,
    )

    object_class_dao = provide(ObjectClassDAO, scope=Scope.REQUEST)
    object_class_dao_legacy = provide(
        ObjectClassDAOLegacy,
        scope=Scope.REQUEST,
    )
    attribute_dao = provide(AttributeDAO, scope=Scope.REQUEST)
    directory_dao = provide(DirectoryDAO, scope=Scope.REQUEST)

    entity_type_dao = provide(EntityTypeDAO, scope=Scope.REQUEST)
    attribute_type_system_flags_use_case = provide(
        AttributeTypeSystemFlagsUseCase,
        scope=Scope.REQUEST,
    )
    attribute_type_use_case = provide(
        AttributeTypeUseCase,
        scope=Scope.REQUEST,
    )

    @provide(scope=Scope.REQUEST)
    def get_attribute_type_use_case_legacy(
        self,
        session: AsyncSession,
    ) -> AttributeTypeUseCaseLegacy:
        """Legacy attribute type use case bound to a single session."""
        at_dao_legacy = AttributeTypeDAOLegacy(session=session)
        return AttributeTypeUseCaseLegacy(
            attribute_type_dao_legacy=at_dao_legacy,
        )

    object_class_use_case = provide(ObjectClassUseCase, scope=Scope.REQUEST)

    @provide(scope=Scope.REQUEST)
    def get_object_class_use_case_legacy(
        self,
        session: AsyncSession,
    ) -> ObjectClassUseCaseLegacy:
        """Legacy object class use case bound to a single session for all DAOs."""  # noqa: E501
        at_dao_legacy = AttributeTypeDAOLegacy(session=session)
        oc_dao_legacy = ObjectClassDAOLegacy(session=session)
        return ObjectClassUseCaseLegacy(
            attribute_type_dao_legacy=at_dao_legacy,
            object_class_dao_legacy=oc_dao_legacy,
        )

    user_password_history_use_cases = provide(
        UserPasswordHistoryUseCases,
        scope=Scope.REQUEST,
    )
    password_ban_word_repository = provide(
        PasswordBanWordRepository,
        scope=Scope.REQUEST,
    )
    password_policy_dao = provide(PasswordPolicyDAO, scope=Scope.REQUEST)
    password_use_cases = provide(PasswordPolicyUseCases, scope=Scope.REQUEST)
    password_policy_validator = provide(
        PasswordPolicyValidator,
        scope=Scope.REQUEST,
    )
    password_validator_settings = provide(
        PasswordValidatorSettings,
        scope=Scope.REQUEST,
    )
    password_policies_adapter = provide(
        PasswordPolicyFastAPIAdapter,
        scope=Scope.REQUEST,
    )
    password_ban_words_use_cases = provide(
        PasswordBanWordUseCases,
        scope=Scope.REQUEST,
    )
    password_ban_words_adapter = provide(
        PasswordBanWordsFastAPIAdapter,
        scope=Scope.REQUEST,
    )
    password_utils = provide(PasswordUtils, scope=scope)

    dns_fastapi_adapter = provide(DNSFastAPIAdapter, scope=Scope.REQUEST)
    dns_use_case = provide(DNSUseCase, scope=Scope.REQUEST)
    dns_state_gateway = provide(DNSStateGateway, scope=Scope.REQUEST)
    network_policy_gateway = provide(NetworkPolicyGateway, scope=Scope.SESSION)
    network_policy_validator_gateway = provide(
        NetworkPolicyValidatorGateway,
        provides=NetworkPolicyValidatorProtocol,
        scope=Scope.SESSION,
    )
    network_policy_validator = provide(
        NetworkPolicyValidatorUseCase,
        scope=Scope.SESSION,
    )

    @provide(scope=scope, provides=AsyncEngine)
    def get_engine(self, settings: Settings) -> AsyncEngine:
        """Get async engine."""
        return settings.engine

    @provide(scope=Scope.APP, provides=async_sessionmaker[AsyncSession])
    def get_session_factory(
        self,
        engine: AsyncEngine,
    ) -> async_sessionmaker[AsyncSession]:
        """Create session factory."""
        return async_sessionmaker(
            engine,
            expire_on_commit=False,
            autoflush=False,
            autocommit=False,
        )

    @provide(scope=Scope.APP)
    async def get_session(
        self,
        engine: AsyncEngine,
        session_factory: async_sessionmaker[AsyncSession],
        settings: Settings,
    ) -> AsyncIterator[AsyncSession]:
        """Get test session with a savepoint."""
        if self._cached_session:
            yield self._cached_session
            return

        connection = await engine.connect()
        trans = await connection.begin()

        schema_name = settings.TEST_POSTGRES_SCHEMA
        await connection.execute(
            text(f"SET search_path = {schema_name}, public;"),
        )

        async_session = session_factory(
            bind=connection,
            info={"mode": "test_transaction"},
            join_transaction_mode="create_savepoint",
        )

        self._cached_session = async_session
        self._session_id = uuid.uuid4()

        yield async_session

        self._cached_session = None
        self._session_id = None

        async_session.expire_all()
        await trans.rollback()
        await async_session.close()
        await connection.close()

    @provide(scope=Scope.SESSION)
    async def get_ldap_session(
        self,
        storage: SessionStorage,
    ) -> AsyncIterator[LDAPSession]:
        """Create ldap session."""
        session = LDAPSession(storage=storage)
        await session.start()
        yield session
        await session.disconnect()

    monitor_use_case = provide(AuditMonitorUseCase, scope=Scope.REQUEST)

    @provide(scope=Scope.REQUEST, provides=MultifactorAPI)
    async def get_mfa_api(self) -> Mock:
        """Create mock mfa."""
        mfa = Mock()
        mfa.ldap_validate_mfa = AsyncMock()
        mfa.get_create_mfa = AsyncMock(return_value="example.com")
        return mfa

    @provide(scope=Scope.REQUEST, provides=LDAPMultiFactorAPI)
    async def get_mfa_ldap_api(self) -> Mock:
        """Create mock mfa."""
        mfa = Mock()
        mfa.ldap_validate_mfa = AsyncMock()
        mfa.get_create_mfa = AsyncMock(return_value="example.com")
        return mfa

    @provide(scope=Scope.APP)
    async def get_redis_for_sessions(
        self,
        settings: Settings,
    ) -> AsyncIterator[SessionStorageClient]:
        """Get redis connection."""
        dsn = settings.SESSION_STORAGE_URL
        db_num = settings.TEST_WORKER_ID * 2
        worker_dsn = f"{dsn.scheme}://{dsn.host}:{dsn.port}/{db_num}"
        client = redis.Redis.from_url(worker_dsn)

        if not await client.ping():
            raise SystemError("Redis is not available")

        yield SessionStorageClient(client)

        await client.flushdb()
        with suppress(RuntimeError):
            await client.aclose()

    @provide(scope=Scope.REQUEST, provides=MasterGatewayProtocol)
    async def get_master_gateway(
        self,
        session: AsyncSession,
        settings: Settings,
    ) -> PGMasterGateway:
        return PGMasterGateway(session, settings)

    master_check_use_case = provide(
        MasterCheckUseCase,
        scope=Scope.REQUEST,
    )

    @provide(scope=Scope.APP)
    async def get_session_storage(
        self,
        client: SessionStorageClient,
        settings: Settings,
    ) -> SessionStorage:
        """Get session storage."""
        return RedisSessionStorage(
            client,
            settings.SESSION_KEY_LENGTH,
            settings.SESSION_KEY_EXPIRE_SECONDS,
        )

    role_dao = provide(RoleDAO, scope=Scope.REQUEST, cache=False)
    ace_dao = provide(AccessControlEntryDAO, scope=Scope.REQUEST)
    ace_migrations_dao = provide(
        AccessControlEntryAttributeTypeRemapDAO,
        scope=Scope.REQUEST,
    )
    access_manager = provide(AccessManager, scope=Scope.REQUEST)
    role_use_case = provide(RoleUseCase, scope=Scope.REQUEST)

    identity_fastapi_adapter = provide(
        AuthFastAPIAdapter,
        scope=Scope.REQUEST,
    )

    auth_manager = provide(
        AuthManager,
        scope=Scope.REQUEST,
    )

    @provide(scope=Scope.REQUEST)
    async def get_identity_provider(
        self,
        request: Request,
        session_storage: SessionStorage,
        settings: Settings,
        identity_provider_gateway: IdentityProviderGateway,
    ) -> IdentityProvider:
        """Create ldap session."""
        ip_from_request = get_ip_from_request(request)
        user_agent = get_user_agent_from_request(request)

        return IdentityProvider(
            session_storage,
            settings,
            identity_provider_gateway,
            ip_from_request=str(ip_from_request),
            user_agent=user_agent,
            session_key=request.cookies.get("id", ""),
        )

    identity_provider_gateway = provide(
        IdentityProviderGateway,
        scope=Scope.REQUEST,
    )
    mfa_fastapi_adapter = provide(MFAFastAPIAdapter, scope=Scope.REQUEST)
    mfa_manager = provide(MFAManager, scope=Scope.REQUEST)
    ldap_entity_type_adapter = provide(
        LDAPEntityTypeFastAPIAdapter,
        scope=Scope.REQUEST,
    )

    kerberos_service = provide(KerberosService, scope=Scope.REQUEST)
    kerberos_fastapi_adapter = provide(
        KerberosFastAPIAdapter,
        scope=Scope.REQUEST,
    )

    @provide(scope=Scope.REQUEST)
    def get_krb_template_render(
        self,
        settings: Settings,
    ) -> KRBTemplateRenderer:
        """Provide KRBTemplateRenderer with settings.TEMPLATES."""
        return KRBTemplateRenderer(settings.TEMPLATES)

    krb_ldap_manager = provide(KRBLDAPStructureManager, scope=Scope.REQUEST)
    audit_policy_dao = provide(AuditPoliciesDAO, scope=Scope.REQUEST)
    audit_use_case = provide(AuditUseCase, scope=Scope.REQUEST)
    audit_destination_dao = provide(AuditDestinationDAO, scope=Scope.REQUEST)
    attribute_value_validator = provide(AttributeValueValidator, scope=scope)

    @provide(scope=Scope.REQUEST, provides=AuditService)
    async def get_audit_service(self) -> AsyncIterator[AsyncMock]:
        """Provide a mock audit service."""
        audit_service = Mock()

        ok_response = Mock()
        ok_response.status_code = 200

        audit_service.get_policies = AsyncMock(return_value=[])
        audit_service.update_policy = AsyncMock(return_value=None)
        audit_service.get_destinations = AsyncMock(return_value=[])
        audit_service.create_destination = AsyncMock(return_value=None)
        audit_service.update_destination = AsyncMock(return_value=None)
        audit_service.delete_destination = AsyncMock(return_value=None)

        if not self._cached_audit_service:
            self._cached_audit_service = audit_service

        yield self._cached_audit_service

        self._cached_audit_service = None

    audit_adapter = provide(AuditPoliciesAdapter, scope=Scope.REQUEST)

    @provide(scope=scope)
    async def get_audit_redis_client(
        self,
        settings: Settings,
    ) -> AsyncIterator[AuditRedisClient]:
        """Get audit redis client."""
        dsn = settings.EVENT_HANDLER_URL
        db_num = settings.TEST_WORKER_ID * 2 + 1
        worker_dsn = f"{dsn.scheme}://{dsn.host}:{dsn.port}/{db_num}"
        client = redis.Redis.from_url(worker_dsn)

        if not await client.ping():
            raise SystemError("Redis is not available")

        yield AuditRedisClient(client)

        with suppress(RuntimeError):
            await client.aclose()

    @provide(scope=Scope.APP)
    async def get_raw_audit_manager(
        self,
        client: AuditRedisClient,
        settings: Settings,
    ) -> AsyncIterator[RawAuditManager]:
        """Get raw audit manager."""
        yield RawAuditManager(
            client,
            settings.RAW_EVENT_STREAM_NAME,
            settings.EVENT_HANDLER_GROUP,
            settings.EVENT_CONSUMER_NAME,
            settings.IS_PROC_EVENT_KEY,
        )

    @provide(scope=Scope.APP)
    async def get_normalized_audit_manager(
        self,
        client: AuditRedisClient,
        settings: Settings,
    ) -> AsyncIterator[NormalizedAuditManager]:
        """Get raw audit manager."""
        yield NormalizedAuditManager(
            client,
            settings.NORMALIZED_EVENT_STREAM_NAME,
            settings.EVENT_SENDER_GROUP,
            settings.EVENT_CONSUMER_NAME,
            settings.IS_PROC_EVENT_KEY,
        )

    shadow_adapter = provide(
        ShadowAdapter,
        scope=Scope.REQUEST,
    )
    request = from_context(provides=Request, scope=Scope.REQUEST)

    @provide(scope=Scope.REQUEST)
    async def get_audit_monitor(
        self,
        session: AsyncSession,
        audit_use_case: "AuditUseCase",
        session_storage: SessionStorage,
        settings: Settings,
        request: Request,
    ) -> AuditMonitor:
        """Create ldap session."""
        ip_from_request = get_ip_from_request(request)
        user_agent = get_user_agent_from_request(request)
        session_key = request.cookies.get("id", "")

        return AuditMonitor(
            session=session,
            audit_use_case=audit_use_case,
            session_storage=session_storage,
            settings=settings,
            ip_from_request=ip_from_request,
            user_agent=user_agent,
            session_key=session_key,
        )

    add_request_context = provide(
        LDAPAddRequestContext,
        scope=Scope.REQUEST,
    )
    bind_request_context = provide(
        LDAPBindRequestContext,
        scope=Scope.REQUEST,
    )
    delete_request_context = provide(
        LDAPDeleteRequestContext,
        scope=Scope.REQUEST,
    )
    extended_request_context = provide(
        LDAPExtendedRequestContext,
        scope=Scope.REQUEST,
    )
    modify_request_context = provide(
        LDAPModifyRequestContext,
        scope=Scope.REQUEST,
    )
    modify_dn_request_context = provide(
        LDAPModifyDNRequestContext,
        scope=Scope.REQUEST,
    )
    search_request_context = provide(
        LDAPSearchRequestContext,
        scope=Scope.REQUEST,
    )
    abandon_request_context = provide(
        LDAPAbandonRequestContext,
        scope=Scope.REQUEST,
    )

    unbind_request_context = provide(
        LDAPUnbindRequestContext,
        scope=Scope.REQUEST,
    )
    session_repository = provide(SessionRepository, scope=Scope.REQUEST)
    session_gateway = provide(SessionFastAPIGateway, scope=Scope.REQUEST)
    attribute_type_fastapi_adapter = provide(
        AttributeTypeFastAPIAdapter,
        scope=Scope.REQUEST,
    )
    object_class_fastapi_adapter = provide(
        ObjectClassFastAPIAdapter,
        scope=Scope.REQUEST,
    )

    entity_type_use_case = provide(EntityTypeUseCase, scope=Scope.REQUEST)

    dhcp_adapter = provide(DHCPAdapter, scope=Scope.REQUEST)
    setup_gateway = provide(SetupGateway, scope=Scope.REQUEST)
    setup_use_case = provide(SetupUseCase, scope=Scope.REQUEST)
    network_policy_adapter = provide(
        NetworkPolicyFastAPIAdapter,
        scope=Scope.REQUEST,
    )
    network_policy_use_case = provide(
        NetworkPolicyUseCase,
        scope=Scope.REQUEST,
    )

    @provide(
        provides=AuthorizationProviderProtocol,
        scope=Scope.REQUEST,
    )
    def authorization_provider_protocol(
        self,
        identity_provider: IdentityProvider,
    ) -> AuthorizationProvider:
        return AuthorizationProvider(identity_provider)

    rootdse_gw = provide(
        SADomainGateway,
        provides=DomainReadProtocol,
        scope=Scope.REQUEST,
    )
    rootdse_reader = provide(RootDSEReader, scope=Scope.REQUEST)
    dcinfo_reader = provide(DCInfoReader, scope=Scope.REQUEST)


@dataclass
class TestCreds:
    """Test credentials class."""

    __test__ = False

    un: str
    pw: str


@dataclass
class TestAdminCreds(TestCreds):
    """Test admin credentials class."""

    __test__ = False


@pytest_asyncio.fixture(scope="session")
async def container(settings: Settings) -> AsyncIterator[AsyncContainer]:
    """Create test container."""
    ctnr = make_async_container(
        TestProvider(),
        MFACredsProvider(),
        context={Settings: settings},
        start_scope=Scope.RUNTIME,
    )
    yield ctnr
    await ctnr.close()


class MutePolicyBindRequest(BindRequest):
    """Mute group validaton."""

    __test__ = False

    @staticmethod
    async def is_user_group_valid(*args, **kwargs) -> bool:  # type: ignore
        """Stub."""
        return True


@pytest_asyncio.fixture
async def kadmin(container: AsyncContainer) -> AsyncIterator[AbstractKadmin]:
    """Get di kadmin."""
    async with container(scope=Scope.APP) as container:
        yield await container.get(AbstractKadmin)


@pytest_asyncio.fixture
async def audit_service(
    container: AsyncContainer,
) -> AsyncIterator[AuditService]:
    """Get di audit_service."""
    async with container(scope=Scope.REQUEST) as container:
        yield await container.get(AuditService)


@pytest.fixture(scope="session")
def event_loop() -> Generator:
    """Create uvloop event loop."""
    loop = uvloop.new_event_loop()
    yield loop
    with suppress(asyncio.CancelledError, RuntimeError):
        asyncio.gather(*asyncio.tasks.all_tasks(loop)).cancel()
        loop.close()


@pytest.fixture(scope="session")
def settings() -> Settings:
    """Get settings."""
    return Settings.from_os()


def is_master(config: pytest.Config) -> bool:
    """Check is master worker."""
    return not hasattr(config, "workerinput")


@pytest.fixture(scope="session", autouse=True)
async def add_schema(
    container: AsyncContainer,
    settings: Settings,
) -> AsyncGenerator:
    """Create schema for each worker and drops it afterwards."""
    if settings.TEST_POSTGRES_SCHEMA == "public":
        yield
        return

    engine = await container.get(AsyncEngine)

    async with engine.begin() as conn:
        await conn.execute(
            schema.CreateSchema(
                settings.TEST_POSTGRES_SCHEMA,
                if_not_exists=True,
            ),
        )

        # NOTE: Create extensions for public schema only one time to
        # avoid conflicts
        if settings.TEST_POSTGRES_SCHEMA.endswith("gw0"):
            await conn.execute(
                text(
                    "CREATE EXTENSION IF NOT EXISTS "
                    '"uuid-ossp" SCHEMA public;',
                ),
            )
            await conn.execute(
                text(
                    'CREATE EXTENSION IF NOT EXISTS "pg_trgm" SCHEMA public;',
                ),
            )

    yield

    async with engine.begin() as conn:
        await conn.execute(
            text(f"DROP SCHEMA {settings.TEST_POSTGRES_SCHEMA} CASCADE;"),
        )


class TestMigrationProvider(Provider):
    """Provider for migrations."""

    async_conn = from_context(provides=AsyncConnection, scope=Scope.RUNTIME)

    @provide(scope=Scope.APP, cache=False)
    def get_session_factory(
        self,
        async_conn: AsyncConnection,
    ) -> AsyncSession:
        """Create session factory."""
        return AsyncSession(async_conn)


@pytest_asyncio.fixture(scope="session", autouse=True)
async def _migrations(
    add_schema: None,  # noqa: ARG001
    container: AsyncContainer,
    settings: Settings,
) -> AsyncGenerator:
    """Run simple migrations."""
    engine = await container.get(AsyncEngine)

    config = AlembicConfig("alembic.ini")
    config.attributes["app_settings"] = settings

    def upgrade(conn: AsyncConnection) -> None:
        config.attributes["connection"] = conn
        command.upgrade(config, "head")

    def downgrade(conn: AsyncConnection) -> None:
        config.attributes["connection"] = conn
        command.downgrade(config, "base")

    test_migration_provider = TestMigrationProvider()
    async with engine.begin() as conn:
        config.attributes["connection"] = conn
        config.attributes["dishka_container"] = make_async_container(
            TestProvider(),
            test_migration_provider,
            context={Settings: settings, AsyncConnection: conn},
            start_scope=Scope.RUNTIME,
        )
        await conn.run_sync(upgrade)  # type: ignore

    yield

    async with engine.begin() as conn:
        config.attributes["dishka_container"] = make_async_container(
            TestProvider(),
            test_migration_provider,
            context={Settings: settings, AsyncConnection: conn},
            start_scope=Scope.RUNTIME,
        )
        await conn.run_sync(downgrade)  # type: ignore


@pytest_asyncio.fixture(scope="function")
async def session(
    container: AsyncContainer,
    handler: PoolClientHandler,
) -> AsyncIterator[AsyncSession]:
    """Get session and acquire after completion."""
    async with container(scope=Scope.APP) as container:
        session = await container.get(AsyncSession)
        handler.container = container
        yield session


@pytest_asyncio.fixture(scope="function")
async def raw_audit_manager(
    container: AsyncContainer,
) -> AsyncIterator[RawAuditManager]:
    """Get raw audit adapter."""
    async with container(scope=Scope.APP) as container:
        yield await container.get(RawAuditManager)


@pytest_asyncio.fixture(scope="function")
async def setup_session(
    session: AsyncSession,
    raw_audit_manager: RawAuditManager,
    password_utils: PasswordUtils,
) -> None:
    """Get session and acquire after completion."""
    role_dao = RoleDAO(session)
    ace_dao = AccessControlEntryDAO(session)
    role_use_case = RoleUseCase(role_dao, ace_dao)
    attribute_value_validator = AttributeValueValidator()
    attribute_type_dao = AttributeTypeDAO(session)
    attribute_type_system_flags_use_case = AttributeTypeSystemFlagsUseCase()
    object_class_dao_legacy = ObjectClassDAOLegacy(session=session)
    attribute_type_dao_legacy = AttributeTypeDAOLegacy(session=session)
    object_class_use_case_legacy = ObjectClassUseCaseLegacy(
        attribute_type_dao_legacy=attribute_type_dao_legacy,
        object_class_dao_legacy=object_class_dao_legacy,
    )
    attribute_type_use_case_legacy = AttributeTypeUseCaseLegacy(
        attribute_type_dao_legacy=attribute_type_dao_legacy,
    )

    object_class_dao = ObjectClassDAO(session)
    directory_dao = DirectoryDAO(session)
    attribute_dao = AttributeDAO(session)
    entity_type_dao = EntityTypeDAO(
        session,
        attribute_value_validator=attribute_value_validator,
        directory_dao=directory_dao,
    )
    entity_type_use_case = EntityTypeUseCase(
        entity_type_dao=entity_type_dao,
        object_class_dao=object_class_dao,
        directory_dao=directory_dao,
    )
    directory_create_use_case = DirectoryCreateUseCase(
        session=session,
        entity_type_use_case=entity_type_use_case,
        role_use_case=role_use_case,
        directory_dao=directory_dao,
        attribute_dao=attribute_dao,
    )
    object_class_use_case = ObjectClassUseCase(
        attribute_type_dao=attribute_type_dao,
        object_class_dao=object_class_dao,
        entity_type_dao=entity_type_dao,
        directory_create_use_case=directory_create_use_case,
    )

    attribute_type_use_case = AttributeTypeUseCase(
        attribute_type_dao=attribute_type_dao,
        attribute_type_system_flags_use_case=attribute_type_system_flags_use_case,
        object_class_dao=object_class_dao,
        directory_create_use_case=directory_create_use_case,
    )

    audit_policy_dao = AuditPoliciesDAO(session)
    audit_destination_dao = AuditDestinationDAO(session)
    audit_use_case = AuditUseCase(
        audit_policy_dao,
        audit_destination_dao,
        raw_audit_manager,
    )
    password_policy_dao = PasswordPolicyDAO(
        session,
        attribute_value_validator=attribute_value_validator,
    )
    password_policy_validator = PasswordPolicyValidator(
        PasswordValidatorSettings(),
        password_utils,
    )
    password_ban_word_repository = PasswordBanWordRepository(session)
    password_use_cases = PasswordPolicyUseCases(
        password_policy_dao,
        password_policy_validator,
        password_ban_word_repository,
    )
    setup_gateway = SetupGateway(
        session,
        password_utils,
        entity_type_use_case=entity_type_use_case,
        attribute_value_validator=attribute_value_validator,
        directory_dao=directory_dao,
    )

    for entity_type_dto in chain(ENTITY_TYPE_DTOS_V1, ENTITY_TYPE_DTOS_V2):
        await entity_type_use_case.create_not_safe(entity_type_dto)
    await session.flush()

    await audit_use_case.create_policies()
    await setup_gateway.setup_enviroment(
        dn="md.test",
        data=TEST_DATA,
        is_system=False,
    )

    for _at_dto in (
        AttributeTypeDTO[None](
            oid="1.2.3.4.5.6.7.8",
            name="attr_with_bvalue",
            ldap_display_name="attrWithBvalue",
            syntax="1.3.6.1.4.1.1466.115.121.1.40",  # Octet String
            single_value=True,
            no_user_modification=False,
            is_system=True,
            system_flags=0,
            is_included_anr=False,
        ),
        AttributeTypeDTO[None](
            oid="1.2.3.4.5.6.7.8.9",
            name="testing_attr",
            ldap_display_name="testingAttr",
            syntax="1.3.6.1.4.1.1466.115.121.1.15",
            single_value=True,
            no_user_modification=False,
            is_system=True,
            system_flags=0,
            is_included_anr=False,
        ),
    ):
        await attribute_type_use_case.create(_at_dto)

    for attr_type_name in (
        "description",
        "posixEmail",
        "userPrincipalName",
        "userAccountControl",
        "cn",
        "objectClass",
    ):
        _at = await attribute_type_use_case_legacy.get(
            attr_type_name,
        )
        if not _at:
            raise ValueError(
                f"setup_session:: AttributeType {attr_type_name} not found",
            )
        await attribute_type_use_case.create(_at)

    for _obj_class_name in (
        "top",
        "person",
        "organizationalPerson",
        "user",
        "domain",
        "container",
        "organization",
        "domainDNS",
        "group",
        "inetOrgPerson",
        "posixAccount",
    ):
        _oc_dto = await object_class_use_case_legacy.get(_obj_class_name)
        _oc_dto.attribute_types_may = [
            _.name  # type: ignore
            for _ in _oc_dto.attribute_types_may
        ]
        _oc_dto.attribute_types_must = [
            _.name  # type: ignore
            for _ in _oc_dto.attribute_types_must
        ]
        await object_class_use_case.create(_oc_dto)  # type: ignore

    # NOTE: after setup environment we need base DN to be created
    await password_use_cases.create_default_domain_policy()

    await role_use_case.create_domain_admins_role()

    await role_use_case._role_dao.create(  # noqa: SLF001
        dto=RoleDTO(
            name="TEST ONLY LOGIN ROLE",
            creator_upn=None,
            is_system=True,
            groups=["cn=admin login only,cn=Groups,dc=md,dc=test"],
            permissions=AuthorizationRules.AUTH_LOGIN,
        ),
    )

    await session.commit()


@pytest_asyncio.fixture(scope="function")
async def ldap_session(
    container: AsyncContainer,
) -> AsyncIterator[LDAPSession]:
    """Yield empty session."""
    async with container(scope=Scope.SESSION) as container:
        yield await container.get(LDAPSession)


@pytest_asyncio.fixture(scope="function")
async def ldap_bound_session(
    ldap_session: LDAPSession,
    session: AsyncSession,
    creds: TestCreds,
    setup_session: None,  # noqa: ARG001
) -> AsyncIterator[LDAPSession]:
    """Yield bound session."""
    user = await get_user(session, creds.un)
    assert user
    await ldap_session.set_user(user)
    yield ldap_session
    return


@pytest_asyncio.fixture(scope="function")
async def network_policy_gateway(
    container: AsyncContainer,
) -> AsyncIterator[NetworkPolicyGateway]:
    """Get network policy gateway."""
    async with container(scope=Scope.SESSION) as container:
        yield await container.get(NetworkPolicyGateway)


@pytest_asyncio.fixture(scope="function")
async def network_policy_use_case(
    container: AsyncContainer,
) -> AsyncIterator[NetworkPolicyUseCase]:
    """Get network policy gateway."""
    async with container(scope=Scope.REQUEST) as container:
        yield await container.get(NetworkPolicyUseCase)


@pytest_asyncio.fixture(scope="function")
async def network_policy_validator(
    container: AsyncContainer,
) -> AsyncIterator[NetworkPolicyValidatorUseCase]:
    """Get network policy validator."""
    async with container(scope=Scope.SESSION) as container:
        yield await container.get(NetworkPolicyValidatorUseCase)


@pytest_asyncio.fixture(scope="session")
async def handler(
    settings: Settings,
    container: AsyncContainer,
) -> AsyncIterator[PoolClientHandler]:
    """Create test handler."""
    settings.set_test_port()
    test_log = logger.bind(name="ldap_test")
    async with container(scope=Scope.APP) as app_scope:
        yield PoolClientHandler(settings, app_scope, test_log)  # type: ignore


@pytest_asyncio.fixture(scope="function")
async def entity_type_dao(
    container: AsyncContainer,
) -> AsyncIterator[EntityTypeDAO]:
    """Get session and acquire after completion."""
    async with container(scope=Scope.REQUEST) as container:
        session = await container.get(AsyncSession)
        attribute_value_validator = await container.get(
            AttributeValueValidator,
        )
        directory_dao = await container.get(DirectoryDAO)
        yield EntityTypeDAO(
            session,
            attribute_value_validator=attribute_value_validator,
            directory_dao=directory_dao,
        )


@pytest_asyncio.fixture(scope="function")
async def entity_type_use_case(
    container: AsyncContainer,
) -> AsyncIterator[EntityTypeUseCase]:
    """Get entity type use case."""
    async with container(scope=Scope.REQUEST) as container:
        yield await container.get(EntityTypeUseCase)


@pytest_asyncio.fixture(scope="function")
async def password_policy_dao(
    container: AsyncContainer,
) -> AsyncIterator[PasswordPolicyDAO]:
    """Get session and acquire after completion."""
    async with container(scope=Scope.APP) as container:
        session = await container.get(AsyncSession)
        attribute_value_validator = await container.get(
            AttributeValueValidator,
        )
        yield PasswordPolicyDAO(
            session,
            attribute_value_validator=attribute_value_validator,
        )


@pytest_asyncio.fixture(scope="function")
async def password_ban_word_repository(
    container: AsyncContainer,
) -> AsyncIterator[PasswordBanWordRepository]:
    """Get password ban word repository."""
    async with container(scope=Scope.APP) as container:
        session = await container.get(AsyncSession)
        yield PasswordBanWordRepository(session)


@pytest_asyncio.fixture(scope="function")
async def password_utils(
    container: AsyncContainer,
) -> AsyncIterator[PasswordUtils]:
    """Get session and acquire after completion."""
    async with container(scope=Scope.APP) as container:
        yield PasswordUtils()


@pytest_asyncio.fixture(scope="function")
async def password_policy_use_cases(
    container: AsyncContainer,
    password_policy_dao: PasswordPolicyDAO,
    password_ban_word_repository: PasswordBanWordRepository,
    password_policy_validator: PasswordPolicyValidator,
) -> AsyncIterator[PasswordPolicyUseCases]:
    """Get session and acquire after completion."""
    async with container(scope=Scope.APP) as container:
        yield PasswordPolicyUseCases(
            password_policy_dao,
            password_policy_validator,
            password_ban_word_repository,
        )


@pytest_asyncio.fixture(scope="function")
async def password_validator_settings(
    container: AsyncContainer,
) -> AsyncIterator[PasswordValidatorSettings]:
    """Get session and acquire after completion."""
    async with container(scope=Scope.APP) as container:
        yield PasswordValidatorSettings()


@pytest_asyncio.fixture(scope="function")
async def password_policy_validator(
    container: AsyncContainer,
    password_validator_settings: PasswordValidatorSettings,
    password_utils: PasswordUtils,
) -> AsyncIterator[PasswordPolicyValidator]:
    """Get session and acquire after completion."""
    async with container(scope=Scope.APP) as container:
        yield PasswordPolicyValidator(
            password_validator_settings,
            password_utils,
        )


@pytest_asyncio.fixture(scope="function")
async def attribute_type_dao(
    container: AsyncContainer,
) -> AsyncIterator[AttributeTypeDAO]:
    """Get session and acquire after completion."""
    async with container(scope=Scope.REQUEST) as container:
        session = await container.get(AsyncSession)
        yield AttributeTypeDAO(session)


@pytest_asyncio.fixture(scope="function")
async def role_dao(container: AsyncContainer) -> AsyncIterator[RoleDAO]:
    """Get session and acquire after completion."""
    async with container(scope=Scope.APP) as container:
        session = await container.get(AsyncSession)
        yield RoleDAO(session)


@pytest_asyncio.fixture(scope="function")
async def access_control_entry_dao(
    container: AsyncContainer,
) -> AsyncIterator[AccessControlEntryDAO]:
    """Get session and aquire after completion."""
    async with container(scope=Scope.APP) as container:
        session = await container.get(AsyncSession)
        yield AccessControlEntryDAO(session)


@pytest.fixture
def access_manager() -> AccessManager:
    """Get access manager."""
    return AccessManager()


@pytest_asyncio.fixture(scope="function")
async def role_use_case(
    container: AsyncContainer,
) -> AsyncIterator[RoleUseCase]:
    """Get role use case."""
    async with container(scope=Scope.APP) as container:
        session = await container.get(AsyncSession)
        role_dao = RoleDAO(session)
        ace_dao = AccessControlEntryDAO(session)
        yield RoleUseCase(role_dao, ace_dao)


@pytest.fixture(scope="session", autouse=True)
def _server(
    event_loop: asyncio.BaseEventLoop,
    handler: PoolClientHandler,
) -> Generator:
    """Run server in background."""
    task = asyncio.ensure_future(handler.start(), loop=event_loop)
    event_loop.run_until_complete(asyncio.sleep(0.1))
    yield
    with suppress(asyncio.CancelledError):
        task.cancel()


@pytest.fixture
async def ldap_client(
    settings: Settings,
    creds: TestCreds,
) -> AsyncIterator[aioldap3.LDAPConnection]:
    """Get LDAP client without credentials."""
    conn = aioldap3.LDAPConnection(
        aioldap3.Server(host=str(settings.HOST), port=settings.PORT),
    )
    await conn.bind(creds.un, creds.pw)
    yield conn
    await conn.unbind()


@pytest.fixture
async def anonymous_ldap_client(
    settings: Settings,
) -> AsyncIterator[aioldap3.LDAPConnection]:
    """Get LDAP client without credentials."""
    conn = aioldap3.LDAPConnection(
        aioldap3.Server(host=str(settings.HOST), port=settings.PORT),
    )
    await conn.bind()
    yield conn
    await conn.unbind()


@pytest_asyncio.fixture(scope="function")
async def app(
    settings: Settings,
    container: AsyncContainer,
) -> AsyncIterator[FastAPI]:
    """App creator fixture."""
    async with container(scope=Scope.APP) as container:
        app = _create_basic_app(settings)
        app.include_router(shadow_router, prefix="/shadow")
        setup_dishka(container, app)
        yield app


@pytest_asyncio.fixture(scope="function")
async def unbound_http_client(
    app: FastAPI,
) -> AsyncIterator[httpx.AsyncClient]:
    """Get async client for fastapi tests.

    :param FastAPI app: asgi app
    :yield Iterator[AsyncIterator[httpx.AsyncClient]]: yield client
    """
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app, root_path="/api"),
        timeout=3,
        base_url="http://test",
    ) as client:
        yield client


@pytest_asyncio.fixture(scope="function")
async def http_client(
    unbound_http_client: httpx.AsyncClient,
    creds: TestCreds,
    setup_session: None,  # noqa: ARG001
) -> httpx.AsyncClient:
    """Authenticate and return client with cookies.

    :param httpx.AsyncClient unbound_http_client: client w/o cookies
    :param TestCreds creds: creds to authn
    :param None setup_session: just a fixture call
    :return httpx.AsyncClient: bound client with cookies
    """
    response = await unbound_http_client.post(
        "auth/",
        data={"username": creds.un, "password": creds.pw},
    )

    assert response.status_code == 200
    assert unbound_http_client.cookies.get("id")

    return unbound_http_client


@pytest_asyncio.fixture(scope="function")
async def http_client_with_login_perm(
    unbound_http_client: httpx.AsyncClient,
    creds_with_login_perm: TestCreds,
    setup_session: None,  # noqa: ARG001
) -> httpx.AsyncClient:
    """Authenticate and return client with cookies.

    :param httpx.AsyncClient unbound_http_client: client w/o cookies
    :param TestCreds creds: creds to authn
    :param None setup_session: just a fixture call
    :return httpx.AsyncClient: bound client with cookies
    """
    response = await unbound_http_client.post(
        "auth/",
        data={
            "username": creds_with_login_perm.un,
            "password": creds_with_login_perm.pw,
        },
    )

    assert response.status_code == 200
    assert unbound_http_client.cookies.get("id")

    return unbound_http_client


@pytest_asyncio.fixture(scope="function")
async def admin_http_client(
    app: FastAPI,
    admin_creds: TestAdminCreds,
    setup_session: None,  # noqa: ARG001
) -> AsyncIterator[httpx.AsyncClient]:
    """Authenticate as admin and return client with cookies.

    :param httpx.AsyncClient unbound_http_client: client w/o cookies
    :param None setup_session: just a fixture call
    :return httpx.AsyncClient: bound client with cookies
    """
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app, root_path="/api"),
        timeout=3,
        base_url="http://test",
    ) as unbound_http_client:
        response = await unbound_http_client.post(
            "auth/",
            data={"username": admin_creds.un, "password": admin_creds.pw},
        )

        assert response.status_code == 200
        assert unbound_http_client.cookies.get("id")

        yield unbound_http_client


@pytest.fixture
def creds(user: dict) -> TestCreds:
    """Get creds from test data."""
    return TestCreds(user["sam_account_name"], user["password"])


@pytest.fixture
def creds_with_login_perm(user_with_login_perm: dict) -> TestCreds:
    """Get creds from test data."""
    return TestCreds(
        user_with_login_perm["sam_account_name"],
        user_with_login_perm["password"],
    )


@pytest.fixture
def admin_creds(admin_user: dict) -> TestAdminCreds:
    """Get admin creds from test data."""
    return TestAdminCreds(
        admin_user["sam_account_name"],
        admin_user["password"],
    )


@pytest.fixture
def user() -> dict:
    """Get user data."""
    return user_data_dict


@pytest.fixture
def admin_user() -> dict:
    """Get admin user data."""
    return admin_user_data_dict


@pytest.fixture
def user_with_login_perm() -> dict:
    """Get user data."""
    return user_with_login_perm_data_dict


@pytest.fixture
async def api_permissions_checker(
    request_container: AsyncContainer,
) -> AsyncIterator[AuthorizationProvider]:
    """Get all api permissions."""
    return await request_container.get(AuthorizationProviderProtocol)


@pytest_asyncio.fixture
async def request_params() -> dict:
    """Return minimal ASGI scope plus response for request-scoped providers."""
    scope = {
        "type": "http",
        "method": "GET",
        "scheme": "http",
        "path": "/",
        "query_string": b"",
        "root_path": "",
        "headers": [],
        "client": ("127.0.0.1", 0),
        "server": ("testserver", 80),
    }
    request = Request(scope)
    response = Response()
    return {Request: request, Response: response}


@pytest_asyncio.fixture
async def request_container(
    container: AsyncContainer,
    request_params: dict,
) -> AsyncIterator[AsyncContainer]:
    """Create request scope with Request context."""
    async with container(scope=Scope.REQUEST, context=request_params) as cont:
        yield cont


@pytest.fixture
def _force_override_tls(settings: Settings) -> Iterator:
    """Override tls status for tests."""
    current_status = settings.USE_CORE_TLS
    settings.USE_CORE_TLS = True
    yield
    settings.USE_CORE_TLS = current_status


@pytest_asyncio.fixture
async def dns_manager(
    container: AsyncContainer,
) -> AsyncIterator[AbstractDNSManager]:
    """Get DI DNS manager."""
    async with container(scope=Scope.REQUEST) as container:
        yield await container.get(AbstractDNSManager)


@pytest_asyncio.fixture
async def dhcp_manager(
    request_container: AsyncContainer,
) -> AsyncIterator[AbstractDHCPManager]:
    """Get DI DHCP manager."""
    yield await request_container.get(AbstractDHCPManager)


@pytest.fixture
async def storage(container: AsyncContainer) -> AsyncIterator[SessionStorage]:
    """Return session storage."""
    async with container() as c:
        yield await c.get(SessionStorage)


@pytest.fixture
async def ctx_bind(
    container: AsyncContainer,
) -> AsyncIterator[LDAPBindRequestContext]:
    """Return session storage."""
    async with container(scope=Scope.REQUEST) as c:
        yield await c.get(LDAPBindRequestContext)


@pytest.fixture
async def ctx_unbind(
    container: AsyncContainer,
) -> AsyncIterator[LDAPUnbindRequestContext]:
    """Return session storage."""
    async with container(scope=Scope.REQUEST) as c:
        yield await c.get(LDAPUnbindRequestContext)


@pytest.fixture
async def ctx_add(
    container: AsyncContainer,
) -> AsyncIterator[LDAPAddRequestContext]:
    """Return session storage."""
    async with container(scope=Scope.REQUEST) as c:
        yield await c.get(LDAPAddRequestContext)


@pytest.fixture
async def ctx_search(
    container: AsyncContainer,
) -> AsyncIterator[LDAPSearchRequestContext]:
    """Return session storage."""
    async with container(scope=Scope.REQUEST) as c:
        yield await c.get(LDAPSearchRequestContext)


def pytest_configure(config: pytest.Config) -> None:
    """Pytest hook to limit xdist workers based on Dragonfly DBs.

    If need to increase the number of Dragonfly DBs, set DFLY_dbnum env
    variable in dragonfly and test services in docker-compose.yml file.
    """
    if is_master(config):
        dragonfly_dbs = int(os.getenv("DFLY_dbnum", "16")) // 2  # noqa: SIM112
        xdist_worker_count = config.option.numprocesses
        if xdist_worker_count and xdist_worker_count > dragonfly_dbs:
            raise pytest.UsageError(
                f"The use of more than {dragonfly_dbs} processes "
                f"is prohibited (requested {xdist_worker_count}).\n"
                "Reduce the value of -n/--numprocesses or "
                "increase the number of Dragonfly databases.\n",
            )


def pytest_xdist_auto_num_workers(config: pytest.Config) -> int:
    """Pytest hook to limit xdist workers for auto mode."""
    if is_master(config):
        cpu_count = os.cpu_count()
        dragonfly_dbs = int(os.getenv("DFLY_dbnum", "16")) // 2  # noqa: SIM112
        return min(cpu_count if cpu_count else 2, dragonfly_dbs)
    return 0