projection_assertions.mojo (4413B)
1 from std.collections import List 2 3 from json import Value, dumps, loads 4 5 6 def registered_projection_operators() -> List[String]: 7 var operators = List[String]() 8 operators.append("equals") 9 operators.append("absent") 10 operators.append("present") 11 operators.append("contains") 12 operators.append("not_equals") 13 operators.append("tolerance") 14 return operators^ 15 16 17 def _has_key(value: Value, key: String) -> Bool: 18 for candidate in value.object_keys(): 19 if candidate == key: 20 return True 21 return False 22 23 24 def _lookup(actual: Value, pointer: String) raises -> Value: 25 var current = actual.copy() 26 if pointer == "" or pointer == "/": 27 return current^ 28 for raw_token in pointer.split("/"): 29 var token = String(raw_token) 30 if token == "": 31 continue 32 if current.is_array(): 33 current = current.array_items()[Int(token)].copy() 34 elif current.is_object(): 35 current = loads(current.get(token)) 36 else: 37 raise Error("projection path does not resolve: " + pointer) 38 return current^ 39 40 41 def _path_exists(actual: Value, pointer: String) -> Bool: 42 try: 43 _ = _lookup(actual, pointer) 44 return True 45 except: 46 return False 47 48 49 def _deep_equal(lhs: Value, rhs: Value) -> Bool: 50 return dumps(lhs) == dumps(rhs) 51 52 53 def _is_number(value: Value) -> Bool: 54 return value.is_int() or value.is_float() 55 56 57 def _as_float(value: Value) -> Float64: 58 if value.is_float(): 59 return value.float_value() 60 return Float64(value.int_value()) 61 62 63 def assert_projection(actual: Value, assertions: List[Value]) raises: 64 for assertion in assertions: 65 if not _has_key(assertion, "operator"): 66 raise Error("projection assertion requires 'operator'") 67 var operator = assertion["operator"].string_value() 68 var registered = registered_projection_operators() 69 var known = False 70 for candidate in registered: 71 if candidate == operator: 72 known = True 73 if not known: 74 raise Error("unknown projection operator: " + operator) 75 76 if operator == "present": 77 if not _path_exists(actual, assertion["path"].string_value()): 78 raise Error( 79 "expected present path: " + assertion["path"].string_value() 80 ) 81 continue 82 if operator == "absent": 83 if _path_exists(actual, assertion["path"].string_value()): 84 raise Error( 85 "expected absent path: " + assertion["path"].string_value() 86 ) 87 continue 88 89 var observed = _lookup(actual, assertion["path"].string_value()) 90 if operator == "equals": 91 if not _deep_equal(observed, assertion["value"]): 92 raise Error( 93 "expected equality at " + assertion["path"].string_value() 94 ) 95 elif operator == "not_equals": 96 if _deep_equal(observed, assertion["value"]): 97 raise Error( 98 "expected inequality at " + assertion["path"].string_value() 99 ) 100 elif operator == "contains": 101 if not observed.is_array(): 102 raise Error( 103 "contains requires an array at " 104 + assertion["path"].string_value() 105 ) 106 var found = False 107 for item in observed.array_items(): 108 if _deep_equal(item, assertion["value"]): 109 found = True 110 if not found: 111 raise Error( 112 "expected contained value at " 113 + assertion["path"].string_value() 114 ) 115 elif operator == "tolerance": 116 if not _is_number(observed) or not _is_number(assertion["value"]): 117 raise Error( 118 "tolerance requires numeric values at " 119 + assertion["path"].string_value() 120 ) 121 var delta = _as_float(observed) - _as_float(assertion["value"]) 122 if delta < 0.0: 123 delta = -delta 124 var tolerance = _as_float(assertion["tolerance"]) 125 if delta > tolerance: 126 raise Error( 127 "value out of tolerance at " 128 + assertion["path"].string_value() 129 )