mirror of
https://github.com/graphql-python/graphene-django.git
synced 2025-07-11 08:42:32 +03:00
Issue #1111: foreign key should also call get_queryset method
This commit is contained in:
parent
e0a5d1c58e
commit
095cbf1cf6
|
@ -254,7 +254,24 @@ def convert_field_to_djangomodel(field, registry=None):
|
||||||
if not _type:
|
if not _type:
|
||||||
return
|
return
|
||||||
|
|
||||||
return Field(_type, description=field.help_text, required=not field.null)
|
class CustomField(Field):
|
||||||
|
def get_resolver(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(CustomField, self).get_resolver(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=field.help_text, required=not field.null)
|
||||||
|
|
||||||
return Dynamic(dynamic_type)
|
return Dynamic(dynamic_type)
|
||||||
|
|
||||||
|
|
|
@ -13,6 +13,9 @@ class Person(models.Model):
|
||||||
class Pet(models.Model):
|
class Pet(models.Model):
|
||||||
name = models.CharField(max_length=30)
|
name = models.CharField(max_length=30)
|
||||||
age = models.PositiveIntegerField()
|
age = models.PositiveIntegerField()
|
||||||
|
owner = models.ForeignKey(
|
||||||
|
"Person", null=True, blank=True, on_delete=models.CASCADE, related_name="pets"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class FilmDetails(models.Model):
|
class FilmDetails(models.Model):
|
||||||
|
@ -91,8 +94,8 @@ class CNNReporter(Reporter):
|
||||||
|
|
||||||
class Article(models.Model):
|
class Article(models.Model):
|
||||||
headline = models.CharField(max_length=100)
|
headline = models.CharField(max_length=100)
|
||||||
pub_date = models.DateField()
|
pub_date = models.DateField(auto_now_add=True)
|
||||||
pub_date_time = models.DateTimeField()
|
pub_date_time = models.DateTimeField(auto_now_add=True)
|
||||||
reporter = models.ForeignKey(
|
reporter = models.ForeignKey(
|
||||||
Reporter, on_delete=models.CASCADE, related_name="articles"
|
Reporter, on_delete=models.CASCADE, related_name="articles"
|
||||||
)
|
)
|
||||||
|
|
355
graphene_django/tests/test_get_queryset.py
Normal file
355
graphene_django/tests/test_get_queryset.py
Normal 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"}}]},
|
||||||
|
}
|
|
@ -15,7 +15,7 @@ from ..compat import IntegerRangeField, MissingType
|
||||||
from ..fields import DjangoConnectionField
|
from ..fields import DjangoConnectionField
|
||||||
from ..types import DjangoObjectType
|
from ..types import DjangoObjectType
|
||||||
from ..utils import DJANGO_FILTER_INSTALLED
|
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():
|
def test_should_query_only_fields():
|
||||||
|
@ -247,8 +247,8 @@ def test_should_node():
|
||||||
|
|
||||||
|
|
||||||
def test_should_query_onetoone_fields():
|
def test_should_query_onetoone_fields():
|
||||||
film = Film(id=1)
|
film = Film.objects.create(id=1)
|
||||||
film_details = FilmDetails(id=1, film=film)
|
film_details = FilmDetails.objects.create(id=1, film=film)
|
||||||
|
|
||||||
class FilmNode(DjangoObjectType):
|
class FilmNode(DjangoObjectType):
|
||||||
class Meta:
|
class Meta:
|
||||||
|
@ -1314,14 +1314,12 @@ def test_should_preserve_prefetch_related(django_assert_num_queries):
|
||||||
class ReporterType(DjangoObjectType):
|
class ReporterType(DjangoObjectType):
|
||||||
class Meta:
|
class Meta:
|
||||||
model = Reporter
|
model = Reporter
|
||||||
interfaces = (graphene.relay.Node,)
|
interfaces = (Node,)
|
||||||
|
|
||||||
class FilmType(DjangoObjectType):
|
class FilmType(DjangoObjectType):
|
||||||
reporters = DjangoConnectionField(ReporterType)
|
|
||||||
|
|
||||||
class Meta:
|
class Meta:
|
||||||
model = Film
|
model = Film
|
||||||
interfaces = (graphene.relay.Node,)
|
interfaces = (Node,)
|
||||||
|
|
||||||
class Query(graphene.ObjectType):
|
class Query(graphene.ObjectType):
|
||||||
films = DjangoConnectionField(FilmType)
|
films = DjangoConnectionField(FilmType)
|
||||||
|
@ -1552,3 +1550,68 @@ def test_connection_should_allow_offset_filtering_with_after():
|
||||||
"allReporters": {"edges": [{"node": {"firstName": "Jane", "lastName": "Roe"}},]}
|
"allReporters": {"edges": [{"node": {"firstName": "Jane", "lastName": "Roe"}},]}
|
||||||
}
|
}
|
||||||
assert result.data == expected
|
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"}],
|
||||||
|
}
|
||||||
|
# assert False
|
||||||
|
|
Loading…
Reference in New Issue
Block a user