class ElasticAppConfig(AppConfig):
    """AppConfig for Search3."""

    name = "elasticsearch_django"
    verbose_name = "Elasticsearch"

    def ready(self) -> None:
        """Validate config and connect signals."""
        super(ElasticAppConfig, self).ready()
        _validate_config(bool(settings.get_setting("strict_validation")))
        _connect_signals() def _connect_signals() -> None:
    """Connect up post_save, post_delete signals for models."""
    for index in settings.get_index_names():
        for model in settings.get_index_models(index):
            _connect_model_signals(model) def _delete_from_search_index(*, instance: SearchDocumentMixin, index: str) -> None:
    """Remove a document from a search index."""
    pre_delete.send(sender=instance.__class__, instance=instance, index=index)
    if settings.auto_sync(instance):
        instance.delete_search_document(index=index) def _on_model_delete(sender: type[Model], **kwargs: Any) -> None:
    """Remove documents from search indexes post_delete."""
    instance = kwargs.pop("instance")
    for index in instance.search_indexes:
        try:
            _delete_from_search_index(instance=instance, index=index)
        except Exception:  # noqa: B902
            logger.exception("Error handling 'on_delete' signal for %s", instance) def _on_model_save(sender: type[Model], **kwargs: Any) -> None:
    """Update document in search index post_save."""
    instance = kwargs.pop("instance")
    update_fields = kwargs.pop("update_fields")
    for index in instance.search_indexes:
        try:
            _update_search_index(
                instance=instance, index=index, update_fields=update_fields
            )
        except Exception:  # noqa: B902
            logger.exception("Error handling 'on_save' signal for %s", instance) def _update_search_index(
    *, instance: SearchDocumentMixin, index: str, update_fields: list[str]
) -> None:
    """Process index / update search index update actions."""
    if not _in_search_queryset(instance=instance, index=index):
        logger.debug(
            "Object (%r) is not in search queryset, ignoring update.", instance
        )
        return

    try:
        if update_fields:
            pre_update.send(
                sender=instance.__class__,
                instance=instance,
                index=index,
                update_fields=update_fields,
            )
            if settings.auto_sync(instance):
                instance.update_search_document(
                    index=index, update_fields=update_fields
                )
        else:
            pre_index.send(sender=instance.__class__, instance=instance, index=index)
            if settings.auto_sync(instance):
                instance.index_search_document(index=index)
    except Exception:  # noqa: B902
        logger.exception("Error handling 'post_save' signal for %s", instance) def _validate_config(strict: bool = False) -> None:
    """Validate settings.SEARCH_SETTINGS."""
    for index in settings.get_index_names():
        _validate_mapping(index, strict=strict)
        for model in settings.get_index_models(index):
            _validate_model(model)
    if settings.get_setting("update_strategy", "full") not in ["full", "partial"]:
        raise ImproperlyConfigured(
            "Invalid SEARCH_SETTINGS: 'update_strategy' value must be "
            "'full' or 'partial'."
        ) def _validate_mapping(index: str, strict: bool = False) -> None:
    """Check that an index mapping JSON file exists."""
    try:
        settings.get_index_mapping(index)
    except IOError:
        if strict:
            raise ImproperlyConfigured("Index '%s' has no mapping file." % index)
        else:
            logger.warning("Index '%s' has no mapping, relying on ES instead.", index) def _validate_model(model: Model) -> None:
    """Check that a model configured for an index subclasses the required classes."""
    if not hasattr(model, "as_search_document"):
        raise ImproperlyConfigured("'%s' must implement `as_search_document`." % model)
    if not hasattr(model.objects, "get_search_queryset"):
        raise ImproperlyConfigured(
            "'%s.objects must implement `get_search_queryset`." % model
        ) class SearchDocumentMixin(object):
    """
    Mixin used by models that are indexed for ES.

    This mixin defines the interface exposed by models that
    are indexed ready for ES. The only method that needs
    implementing is `as_search_document`.

    """

    # Django model field types that can be serialized directly into
    # a known format. All other types will need custom serialization.
    # Used by as_search_document_update method
    SIMPLE_UPDATE_FIELD_TYPES = [
        "ArrayField",
        "AutoField",
        "BooleanField",
        "CharField",
        "DateField",
        "DateTimeField",
        "DecimalField",
        "EmailField",
        "FloatField",
        "IntegerField",
        "TextField",
        "URLField",
        "UUIDField",
    ]

    @property
    def search_indexes(self) -> list[str]:
        """Return the list of indexes for which this model is configured."""
        return get_model_indexes(self.__class__)

    @property
    def search_document_cache_key(self) -> str:
        """Key used for storing search docs in local cache."""
        return "elasticsearch_django:{}.{}.{}".format(
            self._meta.app_label, self._meta.model_name, self.pk  # type: ignore
        )

    @property
    def search_doc_type(self) -> str:
        """Return the doc_type used for the model."""
        raise DeprecationWarning("Mapping types have been removed from ES7.x")

    def as_search_document(self, *, index: str) -> dict:
        """
        Return the object as represented in a named index.

        This is named to avoid confusion - if it was `get_search_document`,
        which would be the logical name, it would not be clear whether it
        referred to getting the local representation of the search document,
        or actually fetching it from the index.

        Kwargs:
            index: string, the name of the index in which the object is to
                appear - this allows different representations in different
                indexes. Defaults to '_all', in which case all indexes use
                the same search document structure.

        Returns a dictionary.

        """
        raise NotImplementedError(
            "{} does not implement 'as_search_document'.".format(
                self.__class__.__name__
            )
        )

    def _is_field_serializable(self, field_name: str) -> bool:
        """Return True if the field can be serialized into a JSON doc."""
        return (
            self._meta.get_field(field_name).get_internal_type()  # type: ignore
            in self.SIMPLE_UPDATE_FIELD_TYPES
        )

    def clean_update_fields(self, index: str, update_fields: list[str]) -> list[str]:
        """
        Clean the list of update_fields based on the index being updated.

        If any field in the update_fields list is not in the set of properties
        defined by the index mapping for this model, then we ignore it. If
        a field _is_ in the mapping, but the underlying model field is a
        related object, and thereby not directly serializable, then this
        method will raise a ValueError.

        """
        search_fields = get_model_index_properties(self, index)
        clean_fields = [f for f in update_fields if f in search_fields]
        ignore = [f for f in update_fields if f not in search_fields]
        if ignore:
            logger.debug(
                "Ignoring fields from partial update: %s",
                [f for f in update_fields if f not in search_fields],
            )
        for f in clean_fields:
            if not self._is_field_serializable(f):
                raise ValueError(
                    "'%s' cannot be automatically serialized into a search "
                    "document property. Please override as_search_document_update.",
                    f,
                )
        return clean_fields

    def as_search_document_update(
        self, *, index: str, update_fields: list[str]
    ) -> dict:
        """
        Return a partial update document based on which fields have been updated.

        If an object is saved with the `update_fields` argument passed
        through, then it is assumed that this is a 'partial update'. In
        this scenario we need a {property: value} dictionary containing
        just the fields we want to update.

        This method handles two possible update strategies - 'full' or 'partial'.
        The default 'full' strategy simply returns the value of `as_search_document`
        - thereby replacing the entire document each time. The 'partial' strategy is
        more intelligent - it will determine whether the fields passed are in the
        search document mapping, and return a partial update document that contains
        only those that are. In addition, if any field that _is_ included cannot
        be automatically serialized (e.g. a RelatedField object), then this method
        will raise a ValueError. In this scenario, you should override this method
        in your subclass.

        >>> def as_search_document_update(self, index, update_fields):
        ...     if 'user' in update_fields:
        ...         update_fields.remove('user')
        ...         doc = super().as_search_document_update(index, update_fields)
        ...         doc['user'] = self.user.get_full_name()
        ...         return doc
        ...     return super().as_search_document_update(index, update_fields)

        You may also wish to subclass this method to perform field-specific logic
        - in this example if only the timestamp is being saved, then ignore the
        update if the timestamp is later than a certain time.

        >>> def as_search_document_update(self, index, update_fields):
        ...     if update_fields == ['timestamp']:
        ...         if self.timestamp > today():
        ...            return {}
        ...     return super().as_search_document_update(index, update_fields)

        """
        if UPDATE_STRATEGY == UPDATE_STRATEGY_FULL:
            return self.as_search_document(index=index)

        if UPDATE_STRATEGY == UPDATE_STRATEGY_PARTIAL:
            # in partial mode we update the intersection of update_fields and
            # properties found in the mapping file.
            return {
                k: getattr(self, k)
                for k in self.clean_update_fields(
                    index=index, update_fields=update_fields
                )
            }

        raise ValueError("Invalid update strategy.")

    def as_search_action(self, *, index: str, action: str) -> dict:
        """
        Return an object as represented in a bulk api operation.

        Bulk API operations have a very specific format. This function will
        call the standard `as_search_document` method on the object and then
        wrap that up in the correct format for the action specified.

        https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html

        Args:
            index: string, the name of the index in which the action is to
                be taken. Bulk operations are only every carried out on a single
                index at a time.
            action: string ['index' | 'update' | 'delete'] - this decides
                how the final document is formatted.

        Returns a dictionary.

        """
        if action not in ("index", "update", "delete"):
            raise ValueError("Action must be 'index', 'update' or 'delete'.")

        document = {
            "_index": index,
            "_op_type": action,
            "_id": self.pk,  # type: ignore
        }

        if action == "index":
            document["_source"] = self.as_search_document(index=index)
        elif action == "update":
            document["doc"] = self.as_search_document(index=index)
        return document

    def fetch_search_document(self, *, index: str) -> dict:
        """Fetch the object's document from a search index by id."""
        if not self.pk:  # type: ignore
            raise ValueError("Object must have a primary key before being indexed.")
        client = get_client()
        return client.get(index=index, id=self.pk)  # type: ignore

    def index_search_document(self, *, index: str) -> None:
        """
        Create or replace search document in named index.

        Checks the local cache to see if the document has changed,
        and if not aborts the update, else pushes to ES, and then
        resets the local cache. Cache timeout is set as "cache_expiry"
        in the settings, and defaults to 60s.

        """
        cache_key = self.search_document_cache_key
        new_doc = self.as_search_document(index=index)
        cached_doc = cache.get(cache_key)
        if new_doc == cached_doc:
            logger.debug("Search document for %r is unchanged, ignoring update.", self)
            return
        cache.set(cache_key, new_doc, timeout=get_setting("cache_expiry", 60))
        get_client().index(index=index, body=new_doc, id=self.pk)  # type: ignore

    def update_search_document(self, *, index: str, update_fields: list[str]) -> None:
        """
        Partial update of a document in named index.

        Partial updates are invoked via a call to save the document
        with 'update_fields'. These fields are passed to the
        as_search_document method so that it can build a partial
        document. NB we don't just call as_search_document and then
        strip the fields _not_ in update_fields as we are trying
        to avoid possibly expensive operations in building the
        source document. The canonical example for this method
        is updating a single timestamp on a model - we don't want
        to have to walk the model relations and build a document
        in this case - we just want to push the timestamp.

        When POSTing a partial update the `as_search_document` doc
        must be passed to the `client.update` wrapped in a "doc" node,
        # noqa: E501, see: https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update.html

        """
        doc = self.as_search_document_update(index=index, update_fields=update_fields)
        if not doc:
            logger.debug("Ignoring object update as document is empty.")
            return
        retry_on_conflict = cast(int, get_setting("retry_on_conflict", 0))
        get_client().update(
            index=index,
            id=self.pk,  # type: ignore
            body={"doc": doc},
            retry_on_conflict=retry_on_conflict,
        )

    def delete_search_document(self, *, index: str) -> None:
        """Delete document from named index."""
        cache.delete(self.search_document_cache_key)
        get_client().delete(index=index, id=self.pk)  # type: ignore class ExampleModel(SearchDocumentMixin, models.Model):
    """Model class for use in tests."""

    simple_field_1 = models.IntegerField()
    simple_field_2 = models.CharField(max_length=100)
    complex_field = models.FileField()

    objects = ExampleModelManager()

    def as_search_document(self, index="_all"):
        return {
            "simple_field_1": self.simple_field_1,
            "simple_field_2": self.simple_field_2,
            "complex_field": str(self.complex_field),
        }
from unittest import mock
from django.core.exceptions import ImproperlyConfigured
from django.test import TestCase
from elasticsearch_django.apps import (
    ElasticAppConfig,
    _connect_signals,
    _delete_from_search_index,
    _on_model_delete,
    _on_model_save,
    _update_search_index,
    _validate_config,
    _validate_mapping,
    _validate_model,
)
from elasticsearch_django.models import SearchDocumentMixin
from .models import ExampleModel
import elasticsearch_django





class SearchAppsConfigTests(TestCase):
    """Tests for the apps module ready function."""

    @mock.patch("elasticsearch_django.apps.settings.get_setting", lambda x: False)
    @mock.patch("elasticsearch_django.apps._validate_config")
    @mock.patch("elasticsearch_django.apps._connect_signals")
    def test_ready(self, mock_signals, mock_config):
        """Test the AppConfig.ready method."""
        # mock_setting = True
        config = ElasticAppConfig("foo_bar", elasticsearch_django)
        config.ready()
        mock_config.assert_called_once_with(False)
        mock_signals.assert_called_once_with()


class SearchAppsValidationTests(TestCase):
    """Tests for the apps module validation functions."""

    def test__validate_model(self):
        """Test _validate_model function."""
        # 1. model doesn't implement as_search_document
        with mock.patch("tests.models.ExampleModel") as tm:
            del tm.as_search_document
            self.assertRaises(ImproperlyConfigured, _validate_model, tm)

        # 2. model.objects doesn't implement get_search_queryset
        with mock.patch("tests.models.ExampleModel") as tm:
            del tm.objects.get_search_queryset
            self.assertRaises(ImproperlyConfigured, _validate_model, tm)

        # model should pass
        with mock.patch("tests.models.ExampleModel") as tm:
            _validate_model(tm)

    @mock.patch("elasticsearch_django.apps.settings")
    def test__validate_mapping(self, mock_settings):
        """Test _validate_model function."""
        _validate_mapping("foo", strict=True)
        mock_settings.get_index_mapping.assert_called_once_with("foo")
        mock_settings.get_index_mapping.side_effect = IOError()
        self.assertRaises(ImproperlyConfigured, _validate_mapping, "foo", strict=True)
        # shouldn't raise error
        _validate_mapping("foo", strict=False)

    @mock.patch("elasticsearch_django.apps.settings")
    @mock.patch("elasticsearch_django.apps._validate_model")
    @mock.patch("elasticsearch_django.apps._validate_mapping")
    def test__validate_config(self, mock_mapping, mock_model, mock_settings):
        """Test _validate_model function."""
        mock_settings.get_index_names.return_value = ["foo"]
        mock_settings.get_setting.return_value = "full"
        mock_settings.get_index_models.return_value = [ExampleModel]
        _validate_config()
        mock_mapping.assert_called_once_with("foo", strict=False)
        mock_model.assert_called_once_with(ExampleModel)

    @mock.patch("elasticsearch_django.apps.settings")
    @mock.patch("elasticsearch_django.apps._validate_model")
    @mock.patch("elasticsearch_django.apps._validate_mapping")
    def test__validate_config_invalid_strategy(
        self, mock_mapping, mock_model, mock_settings
    ):
        """Test _validate_model function with an invalid update_strategy."""
        mock_settings.get_index_names.return_value = ["foo"]
        mock_settings.get_setting.return_value = "foo"
        mock_settings.get_index_models.return_value = [ExampleModel]
        self.assertRaises(ImproperlyConfigured, _validate_config)

    @mock.patch("elasticsearch_django.apps.signals")
    @mock.patch("elasticsearch_django.apps.settings")
    def test__connect_signals(self, mock_settings, mock_signals):
        """Test the _connect_signals function."""
        # this should connect up the signals once, for ExampleModel
        mock_settings.get_index_names.return_value = ["foo"]
        mock_settings.get_index_models.return_value = [ExampleModel]
        _connect_signals()
        mock_signals.post_save.connect.assert_called_once_with(
            _on_model_save, sender=ExampleModel, dispatch_uid="examplemodel.post_save"
        )
        mock_signals.post_delete.connect.assert_called_once_with(

            _on_model_delete,