在没有模拟的情况下在Django中测试Elasticsearch

介绍
这应该是一篇简短的文章-我最近在工作中继承了Django项目，其中Elasticsearch是主要组件。
在代码库中，有大量的辅助函数可构造如下所示的Elasticsearch查询DSL。

def search_media(query):
    """Example helper method to get movies and shows based on a search query
    """
    client = Elasticsearch(settings.ELASTICSEARCH_HOST)
    body = {
        "query": {"multi_match": {"query": query, "fields": ["title", "description"]}}
    }
    response = client.search(index=["movie", "show"], body=body)
    return [h["_source"] for h in response["hits"]["hits"]]

def search_media(query):

"""Example helper method to get movies and shows based on a search query

"""

client = Elasticsearch(settings.ELASTICSEARCH_HOST)

body = {

"query": {"multi_match": {"query": query, "fields": ["title", "description"]}}

}

response = client.search(index=["movie", "show"], body=body)

return [h["_source"] for h in response["hits"]["hits"]]

尽管上面的示例似乎很简单，但是Elasticsearch查询的JSON主体可能变得非常大而复杂。作为Elasticsearch的相对入门者，我几乎没有信心更改任何涉及这些辅助功能的代码。
虽然在代码库中存在预先存在的测试，但所有测试都模拟了这些Elasticsearch帮助器功能。

@mock.patch("path.to.some.elasticsearch_helper")
def test_with_mock(elasticsearch_helper_mock):
    elasticsearch_helper_mock.return_value = []
    results = get_search_results("Some query")
    assert results == []

@mock.patch("path.to.some.elasticsearch_helper")

def test_with_mock(elasticsearch_helper_mock):

elasticsearch_helper_mock.return_value = []

results = get_search_results("Some query")

assert results == []

模拟问题
这些是这些基于模拟的测试的问题。
当编写与数据库或分布式缓存接口的测试时，我发现模拟数据库/缓存往往会产生脆弱的测试，并提供很少的价值，因为它们通常无法捕捉到系统行为的细微差别。测试。
我们经常对这些功能进行的更改类型包括调整Elasticsearch DSL查询JSON，更改全局Elasticsearch配置，修改字段映射或升级Elasticsearch版本。在所有这些情况下，基于模拟的测试实际上并不能提供对错误的任何防护，因为它们（通过设计）绕过了所有与Elasticsearch相关的代码执行路径。
如果编写测试的目的是在更改代码时提高信心，那么我认为这些测试未能达到目标。
注意：我认为模拟是有用的，并且有其地位。在需要验证“管道”的测试中特别好。我个人认为它通常不适合需要数据库/缓存访问的测试。

编写理想的测试
因此，如果不是采用模拟方法，该怎么办？我们真正想要的是测试中的以下属性。
针对真实的ElasticSearch运行的测试。
独立且相互隔离的测试。
解决方案1似乎是不言自明的，我们需要针对正在运行的Elasticsearch集群运行测试。但是另一个重要的特性是我们的测试是独立的。测试A传递永远不会影响测试B的执行或输出。换句话说，我们不希望Elasticsearch状态在两次运行之间持久存在并且彼此影响。实际上，我不确定如何实现这一目标。
镜像现有模式
我很快意识到的一件事是，这实际上是数据库的一个已解决问题，因为在Django中，内置了支持针对数据库编写测试的支持。

import pytest

@pytest.mark.django_db(transaction=True)
def test_user():
    me = User.objects.get(username='me')
    assert me.is_superuser

import pytest

@pytest.mark.django_db(transaction=True)

def test_user():

me = User.objects.get(username='me')

assert me.is_superuser

使用由pytest.mark.django_db装饰器标记的测试，pytest将在测试主体之前和之后运行额外的设置并拆除逻辑。在较高级别上，它将：
设置一个临时测试数据库并在其上运行迁移。
运行测试。
拆除临时测试数据库。
因此，反映这些高级步骤，我们对Elasticsearch测试需要做的是以下高级步骤：
设置测试Elasticsearch集群并创建相关索引。
运行测试。
删除相关索引。
事实证明，我们可以使用pytest的Fixture系统和几行代码来实现该目标。
测试Elasticsearch
基础架构设置
第一步是将测试Elasticsearch集群和实际的Elasticsearch集群分开。这类似于Django的数据库测试，在Django中，它创建了一个以test_开头的独立数据库。我们可以通过使用docker-compose运行两个单独的实例来模仿类似的东西。

# docker-compose.yml

version: "3"

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.8.0
    environment:
      - cluster.name=docker-cluster
      - discovery.type=single-node
      - bootstrap.memory_lock=true
      - xpack.security.enabled=false
      - ES_JAVA_OPTS=-Xms256m -Xmx256m
    ulimits:
      memlock:
        soft: -1
        hard: -1
  elasticsearch_test:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.8.0
    environment:
      - cluster.name=docker-testing-cluster
      - discovery.type=single-node
      - bootstrap.memory_lock=true
      - xpack.security.enabled=false
      - ES_JAVA_OPTS=-Xms128m -Xmx128m
    ulimits:
      memlock:
        soft: -1
        hard: -1
  web:
    build: .
    command: poetry run python manage.py runserver 0.0.0.0:8000
    volumes:
      - .:/app/
    ports:
      - "8000:8000"
    depends_on:
      - elasticsearch
      - elasticsearch_test

# docker-compose.yml

version: "3"

services:

elasticsearch:

image: docker.elastic.co/elasticsearch/elasticsearch:7.8.0

environment:

- cluster.name=docker-cluster

- discovery.type=single-node

- bootstrap.memory_lock=true

- xpack.security.enabled=false

- ES_JAVA_OPTS=-Xms256m -Xmx256m

ulimits:

memlock:

soft: -1

hard: -1

elasticsearch_test:

image: docker.elastic.co/elasticsearch/elasticsearch:7.8.0

environment:

- cluster.name=docker-testing-cluster

- discovery.type=single-node

- bootstrap.memory_lock=true

- xpack.security.enabled=false

- ES_JAVA_OPTS=-Xms128m -Xmx128m

ulimits:

memlock:

soft: -1

hard: -1

web:

build: .

command: poetry run python manage.py runserver 0.0.0.0:8000

volumes:

- .:/app/

ports:

- "8000:8000"

depends_on:

- elasticsearch

- elasticsearch_test

这是通过运行与2个不同的服务相同的docker.elastic.co/elasticsearch/elasticsearch:7.8.0 docker映像来完成的，一个服务将由应用程序使用，一个称为elasticsearch，另一个将被专门用于测试的称为elasticsearch_test。。
虽然运行2 Elasticsearch集群似乎有些浪费，但请记住，这仅用于本地开发，并且可以通过诸如Discovery.type = single-node和ES_JAVA_OPTS之类的设置来缓解。
接下来，我们可以通过添加conftest.py来创建elasticsearch pytest固定装置。

# conftest.py

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

from example.elasticsearch import MOVIE_MAPPING, SHOW_MAPPING

schemas = {
    "movie": MOVIE_MAPPING,
    "show": SHOW_MAPPING,
}

ELASTICSEARCH_TEST_HOST = "http://elasticsearch_test:9200"


def setup_elasticsearch():
    es = Elasticsearch(ELASTICSEARCH_TEST_HOST)
    for index_name, schema in schemas.items():
        body = {
            "settings": {
                "number_of_shards": 1,
                "number_of_replicas": 1,
                "index.store.type": "mmapfs",
            },
            "mappings": schema,
        }
        es.indices.create(index=index_name, body=body)


def teardown_elasticsearch():
    es = Elasticsearch(ELASTICSEARCH_TEST_HOST)
    for index_name in schemas.keys():
        es.indices.delete(index=index_name)


@pytest.fixture
def elasticsearch(settings):
    settings.ELASTICSEARCH_HOST = ELASTICSEARCH_TEST_HOST

    setup_elasticsearch()
    yield Elasticsearch(ELASTICSEARCH_TEST_HOST)
    teardown_elasticsearch()

# conftest.py

import pytest

from django.conf import settings

from elasticsearch import Elasticsearch

from example.elasticsearch import MOVIE_MAPPING, SHOW_MAPPING

schemas = {

"movie": MOVIE_MAPPING,

"show": SHOW_MAPPING,

}

ELASTICSEARCH_TEST_HOST = "http://elasticsearch_test:9200"

def setup_elasticsearch():

es = Elasticsearch(ELASTICSEARCH_TEST_HOST)

for index_name, schema in schemas.items():

body = {

"settings": {

"number_of_shards": 1,

"number_of_replicas": 1,

"index.store.type": "mmapfs",

"mappings": schema,

}

es.indices.create(index=index_name, body=body)

def teardown_elasticsearch():

es = Elasticsearch(ELASTICSEARCH_TEST_HOST)

for index_name in schemas.keys():

es.indices.delete(index=index_name)

@pytest.fixture

def elasticsearch(settings):

settings.ELASTICSEARCH_HOST = ELASTICSEARCH_TEST_HOST

setup_elasticsearch()

yield Elasticsearch(ELASTICSEARCH_TEST_HOST)

teardown_elasticsearch()

如果我们看一下Elasticsearch夹具函数，我们可以看到它反映了我们描述的高级步骤。
我们称为setup_elasticsearch，它将通过适当的测试创建相关的索引。
我们屈服于测试机构。
我们将其称为teardown_elasticsearch，它会删除相关索引。
settings.ELASTICSEARCH_HOST = ELASTICSEARCH_TEST_HOST的额外内容是，确保我们在测试执行期间将settings.ELASTICSEARCH_HOST指向我们的测试Elasticsearch集群。
还要注意，setup_elasticsearch将特定于您要测试的应用程序。在这里，您将要保留全局Elasticsearch设置并创建与您的应用程序相关的索引。
编写测试
有了docker-compose配置和pytest固定装置，我们就可以编写测试了！
因此，这是我们一开始显示的search_media帮助器函数。

def search_media(query):
    """Example helper method to get movies and shows based on a search query
    """
    client = Elasticsearch(settings.ELASTICSEARCH_HOST)
    body = {
        "query": {"multi_match": {"query": query, "fields": ["title", "description"]}}
    }
    response = client.search(index=["movie", "show"], body=body)
    return [h["_source"] for h in response["hits"]["hits"]]

def search_media(query):

"""Example helper method to get movies and shows based on a search query

"""

client = Elasticsearch(settings.ELASTICSEARCH_HOST)

body = {

"query": {"multi_match": {"query": query, "fields": ["title", "description"]}}

}

response = client.search(index=["movie", "show"], body=body)

return [h["_source"] for h in response["hits"]["hits"]]

这是我们可以使用elasticsearch固定装置编写的示例测试。

from example.elasticsearch import search_media


def index_test_fixtures(es, index_name, data):
    created = es.index(index=index_name, body=data)
    assert created["result"] == "created"
    es.indices.refresh(index_name)


class TestElasticsearch:
    def test_elasticsearch(self, elasticsearch):
        # Setup test fixtures
        index_test_fixtures(
            elasticsearch,
            "movie",
            {
                "slug": "episode-5",
                "title": "Star Wars: Episode V - The Empire Strikes Back",
                "description": "After the Rebels are brutally overpowered by the Empire on the ice planet Hoth, Luke Skywalker begins Jedi training with Yoda, while his friends are pursued by Darth Vader and a bounty hunter named Boba Fett all over the galaxy.",
            },
        )
        index_test_fixtures(
            elasticsearch,
            "movie",
            {
                "slug": "episode-3",
                "title": "Star Wars: Episode III - Revenge of the Sith",
                "description": "Three years into the Clone Wars, the Jedi rescue Palpatine from Count Dooku. As Obi-Wan pursues a new threat, Anakin acts as a double agent between the Jedi Council and Palpatine and is lured into a sinister plan to rule the galaxy.",
            },
        )
        index_test_fixtures(
            elasticsearch,
            "movie",
            {
                "slug": "rouge-one",
                "title": "Rogue One: A Star Wars Story",
                "description": "The daughter of an Imperial scientist joins the Rebel Alliance in a risky move to steal the Death Star plans.",
            },
        )

        index_test_fixtures(
            elasticsearch,
            "show",
            {
                "slug": "clone-wars",
                "title": "Star Wars: The Clone Wars",
                "description": "Jedi Knights lead the Grand Army of the Republic against the droid army of the Separatists.",
            },
        )
        index_test_fixtures(
            elasticsearch,
            "show",
            {
                "slug": "the-mandalorian",
                "title": "The Mandalorian",
                "description": "The travels of a lone bounty hunter in the outer reaches of the galaxy, far from the authority of the New Republic.",
            },
        )

        # Test search helper
        results = search_media("Star Wars")
        results = {r["slug"] for r in results}
        assert results == {"episode-5", "episode-3", "rouge-one", "clone-wars"}

        results = search_media("Jedi")
        results = {r["slug"] for r in results}
        assert results == {"episode-5", "episode-3", "clone-wars"}

        results = search_media("Galaxy")
        results = {r["slug"] for r in results}
        assert results == {"episode-5", "episode-3", "the-mandalorian"}

from example.elasticsearch import search_media

def index_test_fixtures(es, index_name, data):

created = es.index(index=index_name, body=data)

assert created["result"] == "created"

es.indices.refresh(index_name)

class TestElasticsearch:

def test_elasticsearch(self, elasticsearch):

# Setup test fixtures

index_test_fixtures(

elasticsearch,

"movie",

{

"slug": "episode-5",

"title": "Star Wars: Episode V - The Empire Strikes Back",

"description": "After the Rebels are brutally overpowered by the Empire on the ice planet Hoth, Luke Skywalker begins Jedi training with Yoda, while his friends are pursued by Darth Vader and a bounty hunter named Boba Fett all over the galaxy.",

)

index_test_fixtures(

elasticsearch,

"movie",

{

"slug": "episode-3",

"title": "Star Wars: Episode III - Revenge of the Sith",

"description": "Three years into the Clone Wars, the Jedi rescue Palpatine from Count Dooku. As Obi-Wan pursues a new threat, Anakin acts as a double agent between the Jedi Council and Palpatine and is lured into a sinister plan to rule the galaxy.",

)

index_test_fixtures(

elasticsearch,

"movie",

{

"slug": "rouge-one",

"title": "Rogue One: A Star Wars Story",

"description": "The daughter of an Imperial scientist joins the Rebel Alliance in a risky move to steal the Death Star plans.",

)

index_test_fixtures(

elasticsearch,

"show",

{

"slug": "clone-wars",

"title": "Star Wars: The Clone Wars",

"description": "Jedi Knights lead the Grand Army of the Republic against the droid army of the Separatists.",

)

index_test_fixtures(

elasticsearch,

"show",

{

"slug": "the-mandalorian",

"title": "The Mandalorian",

"description": "The travels of a lone bounty hunter in the outer reaches of the galaxy, far from the authority of the New Republic.",

)

# Test search helper

results = search_media("Star Wars")

results = {r["slug"] for r in results}

assert results == {"episode-5", "episode-3", "rouge-one", "clone-wars"}

results = search_media("Jedi")

results = {r["slug"] for r in results}

assert results == {"episode-5", "episode-3", "clone-wars"}

results = search_media("Galaxy")

results = {r["slug"] for r in results}

assert results == {"episode-5", "episode-3", "the-mandalorian"}

最重要的一点是测试功能签名中的elasticsearch参数：def test_elasticsearch（self，elasticsearch）:。这是pytest的提示，我们希望该特定测试执行我们先前定义的elasticsearch固定装置。
像大多数测试一样，我们需要首先设置一些测试数据。在可用Elasticsearch对象的情况下，我们可以调用.index将一些资源持久保存到我们的elasticsearch_test集群中。在此特定示例中，我们将为《星球大战》的电影和节目建立索引。
最后，我们可以调用被测试的辅助函数search_media的返回值并进行断言。这里的好处是该测试符合我们最初的目标：它在真实的Elasticsearch集群上运行，并且与其他测试隔离！
结论
我希望这对那些将Elasticsearch纳入其技术堆栈的人有所帮助。请注意，尽管我使用Django和pytest来实现此示例，但是本文中讨论的思想和模式可以很容易地扩展到您选择的网络框架/语言。
测试愉快！

原文：https://yanglinzhao.com/posts/test-elasticsearch-in-django/

M	T	W	T	F	S	S
« Jun
1	2	3	4	5	6	7
8	9	10	11	12	13	14
15	16	17	18	19	20	21
22	23	24	25	26	27	28
29	30

头脑的思考

头脑的思考

在没有模拟的情况下在Django中测试Elasticsearch