Add support for tag in API #708

Signed-off-by: tdruez <[email protected]>

CHANGELOG.rst

@@ -31,6 +31,8 @@ Unreleased
   Using the CLI, the uploaded files can be tagged using the "filename:tag" syntax
   while using the `--input-file` arguments.
   In the UI, tags can be edited from the Project details view "Inputs" panel.
+  On the REST API, a new `upload_file_tag` field is available to use along the
+  `upload_file`.
   https://github.com/nexB/scancode.io/issues/708
 
 v33.0.0 (2024-01-16)

scanpipe/api/serializers.py

@@ -156,6 +156,7 @@ class ProjectSerializer(
         help_text="Execute pipeline now",
     )
     upload_file = serializers.FileField(write_only=True, required=False)
+    upload_file_tag = serializers.CharField(write_only=True, required=False)
     input_urls = StrListField(
         write_only=True,
         required=False,
@@ -182,6 +183,7 @@ class Meta:
             "url",
             "uuid",
             "upload_file",
+            "upload_file_tag",
             "input_urls",
             "webhook_url",
             "created_date",
@@ -265,6 +267,7 @@ def create(self, validated_data):
         This ensures the Project data integrity before running any pipelines.
         """
         upload_file = validated_data.pop("upload_file", None)
+        upload_file_tag = validated_data.pop("upload_file_tag", "")
         input_urls = validated_data.pop("input_urls", [])
         pipeline = validated_data.pop("pipeline", [])
         execute_now = validated_data.pop("execute_now", False)
@@ -273,7 +276,7 @@ def create(self, validated_data):
         project = super().create(validated_data)
 
         if upload_file:
-            project.add_uploads([upload_file])
+            project.add_upload(upload_file, tag=upload_file_tag)
 
         for url in input_urls:
             project.add_input_source(download_url=url)

scanpipe/api/views.py

@@ -273,14 +273,20 @@ def add_input(self, request, *args, **kwargs):
             return Response(message, status=status.HTTP_400_BAD_REQUEST)
 
         upload_file = request.data.get("upload_file")
+        upload_file_tag = request.data.get("upload_file_tag", "")
         input_urls = request.data.get("input_urls", [])
 
         if not (upload_file or input_urls):
             message = {"status": "upload_file or input_urls required."}
             return Response(message, status=status.HTTP_400_BAD_REQUEST)
 
         if upload_file:
-            project.add_uploads([upload_file])
+            project.add_upload(upload_file, tag=upload_file_tag)
+
+        # Add support for providing multiple URLs in a single string.
+        if isinstance(input_urls, str):
+            input_urls = input_urls.split()
+        input_urls = [url for entry in input_urls for url in entry.split()]
 
         for url in input_urls:
             project.add_input_source(download_url=url)

scanpipe/models.py

@@ -1009,14 +1009,21 @@ def add_downloads(self, downloads):
                 filename=downloaded.filename,
             )
 
+    def add_upload(self, uploaded_file, tag=""):
+        """
+        Write the given `upload` to the current project's input/ directory and
+        adds the `input_source`.
+        """
+        self.write_input_file(uploaded_file)
+        self.add_input_source(filename=uploaded_file.name, is_uploaded=True, tag=tag)
+
     def add_uploads(self, uploads):
         """
         Write the given `uploads` to the current project's input/ directory and
         adds the `input_source` for each entry.
         """
-        for uploaded in uploads:
-            self.write_input_file(uploaded)
-            self.add_input_source(filename=uploaded.name, is_uploaded=True)
+        for uploaded_file in uploads:
+            self.add_upload(uploaded_file)
 
     def add_pipeline(self, pipeline_name, execute_now=False):
         """

scanpipe/tests/test_api.py

@@ -294,6 +294,7 @@ def test_scanpipe_api_project_create_base(self, mock_execute_pipeline_task):
             "name": "BetterName",
             "pipeline": "analyze_docker_image",
             "upload_file": io.BytesIO(b"Content"),
+            "upload_file_tag": "tag value",
             "execute_now": True,
         }
         response = self.csrf_client.post(self.project_list_url, data)
@@ -303,6 +304,7 @@ def test_scanpipe_api_project_create_base(self, mock_execute_pipeline_task):
         created_project_detail_url = response.data["url"]
         response = self.csrf_client.get(created_project_detail_url)
         self.assertEqual(["upload_file"], response.data["input_root"])
+        self.assertEqual("tag value", response.data["input_sources"][0]["tag"])
 
         data = {
             "name": "Upload",
@@ -770,18 +772,26 @@ def test_scanpipe_api_project_action_add_input(self):
         self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
 
         data = {
-            "input_urls": "https://example.com/archive.zip",
+            "input_urls": "https://example.com/archive.zip#tag",
         }
         response = self.csrf_client.post(url, data=data)
         self.assertEqual({"status": "Input(s) added."}, response.data)
+        input_source = self.project1.inputsources.get(is_uploaded=False)
+        self.assertEqual("", input_source.filename)
+        self.assertEqual(data["input_urls"], input_source.download_url)
+        self.assertEqual("tag", input_source.tag)
 
         data = {
             "upload_file": io.BytesIO(b"Content"),
+            "upload_file_tag": "tag value",
         }
         response = self.csrf_client.post(url, data=data)
         self.assertEqual({"status": "Input(s) added."}, response.data)
         expected = sorted(["upload_file"])
         self.assertEqual(expected, sorted(self.project1.input_root))
+        input_source = self.project1.inputsources.get(is_uploaded=True)
+        self.assertEqual("upload_file", input_source.filename)
+        self.assertEqual("tag value", input_source.tag)
 
         run = self.project1.add_pipeline("analyze_docker_image")
         run.set_task_started(task_id=uuid.uuid4())

scanpipe/tests/test_models.py

@@ -371,7 +371,7 @@ def test_scanpipe_project_model_get_inputs_with_source(self):
         self.assertEqual([], self.project1.get_inputs_with_source())
 
         uploaded_file = SimpleUploadedFile("file.ext", content=b"content")
-        self.project1.add_uploads([uploaded_file])
+        self.project1.add_upload(uploaded_file)
         self.project1.copy_input_from(self.data_location / "notice.NOTICE")
         self.project1.add_input_source(filename="missing.zip", is_uploaded=True)