Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Support for GraphQL tests for the
pytest runner
Ref: #649
- Loading branch information
1 parent
1574751
commit d1463a7
Showing
5 changed files
with
109 additions
and
38 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,44 +1,79 @@ | ||
| from functools import partial | ||
| from typing import Any, Callable, Dict, Generator, List, Optional, Tuple, cast | ||
| from urllib.parse import urlsplit | ||
|
|
||
| import attr | ||
| import graphql | ||
| import requests | ||
| from hypothesis import strategies as st | ||
| from hypothesis.strategies import SearchStrategy | ||
| from hypothesis_graphql import strategies as gql_st | ||
|
|
||
| from ...checks import not_a_server_error | ||
| from ...hooks import HookDispatcher | ||
| from ...models import Case, CheckFunction, Endpoint | ||
| from ...schemas import BaseSchema | ||
| from ...stateful import Feedback | ||
| from ...utils import GenericResponse | ||
|
|
||
|
|
||
| @attr.s() # pragma: no mutate | ||
| class GraphQLCase: | ||
| path: str = attr.ib() # pragma: no mutate | ||
| data: str = attr.ib() # pragma: no mutate | ||
| class GraphQLCase(Case): | ||
| def as_requests_kwargs(self, base_url: Optional[str] = None) -> Dict[str, Any]: | ||
| base_url = self._get_base_url(base_url) | ||
| return {"method": self.method, "url": base_url, "json": {"query": self.body}} | ||
|
|
||
| def call(self) -> requests.Response: | ||
| return requests.post(self.path, json={"query": self.data}) | ||
| def as_werkzeug_kwargs(self, headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]: | ||
| final_headers = self.headers.copy() if self.headers is not None else {} | ||
| if headers: | ||
| final_headers.update(headers) | ||
| return { | ||
| "method": self.method, | ||
| "path": self.endpoint.schema.get_full_path(self.formatted_path), | ||
| "headers": final_headers, | ||
| "query_string": self.query, | ||
| "json": {"query": self.body}, | ||
| } | ||
|
|
||
|
|
||
| @attr.s(slots=True) # pragma: no mutate | ||
| class GraphQLQuery: | ||
| path: str = attr.ib() # pragma: no mutate | ||
| schema: graphql.GraphQLSchema = attr.ib() # pragma: no mutate | ||
|
|
||
| def as_strategy(self) -> st.SearchStrategy[GraphQLCase]: | ||
| constructor = partial(GraphQLCase, path=self.path) | ||
| return st.builds(constructor, data=gql_st.query(self.schema)) | ||
| def validate_response( | ||
| self, | ||
| response: GenericResponse, | ||
| checks: Tuple[CheckFunction, ...] = (not_a_server_error,), | ||
| ) -> None: | ||
| return super().validate_response(response, checks) | ||
|
|
||
|
|
||
| @attr.s() # pragma: no mutate | ||
| class GraphQLSchema(BaseSchema): | ||
| schema: graphql.GraphQLSchema = attr.ib(init=False) # pragma: no mutate | ||
|
|
||
| def __attrs_post_init__(self) -> None: | ||
| self.schema = graphql.build_client_schema(self.raw_schema) | ||
| def get_full_path(self, path: str) -> str: | ||
| return self.base_path | ||
|
|
||
| @property # pragma: no mutate | ||
| def verbose_name(self) -> str: | ||
| return "GraphQL" | ||
|
|
||
| @property | ||
| def query(self) -> GraphQLQuery: | ||
| return GraphQLQuery(path=self.location or "", schema=self.schema) | ||
| def base_path(self) -> str: | ||
| if self.base_url: | ||
| return urlsplit(self.base_url).path | ||
| return self._get_base_path() | ||
|
|
||
| def _get_base_path(self) -> str: | ||
| return cast(str, urlsplit(self.location).path) | ||
|
|
||
| def get_all_endpoints(self) -> Generator[Endpoint, None, None]: | ||
| yield Endpoint( | ||
| base_url=self.location, path=self.base_path, method="POST", schema=self, definition=None # type: ignore | ||
| ) | ||
|
|
||
| def get_case_strategy( | ||
| self, endpoint: Endpoint, hooks: Optional[HookDispatcher] = None, feedback: Optional[Feedback] = None | ||
| ) -> SearchStrategy: | ||
| constructor = partial(GraphQLCase, endpoint=endpoint) | ||
| schema = graphql.build_client_schema(self.raw_schema) | ||
| return st.builds(constructor, body=gql_st.query(schema)) | ||
|
|
||
| def get_strategies_from_examples(self, endpoint: Endpoint) -> List[SearchStrategy[Case]]: | ||
| return [] | ||
|
|
||
| def get_hypothesis_conversion(self, endpoint: Endpoint, location: str) -> Optional[Callable]: | ||
| return None |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| def test_basic_pytest_graphql(testdir, graphql_endpoint): | ||
| testdir.make_test( | ||
| f""" | ||
| schema = schemathesis.graphql.from_url('{graphql_endpoint}') | ||
| @schema.parametrize() | ||
| @settings(max_examples=10) | ||
| def test_(request, case): | ||
| request.config.HYPOTHESIS_CASES += 1 | ||
| assert case.path == "/graphql" | ||
| assert "patron" in case.body | ||
| response = case.call() | ||
| assert response.status_code == 200 | ||
| case.validate_response(response) | ||
| """, | ||
| ) | ||
| result = testdir.runpytest("-v", "-s") | ||
| result.assert_outcomes(passed=1) | ||
| result.stdout.re_match_lines( | ||
| [r"test_basic_pytest_graphql.py::test_\[POST:/graphql\] PASSED", r"Hypothesis calls: 10"] | ||
| ) |