mirror of
https://github.com/graphql-python/graphene-django.git
synced 2025-12-07 10:14:21 +03:00
91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
from datetime import datetime
|
|
|
|
import pytest
|
|
from mock import mock
|
|
|
|
from graphene import Schema
|
|
|
|
from graphene_django.elasticsearch.filter import filters
|
|
from graphene_django.tests.models import Article, Reporter
|
|
from graphene_django.utils import DJANGO_FILTER_INSTALLED, DJANGO_ELASTICSEARCH_DSL_INSTALLED
|
|
from graphene_django.elasticsearch.tests.filters import ESFilterQuery, ArticleDocument
|
|
|
|
pytestmark = []
|
|
|
|
if not DJANGO_FILTER_INSTALLED or not DJANGO_ELASTICSEARCH_DSL_INSTALLED:
|
|
pytestmark.append(
|
|
pytest.mark.skipif(
|
|
True, reason="django_filters not installed or not compatible"
|
|
)
|
|
)
|
|
|
|
pytestmark.append(pytest.mark.django_db)
|
|
|
|
|
|
def fake_data():
|
|
r1 = Reporter.objects.create(first_name="r1", last_name="r1", email="r1@test.com")
|
|
a1 = Article.objects.create(
|
|
headline="a1",
|
|
pub_date=datetime.now(),
|
|
pub_date_time=datetime.now(),
|
|
reporter=r1,
|
|
editor=r1,
|
|
)
|
|
a2 = Article.objects.create(
|
|
headline="a2",
|
|
pub_date=datetime.now(),
|
|
pub_date_time=datetime.now(),
|
|
reporter=r1,
|
|
editor=r1,
|
|
)
|
|
return a1, a2
|
|
|
|
|
|
def filter_generation(field, query_str, spected_arguments):
|
|
a1, a2 = fake_data()
|
|
|
|
query = """
|
|
query {
|
|
%s(%s) {
|
|
edges {
|
|
node {
|
|
headline
|
|
}
|
|
}
|
|
}
|
|
}
|
|
""" % (field, query_str)
|
|
|
|
mock_count = mock.Mock(return_value=3)
|
|
mock_slice = mock.Mock(return_value=mock.Mock(to_queryset=mock.Mock(
|
|
return_value=Article.objects.filter(pk__in=[a1.id, a2.id])
|
|
)))
|
|
mock_query = mock.Mock(return_value=ArticleDocument.search())
|
|
|
|
with mock.patch('django_elasticsearch_dsl.search.Search.count', mock_count),\
|
|
mock.patch('django_elasticsearch_dsl.search.Search.__getitem__', mock_slice),\
|
|
mock.patch('elasticsearch_dsl.Search.query', mock_query):
|
|
|
|
schema = Schema(query=ESFilterQuery)
|
|
result = schema.execute(query)
|
|
|
|
assert not result.errors
|
|
|
|
mock_query.assert_called_with(filters.StringFilterES(attr='headline').get_q(spected_arguments))
|
|
|
|
assert len(result.data[field]["edges"]) == 2
|
|
assert result.data[field]["edges"][0]["node"]["headline"] == "a1"
|
|
assert result.data[field]["edges"][1]["node"]["headline"] == "a2"
|
|
|
|
|
|
def test_filter_as_field():
|
|
filter_generation("articlesAsField", "headline: \"A text\"", {"headline": "A text"})
|
|
|
|
|
|
def test_filter_in_meta():
|
|
filter_generation("articlesInMeta", "headline: \"A text\"", {"headline": "A text"})
|
|
|
|
|
|
def test_filter_in_meta_dict():
|
|
filter_generation("articlesInMetaDict", "headline: \"A text\"", {"headline": "A text"})
|