Coverage for graphqler / fuzzer / engine / materializers / dos / dos_deep_recursion_materializer.py: 27%
33 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-18 23:20 -0400
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-18 23:20 -0400
1"""Regular mutation materializer:
2Materializes a mutation that is ready to be sent off
3"""
5from ..materializer import Materializer
6from ..utils.materialization_utils import prettify_graphql_payload
7from graphqler.utils.objects_bucket import ObjectsBucket
8from graphqler.utils.api import API
11class DOSDeepRecursionMaterializer(Materializer):
12 def __init__(self, api: API, fail_on_hard_dependency_not_met: bool = False, max_depth: int = 20):
13 super().__init__(api, fail_on_hard_dependency_not_met)
14 self.fail_on_hard_dependency_not_met = fail_on_hard_dependency_not_met
15 self.max_depth = max_depth
17 def get_payload(self, name: str, objects_bucket: ObjectsBucket, graphql_type: str = "") -> tuple[str, dict]:
18 """Materializes the payload with parameters filled in
19 1. Make sure all dependencies are satisfied (hardDependsOn)
20 2. Fill in the inputs ()
22 Args:
23 name (str): The payload name of the Query or Mutation
24 objects_bucket (dict): The bucket of objects that have already been created
25 graphql_type (str): The type of the graphql operation (Query or Mutation)
27 Returns:
28 tuple[str, dict]: The string of the payload, and the used objects list
29 """
30 if graphql_type == "Query":
31 return self._get_query_payload(name, objects_bucket)
32 elif graphql_type == "Mutation":
33 return self._get_mutation_payload(name, objects_bucket)
34 else:
35 raise ValueError("Invalid graphql_type provided")
37 def _get_mutation_payload(self, mutation_name: str, objects_bucket: ObjectsBucket) -> tuple[str, dict]:
38 mutation_info = self.api.mutations[mutation_name]
39 mutation_inputs = self.materialize_inputs(mutation_info, mutation_info["inputs"], objects_bucket, max_depth=self.max_depth)
40 mutation_output = self.materialize_output(mutation_info, mutation_info["output"], objects_bucket, max_depth=self.max_depth)
41 if mutation_inputs.strip() == "":
42 mutation_payload = f"""
43 mutation {{
44 {mutation_name}
45 {mutation_output}
46 }}
47 """
48 else:
49 mutation_payload = f"""
50 mutation {{
51 {mutation_name} (
52 {mutation_inputs}
53 )
54 {mutation_output}
55 }}
56 """
57 pretty_payload = prettify_graphql_payload(mutation_payload)
58 return pretty_payload, self.used_objects
60 def _get_query_payload(self, query_name: str, objects_bucket: ObjectsBucket) -> tuple[str, dict]:
61 query_info = self.api.queries[query_name]
62 query_inputs = self.materialize_inputs(query_info, query_info["inputs"], objects_bucket, max_depth=self.max_depth)
63 query_output = self.materialize_output(query_info, query_info["output"], objects_bucket, max_depth=self.max_depth)
65 if query_inputs != "":
66 query_inputs = f"({query_inputs})"
68 payload = f"""
69 query {{
70 {query_name} {query_inputs}
71 {query_output}
72 }}
73 """
74 pretty_payload = prettify_graphql_payload(payload)
75 return pretty_payload, self.used_objects