fix: unit test for graphene pr#1412 (#1315)

* Issue #1111: foreign key should also call get_queryset method

* fix: test for graphene PR https://github.com/graphql-python/graphene/pull/1412

Co-authored-by: Thomas Leonard <thomas@loftorbital.com>
This commit is contained in:
Thomas Leonard 2022-09-23 09:45:02 +01:00 committed by GitHub
parent b2f83eb277
commit 5d81ba04f9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 445 additions and 5 deletions

View File

@ -308,7 +308,24 @@ def convert_field_to_djangomodel(field, registry=None):
if not _type:
return
return Field(
class CustomField(Field):
def wrap_resolve(self, parent_resolver):
"""
Implements a custom resolver which go through the `get_node` method to insure that
it goes through the `get_queryset` method of the DjangoObjectType.
"""
resolver = super().wrap_resolve(parent_resolver)
def custom_resolver(root, info, **args):
fk_obj = resolver(root, info, **args)
if fk_obj is None:
return None
else:
return _type.get_node(info, fk_obj.pk)
return custom_resolver
return CustomField(
_type,
description=get_django_field_description(field),
required=not field.null,

View File

@ -13,6 +13,9 @@ class Person(models.Model):
class Pet(models.Model):
name = models.CharField(max_length=30)
age = models.PositiveIntegerField()
owner = models.ForeignKey(
"Person", on_delete=models.CASCADE, null=True, blank=True, related_name="pets"
)
class FilmDetails(models.Model):

View File

@ -0,0 +1,355 @@
import pytest
import graphene
from graphene.relay import Node
from graphql_relay import to_global_id
from ..fields import DjangoConnectionField
from ..types import DjangoObjectType
from .models import Article, Reporter
class TestShouldCallGetQuerySetOnForeignKey:
"""
Check that the get_queryset method is called in both forward and reversed direction
of a foreignkey on types.
(see issue #1111)
"""
@pytest.fixture(autouse=True)
def setup_schema(self):
class ReporterType(DjangoObjectType):
class Meta:
model = Reporter
@classmethod
def get_queryset(cls, queryset, info):
if info.context and info.context.get("admin"):
return queryset
raise Exception("Not authorized to access reporters.")
class ArticleType(DjangoObjectType):
class Meta:
model = Article
@classmethod
def get_queryset(cls, queryset, info):
return queryset.exclude(headline__startswith="Draft")
class Query(graphene.ObjectType):
reporter = graphene.Field(ReporterType, id=graphene.ID(required=True))
article = graphene.Field(ArticleType, id=graphene.ID(required=True))
def resolve_reporter(self, info, id):
return (
ReporterType.get_queryset(Reporter.objects, info)
.filter(id=id)
.last()
)
def resolve_article(self, info, id):
return (
ArticleType.get_queryset(Article.objects, info).filter(id=id).last()
)
self.schema = graphene.Schema(query=Query)
self.reporter = Reporter.objects.create(first_name="Jane", last_name="Doe")
self.articles = [
Article.objects.create(
headline="A fantastic article",
reporter=self.reporter,
editor=self.reporter,
),
Article.objects.create(
headline="Draft: My next best seller",
reporter=self.reporter,
editor=self.reporter,
),
]
def test_get_queryset_called_on_field(self):
# If a user tries to access an article it is fine as long as it's not a draft one
query = """
query getArticle($id: ID!) {
article(id: $id) {
headline
}
}
"""
# Non-draft
result = self.schema.execute(query, variables={"id": self.articles[0].id})
assert not result.errors
assert result.data["article"] == {
"headline": "A fantastic article",
}
# Draft
result = self.schema.execute(query, variables={"id": self.articles[1].id})
assert not result.errors
assert result.data["article"] is None
# If a non admin user tries to access a reporter they should get our authorization error
query = """
query getReporter($id: ID!) {
reporter(id: $id) {
firstName
}
}
"""
result = self.schema.execute(query, variables={"id": self.reporter.id})
assert len(result.errors) == 1
assert result.errors[0].message == "Not authorized to access reporters."
# An admin user should be able to get reporters
query = """
query getReporter($id: ID!) {
reporter(id: $id) {
firstName
}
}
"""
result = self.schema.execute(
query, variables={"id": self.reporter.id}, context_value={"admin": True},
)
assert not result.errors
assert result.data == {"reporter": {"firstName": "Jane"}}
def test_get_queryset_called_on_foreignkey(self):
# If a user tries to access a reporter through an article they should get our authorization error
query = """
query getArticle($id: ID!) {
article(id: $id) {
headline
reporter {
firstName
}
}
}
"""
result = self.schema.execute(query, variables={"id": self.articles[0].id})
assert len(result.errors) == 1
assert result.errors[0].message == "Not authorized to access reporters."
# An admin user should be able to get reporters through an article
query = """
query getArticle($id: ID!) {
article(id: $id) {
headline
reporter {
firstName
}
}
}
"""
result = self.schema.execute(
query, variables={"id": self.articles[0].id}, context_value={"admin": True},
)
assert not result.errors
assert result.data["article"] == {
"headline": "A fantastic article",
"reporter": {"firstName": "Jane"},
}
# An admin user should not be able to access draft article through a reporter
query = """
query getReporter($id: ID!) {
reporter(id: $id) {
firstName
articles {
headline
}
}
}
"""
result = self.schema.execute(
query, variables={"id": self.reporter.id}, context_value={"admin": True},
)
assert not result.errors
assert result.data["reporter"] == {
"firstName": "Jane",
"articles": [{"headline": "A fantastic article"}],
}
class TestShouldCallGetQuerySetOnForeignKeyNode:
"""
Check that the get_queryset method is called in both forward and reversed direction
of a foreignkey on types using a node interface.
(see issue #1111)
"""
@pytest.fixture(autouse=True)
def setup_schema(self):
class ReporterType(DjangoObjectType):
class Meta:
model = Reporter
interfaces = (Node,)
@classmethod
def get_queryset(cls, queryset, info):
if info.context and info.context.get("admin"):
return queryset
raise Exception("Not authorized to access reporters.")
class ArticleType(DjangoObjectType):
class Meta:
model = Article
interfaces = (Node,)
@classmethod
def get_queryset(cls, queryset, info):
return queryset.exclude(headline__startswith="Draft")
class Query(graphene.ObjectType):
reporter = Node.Field(ReporterType)
article = Node.Field(ArticleType)
self.schema = graphene.Schema(query=Query)
self.reporter = Reporter.objects.create(first_name="Jane", last_name="Doe")
self.articles = [
Article.objects.create(
headline="A fantastic article",
reporter=self.reporter,
editor=self.reporter,
),
Article.objects.create(
headline="Draft: My next best seller",
reporter=self.reporter,
editor=self.reporter,
),
]
def test_get_queryset_called_on_node(self):
# If a user tries to access an article it is fine as long as it's not a draft one
query = """
query getArticle($id: ID!) {
article(id: $id) {
headline
}
}
"""
# Non-draft
result = self.schema.execute(
query, variables={"id": to_global_id("ArticleType", self.articles[0].id)}
)
assert not result.errors
assert result.data["article"] == {
"headline": "A fantastic article",
}
# Draft
result = self.schema.execute(
query, variables={"id": to_global_id("ArticleType", self.articles[1].id)}
)
assert not result.errors
assert result.data["article"] is None
# If a non admin user tries to access a reporter they should get our authorization error
query = """
query getReporter($id: ID!) {
reporter(id: $id) {
firstName
}
}
"""
result = self.schema.execute(
query, variables={"id": to_global_id("ReporterType", self.reporter.id)}
)
assert len(result.errors) == 1
assert result.errors[0].message == "Not authorized to access reporters."
# An admin user should be able to get reporters
query = """
query getReporter($id: ID!) {
reporter(id: $id) {
firstName
}
}
"""
result = self.schema.execute(
query,
variables={"id": to_global_id("ReporterType", self.reporter.id)},
context_value={"admin": True},
)
assert not result.errors
assert result.data == {"reporter": {"firstName": "Jane"}}
def test_get_queryset_called_on_foreignkey(self):
# If a user tries to access a reporter through an article they should get our authorization error
query = """
query getArticle($id: ID!) {
article(id: $id) {
headline
reporter {
firstName
}
}
}
"""
result = self.schema.execute(
query, variables={"id": to_global_id("ArticleType", self.articles[0].id)}
)
assert len(result.errors) == 1
assert result.errors[0].message == "Not authorized to access reporters."
# An admin user should be able to get reporters through an article
query = """
query getArticle($id: ID!) {
article(id: $id) {
headline
reporter {
firstName
}
}
}
"""
result = self.schema.execute(
query,
variables={"id": to_global_id("ArticleType", self.articles[0].id)},
context_value={"admin": True},
)
assert not result.errors
assert result.data["article"] == {
"headline": "A fantastic article",
"reporter": {"firstName": "Jane"},
}
# An admin user should not be able to access draft article through a reporter
query = """
query getReporter($id: ID!) {
reporter(id: $id) {
firstName
articles {
edges {
node {
headline
}
}
}
}
}
"""
result = self.schema.execute(
query,
variables={"id": to_global_id("ReporterType", self.reporter.id)},
context_value={"admin": True},
)
assert not result.errors
assert result.data["reporter"] == {
"firstName": "Jane",
"articles": {"edges": [{"node": {"headline": "A fantastic article"}}]},
}

View File

@ -15,7 +15,7 @@ from ..compat import IntegerRangeField, MissingType
from ..fields import DjangoConnectionField
from ..types import DjangoObjectType
from ..utils import DJANGO_FILTER_INSTALLED
from .models import Article, CNNReporter, Film, FilmDetails, Reporter
from .models import Article, CNNReporter, Film, FilmDetails, Person, Pet, Reporter
def test_should_query_only_fields():
@ -251,8 +251,8 @@ def test_should_node():
def test_should_query_onetoone_fields():
film = Film(id=1)
film_details = FilmDetails(id=1, film=film)
film = Film.objects.create(id=1)
film_details = FilmDetails.objects.create(id=1, film=film)
class FilmNode(DjangoObjectType):
class Meta:
@ -1697,3 +1697,67 @@ def test_connection_should_succeed_if_last_higher_than_number_of_objects():
}
}
assert result.data == expected
def test_should_query_nullable_foreign_key():
class PetType(DjangoObjectType):
class Meta:
model = Pet
class PersonType(DjangoObjectType):
class Meta:
model = Person
class Query(graphene.ObjectType):
pet = graphene.Field(PetType, name=graphene.String(required=True))
person = graphene.Field(PersonType, name=graphene.String(required=True))
def resolve_pet(self, info, name):
return Pet.objects.filter(name=name).first()
def resolve_person(self, info, name):
return Person.objects.filter(name=name).first()
schema = graphene.Schema(query=Query)
person = Person.objects.create(name="Jane")
pets = [
Pet.objects.create(name="Stray dog", age=1),
Pet.objects.create(name="Jane's dog", owner=person, age=1),
]
query_pet = """
query getPet($name: String!) {
pet(name: $name) {
owner {
name
}
}
}
"""
result = schema.execute(query_pet, variables={"name": "Stray dog"})
assert not result.errors
assert result.data["pet"] == {
"owner": None,
}
result = schema.execute(query_pet, variables={"name": "Jane's dog"})
assert not result.errors
assert result.data["pet"] == {
"owner": {"name": "Jane"},
}
query_owner = """
query getOwner($name: String!) {
person(name: $name) {
pets {
name
}
}
}
"""
result = schema.execute(query_owner, variables={"name": "Jane"})
assert not result.errors
assert result.data["person"] == {
"pets": [{"name": "Jane's dog"}],
}

View File

@ -59,7 +59,8 @@ setup(
keywords="api graphql protocol rest relay graphene",
packages=find_packages(exclude=["tests", "examples", "examples.*"]),
install_requires=[
"graphene>=3.0,<4",
# "graphene>=3.0,<4",
"graphene @ git+https://github.com/loft-orbital/graphene.git@loft-v3-1.0#egg=graphene",
"graphql-core>=3.1.0,<4",
"graphql-relay>=3.1.1,<4",
"Django>=3.2",