Introduction
In the first article, we delved into how elastic search works under the hood.
In our second article, 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;
Test set up
We will be using pytest to test thus we need a pytest.ini file to configure our test settings
[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;
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
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;
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
.
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;
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.
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
Conclusion
We have understood elastic search, implemented it and even added some tests to it ensuring that the workflow we intended is achieved.
Top comments (0)