Implementing integration tests for a Django application with Elastic Search.
Tags: django, elasticsearch, testing, cicd
Published: 2024-03-17 08:08:31

## Introduction

In the [first article](https://dev.to/robinmuhia/elastic-search-in-django-using-elasticsearchdsl-i00), we delved into how elastic search works under the hood.

In our [second article](https://dev.to/robinmuhia/implementation-of-elastic-search-in-django-32d6), we implemented elastic search in a Django application.

In this article, we will implement a CI/CD pipeline that can incorporate elastic search tests.

We will use the repository we cloned in our second article so catch up with those articles before proceeding with this blog.

## Structure

Ensure you have a tests folder that matches the one below;

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/6i2ua0530nop0ny78agm.png)

## Test set up
We will be using pytest to test thus we need a pytest.ini file to configure our test settings

```bash
[pytest]
DJANGO_SETTINGS_MODULE = tests.test_settings
addopts = --ds=tests.test_settings -v  --durations=10 --cache-clear --create-db
python_files = tests.py test_*.py
```
Our test_settings.py file from above has the following configurations;

```python
from elastic_search.config.settings import *  # noqa

ELASTICSEARCH_DSL_SIGNAL_PROCESSOR = "django_elasticsearch_dsl.signals.RealTimeSignalProcessor"
ELASTICSEARCH_DSL = {
    "default": {
        "hosts": ["http://localhost:9200"],
    },
}
```
We need to re-index documents with nested fields so we need the following utility function in our test file. It is located in elasticsearch_test.py file

```python
from django.core.management import call_command


class ElasticSearchMixin:
    def create_elasticsearch_index(self):
        """Create elastic search indices."""
        call_command("search_index", "--rebuild", "-f")
```

In our conftest.py file, we have the following method that is used to clear data in indices between tests;

```python
import pytest
from django.conf import settings
from elasticsearch import Elasticsearch


@pytest.fixture(autouse=True)
def clear_elasticsearch_indices():
    """
    Fixture to clear all Elasticsearch indices after each test case.
    """
    es = Elasticsearch(hosts=settings.ELASTICSEARCH_DSL["default"]["hosts"])

    indices = es.options().indices.get_alias(index="*").keys()

    for index_name in indices:
        es.options(ignore_status=[404, 400]).indices.delete(index=index_name)
    yield
```

Finally, our tests;
We first test to see if our documents are being properly indexed 
.
```python 
import pytest
from django.test import TestCase
from model_bakery import baker

from elastic_search.books.documents import BookDocument
from elastic_search.books.models import Author, Book, Country, Genre
from tests.utils.elasticsearch_test import ElasticSearchMixin

pytestmark = pytest.mark.django_db


@pytest.mark.django_db
class BookDocumentTest(TestCase, ElasticSearchMixin):
    def setUp(self):
        """Set up test suite."""
        self.create_elasticsearch_index()
        self.genre = baker.make(
            Genre,
            name="action",
        )
        self.author = baker.make(Author, name="Dave Muhia")
        self.country = baker.make(Country, name="Kenya")
        self.book = baker.make(
            Book,
            title="Edge of Tommorrow",
            description="Best movie",
            year=2014,
            rating=4.2,
            author=self.author,
            genre=self.genre,
            country=self.country,
        )

    def test_document_indexing(self):
        """Make sure document is properly indexed."""
        book_hits = BookDocument().search().execute()
        assert book_hits is not None
        book_hit = book_hits.hits[0]
        assert book_hit.genre.name == "action"
        assert book_hit.country.name == "Kenya"
        assert book_hit.author.name == "Dave Muhia"
        assert book_hit.title == "Edge of Tommorrow"
        assert book_hit.description == "Best movie"
        assert book_hit.year == 2014
        assert book_hit.rating == 4.2

    def test_get_queryset(self):
        queryset = BookDocument().get_queryset()
        assert len(queryset) == 1

    def test_get_instances_from_related(self):
        """Test retrieval of related models."""
        book_document = BookDocument()

        genre_instance = book_document.get_instances_from_related(self.genre)
        assert genre_instance is not None
        assert len(genre_instance) == 1
        assert genre_instance[0].title == self.book.title

        author_instance = book_document.get_instances_from_related(self.author)
        assert author_instance is not None
        assert len(author_instance) == 1
        assert author_instance[0].title == self.book.title

        country_instance = book_document.get_instances_from_related(self.country)
        assert country_instance is not None
        assert len(country_instance) == 1
        assert country_instance[0].title == self.book.title
```

Then we test our views;

```python
import pytest
from django.test import TestCase
from model_bakery import baker
from rest_framework.test import APITestCase

from elastic_search.books.models import Author, Book, Country, Genre
from tests.utils.elasticsearch_test import ElasticSearchMixin

pytestmark = pytest.mark.django_db


@pytest.mark.django_db
class BookDocumentTest(APITestCase, TestCase, ElasticSearchMixin):
    def setUp(self):
        """Set up test suite."""
        self.create_elasticsearch_index()
        self.genre = baker.make(
            Genre,
            name="action",
        )
        self.author = baker.make(Author, name="Dave Muhia")
        self.country = baker.make(Country, name="Kenya")
        self.book = baker.make(
            Book,
            title="Edge of Tommorrow",
            description="Best movie",
            year=2014,
            rating=4.2,
            author=self.author,
            genre=self.genre,
            country=self.country,
        )
        self.genre2 = baker.make(
            Genre,
            name="action",
        )
        self.author2 = baker.make(Author, name="John Connor")
        self.country2 = baker.make(Country, name="Kenya")
        self.book2 = baker.make(
            Book,
            title="Demon Slayer",
            description="Tengen Uzui is the GOAT. All your hashiras were killed.",
            year=2023,
            rating=5,
            author=self.author2,
            genre=self.genre2,
            country=self.country2,
        )

    def test_search_endpoint(self):
        """Test the search endpoint."""
        base_url = "/api/books/search/"
        # no search params
        response = self.client.get(base_url)
        assert response.status_code == 200
        assert response.data["count"] == 2
        assert response.data["results"][0]["title"] == self.book.title
        assert response.data["results"][1]["title"] == self.book2.title

        # test search
        response = self.client.get(base_url + "?search=Tommorrow")
        assert response.status_code == 200
        assert response.data["count"] == 1
        assert response.data["results"][0]["title"] == self.book.title

        # test fuzzy search on description
        response = self.client.get(base_url + "?search=kilted")
        assert response.status_code == 200
        assert response.data["count"] == 1
        assert response.data["results"][0]["title"] == self.book2.title

        # test filters
        response = self.client.get(base_url + "?search=a&year=2023")
        assert response.status_code == 200
        assert response.data["count"] == 1
        assert response.data["results"][0]["title"] == self.book2.title
```
### CI/CD

Instead of mocking, we want to be able to test our application against an actual elastic search instance. Hence the following CI pipeline.

```yaml
name: build and deploy

on:
  push:
    branches:
      - main
  pull_request:
    branches:
      - main

jobs:
  build:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:latest
        env:
          POSTGRES_DB: elastic
          POSTGRES_PASSWORD: elastic
          POSTGRES_USER: elastic
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
      elasticsearch_test:
        image: docker.elastic.co/elasticsearch/elasticsearch:8.10.2
        env:
          cluster.name: docker-cluster
          discovery.type: single-node
          bootstrap.memory_lock: "true"
          xpack.security.enabled: "false"
          ES_JAVA_OPTS: "-Xms256m -Xmx256m"
        ports:
          - 9200:9200
        options: --health-cmd="curl -f http://localhost:9200/_cluster/health || exit 1" --health-interval=10s --health-timeout=5s --health-retries=3

    steps:
      - name: Checkout code
        uses: actions/checkout@v2

      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: "3.11"

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt

      - name: Run tests
        run: |
          pytest
```

This configuration allows us to do so.

## Viola

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/9gey5an7ziz7v270895c.png)

## Conclusion

We have understood elastic search, implemented it and even added some tests to it ensuring that the workflow we intended is achieved.