Enhance the inputs management API on the project model

Signed-off-by: Thomas Druez <[email protected]>

scanpipe/management/commands/__init__.py

@@ -20,7 +20,6 @@
 # ScanCode.io is a free software code scanning tool from nexB Inc. and others.
 # Visit https://github.com/nexB/scancode.io for support and download.
 
-import shutil
 from pathlib import Path
 
 from django.apps import apps
@@ -75,15 +74,6 @@ def validate_inputs(inputs):
             raise CommandError(f"{input_location} not found or not a file")
 
 
-def copy_inputs(inputs, dest_path):
-    """
-    Copy the provided `inputs` to the `dest_path`.
-    """
-    for input_location in inputs:
-        destination = dest_path / Path(input_location).name
-        shutil.copyfile(input_location, destination)
-
-
 def validate_pipelines(pipelines):
     """
     Raise an error if one of the `pipelines` is not available.

scanpipe/management/commands/add-input.py

@@ -21,7 +21,6 @@
 # Visit https://github.com/nexB/scancode.io for support and download.
 
 from scanpipe.management.commands import ProjectCommand
-from scanpipe.management.commands import copy_inputs
 from scanpipe.management.commands import validate_inputs
 
 
@@ -41,7 +40,8 @@ def handle(self, *inputs, **options):
         super().handle(*inputs, **options)
 
         validate_inputs(inputs)
-        copy_inputs(inputs, self.project.input_path)
+        for input_location in inputs:
+            self.project.copy_input_from(input_location)
 
         msg = (
             f"File(s) copied to the project inputs directory: {self.project.input_path}"

scanpipe/management/commands/create-project.py

@@ -25,7 +25,6 @@
 from django.core.management import call_command
 from django.core.management.base import BaseCommand
 
-from scanpipe.management.commands import copy_inputs
 from scanpipe.management.commands import validate_inputs
 from scanpipe.management.commands import validate_pipelines
 from scanpipe.models import Project
@@ -85,7 +84,8 @@ def handle(self, *args, **options):
         for pipeline_location in pipelines:
             project.add_pipeline(pipeline_location)
 
-        copy_inputs(inputs, project.input_path)
+        for input_location in inputs:
+            project.copy_input_from(input_location)
 
         if run:
             call_command("run", project=project, stderr=self.stderr, stdout=self.stdout)

scanpipe/models.py

@@ -304,6 +304,22 @@ def add_input_file(self, file_object):
             for chunk in file_object.chunks():
                 f.write(chunk)
 
+    def copy_input_from(self, input_location):
+        """
+        Copy the file at `input_location` to this project input/ directory.
+        """
+        from scanpipe.pipes.input import copy_inputs
+
+        copy_inputs([input_location], self.input_path)
+
+    def move_input_from(self, input_location):
+        """
+        Move the file at `input_location` to this project input/ directory.
+        """
+        from scanpipe.pipes.input import move_inputs
+
+        move_inputs([input_location], self.input_path)
+
     def add_pipeline(self, pipeline, start_run=False):
         """
         Create a new Run instance with the provided `pipeline` on this project.

scanpipe/pipelines/scan_codebase.py

@@ -29,8 +29,8 @@
 from scanpipe.pipelines import Pipeline
 from scanpipe.pipelines import step
 from scanpipe.pipes import outputs
+from scanpipe.pipes.input import copy_inputs
 from scanpipe.pipes import scancode
-from scanpipe.management.commands import copy_inputs
 
 
 class ScanCodebase(Pipeline):

scanpipe/pipes/input.py

@@ -0,0 +1,42 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/nexB/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https://github.com/nexB/scancode.io for support and download.
+
+import shutil
+from pathlib import Path
+
+
+def copy_inputs(inputs, dest_path):
+    """
+    Copy the provided `inputs` to the `dest_path`.
+    """
+    for input_location in inputs:
+        destination = dest_path / Path(input_location).name
+        shutil.copyfile(input_location, destination)
+
+
+def move_inputs(inputs, dest_path):
+    """
+    Move the provided `inputs` to the `dest_path`.
+    """
+    for input_location in inputs:
+        destination = dest_path / Path(input_location).name
+        shutil.move(input_location, destination)

scanpipe/tests/test_models.py

@@ -20,6 +20,7 @@
 # ScanCode.io is a free software code scanning tool from nexB Inc. and others.
 # Visit https://github.com/nexB/scancode.io for support and download.
 
+import tempfile
 import uuid
 from datetime import datetime
 from pathlib import Path
@@ -115,6 +116,26 @@ def test_scanpipe_project_model_add_input_file(self):
 
         self.assertEqual(["file.ext"], self.project1.input_files)
 
+    def test_scanpipe_project_model_copy_input_from(self):
+        self.assertEqual([], self.project1.input_files)
+
+        _, input_location = tempfile.mkstemp()
+        input_filename = Path(input_location).name
+
+        self.project1.copy_input_from(input_location)
+        self.assertEqual([input_filename], self.project1.input_files)
+        self.assertTrue(Path(input_location).exists())
+
+    def test_scanpipe_project_model_move_input_from(self):
+        self.assertEqual([], self.project1.input_files)
+
+        _, input_location = tempfile.mkstemp()
+        input_filename = Path(input_location).name
+
+        self.project1.move_input_from(input_location)
+        self.assertEqual([input_filename], self.project1.input_files)
+        self.assertFalse(Path(input_location).exists())
+
     def test_scanpipe_project_model_get_next_run(self):
         self.assertEqual(None, self.project1.get_next_run())