Merge pull request #127 from fossology/feat/api-v2

feat(v2): support API version v2 by default

Author: deveaud-m

fossology/__init__.py

@@ -25,7 +25,13 @@
 
 
 def fossology_token(
-    url, username, password, token_name, token_scope=TokenScope.READ, token_expire=None
+    url,
+    username,
+    password,
+    token_name,
+    token_scope=TokenScope.READ,
+    token_expire=None,
+    version="v2",
 ):
     """Generate an API token using username/password
 
@@ -35,7 +41,7 @@ def fossology_token(
 
     >>> from fossology import fossology_token # doctest: +SKIP
     >>> from fossology.obj import TokenScope # doctest: +SKIP
-    >>> token = fossology_token("https://fossology.example.com", "Me", "MyPassword", "MyToken") # doctest: +SKIP
+    >>> token = fossology_token("https://fossology.example.com/repo", "Me", "MyPassword", "MyToken", version="v2") # doctest: +SKIP
 
 
     :param url: the URL of the Fossology server
@@ -44,30 +50,46 @@ def fossology_token(
     :param name: the name of the token
     :param scope: the scope of the token (default: TokenScope.READ)
     :param expire: the expire date of the token, e.g. 2019-12-25 (default: max. 30 days)
+    :param version: the version of the API to use (default: "v2")
     :type url: string
     :type username: string
     :type password: string
     :type name: string
     :type scope: TokenScope
     :type expire: string
+    :type version: string
     :return: the new token
     :rtype: string
     :raises AuthenticationError: if the username or password is incorrect
     :raises FossologyApiError: if another error occurs
     """
-    data = {
-        "username": username,
-        "password": password,
-        "token_name": token_name,
-        "token_scope": token_scope.value,
-    }
+    if version == "v2":
+        data = {
+            "username": username,
+            "password": password,
+            "tokenName": token_name,
+            "tokenScope": token_scope.value,
+        }
+    else:
+        data = {
+            "username": username,
+            "password": password,
+            "token_name": token_name,
+            "token_scope": token_scope.value,
+        }
     if token_expire:
-        data["token_expire"] = token_expire
+        if version == "v2":
+            data["tokenExpire"] = token_expire
+        else:
+            data["token_expire"] = token_expire
     else:
         now = date.today()
-        data["token_expire"] = str(now + timedelta(days=30))
+        if version == "v2":
+            data["tokenExpire"] = str(now + timedelta(days=30))
+        else:
+            data["token_expire"] = str(now + timedelta(days=30))
     try:
-        response = requests.post(url + "/api/v1/tokens", data=data)
+        response = requests.post(url + "/api/" + version + "/tokens", data=data)
         if response.status_code == 201:
             token = response.json()["Authorization"]
             return token.replace("Bearer ", "")
@@ -96,19 +118,20 @@ class Fossology(
 
     :param url: URL of the Fossology instance
     :param token: The API token generated using the Fossology UI
+    :param version: the version of the API to use (default: "v2")
     :type url: str
     :type token: str
+    :type version: str
     :raises FossologyApiError: if a REST call failed
     :raises AuthenticationError: if the user couldn't be authenticated
     """
 
-    def __init__(self, url, token, name=None):
+    def __init__(self, url, token, version="v2"):
         self.host = url
         self.token = token
         self.users = list()
         self.folders = list()
-
-        self.api = f"{self.host}/api/v2"
+        self.api = f"{self.host}/api/{version}"
         self.session = requests.Session()
         self.session.headers.update({"Authorization": f"Bearer {self.token}"})
         self.info = self.get_info()

fossology/obj.py

@@ -571,6 +571,13 @@ def __str__(self):
 
     @classmethod
     def from_json(cls, json_dict):
+        for key in ("viewInfo", "metaInfo", "packageInfo", "tagInfo", "reuseInfo"):
+            try:
+                json_dict[key.replace("I", "_i")] = json_dict[key]
+                del json_dict[key]
+            except KeyError:
+                pass
+
         return cls(**json_dict)
 
 
@@ -586,6 +593,9 @@ class Upload(object):
     :param description: further information about the upload
     :param uploadname: the name of the upload (default: the name of the upload file)
     :param uploaddate: the date of the upload
+    :param assignee: the user who is assigned to the upload
+    :param assigneeDate: the date of the assignment
+    :param closingDate: the date of the closing
     :param hash: the hash data of the uploaded file
     :param kwargs: handle any other upload information provided by the fossology instance
     :type folderid: int
@@ -594,6 +604,9 @@ class Upload(object):
     :type description: string
     :type uploadname: string
     :type uploaddate: string
+    :type assignee: string
+    :type assigneeDate: string
+    :type closingDate: string
     :type hash: Hash
     :type kwargs: key word argument
     """
@@ -606,8 +619,9 @@ def __init__(
         description,
         uploadname,
         uploaddate,
-        filesize=None,
-        filesha1=None,
+        assignee=None,
+        assigneeDate=None,
+        closingDate=None,
         hash=None,
         **kwargs,
     ):
@@ -617,30 +631,26 @@ def __init__(
         self.description = description
         self.uploadname = uploadname
         self.uploaddate = uploaddate
-        if filesize and filesha1:
-            self.filesize = filesize
-            self.filesha1 = filesha1
-            self.hash = None
-        else:
-            self.filesize = None
-            self.filesha1 = None
-            self.hash = Hash.from_json(hash)
+        self.assignee = (assignee,)
+        self.assigneeDate = (assigneeDate,)
+        self.closeDate = (closingDate,)
+        self.hash = Hash.from_json(hash)
         self.additional_info = kwargs
 
     def __str__(self):
-        if self.filesize:
-            return (
-                f"Upload '{self.uploadname}' ({self.id}, {self.filesize}B, {self.filesha1}) "
-                f"in folder {self.foldername} ({self.folderid})"
-            )
-        else:
-            return (
-                f"Upload '{self.uploadname}' ({self.id}, {self.hash.size}B, {self.hash.sha1}) "
-                f"in folder {self.foldername} ({self.folderid})"
-            )
+        return (
+            f"Upload '{self.uploadname}' ({self.id}, {self.hash.size}B, {self.hash.sha1}) "
+            f"in folder {self.foldername} ({self.folderid})"
+        )
 
     @classmethod
     def from_json(cls, json_dict):
+        for key in ("folderId", "folderName", "uploadName", "uploadDate"):
+            try:
+                json_dict[key.lower()] = json_dict[key]
+                del json_dict[key]
+            except KeyError:
+                pass
         return cls(**json_dict)
 
 

fossology/uploads.py

@@ -35,6 +35,8 @@ def list_uploads_parameters(
     status: ClearingStatus | None = None,
     assignee: str | None = None,
     since: str | None = None,
+    group: str | None = None,
+    limit: str | None = None,
 ) -> dict:
     """Helper function to list of query parameters for GET /uploads endpoint"""
     date_pattern = re.compile("^[0-9]{4}-[0-9]{2}-[0-9]{2}")
@@ -56,6 +58,10 @@ def list_uploads_parameters(
             )
         else:
             params["since"] = since
+    if group:
+        params["groupName"] = group
+    if limit:
+        params["limit"] = limit
     return params
 
 
@@ -285,16 +291,10 @@ def upload_file(
                 upload = self.detail_upload(
                     response.json()["message"], group, wait_time
                 )
-                if upload.filesize:
-                    logger.info(
-                        f"Upload {upload.uploadname} ({upload.filesize}) "
-                        f"has been uploaded on {upload.uploaddate}"
-                    )
-                else:
-                    logger.info(
-                        f"Upload {upload.uploadname} ({upload.hash.size}) "
-                        f"has been uploaded on {upload.uploaddate}"
-                    )
+                logger.info(
+                    f"Upload {upload.uploadname} ({upload.hash.size}) "
+                    f"has been uploaded on {upload.uploaddate}"
+                )
                 return upload
             except TryAgain:
                 description = f"Upload of {source} failed"
@@ -391,6 +391,7 @@ def upload_licenses(
         headers = {}
         if group:
             headers["groupName"] = group
+            params["groupName"] = group  # type: ignore
 
         response = self.session.get(
             f"{self.api}/uploads/{upload.id}/licenses", params=params, headers=headers
@@ -545,6 +546,8 @@ def list_uploads(
             status=status,
             assignee=assignee,
             since=since,
+            group=group,
+            limit=page_size,
         )
         uploads_list = list()
         if all_pages:
@@ -554,6 +557,7 @@ def list_uploads(
             x_total_pages = page
         while page <= x_total_pages:
             headers["page"] = str(page)
+            params["page"] = str(page)
             response = self.session.get(
                 f"{self.api}/uploads", headers=headers, params=params
             )
@@ -643,8 +647,10 @@ def move_upload(self, upload: Upload, folder: Folder, action: str):
         :raises FossologyApiError: if the REST call failed
         :raises AuthorizationError: if the REST call is not authorized
         """
-        headers = {"folderId": str(folder.id), "action": action}
-        response = self.session.put(f"{self.api}/uploads/{upload.id}", headers=headers)
+        params = {"folderId": str(folder.id), "action": action}
+        response = self.session.put(
+            f"{self.api}/uploads/{upload.id}", headers=params, params=params
+        )
 
         if response.status_code == 202:
             logger.info(
@@ -771,7 +777,7 @@ def upload_permissions(
         :raises AuthorizationError: if the REST call is not authorized
         """
         response = self.session.get(f"{self.api}/uploads/{upload.id}/perm-groups")
-
+        print(response.request.url)
         if response.status_code == 200:
             return UploadPermGroups.from_json(response.json())
 

tests/conftest.py

@@ -164,7 +164,22 @@ def foss_token(foss_server: str) -> str:
 @pytest.fixture(scope="session")
 def foss(foss_server: str, foss_token: str, foss_agents: Agents) -> fossology.Fossology:
     try:
-        foss = fossology.Fossology(foss_server, foss_token, "fossy")
+        foss = fossology.Fossology(foss_server, foss_token)
+    except (FossologyApiError, AuthenticationError) as error:
+        exit(error.message)
+
+    # Configure all license agents besides 'ojo'
+    foss.user.agents = foss_agents
+    yield foss
+    foss.close()
+
+
+@pytest.fixture(scope="session")
+def foss_v1(
+    foss_server: str, foss_token: str, foss_agents: Agents
+) -> fossology.Fossology:
+    try:
+        foss = fossology.Fossology(foss_server, foss_token, version="v1")
     except (FossologyApiError, AuthenticationError) as error:
         exit(error.message)
 
@@ -215,6 +230,24 @@ def upload(
     time.sleep(5)
 
 
+@pytest.fixture(scope="function")
+def upload_v1(
+    foss_v1: fossology.Fossology,
+    test_file_path: str,
+) -> Generator:
+    upload = foss_v1.upload_file(
+        foss_v1.rootFolder,
+        file=test_file_path,
+        description="Test upload via fossology-python lib",
+        access_level=AccessLevel.PUBLIC,
+        wait_time=5,
+    )
+    jobs_lookup(foss_v1, upload)
+    yield upload
+    foss_v1.delete_upload(upload)
+    time.sleep(5)
+
+
 @pytest.fixture(scope="session")
 def upload_with_jobs(
     foss: fossology.Fossology, test_file_path: str, foss_schedule_agents: dict

tests/test_items.py

@@ -16,6 +16,12 @@ def test_item_info(foss: Fossology, upload_with_jobs: Upload):
     assert info.meta_info
 
 
+def test_item_info_v1(foss_v1: Fossology, upload_with_jobs: Upload):
+    files, _ = foss_v1.search(license="BSD")
+    info: FileInfo = foss_v1.item_info(upload_with_jobs, files[0].uploadTreeId)
+    assert info.meta_info
+
+
 def test_item_info_with_unknown_item_raises_api_error(
     foss: Fossology, upload_with_jobs: Upload
 ):

tests/test_uploads.py

@@ -7,7 +7,6 @@
 import time
 from datetime import date, timedelta
 from pathlib import Path
-from unittest.mock import Mock
 
 import pytest
 import responses
@@ -31,6 +30,19 @@ def test_upload_sha1(upload: Upload):
     )
 
 
+def test_upload_v1(upload_v1: Upload):
+    assert upload_v1.uploadname == "base-files_11.tar.xz"
+    assert upload_v1.hash.sha1 == "D4D663FC2877084362FB2297337BE05684869B00"
+    assert str(upload_v1) == (
+        f"Upload '{upload_v1.uploadname}' ({upload_v1.id}, {upload_v1.hash.size}B, {upload_v1.hash.sha1}) "
+        f"in folder {upload_v1.foldername} ({upload_v1.folderid})"
+    )
+    assert str(upload_v1.hash) == (
+        f"File SHA1: {upload_v1.hash.sha1} MD5 {upload_v1.hash.md5} "
+        f"SH256 {upload_v1.hash.sha256} Size {upload_v1.hash.size}B"
+    )
+
+
 def test_get_upload_unauthorized(foss: Fossology, upload: Upload):
     with pytest.raises(AuthorizationError) as excinfo:
         foss.detail_upload(
@@ -157,13 +169,14 @@ def test_move_upload_to_non_existing_folder(foss: Fossology, upload: Upload):
 
 @responses.activate
 def test_move_upload_error(foss: Fossology, foss_server: str, upload: Upload):
+    folder = Folder(secrets.randbelow(1000), "Folder", "", foss.rootFolder)
     responses.add(
         responses.PUT,
         f"{foss_server}/api/v2/uploads/{upload.id}",
         status=500,
     )
     with pytest.raises(FossologyApiError):
-        foss.move_upload(upload, Mock(), "move")
+        foss.move_upload(upload, folder, "move")
 
 
 def test_update_upload(foss: Fossology, upload: Upload):