Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_same_type_resolver_maps_are_merged_into_executable_schema():
type_defs = """
type Query {
hello: String
test(data: Int): Boolean
}
"""
query = QueryType()
query.set_field("hello", lambda *_: "World!")
extending_query = QueryType()
@extending_query.field("test")
def resolve_test(*_, data): # pylint: disable=unused-variable
assert data == 4
return True
schema = make_executable_schema(type_defs, [query, extending_query])
result = graphql_sync(schema, "{ hello test(data: 4) }")
assert result.errors is None
assert result.data == {"hello": "World!", "test": True}
def test_directive_can_have_optional_argument():
type_defs = """
directive @test(arg: String) on FIELD_DEFINITION
type Query {
hello: String @test
}
"""
query = QueryType()
query.set_field("hello", lambda *_: "hello world")
schema = make_executable_schema(
type_defs, [query], directives={"test": ReturnValueDirective}
)
result = graphql_sync(schema, "{ hello }")
assert result.errors is None
assert result.data == {"hello": None}
USER
UNKNOWN
}
type User @auth(requires: USER) {
name: String
banned: Boolean @auth(requires: ADMIN)
canPost: Boolean @auth(requires: REVIEWER)
}
type Query {
users: [User]
}
"""
query = QueryType()
@query.field("users")
def _users_resolver(_, __):
return [{"banned": True, "canPost": False, "name": "Ben"}]
schema = make_executable_schema(
type_defs, [query], directives={"auth": AuthDirective}
)
def exec_with_role(role: str):
return graphql_sync(
schema,
"""
query {
users {
name
def federated_manager_():
sdl = StringIO(SDL)
mock_open = MagicMock(return_value=sdl)
with patch("builtins.open", mock_open):
query_type = QueryType()
manager = FederatedManager(
schema_sdl_file="/tmp/schema.graphql", query=query_type
)
yield manager
class UniqueIDDirective(SchemaDirectiveVisitor):
def visit_object(self, object_: GraphQLObjectType) -> GraphQLObjectType:
name, from_ = self.args.values()
def _field_resolver(field, _):
hash_ = hashlib.sha1(object_.name.encode())
for field_name in from_:
hash_.update(str(field[field_name]).encode())
return hash_.hexdigest()
object_.fields[name] = GraphQLField(
description="Unique ID", type_=GraphQLID, resolve=_field_resolver
)
query = QueryType()
query.set_field("people", lambda *_: [{"personID": 1, "name": "Ben"}])
query.set_field(
"locations", lambda *_: [{"locationID": 1, "address": "140 10th St"}]
)
schema = make_executable_schema(
type_defs, [query], directives={"uniqueID": UniqueIDDirective}
)
result = graphql_sync(
schema,
"""
{
people {
uid
personID
name
def test_unsuccesfull_invalid_enum_value_evaluation():
query = QueryType()
query.set_field("testEnum", lambda *_: "INVALID")
schema = make_executable_schema([enum_definition, enum_field], query)
result = graphql_sync(schema, "{ testEnum }")
assert result.errors is not None
def test_different_types_resolver_maps_are_merged_into_executable_schema():
type_defs = """
type Query {
user: User
}
type User {
username: String
}
"""
query = QueryType()
query.set_field("user", lambda *_: Mock(first_name="Joe"))
user = ObjectType("User")
user.set_alias("username", "first_name")
schema = make_executable_schema(type_defs, [query, user])
result = graphql_sync(schema, "{ user { username } }")
assert result.errors is None
assert result.data == {"user": {"username": "Joe"}}
def test_int_enum_is_resolved_from_internal_value():
query = QueryType()
query.set_field("testEnum", lambda *_: 1980)
schema = make_executable_schema([enum_definition, enum_field], [query, int_enum])
result = graphql_sync(schema, "{ testEnum }")
assert result.data["testEnum"] == "EMPIRE"
# This is our schema
type_defs = gql(
"""
type Book {
title: String!
year: Int!
age: Int!
}
type Query {
books: [Book!]!
}
"""
)
query = QueryType() # our Query type
@query.field("books") # Query.books
def resolve_books(*_):
return [
{"title": "The Color of Magic", "year": 1983},
{"title": "The Light Fantastic", "year": 1986},
{"title": "Equal Rites", "year": 1987},
]
book = ObjectType("Book")
@book.field("age")
def resolve_book_age(book, *_):
from datetime import timedelta
from ariadne import QueryType
from django.contrib.auth import get_user_model
from django.core.cache import cache
from django.utils import timezone
from ...threads.models import Attachment, Post, Thread
from ...users.models import DataDownload, DeletedUser
CACHE_KEY = "misago_admin_analytics"
CACHE_LENGTH = 3600 * 4 # 4 hours
User = get_user_model()
analytics = QueryType()
@analytics.field("analytics")
def resolve_analytics(_, info, *, span):
span = clean_span(span)
cache_key = "%s_%s" % (CACHE_KEY, span)
data = cache.get(cache_key)
if not data:
data = get_data_from_db(span)
cache.set(cache_key, data, CACHE_LENGTH)
return data
def clean_span(span):
if span > 360:
return 360