Coverage for graphqler / fuzzer / engine / materializers / materializer.py: 74%
150 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-20 10:09 -0400
1"""Materializer:
2Base class for a regular materializer
3"""
5from ..exceptions.hard_dependency_not_met_exception import HardDependencyNotMetException
6from .utils.materialization_utils import is_valid_object_materialization, clean_output_selectors
7from .getter import Getter
8from graphqler.utils.logging_utils import Logger
9from graphqler.utils.parser_utils import get_base_oftype, is_simple_scalar
10from graphqler.utils.objects_bucket import ObjectsBucket
11from graphqler.utils.api import API
12from graphqler import config
15class Materializer:
16 def __init__(self, api: API, fail_on_hard_dependency_not_met: bool = True, max_depth: int = 5, getter: Getter = Getter()):
17 """Default constructor for a regular materializer
19 Args:
20 api (API): The API object
21 fail_on_hard_dependency_not_met (bool, optional): Whether to fail on hard dependency not met. Defaults to True.
22 getter (Getter, optional): The getters object. Defaults to Getter()
23 """
24 self.api = api
25 self.logger = Logger().get_fuzzer_logger().getChild(__name__) # Get a child logger
26 self.fail_on_hard_dependency_not_met = fail_on_hard_dependency_not_met
27 self.used_objects = {}
28 self.max_depth = max_depth
29 self.getter = getter
31 def get_payload(self, name: str, objects_bucket: ObjectsBucket, graphql_type: str) -> tuple[str, dict]:
32 """Materializes the payload with parameters filled in
34 Args:
35 query_name (str): name of the graphql query or mutation
36 objects_bucket (dict): objects bucket
37 graphql_type (str, optional): one of Query or Mutation. Defaults to ''.
39 Returns:
40 tuple[str, dict]: The string of the payload, and the used objects list
41 """
42 return ("", {})
44 def materialize_output(self,
45 operator_info: dict,
46 output: dict,
47 objects_bucket: ObjectsBucket,
48 max_depth: int = 5,
49 minimal_materialization: bool = False) -> str:
50 """Materializes the output. If returns empty string,
51 then tries to get at least something, bypassing the max depth until the hard cutoff.
53 Args:
54 operator_info (dict): The operator information
55 output_info (dict): The output information
56 objects_bucket (dict): List of objects that have been created or found
57 max_depth (int, optional): Maximum depth for recursive expansion of objects. Defaults to 2.
58 If nothing is returned for this max depth, then we try to get at least something
59 by bypassing the max depth until the hard cutoff.
60 minimal_materialization (bool, optional): Whether to materialize only the minimal fields. Defaults to False.
62 Returns:
63 str: The otput selectors
64 """
65 output_selectors = ""
66 max_depth = max_depth
67 while output_selectors == "":
68 # The initial call to materialize_output_recursive should not include the name and has no objects used yet
69 output_selectors = self.materialize_output_recursive(
70 operator_info=operator_info,
71 output_field=output,
72 used_objects=[],
73 objects_bucket=objects_bucket,
74 include_name=False,
75 minimal_materialization=minimal_materialization,
76 max_depth=max_depth,
77 current_depth=0
78 )
79 if max_depth > config.HARD_CUTOFF_DEPTH:
80 break
81 max_depth += 1
82 cleaned_output_selectors = clean_output_selectors(output_selectors)
83 return cleaned_output_selectors
85 def materialize_output_recursive(self,
86 operator_info: dict,
87 output_field: dict,
88 used_objects: list[str],
89 objects_bucket: ObjectsBucket,
90 include_name: bool,
91 minimal_materialization: bool,
92 max_depth: int,
93 current_depth: int = 0) -> str:
94 """Materializes the output recursively. Some interesting cases:
95 - If we want to stop on an object materializing its fields, we need to not even include the object name
96 IE: {id, firstName, user {}} should just be {id, firstName}
97 Note: This function should be called on a base output type
99 Args:
100 operator_info (dict): Information about the operator that we want to materialize
101 output_field (dict): The field to be output
102 used_objects (list[str]): A list of used objects
103 objects_bucket (dict): List of objects that have been created or found
104 include_name (bool): Whether to include the name of the field or not
105 minimal_materialization (bool): Whether to materialize only the minimal fields
106 max_depth (int): The maximum depth to expand outputs for nested objects
107 current_depth (int): The current depth of the output
109 Returns:
110 str: The built output payload
111 """
112 built_str = ""
114 # When we are including names (IE. fields of an object), we need to include the name of the field
115 if include_name:
116 built_str += output_field["name"]
118 # If there are arguments for this, materialize the arguments
119 if "inputs" in output_field and len(output_field["inputs"]) != 0:
120 inputs = self.materialize_input_fields(operator_info, output_field["inputs"], objects_bucket, max_depth, current_depth)
121 if inputs != "":
122 built_str += f"({inputs})"
124 # Main materialiation logic
125 if output_field["kind"] == "OBJECT":
126 materialized_object_fields = self.materialize_output_object_fields(operator_info, output_field["type"], used_objects, objects_bucket, minimal_materialization, max_depth, current_depth)
127 if materialized_object_fields != "":
128 built_str += " {"
129 built_str += materialized_object_fields
130 built_str += "},"
131 elif output_field["kind"] == "UNION": # For a UNION type, loop through all the UNION types and materialize them into fragments
132 union_types = self.api.unions[output_field["type"]]["possibleTypes"]
133 built_str += " {"
134 for union_type in union_types:
135 materialized_fragment = self.materialize_output_recursive(operator_info, union_type, used_objects, objects_bucket, False, minimal_materialization, max_depth, current_depth)
136 if materialized_fragment != "":
137 built_str += f"... on {union_type['name']} " + materialized_fragment
138 built_str += "},"
139 elif output_field["kind"] == "INTERFACE": # For an INTERFACE type, loop through all the INTERFACE types and materialize them into fragments
140 interface_types = self.api.interfaces[output_field["type"]]["possibleTypes"]
141 built_str += " {"
142 for interface_type in interface_types:
143 materialized_fragment = self.materialize_output_recursive(operator_info, interface_type, used_objects, objects_bucket, False, minimal_materialization, max_depth, current_depth)
144 if materialized_fragment != "":
145 built_str += f"... on {interface_type['name']} " + materialized_fragment
146 built_str += "},"
147 elif (output_field["kind"] == "NON_NULL" or output_field["kind"] == "LIST"): # For a NON_NULL / LIST kind: Don't +1 here because it is an oftype (which doesn't add depth), or else we will double count
148 oftype = output_field["ofType"]
149 materialized_output = self.materialize_output_recursive(operator_info, oftype, used_objects, objects_bucket, False, minimal_materialization, max_depth, current_depth)
150 if materialized_output != "":
151 built_str += materialized_output + ", "
152 else:
153 built_str += ","
155 # If it's a non-scalar but we didn't materialize any fields, then we should return an empty string
156 # Very important for NON_NULL / LIST / OBJECT types
157 chars_to_remove = ",{}. "
158 translation_table = str.maketrans("", "", chars_to_remove)
159 if get_base_oftype(output_field)["kind"] != "SCALAR" and include_name:
160 if built_str == output_field["name"]:
161 return ""
162 elif built_str.translate(translation_table) == output_field["name"]:
163 return ""
164 elif not is_valid_object_materialization(built_str):
165 return ""
167 # A bit of post processing on the built payload
168 if include_name and built_str[-1] != ",":
169 built_str += ","
170 elif not include_name and built_str.strip() == "{}":
171 built_str = ""
173 return built_str
175 def materialize_output_object_fields(self,
176 operator_info: dict,
177 object_name: str,
178 used_objects: list[str],
179 objects_bucket: ObjectsBucket,
180 minimal_materialization: bool,
181 max_depth: int,
182 current_depth: int) -> str:
183 """Loop through an objects fields, and call materialize_output on each of them
185 Args:
186 operator_info (dict): The operator information
187 object_information (dict): The object's information
188 used_objects (list[str]): A list of used objects
189 objects_bucket (dict): List of objects that have been created or found
190 minimal_materialization (bool): Whether to materialize only the minimal fields
191 max_depth (int): The maximum depth to expand outputs for nested objects
192 current_depth (int): The current depth of the output
194 Returns:
195 str: The built output string
196 """
197 built_str = ""
198 object_info = self.api.objects[object_name]
199 fields_to_materialize = object_info["fields"]
201 # If we've seen this object more than the max object cycles, don't use it again
202 # But only do this check while we aren't only materializing non-null fields
203 if used_objects.count(object_name) >= config.MAX_OBJECT_CYCLES:
204 return built_str
206 # Check has any scalar at root level, if it does then we can filter out the rest
207 # We have to do this because sometimes root level might only be an object that outputs another object
208 if minimal_materialization and current_depth == 0:
209 has_scalar = False
210 for field in fields_to_materialize:
211 if is_simple_scalar(field):
212 has_scalar = True
213 break
214 if has_scalar:
215 fields_to_materialize = [field for field in fields_to_materialize if is_simple_scalar(field)]
217 # If we're materializing only minimal fields, then we should only materialize scalar fields as long as we're not at the highest depth
218 if minimal_materialization and current_depth != 0:
219 fields_to_materialize = [field for field in fields_to_materialize if is_simple_scalar(field)]
221 # If we're at max depth, materialize only scalar fields
222 if current_depth >= max_depth:
223 fields_to_materialize = [field for field in fields_to_materialize if is_simple_scalar(field)]
225 # Mark that we've used this object
226 used_objects.append(object_name)
228 # Loop through the fields to materialize each field
229 for field in fields_to_materialize:
230 field_output = self.materialize_output_recursive(operator_info, field, used_objects, objects_bucket, True, minimal_materialization, max_depth, current_depth + 1)
231 if field_output != "" and field_output != "{}":
232 built_str += field_output
233 return built_str
235 def materialize_inputs(self, operator_info: dict, inputs: dict, objects_bucket: ObjectsBucket, max_depth: int) -> str:
236 """Goes through the inputs of the payload
238 Args:
239 operator_info (dict): All information about the operator (either all QUERYs or all MUTATIONs) that we want to materialize
240 inputs (dict): The inputs of to be parsed
241 objects_bucket (dict): The dynamically available objects that are currently in circulation
242 max_depth (int): The maximum depth to proceed to when unravelling nested input objects
244 Returns:
245 str: The input parameters as a string
246 """
247 return self.materialize_input_fields(operator_info, inputs, objects_bucket, max_depth, current_depth=0)
249 def materialize_input_fields(self, operator_info: dict, inputs: dict, objects_bucket: ObjectsBucket, max_depth: int, current_depth: int = 0) -> str:
250 """Goes through the inputs of the payload
252 Args:
253 operator_info (dict): All information about the operator (either all QUERYs or all MUTATIONs) that we want to materialize
254 inputs (dict): The inputs of to be parsed
255 objects_bucket (dict): The dynamically available objects that are currently in circulation
257 Returns:
258 str: The input parameters as a string
259 """
260 built_str = ""
262 # Return early if there are no inputs
263 if inputs is None or len(inputs) == 0 or type(inputs) is not dict:
264 return built_str
266 # Return early if we exceed the max depth
267 if current_depth >= max_depth:
268 return built_str
270 # Go through each input field and materialize it
271 for input_name, input_field in inputs.items():
272 built_str += f"{input_name}: " + self.materialize_input_recursive(operator_info, input_field, objects_bucket, input_name, True, max_depth, current_depth + 1) + ","
273 return built_str
275 def materialize_input_recursive(self,
276 operator_info: dict,
277 input_field: dict,
278 objects_bucket: ObjectsBucket,
279 input_name: str,
280 check_deps: bool,
281 max_depth: int,
282 current_depth: int) -> str:
283 """Materializes a single input field
284 - if the field is one we already know it depends on, just instantly resolve. Or else going down into
285 the oftype will make us lose its name
287 Args:
288 operator_info (dict): All information about the operator (either all QUERYs or all MUTATIONs) that we want to materialize
289 input_field (dict): The field for a mutation (has the)
290 objects_bucket (dict): The dynamically available objects that are currently in circulation
291 input_name (str): The input's name in the overall query (not to be confused with input_field["name"] - which is the field's name in the struct)
292 check_deps (bool): Whether to check the dependencies first or not
294 Returns:
295 str: String of the materialized input field
296 """
297 built_str = ""
298 hard_dependencies: dict = operator_info.get("hardDependsOn", {})
299 soft_dependencies: dict = operator_info.get("softDependsOn", {})
301 # Must first resolve any dependencies we have access to(since if we go down and resolve ofTypes we lose its name)
302 if check_deps and input_field["name"] in hard_dependencies:
303 hard_dependency_object_name = hard_dependencies[input_field["name"]]
304 if objects_bucket.is_object_in_bucket(hard_dependency_object_name):
305 # Use the object from the objects bucket, mark it as used, then continue constructing the string
306 randomly_chosen_object_dependency_val = self.getter.get_closest_value_to_input(input_field["name"], hard_dependency_object_name, objects_bucket)
307 self.used_objects[hard_dependency_object_name] = randomly_chosen_object_dependency_val
308 built_str += f'"{randomly_chosen_object_dependency_val}"'
309 elif hard_dependency_object_name == "UNKNOWN":
310 self.logger.info(f"Using UNKNOWN input for field: {input_field}")
311 built_str += self.materialize_input_recursive(operator_info, input_field, objects_bucket, input_name, False, max_depth, current_depth)
312 else:
313 if self.fail_on_hard_dependency_not_met: # If we are using the dependency graph, then we should be careful dependencies aren't met
314 raise HardDependencyNotMetException(hard_dependency_object_name)
315 else: # Otherwise, in regular non-dependency aware mode, we just materialize the input field
316 self.logger.info("Hard dependency not met -- using random input")
317 built_str += self.materialize_input_recursive(operator_info, input_field, objects_bucket, input_name, False, max_depth, current_depth)
318 elif check_deps and input_field["name"] in soft_dependencies:
319 soft_depedency_name = soft_dependencies[input_field["name"]]
320 if objects_bucket.is_object_in_bucket(soft_depedency_name):
321 # Use the object from the objects bucket, mark it as used, then continue constructing the string
322 randomly_chosen_dependency_val = objects_bucket.get_random_object_field_value(soft_depedency_name, input_field["name"])
323 self.used_objects[soft_depedency_name] = randomly_chosen_dependency_val
324 built_str += f'"{randomly_chosen_dependency_val}"'
325 else:
326 built_str += self.materialize_input_recursive(operator_info, input_field, objects_bucket, input_name, False, max_depth, current_depth)
327 elif input_field["kind"] == "NON_NULL":
328 built_str += self.materialize_input_recursive(operator_info, input_field["ofType"], objects_bucket, input_name, True, max_depth, current_depth)
329 elif input_field["kind"] == "LIST":
330 built_str += f"[{self.materialize_input_recursive(operator_info, input_field['ofType'], objects_bucket, input_name, True, max_depth, current_depth)}]"
331 elif input_field["kind"] == "INPUT_OBJECT":
332 input_object = self.api.input_objects[input_field["type"]]
333 built_str += "{" + self.materialize_input_fields(operator_info, input_object["inputFields"], objects_bucket, max_depth, current_depth) + "}"
334 elif input_field["kind"] == "SCALAR":
335 built_str += self.getter.get_random_scalar(input_name, input_field["type"], objects_bucket)
336 elif input_field["kind"] == "ENUM":
337 built_str += self.getter.get_random_enum_value(self.api.enums[input_field["type"]]["enumValues"])
338 else:
339 built_str += ""
341 return built_str