checks.py

   1 # -*- coding: utf8 -*-
   2 # ============LICENSE_START====================================================
   3 # org.onap.vvp/validation-scripts
   4 # ===================================================================
   5 # Copyright © 2019 AT&T Intellectual Property. All rights reserved.
   6 # ===================================================================
   7 #
   8 # Unless otherwise specified, all software contained herein is licensed
   9 # under the Apache License, Version 2.0 (the "License");
  10 # you may not use this software except in compliance with the License.
  11 # You may obtain a copy of the License at
  12 #
  13 #             http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS,
  17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18 # See the License for the specific language governing permissions and
  19 # limitations under the License.
  20 #
  21 #
  22 #
  23 # Unless otherwise specified, all documentation contained herein is licensed
  24 # under the Creative Commons License, Attribution 4.0 Intl. (the "License");
  25 # you may not use this documentation except in compliance with the License.
  26 # You may obtain a copy of the License at
  27 #
  28 #             https://creativecommons.org/licenses/by/4.0/
  29 #
  30 # Unless required by applicable law or agreed to in writing, documentation
  31 # distributed under the License is distributed on an "AS IS" BASIS,
  32 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  33 # See the License for the specific language governing permissions and
  34 # limitations under the License.
  35 #
  36 # ============LICENSE_END============================================
  37 #
  38 import contextlib
  39 import csv
  40 import io
  41 import json
  42 import os
  43 import subprocess  # nosec
  44 import sys
  45
  46 import pytest
  47 from flake8.main.application import Application
  48
  49 from update_reqs import get_requirements
  50
  51 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
  52 CURRENT_NEEDS_PATH = os.path.join(THIS_DIR, "ice_validator/heat_requirements.json")
  53
  54
  55 def run_pytest(*args, msg="pytest failed"):
  56     original_dir = os.getcwd()
  57     try:
  58         os.chdir(os.path.join(THIS_DIR, "ice_validator"))
  59         if pytest.main(list(args)) != 0:
  60             return [msg]
  61     finally:
  62         os.chdir(original_dir)
  63
  64
  65 class Traceability:
  66
  67     PATH = os.path.join(THIS_DIR, "ice_validator/output/traceability.csv")
  68     TEST_FILE = 6
  69     TEST_NAME = 7
  70     IS_TESTABLE = 5
  71     REQ_ID = 0
  72
  73     def __init__(self):
  74         with open(self.PATH, "r") as f:
  75             rows = csv.reader(f)
  76             next(rows)  # skip header
  77             self.mappings = list(rows)
  78
  79     def unmapped_requirement_errors(self):
  80         """
  81         Returns list of errors where a requirement is testable, but no test was found.
  82         """
  83         testable_mappings = [m for m in self.mappings if m[self.IS_TESTABLE] == "True"]
  84         return [
  85             f"WARN: Missing test for {m[self.REQ_ID]}"
  86             for m in testable_mappings
  87             if not m[self.TEST_NAME]
  88         ]
  89
  90     def mapped_non_testable_requirement_errors(self):
  91         """
  92         Returns list of errors where the requirement isn't testable, but a test was
  93         found.
  94         """
  95         non_testables = [m for m in self.mappings if m[self.IS_TESTABLE] == "False"]
  96         return [
  97             (
  98                 f"No test for {m[0]} is needed, but found: "
  99                 f"{m[self.TEST_FILE]}::{m[self.TEST_NAME]} "
 100             )
 101             for m in non_testables
 102             if m[self.TEST_NAME]
 103         ]
 104
 105
 106 def current_version(needs):
 107     """Extracts and returns the needs under the current version"""
 108     return needs["versions"][needs["current_version"]]["needs"]
 109
 110
 111 def in_scope(_, req_metadata):
 112     """
 113     Checks if requirement is relevant to VVP.
 114
 115     :param: _: not used
 116     :param req_metadata: needs metadata about the requirement
 117     :return: True if the requirement is a testable, Heat requirement
 118     """
 119     return (
 120         "Heat" in req_metadata.get("docname", "")
 121         and "MUST" in req_metadata.get("keyword", "").upper()
 122         and req_metadata.get("validation_mode", "").lower() != "none"
 123     )
 124
 125
 126 def select_items(predicate, source_dict):
 127     """
 128     Creates a new dict from the source dict where the items match the given predicate
 129     :param predicate: predicate function that must accept a two arguments (key & value)
 130     :param source_dict: input dictionary to select from
 131     :return: filtered dict
 132     """
 133     return {k: v for k, v in source_dict.items() if predicate(k, v)}
 134
 135
 136 def check_requirements_up_to_date():
 137     """
 138     Checks if the requirements file packaged with VVP has meaningful differences
 139     to the requirements file published from VNFRQTS.
 140     """
 141     msg = "WARN: heat_requirements.json is out-of-date. Run update_reqs.py to update."
 142     latest_needs = json.load(get_requirements())
 143     with open(CURRENT_NEEDS_PATH, "r") as f:
 144         current_needs = json.load(f)
 145     latest_reqs = select_items(in_scope, current_version(latest_needs))
 146     current_reqs = select_items(in_scope, current_version(current_needs))
 147     if set(latest_reqs.keys()) != set(current_reqs.keys()):
 148         print(msg)
 149     elif not all(
 150         latest["description"] == current_reqs[r_id]["description"]
 151         for r_id, latest in latest_reqs.items()
 152     ):
 153         print(msg)
 154
 155
 156 def check_app_tests_pass():
 157     return run_pytest(
 158         "tests",
 159         "--self-test",
 160         msg="app_tests failed. Run pytest app_tests and fix errors.",
 161     )
 162
 163
 164 def check_self_test_pass():
 165     return run_pytest(
 166         "tests",
 167         "--self-test",
 168         msg="self-test failed. Run pytest --self-test and fix errors.",
 169     )
 170
 171
 172 def check_testable_requirements_are_mapped():
 173     tracing = Traceability()
 174     print("\n".join(tracing.unmapped_requirement_errors()))
 175
 176
 177 def check_non_testable_requirements_are_not_mapped():
 178     tracing = Traceability()
 179     return tracing.mapped_non_testable_requirement_errors()
 180
 181
 182 def check_flake8_passes():
 183     output = io.StringIO()
 184     with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
 185         app = Application()
 186         app.run(["ice_validator"])
 187     output.seek(0)
 188     lines = [f"   {l}" for l in output.readlines()]
 189     return ["flake8 errors detected:"] + lines if lines else []
 190
 191
 192 def check_bandit_passes():
 193     result = subprocess.run(  # nosec
 194         ["bandit", "-c", "bandit.yaml", "-r", ".", "-x", "./.tox/**"],  # nosec
 195         encoding="utf-8",  # nosec
 196         stdout=subprocess.PIPE,  # nosec
 197         stderr=subprocess.PIPE,  # nosec
 198     )  # nosec
 199     msgs = result.stdout.split("\n") if result.returncode != 0 else []
 200     return (
 201         ["bandit errors detected:"] + ["  {}".format(e) for e in msgs] if msgs else []
 202     )
 203
 204
 205 if __name__ == "__main__":
 206
 207     checks = [
 208         check_self_test_pass,
 209         check_non_testable_requirements_are_not_mapped,
 210         check_flake8_passes,
 211         check_bandit_passes,
 212     ]
 213     results = [check() for check in checks]
 214     errors = "\n".join("\n".join(msg) for msg in results if msg)
 215     print(errors or "Everything looks good!")
 216     check_requirements_up_to_date()
 217     check_testable_requirements_are_mapped()
 218     sys.exit(1 if errors else 0)