Make changes according to the review comment

- Refactor the existing deploy method to push artifact's checksums

Author: aslheyrr

README.md

@@ -33,7 +33,6 @@ This library enables you to manage Artifactory resources such as users, groups,
     + [Get the information about a file or folder](#get-the-information-about-a-file-or-folder)
     + [Deploy an artifact](#deploy-an-artifact)
     + [Deploy an artifact with properties](#deploy-an-artifact-with-properties)
-    + [Deploy an artifact with checksums](#deploy-an-artifact-with-checksums)
     + [Deploy an artifact by checksums](#deploy-an-artifact-by-checksums)
     + [Download an artifact](#download-an-artifact)
     + [Retrieve artifact list](#retrieve-artifact-list)
@@ -425,16 +424,6 @@ artifact = art.artifacts.deploy("<LOCAL_FILE_LOCATION>", "<ARTIFACT_PATH_IN_ARTI
 # artifact = art.artifacts.deploy("Desktop/myNewFile.txt", "my-repository/my/new/artifact/directory/file.txt", {"retention": ["30"]})
 ```
 
-#### Deploy an artifact with checksums
-
-```python
-artifact = art.artifacts.deploy("<LOCAL_FILE_LOCATION>", "<ARTIFACT_PATH_IN_ARTIFACTORY>", "<CHECKSUM_ALGORITHMS>")
-# artifact = art.artifacts.deploy("Desktop/myNewFile.txt", "my-repository/my/new/artifact/directory/file.txt", checksum_algorithms=["md5", "sha1", "sha256"] )
-```
-
-Provides locally calculated checksums for files during deployment so that Artifactory can verify the authenticity of artifacts.
-
-
 #### Deploy an artifact by checksums
 
 ```python

pyartifactory/exception.py

@@ -58,7 +58,3 @@ class InvalidTokenDataError(ArtifactoryError):
 
 class BuildNotFoundError(ArtifactoryError):
     """Requested build were not found"""
-
-
-class InvalidAlgorithmError(ArtifactoryError):
-    """The Algoritnm is not supported by Artifactory."""

pyartifactory/models/artifact.py

@@ -10,33 +10,28 @@
 
 from pydantic import BaseModel
 
-from pyartifactory.exception import InvalidAlgorithmError
-
 
 class Checksums(BaseModel):
     """Models a checksum."""
 
-    sha1: Optional[str] = None
-    md5: Optional[str] = None
-    sha256: Optional[str] = None
+    sha1: str
+    md5: str
+    sha256: str
 
     @classmethod
-    def generate(cls, file_: Path, algorithms: List[str]) -> Checksums:
+    def generate(cls, file_: Path) -> Checksums:
         block_size: int = 65536
         mapping: dict[str, Callable[[], Any]] = {"md5": hashlib.md5, "sha1": hashlib.sha1, "sha256": hashlib.sha256}
         results = {}
 
-        for algorithm in algorithms:
-            if algorithm in mapping:
-                hasher = mapping[algorithm]()
-                with file_.absolute().open("rb") as fd:
+        for algorithm, hashing_function in mapping.items():
+            hasher = hashing_function()
+            with file_.absolute().open("rb") as fd:
+                buf = fd.read(block_size)
+                while len(buf) > 0:
+                    hasher.update(buf)
                     buf = fd.read(block_size)
-                    while len(buf) > 0:
-                        hasher.update(buf)
-                        buf = fd.read(block_size)
-                results[algorithm] = hasher.hexdigest()
-            else:
-                raise InvalidAlgorithmError(f"'{algorithm}' is not a supported checksum algorithm")
+            results[algorithm] = hasher.hexdigest()
 
         return cls(**results)
 

pyartifactory/objects/artifact.py

@@ -82,7 +82,6 @@ def deploy(
         artifact_path: Union[Path, str],
         properties: Optional[Dict[str, List[str]]] = None,
         checksum_enabled: bool = False,
-        checksum_algorithms: Optional[List[str]] = None,
     ) -> ArtifactInfoResponse:
         """
         Deploy a file or directory.
@@ -99,30 +98,22 @@ def deploy(
                 for file in files:
                     self.deploy(Path(f"{root}/{file}"), Path(f"{new_root}/{file}"), properties, checksum_enabled)
         else:
-            checksum_headers: Dict[str, str] = {}
-
-            if checksum_enabled:
-                checksum_headers["X-Checksum-Deploy"] = "true"
-                checksum_algorithms = ["sha1", "sha256", "md5"]
-
-            if checksum_algorithms is not None:
-                artifact_check_sums = Checksums.generate(local_file, checksum_algorithms)
-                if artifact_check_sums.md5:
-                    checksum_headers["X-Checksum"] = artifact_check_sums.md5
-                if artifact_check_sums.sha1:
-                    checksum_headers["X-Checksum-Sha1"] = artifact_check_sums.sha1
-                if artifact_check_sums.sha256:
-                    checksum_headers["X-Checksum-Sha256"] = artifact_check_sums.sha256
-
+            properties_param_str = ""
+            if properties is not None:
+                properties_param_str = self._format_properties(properties)
+            route = ";".join(s for s in [artifact_folder.as_posix(), properties_param_str] if s)
+            artifact_check_sums = Checksums.generate(local_file)
+            headers = {
+                "X-Checksum-Sha1": artifact_check_sums.sha1,
+                "X-Checksum-Sha256": artifact_check_sums.sha256,
+                "X-Checksum": artifact_check_sums.md5,
+            }
             if checksum_enabled:
+                headers["X-Checksum-Deploy"] = "true"
                 try:
-                    properties_param_str = ""
-                    if properties is not None:
-                        properties_param_str = self._format_properties(properties)
-                    route = ";".join(s for s in [artifact_folder.as_posix(), properties_param_str] if s)
                     self._put(
                         route=route,
-                        headers=checksum_headers,
+                        headers=headers,
                     )
                 except requests.exceptions.HTTPError as error:
                     if error.response.status_code == 404:
@@ -135,11 +126,7 @@ def deploy(
                     raise ArtifactoryError from error
             else:
                 with local_file.open("rb") as stream:
-                    properties_param_str = ""
-                    if properties is not None:
-                        properties_param_str = self._format_properties(properties)
-                    route = ";".join(s for s in [artifact_folder.as_posix(), properties_param_str] if s)
-                    self._put(route=route, headers=checksum_headers, data=stream)
+                    self._put(route=route, headers=headers, data=stream)
 
             logger.debug("Artifact %s successfully deployed", local_file)
         return self.info(artifact_folder)

tests/test_artifacts.py

@@ -601,7 +601,7 @@ def test_update_property_fail_bad_value():
     ],
 )
 def test_checksum_defined_file(file_path: Path, expected_sha1: str, expected_md5: str, expected_sha256: str):
-    result = Checksums.generate(file_path, ["sha1", "sha256", "md5"])
+    result = Checksums.generate(file_path)
     expected = Checksums(
         sha1=expected_sha1,
         md5=expected_md5,
@@ -681,42 +681,10 @@ def test_deploy_artifact_with_checksum_algorithms_success(file_info: dict, expec
     artifact = artifactory.deploy(
         Path(LOCAL_FILE_LOCATION),
         Path(ARTIFACT_PATH),
-        checksum_algorithms=file_info["originalChecksums"].keys(),
     )
     assert expected == artifact.originalChecksums
 
 
-@pytest.mark.parametrize(
-    "file_info",
-    [
-        pytest.param(
-            {
-                **FILE_INFO_RESPONSE.copy(),
-                "originalChecksums": {"sha224": "d14a028c2a3a2bc9476102bb288234c415a2b01f828ea62ac5b3e42f"},
-            },
-            id="sha224",
-        ),
-    ],
-)
-@responses.activate
-def test_deploy_artifact_with_checksum_algorithms_error(file_info: dict):
-    responses.add(responses.PUT, f"{URL}/{ARTIFACT_PATH}", status=409)
-
-    responses.add(
-        responses.GET,
-        f"{URL}/api/storage/{ARTIFACT_PATH}",
-        json=file_info,
-        status=200,
-    )
-    with pytest.raises(Exception):
-        artifactory = ArtifactoryArtifact(AuthModel(url=URL, auth=AUTH))
-        artifactory.deploy(
-            Path(LOCAL_FILE_LOCATION),
-            Path(ARTIFACT_PATH),
-            checksum_algorithms=file_info["originalChecksums"].keys(),
-        )
-
-
 @responses.activate
 def test_deploy_artifact_with_checksum_success(mocker):
     responses.add(responses.PUT, f"{URL}/{ARTIFACT_PATH}", status=200)