mirror of
https://github.com/microsoft/onefuzz.git
synced 2025-06-12 18:18:08 +00:00
79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
#!/usr/bin/env python
|
|
#
|
|
# Copyright (c) Microsoft Corporation.
|
|
# Licensed under the MIT License.
|
|
|
|
import json
|
|
import logging
|
|
from typing import Optional, Union
|
|
|
|
from memoization import cached
|
|
from onefuzztypes.models import RegressionReport, Report
|
|
from onefuzztypes.primitives import Container
|
|
from pydantic import ValidationError
|
|
|
|
from .azure.containers import get_blob
|
|
from .azure.storage import StorageType
|
|
|
|
|
|
def parse_report_or_regression(
|
|
content: Union[str, bytes], file_path: Optional[str] = None
|
|
) -> Optional[Union[Report, RegressionReport]]:
|
|
if isinstance(content, bytes):
|
|
try:
|
|
content = content.decode()
|
|
except UnicodeDecodeError as err:
|
|
logging.error(
|
|
f"unable to parse report ({file_path}): "
|
|
f"unicode decode of report failed - {err}"
|
|
)
|
|
return None
|
|
|
|
try:
|
|
data = json.loads(content)
|
|
except json.decoder.JSONDecodeError as err:
|
|
logging.error(
|
|
f"unable to parse report ({file_path}): json decoding failed - {err}"
|
|
)
|
|
return None
|
|
|
|
regression_err = None
|
|
try:
|
|
return RegressionReport.parse_obj(data)
|
|
except ValidationError as err:
|
|
regression_err = err
|
|
|
|
try:
|
|
return Report.parse_obj(data)
|
|
except ValidationError as err:
|
|
logging.error(
|
|
f"unable to parse report ({file_path}) as a report or regression. "
|
|
f"regression error: {regression_err} report error: {err}"
|
|
)
|
|
return None
|
|
|
|
|
|
# cache the last 1000 reports
|
|
@cached(max_size=1000)
|
|
def get_report_or_regression(
|
|
container: Container, filename: str
|
|
) -> Optional[Union[Report, RegressionReport]]:
|
|
file_path = "/".join([container, filename])
|
|
if not filename.endswith(".json"):
|
|
logging.error("get_report invalid extension: %s", file_path)
|
|
return None
|
|
|
|
blob = get_blob(container, filename, StorageType.corpus)
|
|
if blob is None:
|
|
logging.error("get_report invalid blob: %s", file_path)
|
|
return None
|
|
|
|
return parse_report_or_regression(blob, file_path=file_path)
|
|
|
|
|
|
def get_report(container: Container, filename: str) -> Optional[Report]:
|
|
result = get_report_or_regression(container, filename)
|
|
if isinstance(result, Report):
|
|
return result
|
|
return None
|