Skip to content

Commit

Permalink
feat: [VRD-818] Implement BuildScan class (#3757)
Browse files Browse the repository at this point in the history
  • Loading branch information
liuverta committed Apr 25, 2023
1 parent edc994d commit ab71768
Show file tree
Hide file tree
Showing 5 changed files with 190 additions and 2 deletions.
6 changes: 5 additions & 1 deletion client/verta/docs/_templates/class.rst
Expand Up @@ -5,6 +5,10 @@
.. autoclass:: {{ objname }}
:members:
:inherited-members:
{% if objname.endswith("Enum") -%}
:undoc-members:
:member-order: bysource
{% endif -%}
{% if 'ExperimentRun' != objname -%}
:exclude-members: log_code, get_code
{%- endif %}
{% endif -%}
26 changes: 26 additions & 0 deletions client/verta/tests/unit_tests/deployment/test_build_scan.py
@@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-

from hypothesis import given, HealthCheck, settings
import pytest

from tests.unit_tests.strategies import build_scan_dict

from verta._internal_utils import time_utils
from verta.endpoint.build import BuildScan, ScanProgressEnum, ScanResultEnum


@given(build_scan_dict=build_scan_dict())
def test_instantiation(build_scan_dict):
"""Verify a BuildScan object can be instantated from a dict."""
build_scan = BuildScan(build_scan_dict)

assert build_scan.date_updated == time_utils.datetime_from_iso(
build_scan_dict["date_updated"],
)
assert build_scan.progress == ScanProgressEnum(build_scan_dict["scan_status"])
if build_scan.progress == ScanProgressEnum.SCANNED:
assert build_scan.passed == (build_scan.result == ScanResultEnum.SAFE)
assert build_scan.result == ScanResultEnum(build_scan_dict["safety_status"])
else:
assert build_scan.passed is False
assert build_scan.result is None
50 changes: 49 additions & 1 deletion client/verta/tests/unit_tests/strategies.py
Expand Up @@ -10,7 +10,7 @@
from verta._internal_utils._utils import _VALID_FLAT_KEY_CHARS
from verta._protos.public.common import CommonService_pb2
from verta._protos.public.modeldb.versioning import Code_pb2, Dataset_pb2
from verta.endpoint import KafkaSettings
from verta.endpoint import KafkaSettings, build


@st.composite
Expand Down Expand Up @@ -169,3 +169,51 @@ def build_dict(draw) -> Dict[str, Any]:
"message": draw(st.text()),
"status": draw(st.text()),
}


@st.composite
def _build_scan_detail(draw) -> Dict[str, Any]:
"""For use in build_scan_dict."""
return {
"name": draw(st.text()),
"package": draw(st.text()),
"description": draw(st.text()),
"severity": draw(
st.sampled_from(
[
"critical",
"high",
"medium",
"low",
"informational",
"unknown",
],
)
),
}


@st.composite
def build_scan_dict(draw) -> Dict[str, Any]:
"""Generate a Verta build scan, as returned by /api/v1/deployment/builds/{build_id}/scan."""
d = {
"creator_request": {
"scan_external": draw(st.booleans()),
},
"date_updated": draw(st.datetimes()).isoformat(timespec="milliseconds") + "Z",
"details": None,
"id": draw(st.integers(min_value=1)),
"scan_status": draw(st.sampled_from(list(build.ScanProgressEnum))).value,
}
if d["scan_status"] == build.ScanProgressEnum.SCANNED:
d["safety_status"] = draw(st.sampled_from(list(build.ScanResultEnum))).value
if d["creator_request"]["scan_external"]:
d["scan_external_status"] = {
"safety_status": d["safety_status"],
"url": draw(st.text()),
}
else:
d["scanner"] = draw(st.text())
d["details"] = draw(st.lists(_build_scan_detail()))

return d
4 changes: 4 additions & 0 deletions client/verta/verta/endpoint/build/__init__.py
Expand Up @@ -5,11 +5,15 @@
from verta._internal_utils import documentation

from ._build import Build
from ._build_scan import BuildScan, ScanProgressEnum, ScanResultEnum


documentation.reassign_module(
[
Build,
BuildScan,
ScanProgressEnum,
ScanResultEnum,
],
module_name=__name__,
)
106 changes: 106 additions & 0 deletions client/verta/verta/endpoint/build/_build_scan.py
@@ -0,0 +1,106 @@
# -*- coding: utf-8 -*-

from datetime import datetime
from enum import Enum
from typing import Optional

from verta._internal_utils import _utils, time_utils


class ScanProgressEnum(str, Enum):
"""The current progress of a build scan.
For all intents and purposes, this can be treated as a :class:`str`.
Examples
--------
.. code-block:: python
assert build.get_scan().progress == "scanned"
"""

UNSCANNED = "unscanned"
SCANNING = "scanning"
SCANNED = "scanned"
ERROR = "error"


class ScanResultEnum(str, Enum):
"""The result of a build scan.
For all intents and purposes, this can be treated as a :class:`str`.
Examples
--------
.. code-block:: python
assert build.get_scan().result == "safe"
"""

UNKNOWN = "unknown"
SAFE = "safe"
UNSAFE = "unsafe"


class BuildScan:
"""A scan of a Verta model build.
There should not be a need to instantiate this class directly; please use
:meth:`Build.get_scan() <verta.endpoint.build.Build.get_scan>` instead.
Attributes
----------
date_updated : timezone-aware :class:`~datetime.datetime`
The date and time when this scan was performed/updated.
progress : :class:`ScanProgressEnum`
The current progress of this scan.
result : :class:`ScanResultEnum` or None
The result of this scan. ``None`` is returned if this scan is not yet
finished, and therefore has no result.
passed : bool
Whether this scan finished and passed. This property is for
convenience, equivalent to
.. code-block:: python
(build_scan.progress == "scanned") and (build_scan.result == "safe")
"""

def __init__(self, json):
self._json = json

def __repr__(self):
detail_str = f'progress "{self.progress.value}"'
if self.result is not None:
detail_str += f', result "{self.result.value}"'

return f"<BuildScan ({detail_str})>"

@classmethod
def _get(cls, conn: _utils.Connection, build_id: int):
url = f"{conn.scheme}://{conn.socket}/api/v1/deployment/builds/{build_id}/scan"
response = _utils.make_request("GET", url, conn)
_utils.raise_for_http_error(response)

return cls(response.json())

@property
def date_updated(self) -> datetime:
return time_utils.datetime_from_iso(self._json["date_updated"])

@property
def progress(self) -> ScanProgressEnum:
return ScanProgressEnum(self._json["scan_status"])

@property
def result(self) -> Optional[ScanResultEnum]:
if self.progress != ScanProgressEnum.SCANNED:
return None
return ScanResultEnum(self._json["safety_status"])

@property
def passed(self) -> bool:
return self.result == ScanResultEnum.SAFE

0 comments on commit ab71768

Please sign in to comment.