conftest.py

import typing
import uuid

import pytest
from django.conf import settings
from django_test_migrations.migrator import Migrator
from mypy_boto3_dynamodb.service_resource import DynamoDBServiceResource, Table
from pytest_django.fixtures import SettingsWrapper

from environments.dynamodb import (
    DynamoEnvironmentV2Wrapper,
    DynamoEnvironmentWrapper,
    DynamoIdentityWrapper,
)
from environments.dynamodb.types import IdentityOverrideV2
from environments.dynamodb.utils import (
    get_environments_v2_identity_override_document_key,
)
from environments.models import Environment
from features.models import Feature, FeatureState
from tests.types import MigratorFactory
from util.mappers import (
    map_environment_to_environment_document,
    map_environment_to_environment_v2_document,
    map_feature_state_to_engine,
    map_identity_override_to_identity_override_document,
)


@pytest.fixture()
def edge_identity_dynamo_wrapper_mock(mocker):  # type: ignore[no-untyped-def]
    return mocker.patch(
        "edge_api.identities.models.EdgeIdentity.dynamo_wrapper",
    )


@pytest.fixture()
def flagsmith_environment_api_key_table(dynamodb: "DynamoDBServiceResource") -> "Table":
    return dynamodb.create_table(
        TableName="flagsmith_environment_api_key",
        KeySchema=[{"AttributeName": "key", "KeyType": "HASH"}],
        AttributeDefinitions=[
            {"AttributeName": "key", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )


@pytest.fixture()
def flagsmith_environment_table(dynamodb: "DynamoDBServiceResource") -> "Table":
    return dynamodb.create_table(
        TableName="flagsmith_environments",
        KeySchema=[{"AttributeName": "api_key", "KeyType": "HASH"}],
        AttributeDefinitions=[
            {"AttributeName": "api_key", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )


@pytest.fixture()
def flagsmith_project_metadata_table(dynamodb: "DynamoDBServiceResource") -> "Table":
    return dynamodb.create_table(
        TableName="flagsmith_project_metadata",
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "N"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )


@pytest.fixture
def dynamodb_identity_wrapper(
    settings: SettingsWrapper,
    flagsmith_identities_table: Table,
) -> DynamoIdentityWrapper:
    settings.IDENTITIES_TABLE_NAME_DYNAMO = flagsmith_identities_table.name
    return DynamoIdentityWrapper()


@pytest.fixture
def dynamodb_wrapper_v2(
    settings: SettingsWrapper,
    flagsmith_environments_v2_table: Table,
) -> DynamoEnvironmentV2Wrapper:
    settings.ENVIRONMENTS_V2_TABLE_NAME_DYNAMO = flagsmith_environments_v2_table.name
    return DynamoEnvironmentV2Wrapper()


@pytest.fixture()
def dynamo_enabled_project_environment_one_document(
    flagsmith_environment_table: Table,
    flagsmith_environments_v2_table: Table,
    dynamo_enabled_project_environment_one: Environment,
) -> dict[str, typing.Any]:
    environment_dict = map_environment_to_environment_document(
        dynamo_enabled_project_environment_one
    )
    flagsmith_environment_table.put_item(
        Item=environment_dict,
    )

    flagsmith_environments_v2_table.put_item(
        Item=map_environment_to_environment_v2_document(
            dynamo_enabled_project_environment_one
        )
    )

    return environment_dict


@pytest.fixture()
def identity_override_document(
    flagsmith_environments_v2_table: Table,
    environment: Environment,
    feature: Feature,
) -> dict[str, typing.Any]:
    identity_uuid = str(uuid.uuid4())
    identifier = "identity-with-dynamo-override"

    identity_override = IdentityOverrideV2(
        environment_id=str(environment.id),
        environment_api_key=environment.api_key,
        document_key=get_environments_v2_identity_override_document_key(
            feature_id=feature.id, identity_uuid=identity_uuid
        ),
        identifier=identifier,
        identity_uuid=identity_uuid,
        feature_state=map_feature_state_to_engine(
            FeatureState(
                feature=feature,
                enabled=True,
                environment=environment,
            ),
        ),
    )

    identity_override_document = map_identity_override_to_identity_override_document(
        identity_override
    )
    flagsmith_environments_v2_table.put_item(Item=identity_override_document)
    return identity_override_document


@pytest.fixture()
def dynamo_environment_wrapper(
    flagsmith_environment_table: Table,
    settings: SettingsWrapper,
) -> DynamoEnvironmentWrapper:
    settings.ENVIRONMENTS_TABLE_NAME_DYNAMO = flagsmith_environment_table.name
    wrapper = DynamoEnvironmentWrapper()
    wrapper.table_name = flagsmith_environment_table.name
    return wrapper


@pytest.fixture()
def migrator(migrator_factory: MigratorFactory) -> Migrator:
    if settings.SKIP_MIGRATION_TESTS:  # pragma: no cover
        pytest.skip("Skip migration tests to speed up tests where necessary")
    migrator: Migrator = migrator_factory()
    return migrator


@pytest.fixture()
def analytics_migrator(migrator_factory: MigratorFactory) -> Migrator:
    if settings.SKIP_MIGRATION_TESTS:  # pragma: no cover
        pytest.skip("Skip migration tests to speed up tests where necessary")
    migrator: Migrator = migrator_factory("analytics")
    return migrator