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

1"""Regular mutation materializer: 

2Materializes a mutation that is ready to be sent off 

3""" 

4 

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 

9 

10 

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 

16 

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 () 

21 

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) 

26 

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") 

36 

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 

59 

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) 

64 

65 if query_inputs != "": 

66 query_inputs = f"({query_inputs})" 

67 

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