/
test_application_async.py
105 lines (89 loc) · 3.86 KB
/
test_application_async.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
# Copyright 2010 New Relic, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import pytest
from testing_support.fixtures import dt_enabled, validate_transaction_metrics
from testing_support.validators.validate_span_events import validate_span_events
from newrelic.api.background_task import background_task
@pytest.fixture(scope="session")
def graphql_run_async():
"""Wrapper function to simulate framework_graphql test behavior."""
def execute(schema, query, *args, **kwargs):
from ariadne import graphql
return graphql(schema, {"query": query}, *args, **kwargs)
return execute
@dt_enabled
def test_query_and_mutation_async(app, graphql_run_async):
from graphql import __version__ as version
FRAMEWORK_METRICS = [
("Python/Framework/Ariadne/None", 1),
("Python/Framework/GraphQL/%s" % version, 1),
]
_test_mutation_scoped_metrics = [
("GraphQL/resolve/Ariadne/storage", 1),
("GraphQL/resolve/Ariadne/storage_add", 1),
("GraphQL/operation/Ariadne/query/<anonymous>/storage", 1),
("GraphQL/operation/Ariadne/mutation/<anonymous>/storage_add.string", 1),
]
_test_mutation_unscoped_metrics = [
("OtherTransaction/all", 1),
("GraphQL/all", 2),
("GraphQL/Ariadne/all", 2),
("GraphQL/allOther", 2),
("GraphQL/Ariadne/allOther", 2),
] + _test_mutation_scoped_metrics
_expected_mutation_operation_attributes = {
"graphql.operation.type": "mutation",
"graphql.operation.name": "<anonymous>",
}
_expected_mutation_resolver_attributes = {
"graphql.field.name": "storage_add",
"graphql.field.parentType": "Mutation",
"graphql.field.path": "storage_add",
"graphql.field.returnType": "StorageAdd",
}
_expected_query_operation_attributes = {
"graphql.operation.type": "query",
"graphql.operation.name": "<anonymous>",
}
_expected_query_resolver_attributes = {
"graphql.field.name": "storage",
"graphql.field.parentType": "Query",
"graphql.field.path": "storage",
"graphql.field.returnType": "[String]",
}
@validate_transaction_metrics(
"query/<anonymous>/storage",
"GraphQL",
scoped_metrics=_test_mutation_scoped_metrics,
rollup_metrics=_test_mutation_unscoped_metrics + FRAMEWORK_METRICS,
background_task=True,
)
@validate_span_events(exact_agents=_expected_mutation_operation_attributes)
@validate_span_events(exact_agents=_expected_mutation_resolver_attributes)
@validate_span_events(exact_agents=_expected_query_operation_attributes)
@validate_span_events(exact_agents=_expected_query_resolver_attributes)
@background_task()
def _test():
async def coro():
ok, response = await graphql_run_async(app, 'mutation { storage_add(string: "abc") { string } }')
assert ok and not response.get("errors")
ok, response = await graphql_run_async(app, "query { storage }")
assert ok and not response.get("errors")
# These are separate assertions because pypy stores 'abc' as a unicode string while other Python versions do not
assert "storage" in str(response.get("data"))
assert "abc" in str(response.get("data"))
loop = asyncio.new_event_loop()
loop.run_until_complete(coro())
_test()