Add vulture to ci/quality.sh scripts to check for dead code. (#622)

components/collector/ci/quality.sh

@@ -4,4 +4,5 @@ mypy src
 pylint src
 safety check --bare -r requirements.txt -r requirements-dev.txt
 bandit --quiet --recursive src/
-
+NAMES_TO_IGNORE='AxeCSV*,AzureDevops*,Bandit*,Calendar*,CxSAST*,Gitlab*,HQ*,Jacoco*,Jenkins*,Jira*,JUnit*,ManualNumber*,OJAudit*,OpenVAS*,OWASPDependencyCheck*,OWASPZAP*,PerformanceTestRunner*,PyupioSafety*,QualityTime*,RobotFramework*,SonarQube*,Trello*,Wekan*'
+vulture --min-confidence 0 --ignore-names $NAMES_TO_IGNORE src/ tests/

components/collector/requirements-dev.txt

@@ -6,3 +6,4 @@ nose==1.3.7
 pip==19.2.3
 pylint==2.4.0
 safety==1.8.5
+vulture==1.1

components/collector/src/source_collectors/__init__.py

@@ -12,7 +12,7 @@
 from .jenkins_test_report import (
     JenkinsTestReportSourceUpToDateness, JenkinsTestReportFailedTests, JenkinsTestReportTests)
 from .jira import JiraIssues, JiraManualTestDuration, JiraReadyUserStoryPoints
-from .junit import JUnitFailedTests, JunitSourceUpToDateness, JUnitTests
+from .junit import JUnitFailedTests, JUnitSourceUpToDateness, JUnitTests
 from .manual_number import ManualNumber
 from .ojaudit import OJAuditViolations
 from .openvas import OpenVASSecurityWarnings, OpenVASSourceUpToDateness

components/collector/src/source_collectors/azure_devops.py

@@ -1,5 +1,6 @@
 """Azure Devops Server metric collector."""
 
+from abc import ABC
 from typing import List
 
 import requests
@@ -8,7 +9,7 @@
 from .source_collector import SourceCollector
 
 
-class AzureDevopsBase(SourceCollector):
+class AzureDevopsBase(SourceCollector, ABC):  # pylint: disable=abstract-method
     """Base class for Azure DevOps collectors."""
 
     MAX_IDS_PER_WORKITEMS_API_CALL = 200  # See

components/collector/src/source_collectors/calendar.py

@@ -1,22 +1,17 @@
 """Calendar metric source."""
 
-import io
 from datetime import datetime
 from typing import cast, List
 
 import requests
 
-from utilities.type import URL
-from .source_collector import SourceCollector
+from utilities.type import Value
+from .source_collector import LocalSourceCollector
 
 
-class CalendarSourceUpToDateness(SourceCollector):
+class CalendarSourceUpToDateness(LocalSourceCollector):
     """Collector class to get the number of days since a user-specified date."""
 
-    def _get_source_responses(self, api_url: URL) -> List[requests.Response]:
-        """Return a random number as the response."""
+    def _parse_source_responses_value(self, responses: List[requests.Response]) -> Value:
         days = (datetime.now() - datetime.fromisoformat(cast(str, self._parameter("date")))).days
-        response = requests.Response()
-        response.raw = io.BytesIO(bytes(str(days), "utf-8"))
-        response.status_code = requests.status_codes.codes["OK"]
-        return [response]
+        return str(days)

components/collector/src/source_collectors/cxsast.py

@@ -1,5 +1,6 @@
 """Collectors for the Checkmarx CxSAST product."""
 
+from abc import ABC
 from datetime import datetime
 from typing import cast, List
 
@@ -13,7 +14,7 @@
 from .source_collector import SourceCollector
 
 
-class CxSASTBase(SourceCollector):
+class CxSASTBase(SourceCollector, ABC):  # pylint: disable=abstract-method
     """Base class for CxSAST collectors."""
 
     TOKEN_RESPONSE, PROJECT_RESPONSE, SCAN_RESPONSE = range(3)

components/collector/src/source_collectors/gitlab.py

@@ -1,5 +1,6 @@
 """Gitlab metric source."""
 
+from abc import ABC
 from datetime import datetime, timedelta, timezone
 from typing import cast, List, Optional, Tuple
 from urllib.parse import quote
@@ -11,7 +12,7 @@
 from .source_collector import SourceCollector
 
 
-class GitlabBase(SourceCollector):
+class GitlabBase(SourceCollector, ABC):  # pylint: disable=abstract-method
     """Baseclass for Gitlab collectors."""
 
     def _gitlab_api_url(self, api: str) -> URL:

components/collector/src/source_collectors/jira.py

@@ -1,5 +1,6 @@
 """Jira metric collector."""
 
+from abc import ABC
 from typing import cast, List
 from urllib.parse import quote
 
@@ -9,7 +10,7 @@
 from .source_collector import SourceCollector
 
 
-class JiraBase(SourceCollector):
+class JiraBase(SourceCollector, ABC):  # pylint: disable=abstract-method
     """Base class for Jira collectors."""
 
     def _api_url(self) -> URL:

components/collector/src/source_collectors/junit.py

@@ -13,7 +13,6 @@
 class JUnitTests(SourceCollector):
     """Collector for JUnit tests."""
 
-    junit_test_report_counts = dict(errored="errors", failed="failures", passed="tests", skipped="skipped")
     junit_status_nodes = dict(errored="error", failed="failure", skipped="skipped")
 
     def _parse_source_responses_value(self, responses: List[requests.Response]) -> Value:
@@ -66,7 +65,7 @@ def entity(case_node, test_case_status: str) -> Entity:
         return entities
 
 
-class JunitSourceUpToDateness(SourceCollector):
+class JUnitSourceUpToDateness(SourceCollector):
     """Collector to collect the Junit report age."""
 
     def _parse_source_responses_value(self, responses: List[requests.Response]) -> Value:

components/collector/src/source_collectors/owasp_dependency_check.py

@@ -1,6 +1,7 @@
 """OWASP Dependency Check metric collector."""
 
 import hashlib
+from abc import ABC
 from typing import List, Tuple
 from xml.etree.ElementTree import Element  # nosec, Element is not available from defusedxml, but only used as type
 
@@ -12,7 +13,7 @@
 from .source_collector import SourceCollector
 
 
-class OWASPDependencyCheckBase(SourceCollector):
+class OWASPDependencyCheckBase(SourceCollector, ABC):  # pylint: disable=abstract-method
     """Base class for OWASP Dependency Check collectors."""
 
     allowed_root_tags = [f"{{https://jeremylong.github.io/DependencyCheck/dependency-check.{version}.xsd}}analysis"

components/collector/src/source_collectors/robot_framework.py

@@ -1,5 +1,6 @@
 """Robot Framework metric collector."""
 
+from abc import ABC
 from typing import List
 
 from dateutil.parser import parse
@@ -10,7 +11,7 @@
 from .source_collector import SourceCollector
 
 
-class RobotFrameworkBaseClass(SourceCollector):
+class RobotFrameworkBaseClass(SourceCollector, ABC):  # pylint: disable=abstract-method
     """Base class for Robot Framework collectors."""
 
     def _landing_url(self, responses: List[requests.Response]) -> URL:

components/collector/src/source_collectors/source_collector.py

@@ -2,6 +2,7 @@
 
 import logging
 import traceback
+from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
 from http import HTTPStatus
 from typing import cast, Dict, List, Optional, Set, Tuple, Type, Union
@@ -12,7 +13,7 @@
 from utilities.type import ErrorMessage, Response, Entities, URL, Value
 
 
-class SourceCollector:
+class SourceCollector(ABC):
     """Base class for source collectors. Source collectors are subclasses of this class that know how to collect the
     measurement data for one specific metric from one specific source."""
 
@@ -114,11 +115,12 @@ def __safely_parse_source_responses(
                 error = stable_traceback(traceback.format_exc())
         return value, total, entities[:self.MAX_ENTITIES], error
 
+    @abstractmethod
     def _parse_source_responses_value(self, responses: List[requests.Response]) -> Value:
         # pylint: disable=no-self-use
-        """Parse the responses to get the measurement for the metric. This method can be overridden by collectors
+        """Parse the responses to get the measurement for the metric. This method must be overridden by collectors
         to parse the retrieved sources data."""
-        return str(responses[0].text)
+        return None  # pragma: nocover
 
     def _parse_source_responses_total(self, responses: List[requests.Response]) -> Value:
         # pylint: disable=no-self-use,unused-argument
@@ -142,7 +144,7 @@ def has_invalid_credentials(self) -> bool:
         return self.__has_invalid_credentials
 
 
-class LocalSourceCollector(SourceCollector):
+class LocalSourceCollector(SourceCollector, ABC):  # pylint: disable=abstract-method
     """Base class for source collectors that do not need to access the network but return static or user-supplied
     data."""
 

components/collector/src/source_collectors/trello.py

@@ -1,5 +1,6 @@
 """Trello metric collector."""
 
+from abc import ABC
 from datetime import datetime
 from typing import cast, List
 
@@ -11,7 +12,7 @@
 from .source_collector import SourceCollector
 
 
-class TrelloBase(SourceCollector):
+class TrelloBase(SourceCollector, ABC):  # pylint: disable=abstract-method
     """Base class for Trello collectors."""
 
     def _landing_url(self, responses: List[requests.Response]) -> URL:

components/collector/src/source_collectors/wekan.py

@@ -1,5 +1,6 @@
 """Wekan metric collector."""
 
+from abc import ABC
 from datetime import datetime
 from typing import cast, List
 
@@ -12,7 +13,7 @@
 from .source_collector import SourceCollector
 
 
-class WekanBase(SourceCollector):
+class WekanBase(SourceCollector, ABC):  # pylint: disable=abstract-method
     """Base class for Wekan collectors."""
 
     def _landing_url(self, responses: List[requests.Response]) -> URL:

components/collector/src/utilities/type.py

@@ -1,6 +1,6 @@
 """Quality-time specific types."""
 
-from typing import Any, Dict, List, NewType, Optional, Sequence, Union
+from typing import Any, Dict, List, NewType, Optional, Union
 
 
 Entity = Dict[str, Union[int, str]]  # pylint: disable=invalid-name
@@ -11,7 +11,4 @@
 Namespaces = Dict[str, str]  # Namespace prefix to Namespace URI mapping
 Response = Dict[str, Any]
 URL = NewType("URL", str)
-Subject = Dict[str, Union[str, Sequence[URL]]]
-Report = Dict[str, Sequence[Subject]]
 Value = Optional[str]
-Parameter = Union[str, List[str]]

components/collector/tests/unittests/metric_collectors_tests/test_metrics_collector.py

@@ -39,10 +39,10 @@ def next_collection(self) -> datetime:
         self.datamodel_response.json.return_value = dict(
             sources=dict(source=dict(parameters=dict(url=dict(mandatory=True, metrics=["metric"])))))
         self.metrics_response = Mock()
-        logging.getLogger().disabled = True
+        logging.disable()
 
     def tearDown(self):
-        logging.getLogger().disabled = False
+        logging.disable(logging.NOTSET)
 
     def test_fetch_without_sources(self):
         """Test fetching measurement for a metric without sources."""

components/collector/tests/unittests/source_collectors_tests/test_junit.py

@@ -55,7 +55,7 @@ def test_failed_tests(self):
 
 
 class JUnitSourceUpToDatenessTest(SourceCollectorTestCase):
-    """Unit test for the source up-to-dateness metric."""
+    """Unit tests for the source up-to-dateness metric."""
 
     def setUp(self):
         super().setUp()

components/server/.vulture_white_list.py

@@ -0,0 +1,8 @@
+enable_cors_generic_route  # unused function (src/routes/cors.py:8)
+api  # unused variable (src/routes/plugins/authentication_plugin.py:13)
+_.name  # unused attribute (src/routes/plugins/authentication_plugin.py:16)
+apply  # unused function (src/routes/plugins/authentication_plugin.py:18)
+api  # unused variable (src/routes/plugins/injection_plugin.py:10)
+_.name  # unused attribute (src/routes/plugins/injection_plugin.py:15)
+setup  # unused function (src/routes/plugins/injection_plugin.py:17)
+apply  # unused function (src/routes/plugins/injection_plugin.py:25)

components/server/ci/quality.sh

@@ -4,4 +4,4 @@ mypy src
 pylint src
 safety check --bare -r requirements.txt -r requirements-dev.txt
 bandit --quiet --recursive src/
-
+vulture --min-confidence 0 src/ tests/ .vulture_white_list.py

components/server/requirements-dev.txt

@@ -5,3 +5,4 @@ nose==1.3.7
 pip==19.2.3
 pylint==2.4.0
 safety==1.8.5
+vulture==1.1

components/server/src/initialization/datamodel.py

@@ -1,4 +1,4 @@
-"""Datamodel loader."""
+"""Data model loader."""
 
 import json
 import logging
@@ -12,15 +12,15 @@
 
 def import_datamodel(database: Database) -> None:
     """Read the data model and store it in the database."""
-    datamodel_path = pathlib.Path(os.path.dirname(os.path.abspath(__file__)), "..", "data", "datamodel.json")
-    with open(datamodel_path) as json_datamodel:
-        data_model = json.load(json_datamodel)
+    data_model_path = pathlib.Path(os.path.dirname(os.path.abspath(__file__)), "..", "data", "datamodel.json")
+    with open(data_model_path) as json_data_model:
+        data_model = json.load(json_data_model)
     latest = latest_datamodel(database)
     if latest:
         del latest["timestamp"]
         del latest["_id"]
         if data_model == latest:
-            logging.info("Skipping loading the datamodel; it is unchanged")
+            logging.info("Skipping loading the data model; it is unchanged")
             return
     insert_new_datamodel(database, data_model)
-    logging.info("Datamodel loaded")
+    logging.info("Data model loaded")

components/server/src/initialization/report.py

@@ -23,7 +23,7 @@ def initialize_reports_overview(database: Database) -> None:
 
 def import_report(database: Database, filename: pathlib.Path) -> None:
     """Read the report and store it in the database."""
-    with open(filename) as json_report:
+    with open(str(filename)) as json_report:
         imported_report = json.load(json_report)
     stored_report = latest_report(database, imported_report["report_uuid"])
     if stored_report:

components/server/src/routes/auth.py

@@ -13,7 +13,7 @@
 
 from database import sessions
 from utilities.functions import uuid
-from utilities.ldap import LDAPUserObject
+from utilities.ldap import LDAPObject
 
 
 def generate_session() -> Tuple[str, datetime]:
@@ -50,7 +50,7 @@ def login(database: Database) -> Dict[str, bool]:
             ldap_root_dn, ldap.SCOPE_SUBTREE, f"(|(uid={username})(cn={username}))", ['dn', 'uid', 'cn'])
         if result:
             logging.info("LDAP search result: %s", result)
-            username = LDAPUserObject(result[0]).cn
+            username = LDAPObject(result[0][1]).cn
         else:
             raise ldap.INVALID_CREDENTIALS
         ldap_server.simple_bind_s(f"cn={username},{ldap_root_dn}", credentials.get("password"))

components/server/src/routes/changelog.py

@@ -6,7 +6,7 @@
 from database import measurements, reports
 
 
-def get_changelog(database: Database, nr_changes: str, **uuids):
+def get_changelog(database: Database, nr_changes: str, **uuids: str):
     """Return the recent most nr_changes changes from the changelog."""
     limit = int(nr_changes)
     changes = [dict(delta=change["delta"], timestamp=change["start"])

components/server/src/routes/datamodel.py

@@ -1,4 +1,4 @@
-"""Datamodel routes."""
+"""Data model routes."""
 
 import bottle
 from pymongo.database import Database

components/server/src/routes/report.py

@@ -35,9 +35,8 @@ def get_data(database: Database, report_uuid: str, subject_uuid: str = None, met
     """Return applicable report, subject, metric, source, and their uuids and names."""
     data = namedtuple(
         "data",
-        "datamodel, report, report_uuid, report_name, subject, subject_uuid, subject_name, "
+        "datamodel, report, report_name, subject, subject_uuid, subject_name, "
         "metric, metric_uuid, metric_name, source, source_uuid, source_name")
-    data.report_uuid = report_uuid
     data.report = latest_report(database, report_uuid)
     data.report_name = data.report.get("title") or ""
     data.source_uuid = source_uuid

components/server/src/utilities/ldap.py

@@ -4,16 +4,11 @@
 class LDAPObject:  # pylint: disable=too-few-public-methods
     """Class helper that unpacks a python-ldap search result."""
 
-    def __init__(self, t) -> None:
-        # pylint: disable=invalid-name
-        self.dn, attrs = t
-        for key, value in attrs.items():
-            setattr(self, key, [i.decode('utf-8') for i in value] if len(value) > 1 else value[0].decode('utf-8'))
-
-
-class LDAPUserObject(LDAPObject):  # pylint: disable=too-few-public-methods
-    """Class helper that represents a LDAP user object."""
-    # pylint: disable=invalid-name
-    dn: str = ""
-    cn: str = ""
-    uid: str = ""
+    def __init__(self, entry) -> None:
+        for key, values in entry.items():
+            string_values = [value.decode('utf-8') for value in values]
+            setattr(self, key, string_values if len(string_values) > 1 else string_values[0])
+
+    def __getattr__(self, key: str) -> str:
+        # Return a default value for non-existing keys
+        return ""  # pragma: nocover