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

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 .getter import Getter 

8from graphqler.config import MAX_OUTPUT_SELECTOR_DEPTH, MAX_INPUT_DEPTH 

9from graphqler.utils.api import API 

10from graphqler.utils.objects_bucket import ObjectsBucket 

11 

12 

13class RegularPayloadMaterializer(Materializer): 

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

15 self.getters = Getter() 

16 self.api = api 

17 self.fail_on_hard_dependency_not_met = fail_on_hard_dependency_not_met 

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

19 

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

21 """Materializes the mutation with parameters filled in 

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

23 2. Fill in the inputs () 

24 

25 Args: 

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

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

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

29 

30 Returns: 

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

32 """ 

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

34 if graphql_type == "Query": 

35 return self._get_query_payload(name, 

36 objects_bucket, 

37 max_input_depth=MAX_INPUT_DEPTH, 

38 max_output_depth=MAX_OUTPUT_SELECTOR_DEPTH, 

39 minimal_materialization=True) 

40 elif graphql_type == "Mutation": 

41 return self._get_mutation_payload(name, 

42 objects_bucket, 

43 max_input_depth=MAX_INPUT_DEPTH, 

44 max_output_depth=MAX_OUTPUT_SELECTOR_DEPTH, 

45 minimal_materialization=True) 

46 else: 

47 raise ValueError("Invalid graphql_type provided") 

48 

49 def _get_query_payload(self, 

50 query_name: str, 

51 objects_bucket: ObjectsBucket, 

52 max_input_depth: int = MAX_INPUT_DEPTH, 

53 max_output_depth: int = MAX_OUTPUT_SELECTOR_DEPTH, 

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

55 query_info = self.api.queries[query_name] 

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

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

58 

59 if query_inputs != "": 

60 query_inputs = f"({query_inputs})" 

61 

62 payload = f""" 

63 query {{ 

64 {query_name} {query_inputs} 

65 {query_output} 

66 }} 

67 """ 

68 pretty_payload = prettify_graphql_payload(payload) 

69 return pretty_payload, self.used_objects 

70 

71 def _get_mutation_payload(self, 

72 mutation_name: str, 

73 objects_bucket: ObjectsBucket, 

74 max_input_depth: int = MAX_INPUT_DEPTH, 

75 max_output_depth: int = MAX_OUTPUT_SELECTOR_DEPTH, 

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

77 mutation_info = self.api.mutations[mutation_name] 

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

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

80 

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

82 mutation_payload = f""" 

83 mutation {{ 

84 {mutation_name} 

85 {mutation_output} 

86 }} 

87 """ 

88 else: 

89 mutation_payload = f""" 

90 mutation {{ 

91 {mutation_name} ( 

92 {mutation_inputs} 

93 ) 

94 {mutation_output} 

95 }} 

96 """ 

97 pretty_payload = prettify_graphql_payload(mutation_payload) 

98 return pretty_payload, self.used_objects