Coverage for graphqler / fuzzer / engine / materializers / getter.py: 53%

1"""Getters module: 

2Used by the materializer to understand how to get values for fields 

3-- Can be overridden to provide custom values by different materializers (IE. DOS, Injection, ect) 

4""" 

5 

6from datetime import datetime, timedelta 

7from graphqler.utils.objects_bucket import ObjectsBucket 

8import random 

9import string 

10 

11 

12class Getter: 

13 def __init__(self): 

14 pass 

15 

16 def get_random_string(self, input_name: str) -> str: 

17 # Maybe we can use the input name somehow? (Like if the input name contains "name") 

18 if "email" in input_name: 

19 return f"\"{''.join(random.choices(string.ascii_lowercase, k=10))}@{''.join(random.choices(string.ascii_lowercase, k=10))}.com\"" 

20 elif "name" in input_name: 

21 return f"\"{''.join(random.choices(string.ascii_uppercase + string.digits, k=10))}\"" 

22 else: 

23 return f"\"{''.join(random.choices(string.ascii_uppercase + string.digits, k=10))}\"" 

24 

25 def get_random_int(self, input_name: str) -> int: 

26 return random.randint(0, 100) 

27 

28 def get_random_float(self, input_name: str) -> float: 

29 if input_name == "latitude" or input_name == "longitude": 

30 return random.uniform(-180.0, 180.0) 

31 return random.uniform(0.0, 1000.0) 

32 

33 def get_random_bool(self, input_name: str) -> str: 

34 return str(bool(random.getrandbits(1))).lower() 

35 

36 def get_random_id(self, input_name: str, objects_bucket: ObjectsBucket) -> str: 

37 random_id = self.get_random_id_from_bucket(input_name, objects_bucket) 

38 if random_id.strip() == "": 

39 random_id = "".join(random.choices(string.ascii_uppercase + string.digits, k=10)) 

40 return '"' + str(random_id) + '"' 

41 

42 def get_random_date(self, input_name: str) -> str: 

43 return f"\"{datetime.today().strftime('%Y-%m-%d')}\"" 

44 

45 def get_random_time(self, input_name: str) -> str: 

46 random_date_interval = random.randint(-10, 10) 

47 calculated_date = datetime.today() + timedelta(days=random_date_interval) 

48 return f"\"{calculated_date.strftime('%Y-%m-%d')}TT00:00:00+00:00\"" 

49 

50 def get_random_long(self, input_name: str) -> str: 

51 return str(random.randint(0, 1000000)) 

52 

53 def get_random_json(self, input_name: str) -> str: 

54 return "{}" 

55 

56 def get_random_datetime(self, input_name: str) -> str: 

57 # Current datetime) 

58 now = datetime.now() 

59 # Range: 3 days before to 3 days after the current datetime 

60 start_date = now - timedelta(days=3) 

61 end_date = now + timedelta(days=3) 

62 

63 # Calculate the total number of seconds in the range 

64 time_delta = end_date - start_date 

65 total_seconds = int(time_delta.total_seconds()) 

66 

67 # Generate a random number of seconds within the range 

68 random_seconds = random.randint(0, total_seconds) 

69 

70 # Add the random seconds to the start date to get a random datetime 

71 random_date = start_date + timedelta(seconds=random_seconds) 

72 

73 # Format the datetime in ISO 8601 format with a 'Z' suffix for UTC 

74 graphql_datetime = random_date.strftime("%Y-%m-%dT%H:%M:%SZ") 

75 

76 return f'"{graphql_datetime}"' 

77 

78 def get_random_custom_scalar(self, input_name: str, scalar_type: str, objects_bucket: ObjectsBucket) -> str: 

79 # Must be a custom scalar, check if it's an ID, if not then just fail 

80 if scalar_type.lower().endswith("id") or scalar_type.lower().endswith("ids"): 

81 return str(self.get_random_id(input_name, objects_bucket)) 

82 elif scalar_type.lower() == "time": 

83 return self.get_random_time(input_name) 

84 elif scalar_type.lower() == "long": 

85 return self.get_random_long(input_name) 

86 elif scalar_type.lower() == "datetime": 

87 return self.get_random_datetime(input_name) 

88 elif scalar_type.lower() == "json": 

89 return self.get_random_json(input_name) 

90 else: 

91 raise Exception(f"This custom scalar is not supported at this time: {input_name}:{scalar_type}") 

92 

93 # Gets a random scalar for the scalar type given 

94 def get_random_scalar(self, input_name: str, scalar_type: str, objects_bucket: ObjectsBucket) -> str: 

95 """Gets a random scalar based on the scalar type, the return value will 

96 be a string regardless if of it's type as this function meant to be used 

97 during materialization 

98 Args: 

99 input_name (str): The input field's name 

100 scalar_type (str): The scalar type (IE. Int, Float, String, Boolean, ID, ect) 

101 objects_bucket (dict): The objects bucket to look up for any random IDs we want to choose 

102 

103 Returns: 

104 str: The returned scalar 

105 """ 

106 if scalar_type == "String": 

107 return self.get_random_string(input_name) 

108 elif scalar_type == "Int": 

109 return str(self.get_random_int(input_name)) 

110 elif scalar_type == "Float": 

111 return str(self.get_random_float(input_name)) 

112 elif scalar_type == "Boolean": 

113 return str(self.get_random_bool(input_name)) 

114 elif scalar_type == "Date": 

115 return self.get_random_date(input_name) 

116 elif scalar_type == "ID": 

117 return self.get_random_id(input_name, objects_bucket) 

118 elif scalar_type == "Cursor": 

119 if input_name == "after" or input_name == "from": 

120 return "null" 

121 else: 

122 return str(1) 

123 else: 

124 return self.get_random_custom_scalar(input_name, scalar_type, objects_bucket) 

125 

126 def get_random_enum_value(self, enum_values: list[dict]) -> str: 

127 """Gets a random enum from the enumValue list 

128 Args: 

129 enum_values (list[dict]): The enumValues list 

130 

131 Returns: 

132 str: The name of the randomly chosen enum, or None if none was found 

133 """ 

134 non_deprecated_enum_values = [enum for enum in enum_values if not enum.get("isDeprecated", False)] 

135 if non_deprecated_enum_values: 

136 enum = random.choice(non_deprecated_enum_values) 

137 return enum["name"] 

138 else: 

139 raise Exception("No non-deprecated enum values found for this enum") 

140 

141 def get_random_id_from_bucket(self, input_name: str, objects_bucket: ObjectsBucket) -> str: 

142 """Tries to get an ID from the bucket based on the input_name first, then just randomly chooses an ID from the bucket, 

143 if the bucket is empty, then just returns an empty string 

144 Gets a random ID from the bucket, or just "" if there are no IDs in the bucket 

145 Args: 

146 input_name (str): The input name 

147 objects_bucket (dict): Object bucket 

148 

149 Returns: 

150 str: an ID 

151 """ 

152 # If it's empty, just return a random ID 

153 if objects_bucket.is_empty(): 

154 return "" 

155 

156 random_id = objects_bucket.get_random_scalar_from_bucket_by_type("ID") 

157 return str(random_id) 

158 

159 def get_closest_value_to_input(self, input_name: str, object_name: str, objects_bucket: ObjectsBucket) -> str | int | float | bool: 

160 """Gets the closest value to the input name given the object name, the input name, and the objects bucket 

161 Args: 

162 input_name (str): The input name 

163 object_name (str): The object name 

164 objects_bucket (dict): The objects bucket 

165 

166 Returns: 

167 str: The closest field name 

168 """ 

169 # Get the object from the bucket 

170 found_value = objects_bucket.get_random_object_field_value(object_name, input_name) 

171 if found_value is not None: 

172 return found_value 

173 

174 # Get the object name without the input name 

175 # IE. If the input name is "name" and the object name is "PersonName", the lookup name would be just 'name' in the object 

176 new_field_name = input_name.lower().replace(object_name.lower(), "").replace("_", "").replace(" ", "") 

177 

178 # Try finding the value again 

179 found_value = objects_bucket.get_random_object_field_value(object_name, new_field_name) 

180 if found_value is not None: 

181 return found_value 

182 

183 raise Exception(f"Could not find a value for the input name: {input_name} in the object: {object_name}")