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_custom_resolver_is_called_with_arguments_passed_with_query():
type_defs = """
type Query {
test(returnValue: Int!): Int
}
"""
query = ResolverMap("Query")
@query.field("test")
def resolve_test(*_, returnValue): # pylint: disable=unused-variable
assert returnValue == 4
return "42"
schema = make_executable_schema(type_defs, query)
result = graphql_sync(schema, "{ test(returnValue: 4) }")
assert result.errors is None
assert result.data == {"test": 42}
def test_custom_resolver_is_called_with_input_type_value_as_dict():
type_defs = """
type Query {
test(data: TestInput): Int
}
input TestInput {
value: Int
}
"""
query = ResolverMap("Query")
@query.field("test")
def resolve_test(*_, data): # pylint: disable=unused-variable
assert data == {"value": 4}
return "42"
schema = make_executable_schema(type_defs, query)
result = graphql_sync(schema, "{ test(data: { value: 4 }) }")
assert result.errors is None
assert result.data == {"test": 42}
def test_executing_mutation_takes_scalar_args_and_returns_scalar_sum():
type_defs = """
type Query {
_: String
}
type Mutation {
sum(a: Int, b: Int): Int
}
"""
mutation = ResolverMap("Mutation")
mutation.field("sum")(lambda *_, a, b: a + b)
schema = make_executable_schema(type_defs, mutation)
result = graphql_sync(schema, "mutation { sum(a: 1, b: 2) }")
assert result.errors is None
assert result.data == {"sum": 3}
def test_custom_resolver_is_called_to_resolve_custom_type_field_value():
type_defs = """
type Query {
test: Custom
}
type Custom {
node: String
}
"""
query = ResolverMap("Query")
query.field("test")(lambda *_: {"node": "custom"})
custom = ResolverMap("Custom")
custom.field("node")(lambda *_: "deep")
schema = make_executable_schema(type_defs, [query, custom])
result = graphql_sync(schema, "{ test { node } }")
assert result.errors is None
assert result.data == {"test": {"node": "deep"}}
def test_default_resolver_resolves_value_from_dict_item():
type_defs = """
type Query {
test: Custom
}
type Custom {
node: String
}
"""
query = ResolverMap("Query")
query.field("test")(lambda *_: {"node": "custom"})
schema = make_executable_schema(type_defs, query)
result = graphql_sync(schema, "{ test { node } }")
assert result.errors is None
assert result.data == {"test": {"node": "custom"}}
def test_custom_resolver_is_called_to_resolve_custom_type_field_value():
type_defs = """
type Query {
test: Custom
}
type Custom {
node: String
}
"""
query = ResolverMap("Query")
query.field("test")(lambda *_: {"node": "custom"})
custom = ResolverMap("Custom")
custom.field("node")(lambda *_: "deep")
schema = make_executable_schema(type_defs, [query, custom])
result = graphql_sync(schema, "{ test { node } }")
assert result.errors is None
assert result.data == {"test": {"node": "deep"}}
def test_custom_and_default_resolvers_are_combined_to_resolve_custom_type_fields():
type_defs = """
type Query {
test: Custom
}
type Custom {
node: String
default: String
}
"""
query = ResolverMap("Query")
query.field("test")(lambda *_: {"node": "custom", "default": "ok"})
custom = ResolverMap("Custom")
custom.field("node")(lambda *_: "deep")
schema = make_executable_schema(type_defs, [query, custom])
result = graphql_sync(schema, "{ test { node default } }")
assert result.errors is None
assert result.data == {"test": {"node": "deep", "default": "ok"}}
def test_default_resolver_resolves_value_from_object_attr():
type_defs = """
type Query {
test: Custom
}
type Custom {
node: String
}
"""
query = ResolverMap("Query")
query.field("test")(lambda *_: Mock(node="custom"))
schema = make_executable_schema(type_defs, query)
result = graphql_sync(schema, "{ test { node } }")
assert result.errors is None
assert result.data == {"test": {"node": "custom"}}
}
input StaffInput {
name: String
}
type Staff {
name: String
}
type Mutation {
addStaff(data: StaffInput): Staff
}
"""
mutation = ResolverMap("Mutation")
@mutation.field("addStaff")
def resolve_add_staff(*_, data): # pylint: disable=unused-variable
assert data == {"name": "Bob"}
return data
schema = make_executable_schema(type_defs, mutation)
result = graphql_sync(
schema, 'mutation { addStaff(data: { name: "Bob" }) { name } }'
)
assert result.errors is None
assert result.data == {"addStaff": {"name": "Bob"}}
hello: String!
notes: [Note!]!
notesContaining(query: String!): [Note!]!
}
type Mutation {
createNote(title: String!, body: String!): Note!
sendMessage(message: String!): Boolean!
}
type Subscription {
messages: String!
}
"""
)
mutation = ResolverMap("Mutation")
pubsub = EventEmitter()
query = ResolverMap("Query")
subscription = SubscriptionAwareResolverMap("Subscription")
@query.field("hello")
async def say_hello(root, info):
await asyncio.sleep(3)
return "Hello!"
@query.field("notes")
async def get_all_notes(root, info):
await init_database() # FIXME: channels needs to expose the ready_callable from daphne
notes = await Note.query.gino.all()
return notes