Lint issues and unittest issues

compliance-api/Makefile

@@ -71,7 +71,7 @@ pylint: ## Linting with pylint
 	. venv/bin/activate && pylint --rcfile=setup.cfg  src/$(PROJECT_NAME)
 
 flake8: ## Linting with flake8
-	. venv/bin/activate && flake8 src/$(PROJECT_NAME) tests
+	. venv/bin/activate && flake8 --ignore=Q000 src/$(PROJECT_NAME) tests
 
 lintfix: ## Linting fix
 	. venv/bin/activate && autopep8 -i -a src/**/*.py tests/**/*.py

compliance-api/src/compliance_api/__init__.py

@@ -4,8 +4,8 @@
 """
 
 import os
-
 from http import HTTPStatus
+
 import secure
 from flask import Flask, current_app, g, request
 from flask_cors import CORS
@@ -16,6 +16,7 @@
 from compliance_api.utils.cache import cache
 from compliance_api.utils.util import allowedorigins
 
+
 # Security Response headers
 csp = (
     secure.ContentSecurityPolicy()
@@ -39,18 +40,17 @@
 def create_app(run_mode=os.getenv("FLASK_ENV", "development")):
     """Create flask app."""
     # pylint: disable=import-outside-toplevel
-    from compliance_api.resources import (
-        API_BLUEPRINT,
-        OPS_BLUEPRINT,
-    )
+    from compliance_api.resources import API_BLUEPRINT, OPS_BLUEPRINT
 
     # Flask app initialize
     app = Flask(__name__)
 
     # All configuration are in config file
     app.config.from_object(get_named_config(run_mode))
 
-    CORS(app, resources={r"/*": {"origins": allowedorigins()}}, supports_credentials=True)
+    CORS(
+        app, resources={r"/*": {"origins": allowedorigins()}}, supports_credentials=True
+    )
 
     # Setup jwt for keycloak
     if os.getenv("FLASK_ENV", "production") != "testing":
@@ -91,6 +91,7 @@ def handle_error(err):
             raise err
         current_app.logger.error(str(err))
         return "Internal server error", HTTPStatus.INTERNAL_SERVER_ERROR
+
     # Return App for run in run.py file
     return app
 
@@ -109,12 +110,13 @@ def get_roles(a_dict):
     app_context.config["JWT_ROLE_CALLBACK"] = get_roles
     jwt_manager.init_app(app_context)
 
+
 def register_shellcontext(app):
     """Register shell context objects."""
     from compliance_api import models  # pylint: disable=import-outside-toplevel
 
     def shell_context():
         """Shell context objects."""
-        return {'app': app, 'jwt': jwt, 'db': db, 'models': models}  # pragma: no cover
+        return {"app": app, "jwt": jwt, "db": db, "models": models}  # pragma: no cover
 
     app.shell_context_processor(shell_context)

compliance-api/src/compliance_api/auth.py

@@ -17,6 +17,7 @@
 from flask import g, request
 from flask_jwt_oidc import JwtManager
 
+
 jwt = (
     JwtManager()
 )  # pylint: disable=invalid-name; lower case name as used by convention in most Flask apps
@@ -40,6 +41,4 @@ def decorated(*args, **kwargs):
         return decorated
 
 
-auth = (
-    Auth()
-)
+auth = Auth()

compliance-api/src/compliance_api/config.py

@@ -24,75 +24,78 @@
 
 from dotenv import find_dotenv, load_dotenv
 
+
 # this will load all the envars from a .env file located in the project root (api)
 load_dotenv(find_dotenv())
 
 
-def get_named_config(config_name: str = 'development'):
+def get_named_config(config_name: str = "development"):
     """Return the configuration object based on the name.
 
     :raise: KeyError: if an unknown configuration is requested
     """
-    if config_name in ['production', 'staging', 'default']:
+    if config_name in ["production", "staging", "default"]:
         config = ProdConfig()
-    elif config_name == 'testing':
+    elif config_name == "testing":
         config = TestConfig()
-    elif config_name == 'development':
+    elif config_name == "development":
         config = DevConfig()
-    elif config_name == 'docker':
+    elif config_name == "docker":
         config = DockerConfig()
     else:
         raise KeyError("Unknown configuration '{config_name}'")
     return config
 
 
-class _Config():  # pylint: disable=too-few-public-methods
+class _Config:  # pylint: disable=too-few-public-methods
     """Base class configuration that should set reasonable defaults for all the other configurations."""
 
     PROJECT_ROOT = os.path.abspath(os.path.dirname(__file__))
 
-    SECRET_KEY = 'a secret'
+    SECRET_KEY = "a secret"
 
     TESTING = False
     DEBUG = False
 
     # POSTGRESQL
-    DB_USER = os.getenv('DATABASE_USERNAME', '')
-    DB_PASSWORD = os.getenv('DATABASE_PASSWORD', '')
-    DB_NAME = os.getenv('DATABASE_NAME', '')
-    DB_HOST = os.getenv('DATABASE_HOST', '')
-    DB_PORT = os.getenv('DATABASE_PORT', '5432')
-    SQLALCHEMY_DATABASE_URI = f'postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{int(DB_PORT)}/{DB_NAME}'
+    DB_USER = os.getenv("DATABASE_USERNAME", "")
+    DB_PASSWORD = os.getenv("DATABASE_PASSWORD", "")
+    DB_NAME = os.getenv("DATABASE_NAME", "")
+    DB_HOST = os.getenv("DATABASE_HOST", "")
+    DB_PORT = os.getenv("DATABASE_PORT", "5432")
+    SQLALCHEMY_DATABASE_URI = (
+        f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{int(DB_PORT)}/{DB_NAME}"
+    )
     SQLALCHEMY_ECHO = True
     SQLALCHEMY_TRACK_MODIFICATIONS = False
-    SQLALCHEMY_RECORD_QUERIES=True
+    SQLALCHEMY_RECORD_QUERIES = True
 
     # JWT_OIDC Settings
-    JWT_OIDC_WELL_KNOWN_CONFIG = os.getenv('JWT_OIDC_WELL_KNOWN_CONFIG')
-    JWT_OIDC_ALGORITHMS = os.getenv('JWT_OIDC_ALGORITHMS', 'RS256')
-    JWT_OIDC_JWKS_URI = os.getenv('JWT_OIDC_JWKS_URI')
-    JWT_OIDC_ISSUER = os.getenv('JWT_OIDC_ISSUER')
-    JWT_OIDC_AUDIENCE = os.getenv('JWT_OIDC_AUDIENCE', 'account')
-    JWT_OIDC_CACHING_ENABLED = os.getenv('JWT_OIDC_CACHING_ENABLED', 'True')
+    JWT_OIDC_WELL_KNOWN_CONFIG = os.getenv("JWT_OIDC_WELL_KNOWN_CONFIG")
+    JWT_OIDC_ALGORITHMS = os.getenv("JWT_OIDC_ALGORITHMS", "RS256")
+    JWT_OIDC_JWKS_URI = os.getenv("JWT_OIDC_JWKS_URI")
+    JWT_OIDC_ISSUER = os.getenv("JWT_OIDC_ISSUER")
+    JWT_OIDC_AUDIENCE = os.getenv("JWT_OIDC_AUDIENCE", "account")
+    JWT_OIDC_CACHING_ENABLED = os.getenv("JWT_OIDC_CACHING_ENABLED", "True")
     JWT_OIDC_JWKS_CACHE_TIMEOUT = 300
 
     # Service account details
-    KEYCLOAK_BASE_URL = os.getenv('KEYCLOAK_BASE_URL')
-    KEYCLOAK_REALMNAME = os.getenv('KEYCLOAK_REALMNAME', 'compliance')
-    KEYCLOAK_SERVICE_ACCOUNT_ID = os.getenv('MET_ADMIN_CLIENT_ID')
-    KEYCLOAK_SERVICE_ACCOUNT_SECRET = os.getenv('MET_ADMIN_CLIENT_SECRET')
+    KEYCLOAK_BASE_URL = os.getenv("KEYCLOAK_BASE_URL")
+    KEYCLOAK_REALMNAME = os.getenv("KEYCLOAK_REALMNAME", "compliance")
+    KEYCLOAK_SERVICE_ACCOUNT_ID = os.getenv("MET_ADMIN_CLIENT_ID")
+    KEYCLOAK_SERVICE_ACCOUNT_SECRET = os.getenv("MET_ADMIN_CLIENT_SECRET")
     # TODO separate out clients for APIs and user management.
     # TODO API client wont need user management roles in keycloak.
-    KEYCLOAK_ADMIN_USERNAME = os.getenv('MET_ADMIN_CLIENT_ID')
-    KEYCLOAK_ADMIN_SECRET = os.getenv('MET_ADMIN_CLIENT_SECRET')
+    KEYCLOAK_ADMIN_USERNAME = os.getenv("MET_ADMIN_CLIENT_ID")
+    KEYCLOAK_ADMIN_SECRET = os.getenv("MET_ADMIN_CLIENT_SECRET")
 
 
 class DevConfig(_Config):  # pylint: disable=too-few-public-methods
     """Dev Config."""
 
     TESTING = False
     DEBUG = True
-    print(f'SQLAlchemy URL (DevConfig): {_Config.SQLALCHEMY_DATABASE_URI}')
+    print(f"SQLAlchemy URL (DevConfig): {_Config.SQLALCHEMY_DATABASE_URI}")
 
 
 class TestConfig(_Config):  # pylint: disable=too-few-public-methods
@@ -104,45 +107,101 @@ class TestConfig(_Config):  # pylint: disable=too-few-public-methods
     TESTING = True
 
     # POSTGRESQL
-    DB_USER = os.getenv('DATABASE_TEST_USERNAME', 'postgres')
-    DB_PASSWORD = os.getenv('DATABASE_TEST_PASSWORD', 'postgres')
-    DB_NAME = os.getenv('DATABASE_TEST_NAME', 'testdb')
-    DB_HOST = os.getenv('DATABASE_TEST_HOST', 'localhost')
-    DB_PORT = os.getenv('DATABASE_TEST_PORT', '5432')
-    SQLALCHEMY_DATABASE_URI = f'postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{int(DB_PORT)}/{DB_NAME}'
+    DB_USER = os.getenv("DATABASE_TEST_USERNAME", "postgres")
+    DB_PASSWORD = os.getenv("DATABASE_TEST_PASSWORD", "postgres")
+    DB_NAME = os.getenv("DATABASE_TEST_NAME", "testdb")
+    DB_HOST = os.getenv("DATABASE_TEST_HOST", "localhost")
+    DB_PORT = os.getenv("DATABASE_TEST_PORT", "5432")
+    SQLALCHEMY_DATABASE_URI = (
+        f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{int(DB_PORT)}/{DB_NAME}"
+    )
 
     JWT_OIDC_TEST_MODE = True
     # JWT_OIDC_ISSUER = _get_config('JWT_OIDC_TEST_ISSUER')
-    JWT_OIDC_TEST_AUDIENCE = os.getenv('JWT_OIDC_TEST_AUDIENCE')
-    JWT_OIDC_TEST_CLIENT_SECRET = os.getenv('JWT_OIDC_TEST_CLIENT_SECRET')
-    JWT_OIDC_TEST_ISSUER = os.getenv('JWT_OIDC_TEST_ISSUER')
-    JWT_OIDC_WELL_KNOWN_CONFIG = os.getenv('JWT_OIDC_WELL_KNOWN_CONFIG')
-    JWT_OIDC_TEST_ALGORITHMS = os.getenv('JWT_OIDC_TEST_ALGORITHMS')
-    JWT_OIDC_TEST_JWKS_URI = os.getenv('JWT_OIDC_TEST_JWKS_URI', default=None)
+    JWT_OIDC_TEST_AUDIENCE = os.getenv("JWT_OIDC_TEST_AUDIENCE")
+    JWT_OIDC_TEST_CLIENT_SECRET = os.getenv("JWT_OIDC_TEST_CLIENT_SECRET")
+    JWT_OIDC_TEST_ISSUER = os.getenv("JWT_OIDC_TEST_ISSUER")
+    JWT_OIDC_WELL_KNOWN_CONFIG = os.getenv("JWT_OIDC_WELL_KNOWN_CONFIG")
+    JWT_OIDC_TEST_ALGORITHMS = os.getenv("JWT_OIDC_TEST_ALGORITHMS")
+    JWT_OIDC_TEST_JWKS_URI = os.getenv("JWT_OIDC_TEST_JWKS_URI", default=None)
+
+    JWT_OIDC_TEST_KEYS = {
+        'keys': [
+            {
+                'kid': 'epictrack',
+                'kty': 'RSA',
+                'alg': 'RS256',
+                'use': 'sig',
+                'n': 'AN-fWcpCyE5KPzHDjigLaSUVZI0uYrcGcc40InVtl-rQRDmAh-C2W8H4_Hxhr5VLc6crsJ2LiJTV_E72S03pzpOOaaYV6-'
+                     'TzAjCou2GYJIXev7f6Hh512PuG5wyxda_TlBSsI-gvphRTPsKCnPutrbiukCYrnPuWxX5_cES9eStR',
+                'e': 'AQAB'
+            }
+        ]
+    }
+
+    JWT_OIDC_TEST_PRIVATE_KEY_JWKS = {
+        'keys': [
+            {
+                'kid': 'forms-flow-ai',
+                'kty': 'RSA',
+                'alg': 'RS256',
+                'use': 'sig',
+                'n': 'AN-fWcpCyE5KPzHDjigLaSUVZI0uYrcGcc40InVtl-rQRDmAh-C2W8H4_Hxhr5VLc6crsJ2LiJTV_E72S03pzpOOaaYV6-'
+                     'TzAjCou2GYJIXev7f6Hh512PuG5wyxda_TlBSsI-gvphRTPsKCnPutrbiukCYrnPuWxX5_cES9eStR',
+                'e': 'AQAB',
+                'd': 'C0G3QGI6OQ6tvbCNYGCqq043YI_8MiBl7C5dqbGZmx1ewdJBhMNJPStuckhskURaDwk4-'
+                     '8VBW9SlvcfSJJrnZhgFMjOYSSsBtPGBIMIdM5eSKbenCCjO8Tg0BUh_'
+                     'xa3CHST1W4RQ5rFXadZ9AeNtaGcWj2acmXNO3DVETXAX3x0',
+                'p': 'APXcusFMQNHjh6KVD_hOUIw87lvK13WkDEeeuqAydai9Ig9JKEAAfV94W6Aftka7tGgE7ulg1vo3eJoLWJ1zvKM',
+                'q': 'AOjX3OnPJnk0ZFUQBwhduCweRi37I6DAdLTnhDvcPTrrNWuKPg9uGwHjzFCJgKd8KBaDQ0X1rZTZLTqi3peT43s',
+                'dp': 'AN9kBoA5o6_Rl9zeqdsIdWFmv4DB5lEqlEnC7HlAP-3oo3jWFO9KQqArQL1V8w2D4aCd0uJULiC9pCP7aTHvBhc',
+                'dq': 'ANtbSY6njfpPploQsF9sU26U0s7MsuLljM1E8uml8bVJE1mNsiu9MgpUvg39jEu9BtM2tDD7Y51AAIEmIQex1nM',
+                'qi': 'XLE5O360x-MhsdFXx8Vwz4304-MJg-oGSJXCK_ZWYOB_FGXFRTfebxCsSYi0YwJo-oNu96bvZCuMplzRI1liZw'
+            }
+        ]
+    }
+
+    JWT_OIDC_TEST_PRIVATE_KEY_PEM = """-----BEGIN RSA PRIVATE KEY-----
+    MIICXQIBAAKBgQDfn1nKQshOSj8xw44oC2klFWSNLmK3BnHONCJ1bZfq0EQ5gIfg
+    tlvB+Px8Ya+VS3OnK7Cdi4iU1fxO9ktN6c6TjmmmFevk8wIwqLthmCSF3r+3+h4e
+    ddj7hucMsXWv05QUrCPoL6YUUz7Cgpz7ra24rpAmK5z7lsV+f3BEvXkrUQIDAQAB
+    AoGAC0G3QGI6OQ6tvbCNYGCqq043YI/8MiBl7C5dqbGZmx1ewdJBhMNJPStuckhs
+    kURaDwk4+8VBW9SlvcfSJJrnZhgFMjOYSSsBtPGBIMIdM5eSKbenCCjO8Tg0BUh/
+    xa3CHST1W4RQ5rFXadZ9AeNtaGcWj2acmXNO3DVETXAX3x0CQQD13LrBTEDR44ei
+    lQ/4TlCMPO5bytd1pAxHnrqgMnWovSIPSShAAH1feFugH7ZGu7RoBO7pYNb6N3ia
+    C1idc7yjAkEA6Nfc6c8meTRkVRAHCF24LB5GLfsjoMB0tOeEO9w9Ous1a4o+D24b
+    AePMUImAp3woFoNDRfWtlNktOqLel5PjewJBAN9kBoA5o6/Rl9zeqdsIdWFmv4DB
+    5lEqlEnC7HlAP+3oo3jWFO9KQqArQL1V8w2D4aCd0uJULiC9pCP7aTHvBhcCQQDb
+    W0mOp436T6ZaELBfbFNulNLOzLLi5YzNRPLppfG1SRNZjbIrvTIKVL4N/YxLvQbT
+    NrQw+2OdQACBJiEHsdZzAkBcsTk7frTH4yGx0VfHxXDPjfTj4wmD6gZIlcIr9lZg
+    4H8UZcVFN95vEKxJiLRjAmj6g273pu9kK4ymXNEjWWJn
+    -----END RSA PRIVATE KEY-----"""
 
 
 class DockerConfig(_Config):  # pylint: disable=too-few-public-methods
     """In support of testing only.used by the py.test suite."""
 
     # POSTGRESQL
-    DB_USER = os.getenv('DATABASE_DOCKER_USERNAME')
-    DB_PASSWORD = os.getenv('DATABASE_DOCKER_PASSWORD')
-    DB_NAME = os.getenv('DATABASE_DOCKER_NAME')
-    DB_HOST = os.getenv('DATABASE_DOCKER_HOST')
-    DB_PORT = os.getenv('DATABASE_DOCKER_PORT', '5432')
-    SQLALCHEMY_DATABASE_URI = f'postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{int(DB_PORT)}/{DB_NAME}'
+    DB_USER = os.getenv("DATABASE_DOCKER_USERNAME")
+    DB_PASSWORD = os.getenv("DATABASE_DOCKER_PASSWORD")
+    DB_NAME = os.getenv("DATABASE_DOCKER_NAME")
+    DB_HOST = os.getenv("DATABASE_DOCKER_HOST")
+    DB_PORT = os.getenv("DATABASE_DOCKER_PORT", "5432")
+    SQLALCHEMY_DATABASE_URI = (
+        f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{int(DB_PORT)}/{DB_NAME}"
+    )
 
-    print(f'SQLAlchemy URL (Docker): {SQLALCHEMY_DATABASE_URI}')
+    print(f"SQLAlchemy URL (Docker): {SQLALCHEMY_DATABASE_URI}")
 
 
 class ProdConfig(_Config):  # pylint: disable=too-few-public-methods
     """Production Config."""
 
-    SECRET_KEY = os.getenv('SECRET_KEY', None)
+    SECRET_KEY = os.getenv("SECRET_KEY", None)
 
     if not SECRET_KEY:
         SECRET_KEY = os.urandom(24)
-        print('WARNING: SECRET_KEY being set as a one-shot', file=sys.stderr)
+        print("WARNING: SECRET_KEY being set as a one-shot", file=sys.stderr)
 
     TESTING = False
     DEBUG = False

compliance-api/src/compliance_api/exceptions/__init__.py

@@ -19,7 +19,7 @@
 error - a description of the error {code / description: classname / full text}
 status_code - where possible use HTTP Error Codes
 """
-from werkzeug.exceptions import UnprocessableEntity, Forbidden, NotFound, BadRequest, Conflict
+from werkzeug.exceptions import BadRequest, Conflict, Forbidden, NotFound, UnprocessableEntity
 from werkzeug.wrappers.response import Response
 
 
@@ -44,7 +44,7 @@ def __init__(self, message, *args, **kwargs):
 
 
 class BadRequestError(BadRequest):
-    """Exception raised when there are issues with the api input"""
+    """Exception raised when there are issues with the api input."""
 
     def __init__(self, message, *args, **kwargs):
         """Return a valid BadRequestError."""
@@ -54,7 +54,7 @@ def __init__(self, message, *args, **kwargs):
 
 
 class ResourceNotFoundError(NotFound):
-    """Exception raised when resource not found"""
+    """Exception raised when resource not found."""
 
     def __init__(self, message, *args, **kwargs):
         """Return a valid ResourceExistsError."""
@@ -64,7 +64,7 @@ def __init__(self, message, *args, **kwargs):
 
 
 class PermissionDeniedError(Forbidden):
-    """Exception raised when resource not found"""
+    """Exception raised when resource not found."""
 
     def __init__(self, message, *args, **kwargs):
         """Return a valid ResourceExistsError."""
@@ -74,7 +74,7 @@ def __init__(self, message, *args, **kwargs):
 
 
 class UnprocessableEntityError(UnprocessableEntity):
-    """Exception raised when resource is not processable"""
+    """Exception raised when resource is not processable."""
 
     def __init__(self, message, *args, **kwargs):
         """Return a valid UnprocessableEntityError."""

compliance-api/src/compliance_api/models/base_model.py

@@ -19,6 +19,7 @@
 
 from .db import db
 
+
 TENANT_ID = 'tenant_id'
 
 

compliance-api/src/compliance_api/models/db.py

@@ -4,6 +4,8 @@
 from flask_marshmallow import Marshmallow
 from flask_migrate import Migrate
 from flask_sqlalchemy import SQLAlchemy
+
+
 # DB initialize in __init__ file
 # db variable use for create models from here
 db = SQLAlchemy()