feat: add new upload parameters for version 1.3.4, related to #52

fossology/uploads.py

@@ -3,11 +3,12 @@
 import json
 import logging
 import time
+import re
 
 from tenacity import TryAgain, retry, retry_if_exception_type, stop_after_attempt
 
 from fossology.exceptions import AuthorizationError, FossologyApiError
-from fossology.obj import Licenses, Summary, Upload, get_options
+from fossology.obj import ClearingStatus, Licenses, Summary, Upload, get_options
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@@ -405,26 +406,38 @@ def delete_upload(self, upload, group=None):
 
     def list_uploads(
         self,
-        folder=None,
-        group=None,
-        recursive=True,
+        folder: int = None,
+        group: str = None,
+        recursive: bool = True,
+        name: str = None,
+        status: ClearingStatus = None,
+        assignee: str = None,
+        since: str = None,
         page_size=100,
         page=1,
         all_pages=False,
     ):
-        """Get all uploads available to the registered user
+        """Get uploads according to filtering criteria (or all available)
 
         API Endpoint: GET /uploads
 
         :param folder: only list uploads from the given folder
         :param group: list uploads from a specific group (not only your own uploads) (default: None)
         :param recursive: wether to list uploads from children folders or not (default: True)
+        :param name: filter pattern for name and description
+        :param status: status of uploads
+        :param assignee: user name to which uploads are assigned to or "-me-" or "-unassigned-"
+        :param since: uploads since given date in YYYY-MM-DD format
         :param page_size: limit the number of uploads per page (default: 100)
         :param page: the number of the page to fetch uploads from (default: 1)
         :param all_pages: get all uploads (default: False)
         :type folder: Folder
         :type group: string
         :type recursive: boolean
+        :type name: str
+        :type status: ClearingStatus
+        :type assignee: str
+        :type since: str
         :type page_size: int
         :type page: int
         :type all_pages: boolean
@@ -433,6 +446,7 @@ def list_uploads(
         :raises FossologyApiError: if the REST call failed
         :raises AuthorizationError: if the user can't access the group
         """
+        date_pattern = re.compile("^[0-9]{4}-[0-9]{2}-[0-9]{2}")
         params = {}
         headers = {"limit": str(page_size)}
         if group:
@@ -441,6 +455,19 @@ def list_uploads(
             params["folderId"] = folder.id
         if not recursive:
             params["recursive"] = "false"
+        if name:
+            params["name"] = name
+        if status:
+            params["status"] = status.value
+        if assignee:
+            params["assignee"] = assignee
+        if since:
+            if not date_pattern.match(since):
+                logger.error(
+                    f"Date format for 'since' query parameter {since} does not match expected format YYYY-MM-DD"
+                )
+            else:
+                params["since"] = since
 
         uploads_list = list()
         if all_pages:

tests/test_uploads.py

@@ -7,9 +7,10 @@
 import pytest
 import responses
 
+from datetime import date, timedelta
 from fossology import Fossology, versiontuple
 from fossology.exceptions import AuthorizationError, FossologyApiError
-from fossology.obj import AccessLevel, Folder, SearchTypes, Upload
+from fossology.obj import AccessLevel, ClearingStatus, Folder, SearchTypes, Upload
 
 
 def test_upload_sha1(foss: Fossology, upload: Upload):
@@ -97,6 +98,21 @@ def test_get_uploads(foss: Fossology, upload_folder: Folder, test_file_path: str
         assert len(foss.list_uploads(folder=upload_subfolder)[0]) == 1
 
 
+def test_filter_uploads(foss: Fossology, upload: Upload):
+    today = date.today().isoformat()
+    tomorrow = (date.today() + timedelta(days=1)).isoformat()
+    # Uploads filtering has been enhance with API version 1.3.4
+    if versiontuple(foss.version) >= versiontuple("1.3.4"):
+        assert len(foss.list_uploads(assignee="-me-")[0]) == 0
+        assert len(foss.list_uploads(assignee="-unassigned-")[0]) == 1
+        assert len(foss.list_uploads(name="Non-existing upload")[0]) == 0
+        assert len(foss.list_uploads(name="Test upload")[0]) == 1
+        assert len(foss.list_uploads(status=ClearingStatus.OPEN)[0]) == 1
+        assert len(foss.list_uploads(status=ClearingStatus.CLOSED)[0]) == 0
+        assert len(foss.list_uploads(since=today)[0]) == 1
+        assert len(foss.list_uploads(since=tomorrow)[0]) == 0
+
+
 def test_upload_from_vcs(foss: Fossology):
     vcs = {
         "vcsType": "git",