Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_eq():
assert Ok(1) == Ok(1)
assert Err(1) == Err(1)
assert Ok(1) != Err(1)
assert Ok(1) != Ok(2)
assert Err(1) != Err(2)
assert not (Ok(1) != Ok(1))
assert Ok(1) != "abc"
assert Ok("0") != Ok(0)
def test_hash():
assert len({Ok(1), Err("2"), Ok(1), Err("2")}) == 2
assert len({Ok(1), Ok(2)}) == 2
assert len({Ok("a"), Err("a")}) == 2
def handle_primitive(cls: Type[T]) -> Result[str, T]:
if cls in (str, int, float):
try:
return Ok(cast(T, cast(Any, cls)(value)))
except ValueError:
return Err(f"Could not parse {cls} from {value}")
elif cls is Path:
return Ok(cast(T, make_absolute_path(Path(value))))
else:
return unhandled_type(cls)
def _get_field(fld: Field[object]) -> Result[str, object]:
if fld.name in d:
return _try_yaml_to_typed(
fld.type, d[fld.name], yaml_file_path, child(fld.name)
)
else:
if all_optional:
return Ok(None)
else:
assert fld.default not in (
MISSING,
NO_DEFAULT,
), f"At {desc}: Did not find field {fld.name} (and it has no default)"
return Ok(fld.default)
def g(elements: Sequence[TElement]) -> Failable[Sequence[TElement]]:
out: List[TElement] = []
for i, em in enumerate(elements):
b = get_value_for_element(t, elements, i)
if b.is_err():
return Err(as_err(b))
elif check_cast(bool, b.unwrap()):
out.append(em)
return Ok(out)
return lambda t, u, v: Ok(f(t, u, v))
def MemoryPressure(self) -> FailableFloat:
ghh = self.trace_gc.GlobalHeapHistory
if ghh is None:
return Err("No GlobalHeapHistory")
elif ghh.HasMemoryPressure:
return Ok(ghh.MemoryPressure)
else:
return Err("GlobalHeapHistory#HasMemoryPressure was false")
return lambda t: Ok(f(t))
def handle_primitive(cls: Type[T]) -> Result[str, T]:
if cls in (str, int, float):
try:
return Ok(cast(T, cast(Any, cls)(value)))
except ValueError:
return Err(f"Could not parse {cls} from {value}")
elif cls is Path:
return Ok(cast(T, make_absolute_path(Path(value))))
else:
return unhandled_type(cls)
def get_processed_trace(
clr: Clr, test_result: TestResult, need_mechanisms_and_reasons: bool, need_join_info: bool
) -> Result[str, ProcessedTrace]:
test_status = option_to_result(
test_result.load_test_status(), lambda: "Need a test status file"
)
if test_result.trace_path is None:
if need_join_info:
return Err("Can't get join info without a trace.")
else:
return Ok(
ProcessedTrace(
clr=clr,
test_result=test_result,
test_status=test_status,
process_info=None,
process_names=cast(ThreadToProcessToName, None),
process_query=None,
join_info=Err("Did not request join info"),
mechanisms_and_reasons=None,
gcs_result=Err("Did not collect a trace"),
)
)
else:
return Ok(
_get_processed_trace_from_process(
clr,