Feature/more tests

owasp_dt/models/vulnerability.py

@@ -33,8 +33,8 @@ class Vulnerability:
     Attributes:
         vuln_id (str):
         source (str):
-        friendly_vuln_id (str):
         uuid (UUID):
+        friendly_vuln_id (Union[Unset, str]):
         title (Union[Unset, str]):
         sub_title (Union[Unset, str]):
         description (Union[Unset, str]):
@@ -76,8 +76,8 @@ class Vulnerability:
 
     vuln_id: str
     source: str
-    friendly_vuln_id: str
     uuid: UUID
+    friendly_vuln_id: Union[Unset, str] = UNSET
     title: Union[Unset, str] = UNSET
     sub_title: Union[Unset, str] = UNSET
     description: Union[Unset, str] = UNSET
@@ -122,10 +122,10 @@ def to_dict(self) -> dict[str, Any]:
 
         source = self.source
 
-        friendly_vuln_id = self.friendly_vuln_id
-
         uuid = str(self.uuid)
 
+        friendly_vuln_id = self.friendly_vuln_id
+
         title = self.title
 
         sub_title = self.sub_title
@@ -243,10 +243,11 @@ def to_dict(self) -> dict[str, Any]:
             {
                 "vulnId": vuln_id,
                 "source": source,
-                "friendlyVulnId": friendly_vuln_id,
                 "uuid": uuid,
             }
         )
+        if friendly_vuln_id is not UNSET:
+            field_dict["friendlyVulnId"] = friendly_vuln_id
         if title is not UNSET:
             field_dict["title"] = title
         if sub_title is not UNSET:
@@ -342,10 +343,10 @@ def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
 
         source = d.pop("source")
 
-        friendly_vuln_id = d.pop("friendlyVulnId")
-
         uuid = UUID(d.pop("uuid"))
 
+        friendly_vuln_id = d.pop("friendlyVulnId", UNSET)
+
         title = d.pop("title", UNSET)
 
         sub_title = d.pop("subTitle", UNSET)
@@ -482,8 +483,8 @@ def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
         vulnerability = cls(
             vuln_id=vuln_id,
             source=source,
-            friendly_vuln_id=friendly_vuln_id,
             uuid=uuid,
+            friendly_vuln_id=friendly_vuln_id,
             title=title,
             sub_title=sub_title,
             description=description,

patch.json

@@ -133,6 +133,7 @@
         }
       },
       "Vulnerability": {
+        "required" : [ "source", "uuid", "vulnId" ],
         "properties": {
           "findingAttribution": {
             "$ref": "#/components/schemas/FindingAttrib"

pyproject.toml

@@ -24,6 +24,8 @@ test = [
     "pytest>=7",
     "pytest-depends",
     "pytest-cov",
-    "dotenv",
+    "dotenv== 0.9.9",
     "openapi-python-client",
+    "tinystream==0.1.18",
+    "is_empty==1.0.1",
 ]

schema.json

@@ -15231,7 +15231,6 @@
       },
       "Vulnerability": {
         "required": [
-          "friendlyVulnId",
           "source",
           "uuid",
           "vulnId"

test/__init__.py

@@ -1,30 +1,43 @@
+import math
+import time
 from pathlib import Path
+from typing import Callable
 
 from dotenv import load_dotenv
-from owasp_dt import Client
-import owasp_dt
+
 from test import config
-__base_dir = Path(__file__).parent
-
-test_project_name = "test-api"
-
-def create_client_from_env() -> owasp_dt.Client:
-    base_url = config.reqenv("OWASP_DTRACK_URL")
-    return Client(
-        base_url=f"{base_url}/api",
-        headers={
-            "X-Api-Key": config.reqenv("OWASP_DTRACK_API_KEY")
-        },
-        verify_ssl=config.getenv("OWASP_DTRACK_VERIFY_SSL", "1", config.parse_true),
-        raise_on_unexpected_status=False,
-        httpx_args={
-            "proxy": config.getenv("HTTPS_PROXY", lambda: config.getenv("HTTP_PROXY", None)),
-            #"no_proxy": getenv("NO_PROXY", "")
-        }
-    )
+
+base_dir = Path(__file__).parent
+
+project_name = "test-api"
+upload_token: str | None = None
+project_uuid: str | None = None
+mit_license_uuid: str | None = None
+
+def retry(callable: Callable, seconds: float, wait_time: float = 3):
+    retries = math.ceil(seconds / wait_time)
+    #start_date = datetime.now()
+    exception = None
+    ret = None
+    for i in range(retries):
+        try:
+            exception = None
+            ret = callable()
+            break
+        except Exception as e:
+            exception = e
+        time.sleep(wait_time)
+
+    if exception:
+        raise exception
+        #raise Exception(f"{exception} after {datetime.now()-start_date}")
+
+    return ret
+
 
 def setup_module():
-    assert load_dotenv(__base_dir / "test.env")
+    assert load_dotenv(base_dir / "test.env")
+
 
 def teardown_module():
     pass

test/api.py

@@ -0,0 +1,44 @@
+from typing import Generator, Callable, TypeVar
+
+from owasp_dt import Client
+from owasp_dt.api.project_property import create_property_1, update_property
+from owasp_dt.models import ProjectProperty
+from owasp_dt.types import Response
+from test import config
+
+
+def create_client_from_env() -> Client:
+    base_url = config.reqenv("OWASP_DTRACK_URL")
+    return Client(
+        base_url=f"{base_url}/api",
+        headers={
+            "X-Api-Key": config.reqenv("OWASP_DTRACK_API_KEY")
+        },
+        verify_ssl=config.getenv("OWASP_DTRACK_VERIFY_SSL", "1", config.parse_true),
+        raise_on_unexpected_status=False,
+        httpx_args={
+            "proxy": config.getenv("HTTPS_PROXY", lambda: config.getenv("HTTP_PROXY", None)),
+            #"no_proxy": getenv("NO_PROXY", "")
+        }
+    )
+
+def upsert_project_property(client: Client, uuid: str, property: ProjectProperty):
+    resp = create_property_1.sync_detailed(client=client, uuid=uuid, body=property)
+    if resp.status_code == 409:
+        resp = update_property.sync_detailed(client=client, uuid=uuid, body=property)
+
+    assert resp.status_code in [200, 201]
+
+T = TypeVar('T')
+
+def page_result(cb: Callable[[int], Response[list[T]]]) -> Generator[list[T]]:
+    page_number = 0
+    while True:
+        page_number += 1
+        resp = cb(page_number)
+        assert resp.status_code == 200
+        items = resp.parsed
+        if len(items) == 0:
+            break
+        else:
+            yield items

test/conftest.py

@@ -1,8 +1,8 @@
 import pytest
 
-from test import create_client_from_env
+from test import api
 
 
 @pytest.fixture
 def client():
-    yield create_client_from_env()
+    yield api.create_client_from_env()

test/test_projects.py

@@ -0,0 +1,20 @@
+import pytest
+
+import owasp_dt
+import test
+from owasp_dt.api.metrics import get_project_current_metrics
+from owasp_dt.api.project import get_projects
+
+
+@pytest.mark.depends(on=['test/test_upload.py::test_upload_sbom'])
+def test_search_project_by_name(client: owasp_dt.Client):
+    resp = get_projects.sync_detailed(client=client, name=test.project_name)
+    projects = resp.parsed
+    assert len(projects) > 0
+    assert projects[0].uuid is not None
+    test.project_uuid = projects[0].uuid
+
+@pytest.mark.depends(on=['test/test_upload.py::test_get_scan_status', 'test_search_project_by_name'])
+def test_get_project_metrics(client: owasp_dt.Client):
+    resp = get_project_current_metrics.sync_detailed(client=client, uuid=test.project_uuid)
+    metrics = resp.parsed

test/test_property.py

@@ -0,0 +1,37 @@
+import pytest
+from tinystream import Opt
+
+import owasp_dt
+import test
+from owasp_dt.api.project import get_project
+from owasp_dt.models import ProjectPropertyPropertyType, ProjectProperty
+from test import api
+
+
+@pytest.mark.depends(on=['test/test_projects.py::test_search_project_by_name'])
+def test_upsert_project_property(client: owasp_dt.Client):
+    property = ProjectProperty(
+        group_name="owasp-dtrack-python-client",
+        property_name="test",
+        property_type=ProjectPropertyPropertyType.STRING,
+        property_value="set",
+        description="Custom property test"
+    )
+    api.upsert_project_property(client=client, uuid=test.project_uuid, property=property)
+
+    def _filter_property(property:ProjectProperty):
+        return property.group_name == "owasp-dtrack-python-client" and property.property_name == "test"
+
+    resp = get_project.sync_detailed(client=client, uuid=test.project_uuid)
+    project = resp.parsed
+    opt_property = Opt(project).map_key("properties").stream().filter(_filter_property).next()
+    assert opt_property.present
+    assert opt_property.get().property_value == "set"
+
+    property.property_value = "new_value"
+    api.upsert_project_property(client=client, uuid=test.project_uuid, property=property)
+    resp = get_project.sync_detailed(client=client, uuid=test.project_uuid)
+    project = resp.parsed
+    opt_property = Opt(project).map_key("properties").stream().filter(_filter_property).next()
+    assert opt_property.present
+    assert opt_property.get().property_value == "new_value"

test/test_teams.py

@@ -0,0 +1,33 @@
+import pytest
+from tinystream import Stream
+
+import owasp_dt
+from owasp_dt.api.team import get_teams, create_team, delete_team
+from owasp_dt.models import Team
+
+test_team = Team(
+    uuid="",
+    name="test-team",
+)
+
+def test_create_team(client: owasp_dt.Client):
+    global test_team
+    resp = create_team.sync_detailed(client=client, body=test_team)
+    assert resp.status_code == 201
+    test_team = resp.parsed
+
+@pytest.mark.depends(on=["test_create_team"])
+def test_created_team(client: owasp_dt.Client):
+    teams = get_teams.sync(client=client)
+    assert Stream(teams).filter(lambda team: team.name == "test-team").count() == 1
+
+
+@pytest.mark.depends(on=["test_created_team"])
+def test_delete_team(client: owasp_dt.Client):
+    resp = delete_team.sync_detailed(client=client, body=test_team)
+    assert resp.status_code == 204
+
+@pytest.mark.depends(on=["test_delete_team"])
+def test_deleted_team(client: owasp_dt.Client):
+    teams = get_teams.sync(client=client)
+    assert Stream(teams).filter(lambda team: team.name == "test-team").count() == 0

test/test_upload.py

@@ -0,0 +1,37 @@
+from pathlib import Path
+from time import sleep
+
+import pytest
+
+import owasp_dt
+import test
+from owasp_dt.api.bom import upload_bom
+from owasp_dt.api.event import is_token_being_processed_1
+from owasp_dt.models import UploadBomBody, IsTokenBeingProcessedResponse
+
+def test_upload_sbom(client: owasp_dt.Client):
+    with open(test.base_dir / "files/test.sbom.xml") as sbom_file:
+        resp = upload_bom.sync_detailed(client=client, body=UploadBomBody(
+            project_name=test.project_name,
+            auto_create=True,
+            bom=sbom_file.read()
+        ))
+        upload = resp.parsed
+        assert upload is not None, "API call failed. Check client permissions."
+        assert upload.token is not None
+        test.upload_token = upload.token
+
+
+@pytest.mark.depends(on=['test_upload_sbom'])
+def test_get_scan_status(client: owasp_dt.Client):
+    max_tries = 10
+    i = 0
+    for i in range(max_tries):
+        resp = is_token_being_processed_1.sync_detailed(client=client, uuid=test.upload_token)
+        status = resp.parsed
+        assert isinstance(status, IsTokenBeingProcessedResponse)
+        if not status.processing:
+            break
+        sleep(1)
+
+    assert i < max_tries, f"Scan not finished within {max_tries} seconds"