Coverage for graphqler / fuzzer / engine / materializers / maximal_payload_materializer.py: 95%

1 

2"""Regular mutation materializer: 

3Materializes a mutation that is ready to be sent off 

4""" 

5 

6from .materializer import Materializer 

7from .utils.materialization_utils import prettify_graphql_payload 

8from .getter import Getter 

9from graphqler.config import MAX_OUTPUT_SELECTOR_DEPTH, MAX_INPUT_DEPTH 

10from graphqler.utils.api import API 

11from graphqler.utils.objects_bucket import ObjectsBucket 

12 

13 

14class MaximalPayloadMaterializer(Materializer): 

15 def __init__(self, api: API, fail_on_hard_dependency_not_met: bool = True): 

16 self.getters = Getter() 

17 self.api = api 

18 self.fail_on_hard_dependency_not_met = fail_on_hard_dependency_not_met 

19 super().__init__(self.api, self.fail_on_hard_dependency_not_met, max_depth=MAX_OUTPUT_SELECTOR_DEPTH, getter=self.getters) 

20 

21 def get_payload(self, name: str, objects_bucket: ObjectsBucket, graphql_type: str) -> tuple[str, dict]: 

22 """Materializes the mutation with parameters filled in 

23 1. Make sure all dependencies are satisfied (hardDependsOn) 

24 2. Fill in the inputs () 

25 

26 Args: 

27 name (str): The name of either the mutation or query 

28 objects_bucket (dict): The bucket of objects that have already been created 

29 graphql_type (str): The type of the graphql operation (Query or Mutation) 

30 

31 Returns: 

32 tuple[str, dict]: The string of the payload, and the used objects list 

33 """ 

34 self.used_objects = {} # Reset the used_objects list per run (from parent class) 

35 if graphql_type == "Query": 

36 return self._get_query_payload(name, 

37 objects_bucket, 

38 max_input_depth=MAX_INPUT_DEPTH, 

39 max_output_depth=MAX_OUTPUT_SELECTOR_DEPTH, 

40 minimal_materialization=False) 

41 elif graphql_type == "Mutation": 

42 return self._get_mutation_payload(name, 

43 objects_bucket, 

44 max_input_depth=MAX_INPUT_DEPTH, 

45 max_output_depth=MAX_OUTPUT_SELECTOR_DEPTH, 

46 minimal_materialization=False) 

47 else: 

48 raise ValueError("Invalid graphql_type provided") 

49 

50 def _get_query_payload(self, 

51 query_name: str, 

52 objects_bucket: ObjectsBucket, 

53 max_input_depth: int = MAX_INPUT_DEPTH, 

54 max_output_depth: int = MAX_OUTPUT_SELECTOR_DEPTH, 

55 minimal_materialization: bool = False) -> tuple[str, dict]: 

56 query_info = self.api.queries[query_name] 

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

58 query_output = self.materialize_output(query_info, query_info["output"], objects_bucket, max_depth=max_output_depth, minimal_materialization=minimal_materialization) 

59 

60 if query_inputs != "": 

61 query_inputs = f"({query_inputs})" 

62 

63 payload = f""" 

64 query {{ 

65 {query_name} {query_inputs} 

66 {query_output} 

67 }} 

68 """ 

69 pretty_payload = prettify_graphql_payload(payload) 

70 return pretty_payload, self.used_objects 

71 

72 def _get_mutation_payload(self, 

73 mutation_name: str, 

74 objects_bucket: ObjectsBucket, 

75 max_input_depth: int = MAX_INPUT_DEPTH, 

76 max_output_depth: int = MAX_OUTPUT_SELECTOR_DEPTH, 

77 minimal_materialization: bool = False) -> tuple[str, dict]: 

78 mutation_info = self.api.mutations[mutation_name] 

79 mutation_inputs = self.materialize_inputs(mutation_info, mutation_info["inputs"], objects_bucket, max_depth=max_input_depth) 

80 mutation_output = self.materialize_output(mutation_info, mutation_info["output"], objects_bucket, max_depth=max_output_depth, minimal_materialization=minimal_materialization) 

81 

82 if mutation_inputs.strip() == "": 

83 mutation_payload = f""" 

84 mutation {{ 

85 {mutation_name} 

86 {mutation_output} 

87 }} 

88 """ 

89 else: 

90 mutation_payload = f""" 

91 mutation {{ 

92 {mutation_name} ( 

93 {mutation_inputs} 

94 ) 

95 {mutation_output} 

96 }} 

97 """ 

98 pretty_payload = prettify_graphql_payload(mutation_payload) 

99 return pretty_payload, self.used_objects