[colic] Add support of scancode_cli and it's corresponding tests to colic backend

A faster version of scancode is now added to colic backend
scancode_cli usage:

colic https://github.com/chaoss/grimoirelab-toolkit
--git-path /tmp/scancode_cli
--exec-path /home/scancode-toolkit/etc/scripts/scancli.py
--category code_license_scancode_cli

Added tests for scancode_cli analyzer which also required updating tests
for in-place implementation of colic backend.

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

Author: inishchith

graal/backends/core/analyzers/scancode.py

@@ -22,38 +22,44 @@
 
 import json
 import subprocess
-
 from graal.graal import (GraalError,
                          GraalRepository)
 from .analyzer import Analyzer
 
 
+SCANCODE_CLI_EXEC = "etc/scripts/scancli.py"
+CONFIGURE_EXEC = 'configure'
+
+
 class ScanCode(Analyzer):
     """A wrapper for nexB/scancode-toolkit.
 
     This class allows to call scancode-toolkit over a file, parses
     the result of the analysis and returns it as a dict.
 
     :param exec_path: path of the scancode executable
+    :param cli: True, if scancode_cli is used
     """
-    version = '0.1.0'
+    version = '0.2.0'
 
-    def __init__(self, exec_path):
+    def __init__(self, exec_path, cli=False):
         if not GraalRepository.exists(exec_path):
             raise GraalError(cause="executable path %s not valid" % exec_path)
 
         self.exec_path = exec_path
+        self.cli = cli
 
-    def analyze(self, **kwargs):
-        """Add information about license
+        if self.cli:
+            exec_path = self.exec_path.replace(SCANCODE_CLI_EXEC, CONFIGURE_EXEC)
+            _ = subprocess.check_output([exec_path]).decode("utf-8")
 
-        :param file_path: file path
+    def __analyze_scancode(self, file_path):
+        """Add information about license using scancode
 
-        :returns result: dict of the results of the analysis
+        :param file_path: file path (in case of scancode)
+        :param file_paths: file paths (in case of scancode_cli for concurrent execution on files)
         """
         result = {'licenses': []}
-        file_path = kwargs['file_path']
-
         try:
             msg = subprocess.check_output([self.exec_path, '--json-pp', '-', '--license', file_path]).decode("utf-8")
         except subprocess.CalledProcessError as e:
@@ -62,9 +68,61 @@ def analyze(self, **kwargs):
             subprocess._cleanup()
 
         licenses_raw = json.loads(msg)
-        if 'files' not in licenses_raw:
-            return result
+        if 'files' in licenses_raw:
+            result['licenses'] = licenses_raw['files'][0]['licenses']
 
-        result['licenses'] = licenses_raw['files'][0]['licenses']
+        return result
+
+    def __analyze_scancode_cli(self, file_paths):
+        """Add information about license using scancode-cli
+
+        :param file_paths: file paths
+        :returns result: dict of the results of the analysis
+        """
+        result = {'files': []}
+
+        try:
+            cmd_scancli = ['python3', self.exec_path]
+            cmd_scancli.extend(file_paths)
+            msg = subprocess.check_output(cmd_scancli).decode("utf-8")
+        except subprocess.CalledProcessError as e:
+            raise GraalError(cause="Scancode failed at %s, %s" % (file_paths,
+                                                                  e.output.decode("utf-8")))
+        finally:
+            subprocess._cleanup()
+
+        output_content = ''
+        outputs_json = []
+        for line in msg.split('\n'):
+            if line == '':
+                if output_content:
+                    output_json = json.loads(output_content)[1:]
+                    outputs_json.append(output_json)
+                    output_content = ''
+            else:
+                output_content += line
+
+        if output_content:
+            output_json = json.loads(output_content)[1:]
+            outputs_json.append(output_json)
+
+        for output_json in outputs_json:
+            file_info = output_json[0]['files'][0]
+            result['files'].append(file_info)
+
+        return result
+
+    def analyze(self, **kwargs):
+        """Add information about license
+
+        :param file_path: file path
+        :param file_paths: file paths ( in case of scancode_cli for concurrent execution on files )
+
+        :returns result: dict of the results of the analysis
+        """
+        if self.cli:
+            result = self.__analyze_scancode_cli(kwargs['file_paths'])
+        else:
+            result = self.__analyze_scancode(kwargs['file_path'])
 
         return result

graal/backends/core/colic.py

@@ -33,9 +33,11 @@
 
 NOMOS = 'nomos'
 SCANCODE = 'scancode'
+SCANCODE_CLI = 'scancode_cli'
 
 CATEGORY_COLIC_NOMOS = 'code_license_' + NOMOS
 CATEGORY_COLIC_SCANCODE = 'code_license_' + SCANCODE
+CATEGORY_COLIC_SCANCODE_CLI = 'code_license_' + SCANCODE_CLI
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +46,7 @@ class CoLic(Graal):
     """CoLic backend.
 
     This class extends the Graal backend. It gathers license information
-    using Nomos
+    using Nomos, Scancode or Scancode-cli
 
     :param uri: URI of the Git repository
     :param git_path: path to the repository or to the log file
@@ -59,9 +61,11 @@ class CoLic(Graal):
     :raises RepositoryError: raised when there was an error cloning or
         updating the repository.
     """
-    version = '0.4.0'
+    version = '0.5.0'
 
-    CATEGORIES = [CATEGORY_COLIC_NOMOS, CATEGORY_COLIC_SCANCODE]
+    CATEGORIES = [CATEGORY_COLIC_NOMOS,
+                  CATEGORY_COLIC_SCANCODE,
+                  CATEGORY_COLIC_SCANCODE_CLI]
 
     def __init__(self, uri, git_path, exec_path, worktreepath=DEFAULT_WORKTREE_PATH,
                  entrypoint=None, in_paths=None, out_paths=None,
@@ -84,6 +88,8 @@ def fetch(self, category=CATEGORY_COLIC_NOMOS, paths=None,
 
         if category == CATEGORY_COLIC_SCANCODE:
             self.analyzer_kind = SCANCODE
+        elif category == CATEGORY_COLIC_SCANCODE_CLI:
+            self.analyzer_kind = SCANCODE_CLI
         elif category == CATEGORY_COLIC_NOMOS:
             self.analyzer_kind = NOMOS
         else:
@@ -101,13 +107,17 @@ def fetch(self, category=CATEGORY_COLIC_NOMOS, paths=None,
     def metadata_category(item):
         """Extracts the category from a Code item.
 
-        This backend generates two types of item which can be:
-        'code_license_nomos' or 'code_license_scancode'.
+        This backend generates the following types of item:
+        - 'code_license_nomos'
+        - 'code_license_scancode'
+        - 'code_license_scancode_cli'
         """
         if item['analyzer'] == NOMOS:
             return CATEGORY_COLIC_NOMOS
         elif item['analyzer'] == SCANCODE:
             return CATEGORY_COLIC_SCANCODE
+        elif item['analyzer'] == SCANCODE_CLI:
+            return CATEGORY_COLIC_SCANCODE_CLI
         else:
             raise GraalError(cause="Unknown analyzer %s" % item['analyzer'])
 
@@ -135,6 +145,7 @@ def _analyze(self, commit):
         :param commit: a Perceval commit item
         """
         analysis = []
+        files_to_process = []
 
         for committed_file in commit['files']:
 
@@ -148,9 +159,18 @@ def _analyze(self, commit):
             if not GraalRepository.exists(local_path):
                 continue
 
-            license_info = self.analyzer.analyze(local_path)
-            license_info.update({'file_path': file_path})
-            analysis.append(license_info)
+            if self.analyzer_kind == NOMOS or self.analyzer_kind == SCANCODE:
+                license_info = self.analyzer.analyze(local_path)
+                license_info.update({'file_path': file_path})
+                analysis.append(license_info)
+            elif self.analyzer_kind == SCANCODE_CLI:
+                files_to_process.append((file_path, local_path))
+
+        if files_to_process:
+            local_paths = [path[1] for path in files_to_process]
+            analysis = self.analyzer.analyze(local_paths)
+            for i in range(len(analysis['files'])):
+                analysis['files'][i]['file_path'] = files_to_process[i][0]
 
         return analysis
 
@@ -170,17 +190,20 @@ class LicenseAnalyzer:
     """Class to analyse the content of files
 
     :param exec_path: path of the license analyzer executable
-    :param kind: the analyzer kind (e.g., NOMOS, SCANCODE)
+    :param kind: the analyzer kind (e.g., NOMOS, SCANCODE, SCANCODE_CLI)
     """
 
     def __init__(self, exec_path, kind=NOMOS):
+        self.kind = kind
         if kind == SCANCODE:
             self.analyzer = ScanCode(exec_path)
+        elif kind == SCANCODE_CLI:
+            self.analyzer = ScanCode(exec_path, cli=True)
         else:
             self.analyzer = Nomos(exec_path)
 
     def analyze(self, file_path):
-        """Analyze the content of a file using Nomos
+        """Analyze the content of a file using Nomos/Scancode
 
         :param file_path: file path
 
@@ -189,7 +212,11 @@ def analyze(self, file_path):
           'licenses': [..]
         }
         """
-        kwargs = {'file_path': file_path}
+        if self.kind == SCANCODE_CLI:
+            kwargs = {'file_paths': file_path}
+        else:
+            kwargs = {'file_path': file_path}
+
         analysis = self.analyzer.analyze(**kwargs)
 
         return analysis

tests/test_colic.py

@@ -33,16 +33,18 @@
 from graal.backends.core.analyzers.scancode import ScanCode
 from graal.backends.core.colic import (CATEGORY_COLIC_NOMOS,
                                        CATEGORY_COLIC_SCANCODE,
+                                       CATEGORY_COLIC_SCANCODE_CLI,
                                        NOMOS,
                                        SCANCODE,
+                                       SCANCODE_CLI,
                                        CoLic,
                                        LicenseAnalyzer,
                                        CoLicCommand)
 from perceval.utils import DEFAULT_DATETIME
 from test_graal import TestCaseGraal
 from base_analyzer import (ANALYZER_TEST_FILE,
                            TestCaseAnalyzer)
-from utils import NOMOS_PATH, SCANCODE_PATH
+from utils import NOMOS_PATH, SCANCODE_PATH, SCANCODE_CLI_PATH
 
 
 class TestCoLicBackend(TestCaseGraal):
@@ -122,7 +124,7 @@ def test_fetch_nomossa(self):
 
         cl = CoLic('http://example.com', self.git_path, NOMOS_PATH, self.worktree_path,
                    in_paths=['perceval/backends/core/github.py'])
-        commits = [commit for commit in cl.fetch()]
+        commits = [commit for commit in cl.fetch(category=CATEGORY_COLIC_NOMOS)]
 
         self.assertEqual(len(commits), 1)
         self.assertFalse(os.path.exists(cl.worktreepath))
@@ -159,6 +161,36 @@ def test_fetch_scancode(self):
             self.assertFalse('parents' in commit['data'])
             self.assertFalse('refs' in commit['data'])
 
+    def test_fetch_scancode_cli(self):
+        """Test whether commits are properly processed"""
+
+        cl = CoLic('http://example.com', self.git_path, SCANCODE_CLI_PATH, self.worktree_path,
+                   in_paths=['perceval/backends/core/github.py'])
+        commits = [commit for commit in cl.fetch(category=CATEGORY_COLIC_SCANCODE_CLI)]
+
+        self.assertEqual(len(commits), 1)
+        self.assertFalse(os.path.exists(cl.worktreepath))
+
+        for commit in commits:
+            self.assertEqual(commit['backend_name'], 'CoLic')
+            self.assertEqual(commit['category'], CATEGORY_COLIC_SCANCODE_CLI)
+            self.assertEqual(commit['data']['analysis']['files'][0]['file_path'],
+                             'perceval/backends/core/github.py')
+            self.assertTrue('Author' in commit['data'])
+            self.assertTrue('Commit' in commit['data'])
+            self.assertFalse('files' in commit['data'])
+            self.assertFalse('parents' in commit['data'])
+            self.assertFalse('refs' in commit['data'])
+
+    def test_fetch_unknown(self):
+        """Test whether commits are properly processed"""
+
+        cl = CoLic('http://example.com', self.git_path, SCANCODE_CLI_PATH, self.worktree_path,
+                   in_paths=['perceval/backends/core/github.py'])
+
+        with self.assertRaises(GraalError):
+            _ = cl.fetch(category="unknown")
+
     def test_metadata_category(self):
         """Test metadata_category"""
 
@@ -186,6 +218,18 @@ def test_metadata_category(self):
         }
         self.assertEqual(CoLic.metadata_category(item), CATEGORY_COLIC_NOMOS)
 
+        item = {
+            "Author": "Valerio Cosentino <[email protected]>",
+            "AuthorDate": "Fri May 18 18:26:48 2018 +0200",
+            "Commit": "Valerio Cosentino <[email protected]>",
+            "CommitDate": "Fri May 18 18:26:48 2018 +0200",
+            "analysis": [],
+            "analyzer": "scancode_cli",
+            "commit": "075f0c6161db5a3b1c8eca45e08b88469bb148b9",
+            "message": "[perceval] first commit"
+        }
+        self.assertEqual(CoLic.metadata_category(item), CATEGORY_COLIC_SCANCODE_CLI)
+
         item = {
             "Author": "Valerio Cosentino <[email protected]>",
             "AuthorDate": "Fri May 18 18:26:48 2018 +0200",
@@ -218,6 +262,10 @@ def test_init(self):
         self.assertIsInstance(license_analyzer, LicenseAnalyzer)
         self.assertIsInstance(license_analyzer.analyzer, ScanCode)
 
+        license_analyzer = LicenseAnalyzer(SCANCODE_CLI_PATH, SCANCODE_CLI)
+        self.assertIsInstance(license_analyzer, LicenseAnalyzer)
+        self.assertIsInstance(license_analyzer.analyzer, ScanCode)
+
         with self.assertRaises(GraalError):
             _ = LicenseAnalyzer("/tmp/analyzer", SCANCODE)
 
@@ -236,6 +284,12 @@ def test_analyze(self):
 
         self.assertIn('licenses', analysis)
 
+        file_paths = [os.path.join(self.tmp_data_path, ANALYZER_TEST_FILE)]
+        license_analyzer = LicenseAnalyzer(SCANCODE_CLI_PATH, kind=SCANCODE_CLI)
+        analysis = license_analyzer.analyze(file_paths)
+
+        self.assertIn('licenses', analysis['files'][0])
+
 
 class TestCoLicCommand(unittest.TestCase):
     """CoLicCommand tests"""