Coverage for graphqler / fuzzer / engine / materializers / dos / dos_batch_materializer.py: 22%

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.api import API 

8from graphqler.utils.objects_bucket import ObjectsBucket 

9 

10 

11class DOSBatchMaterializer(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[list, 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[list, 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 payload = f""" 

43 mutation {{ 

44 {mutation_name} 

45 {mutation_output} 

46 }} 

47 """ 

48 else: 

49 payload = f""" 

50 mutation {{ 

51 {mutation_name} ( 

52 {mutation_inputs} 

53 ) 

54 {mutation_output} 

55 }} 

56 """ 

57 pretty_payload = prettify_graphql_payload(payload) 

58 stripped_payload = pretty_payload.strip().replace("\n", "").replace(" ", "") 

59 batch = [] 

60 # Create 3 of these objects 

61 for _ in range(0, 3): 

62 batch.append({"query": stripped_payload}) 

63 return batch, self.used_objects 

64 

65 def _get_query_payload(self, query_name: str, objects_bucket: ObjectsBucket) -> tuple[list, dict]: 

66 query_info = self.api.queries[query_name] 

67 query_inputs = self.materialize_inputs(query_info, query_info["inputs"], objects_bucket, max_depth=self.max_depth) 

68 query_output = self.materialize_output(query_info, query_info["output"], objects_bucket, max_depth=self.max_depth) 

69 

70 if query_inputs != "": 

71 query_inputs = f"({query_inputs})" 

72 

73 payload = f""" 

74 query {{ 

75 {query_name} {query_inputs} 

76 {query_output} 

77 }} 

78 """ 

79 pretty_payload = prettify_graphql_payload(payload) 

80 stripped_payload = pretty_payload.strip().replace("\n", "").replace(" ", "") 

81 batch = [] 

82 # Create 3 of these objects 

83 for _ in range(0, 3): 

84 batch.append({"query": stripped_payload}) 

85 return batch, self.used_objects