Coverage for graphqler / fuzzer / engine / materializers / injection_materializer.py: 32%

1from .materializer import Materializer 

2from .utils.materialization_utils import prettify_graphql_payload 

3from graphqler.utils.api import API 

4from graphqler.utils.objects_bucket import ObjectsBucket 

5from .getter import Getter 

6 

7from typing import override 

8 

9 

10class InjectionMaterializer(Materializer): 

11 def __init__(self, api: API, fail_on_hard_dependency_not_met: bool = False, max_depth: int = 20, injection_getter: Getter = Getter()): 

12 super().__init__(api, fail_on_hard_dependency_not_met, max_depth, injection_getter) 

13 self.max_depth = max_depth 

14 self.getter = injection_getter 

15 self.api = api 

16 self.fail_on_hard_dependency_not_met = fail_on_hard_dependency_not_met 

17 

18 @override 

19 def get_payload(self, name: str, objects_bucket: ObjectsBucket, graphql_type: str = "", minimal_materialization: bool = False) -> tuple[str, dict]: 

20 """Materializes the payload with parameters filled in 

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

22 2. Fill in the inputs () 

23 

24 Args: 

25 name (str): The payload name of the Query or Mutation 

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

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

28 minimal_materialization (bool): Whether to minimize the materialization 

29 

30 Returns: 

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

32 """ 

33 if graphql_type == "Query": 

34 return self._get_query_payload(name, objects_bucket) 

35 elif graphql_type == "Mutation": 

36 return self._get_mutation_payload(name, objects_bucket) 

37 else: 

38 raise ValueError("Invalid graphql_type provided") 

39 

40 def _get_mutation_payload(self, mutation_name: str, objects_bucket: ObjectsBucket) -> tuple[str, dict]: 

41 mutation_info = self.api.mutations[mutation_name] 

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

43 mutation_output = self.materialize_output(mutation_info, mutation_info["output"], objects_bucket, max_depth=self.max_depth) 

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

45 mutation_payload = f""" 

46 mutation {{ 

47 {mutation_name} 

48 {mutation_output} 

49 }} 

50 """ 

51 else: 

52 mutation_payload = f""" 

53 mutation {{ 

54 {mutation_name} ( 

55 {mutation_inputs} 

56 ) 

57 {mutation_output} 

58 }} 

59 """ 

60 pretty_payload = prettify_graphql_payload(mutation_payload) 

61 return pretty_payload, self.used_objects 

62 

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

64 query_info = self.api.queries[query_name] 

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

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

67 

68 if query_inputs != "": 

69 query_inputs = f"({query_inputs})" 

70 

71 payload = f""" 

72 query {{ 

73 {query_name} {query_inputs} 

74 {query_output} 

75 }} 

76 """ 

77 pretty_payload = prettify_graphql_payload(payload) 

78 return pretty_payload, self.used_objects