Add support for "tagging" URL inputs using #<fragment> #708 (#1062)

* Add support for "tagging" URL inputs using #<fragment> #708

Signed-off-by: tdruez <tdruez@nexb.com>

* Add support for providing multiple URLs in a single str in API #708

Signed-off-by: tdruez <tdruez@nexb.com>

---------

Signed-off-by: tdruez <tdruez@nexb.com>

Author: tdruez

CHANGELOG.rst

@@ -24,6 +24,13 @@ Unreleased
 - Improve the inspect_manifest pipeline to accept archives as inputs.
   https://github.com/nexB/scancode.io/issues/1034
 
+- Add support for "tagging" download URL inputs using the "#<fragment>" section of the
+  URL.
+  This feature is particularly useful in the map_develop_to_deploy pipeline when
+  download URLs are utilized as inputs. Tags such as "from" and "to" can be specified
+  by adding "#from" or "#to" fragments at the end of the download URLs.
+  https://github.com/nexB/scancode.io/issues/708
+
 v33.0.0 (2024-01-16)
 --------------------
 

scanpipe/api/serializers.py

@@ -97,7 +97,7 @@ class StrListField(serializers.ListField):
 
     def to_internal_value(self, data):
         if isinstance(data, str):
-            data = [data]
+            data = data.split()
         return super().to_internal_value(data)
 
 
@@ -250,6 +250,10 @@ def get_codebase_relations_summary(self, project):
         queryset = project.codebaserelations.all()
         return count_group_by(queryset, "map_type")
 
+    def validate_input_urls(self, value):
+        """Add support for providing multiple URLs in a single string."""
+        return [url for entry in value for url in entry.split()]
+
     def create(self, validated_data):
         """
         Create a new `project` with `upload_file` and `pipeline` as optional.

scanpipe/models.py

@@ -32,6 +32,7 @@
 from operator import itemgetter
 from pathlib import Path
 from traceback import format_tb
+from urllib.parse import urlparse
 
 from django.apps import apps
 from django.conf import settings
@@ -983,11 +984,18 @@ def add_input_source(self, download_url="", filename="", is_uploaded=False):
         if not download_url and not filename:
             raise Exception("Provide at least a value for download_url or filename.")
 
+        # Add tag can be provided using the "#<fragment>" part of the URL
+        tag = ""
+        if download_url:
+            parsed_url = urlparse(download_url)
+            tag = parsed_url.fragment
+
         return InputSource.objects.create(
             project=self,
             download_url=download_url,
             filename=filename,
             is_uploaded=is_uploaded,
+            tag=tag,
         )
 
     def add_downloads(self, downloads):

scanpipe/pipelines/deploy_to_develop.py

@@ -37,6 +37,9 @@ class DeployToDevelop(Pipeline):
     prefixes as inputs:
     - "from-[FILENAME]" archive containing the development source code
     - "to-[FILENAME]" archive containing the deployment compiled code
+
+    Alternatively, when using download URLs as inputs, the from and to tag can be
+    provided adding a "#from" / "#to" fragment at the end of the download URLs.
     """
 
     @classmethod

scanpipe/pipes/d2d.py

@@ -58,9 +58,16 @@
 
 
 def get_inputs(project):
-    """Locate the ``from`` and ``to`` input files in project inputs/ directory."""
+    """
+    Locate the ``from`` and ``to`` input files in project inputs/ directory.
+    The input source can be flagged using a "from-" / "to-" prefix in the filename or
+    by adding a "#from" / "#to" fragment at the end of the download URL.
+    """
     from_files = list(project.inputs("from*"))
+    from_files.extend([input.path for input in project.inputsources.filter(tag="from")])
+
     to_files = list(project.inputs("to*"))
+    to_files.extend([input.path for input in project.inputsources.filter(tag="to")])
 
     if len(from_files) < 1:
         raise FileNotFoundError("from* input files not found.")

scanpipe/tests/pipes/test_d2d.py

@@ -75,6 +75,25 @@ def test_scanpipe_pipes_d2d_get_inputs(self):
         self.assertEqual(2, len(from_files))
         self.assertEqual(2, len(to_files))
 
+        _, input_location = tempfile.mkstemp(prefix="")
+        self.project1.copy_input_from(input_location)
+        url_with_fragment = "https://download.url#from"
+        input_source1 = self.project1.add_input_source(
+            download_url=url_with_fragment, filename=Path(input_location).name
+        )
+        _, input_location = tempfile.mkstemp(prefix="")
+        self.project1.copy_input_from(input_location)
+        url_with_fragment = "https://download.url#to"
+        input_source2 = self.project1.add_input_source(
+            download_url=url_with_fragment, filename=Path(input_location).name
+        )
+
+        from_files, to_files = d2d.get_inputs(self.project1)
+        self.assertEqual(3, len(from_files))
+        self.assertEqual(3, len(to_files))
+        self.assertIn(input_source1.path, from_files)
+        self.assertIn(input_source2.path, to_files)
+
     def test_scanpipe_pipes_d2d_get_extracted_path(self):
         path = "not/an/extracted/path/"
         r1 = make_resource_file(self.project1, path)

scanpipe/tests/test_api.py

@@ -355,6 +355,41 @@ def test_scanpipe_api_project_create_base(self, mock_execute_pipeline_task):
         ]
         self.assertEqual(expected, response.data["input_sources"])
 
+    def test_scanpipe_api_project_create_input_urls(self):
+        url1 = "https://example.com/1.zip#from"
+        url2 = "https://example.com/2.zip#to"
+        data = {
+            "name": "Inputs as list",
+            "input_urls": [url1, url2],
+        }
+        response = self.csrf_client.post(self.project_list_url, data)
+        self.assertEqual(status.HTTP_201_CREATED, response.status_code)
+        self.assertEqual(2, len(response.data["input_sources"]))
+
+        data = {
+            "name": "Inputs as string",
+            "input_urls": f"{url1} {url2}",
+        }
+        response = self.csrf_client.post(self.project_list_url, data)
+        self.assertEqual(status.HTTP_201_CREATED, response.status_code)
+        self.assertEqual(2, len(response.data["input_sources"]))
+
+        data = {
+            "name": "Inputs as list of string",
+            "input_urls": [f"{url1} {url2}"],
+        }
+        response = self.csrf_client.post(self.project_list_url, data)
+        self.assertEqual(status.HTTP_201_CREATED, response.status_code)
+        self.assertEqual(2, len(response.data["input_sources"]))
+
+        data = {
+            "name": "Inputs as mixed content",
+            "input_urls": [f"{url1} {url2}", "https://example.com/3.zip"],
+        }
+        response = self.csrf_client.post(self.project_list_url, data)
+        self.assertEqual(status.HTTP_201_CREATED, response.status_code)
+        self.assertEqual(3, len(response.data["input_sources"]))
+
     def test_scanpipe_api_project_create_multiple_pipelines(self):
         data = {
             "name": "Single string",

scanpipe/tests/test_models.py

@@ -464,6 +464,10 @@ def test_scanpipe_project_model_add_input_source(self):
         input_sources = self.project1.inputsources.all()
         self.assertEqual(2, len(input_sources))
 
+        url_with_fragment = "https://download.url#tag_value"
+        input_source = self.project1.add_input_source(download_url=url_with_fragment)
+        self.assertEqual("tag_value", input_source.tag)
+
     def test_scanpipe_project_model_add_downloads(self):
         file_location = self.data_location / "notice.NOTICE"
         copy_input(file_location, self.project1.tmp_path)