Merge pull request #599 from NHSDigital/APM-7323-Sonarqube-fix

Refactor YAML parameters and improve variable handling

Author: sathiya-nhs

ansible/collections/ansible_collections/nhsd/apigee/plugins/action/add_policy_to_pre_flow.py

@@ -32,7 +32,7 @@ def run(self, tmp=None, task_vars=None):
         proxies_dir = args.dist_dir.joinpath(
             "proxies", args.proxy_dir, "apiproxy/proxies"
         )
-        proxies_files = [f for f in proxies_dir.glob("*.xml")]
+        proxies_files = list(proxies_dir.glob("*.xml"))
 
         if len(proxies_files) != 1:
             return {
@@ -57,13 +57,12 @@ def run(self, tmp=None, task_vars=None):
             name = step.find("Name")
             if name.text == args.policy_name:
                 return {"changed": False}
-                break
 
         result = {"changed": True}
         if diff_mode:
             result["diff"] = {
                 "before": etree.tostring(tree, pretty_print=True).decode(),
-                "before_header": str(proxies_file)
+                "before_header": str(proxies_file),
             }
 
         step = etree.Element("Step")
@@ -76,8 +75,8 @@ def run(self, tmp=None, task_vars=None):
             result["diff"].update(
                 {
                     "after": etree.tostring(tree, pretty_print=True).decode(),
-                    "after_header": str(proxies_file)
-                 }
+                    "after_header": str(proxies_file),
+                }
             )
 
         if check_mode or result["changed"] is False:

ansible/collections/ansible_collections/nhsd/apigee/plugins/module_utils/apigee_action.py

@@ -4,10 +4,10 @@
 
 
 class ApigeeAction(ansible.plugins.action.ActionBase):
-    def validate_args(self, Validator: pydantic.BaseModel):
+    def validate_args(self, validator: type[pydantic.BaseModel]):
         """Returns two-length tuple of validated_args and errors dicts."""
         try:
-            args = Validator(**self._task.args)
+            args = validator(**self._task.args)
             return args, {}
         except pydantic.ValidationError as e:
             return (

ansible/collections/ansible_collections/nhsd/apigee/plugins/module_utils/models/ansible/validate_manifest.py

@@ -53,14 +53,14 @@ def check_service_name(cls, service_name, values):
     def prepend_dist_dir_to_spec_paths(cls, manifest, values):
         dist_dir = values.get("dist_dir")
         print(dist_dir)
-        if not dist_dir:
-            return manifest
-        apigee = manifest["apigee"]
-        for env_dict in apigee["environments"]:
-            for spec_dict in env_dict["specs"]:
-                path = spec_dict.get("path")
-                if path is not None:
-                    spec_dict["path"] = os.path.join(dist_dir, path)
+        if dist_dir:
+            apigee = manifest["apigee"]
+            for env_dict in apigee["environments"]:
+                for spec_dict in env_dict["specs"]:
+                    path = spec_dict.get("path")
+                    if path is not None:
+                        spec_dict["path"] = os.path.join(dist_dir, path)
+
         return manifest
 
     @pydantic.validator("manifest")

ansible/collections/ansible_collections/nhsd/apigee/plugins/module_utils/models/apigee/product.py

@@ -1,5 +1,5 @@
 import json
-from typing import Union, Literal, List, Dict, Any, Type
+from typing import Optional, Union, Literal, List, Dict, Any, Type
 from typing_extensions import Annotated
 from pydantic import (
     Field,
@@ -110,29 +110,34 @@ def _literal_name(class_):
     + ")$)"
 )
 
+
 class ApigeeProductAttributeOther(BaseModel):
     name: constr(regex=PRODUCT_ATTRIBUTE_REGEX)
     value: str
 
+
 ApigeeProductAttributeSpecial = Annotated[
-    Union [
+    Union[
         ApigeeProductAttributeAccess,
         ApigeeProductAttributeRateLimit,
         ApigeeProductAttributeRateLimiting,
     ],
-    Field(discriminator="name")
+    Field(discriminator="name"),
 ]
 
+
 def _count_cls(items: List[Any], cls: Type):
     return sum(isinstance(item, cls) for item in items)
 
 
 class ApigeeProduct(BaseModel):
     name: str
     approvalType: Literal["auto", "manual"] = "manual"
-    attributes: List[Union[ApigeeProductAttributeSpecial, ApigeeProductAttributeOther]] = [{"name": "access", "value": "private"}]
-    description: str = None
-    displayName: str = None
+    attributes: List[
+        Union[ApigeeProductAttributeSpecial, ApigeeProductAttributeOther]
+    ] = [{"name": "access", "value": "private"}]
+    description: Optional[str] = None
+    displayName: Optional[str] = None
 
     # Note: This value is manually inserted by apigee_environment
     # object that contains this product. So if you do not provide a

ansible/collections/ansible_collections/nhsd/apigee/plugins/module_utils/models/apigee/rate_limiting_config.py

@@ -2,13 +2,13 @@
 Pydantic class for the rateliming config JSON, attached to products
 and apps to control the ApplyRateLimiting shared flow.
 """
-from typing import Literal
+
+from typing import Literal, Optional
 
 from pydantic import BaseModel, conint, constr, Extra
 
 
 class ExcludeNoneModel(BaseModel):
-
     """
     Providing default values for ratelimiting here would mean that
     changing defaults required a redeploy for all proxies.
@@ -21,26 +21,27 @@ class ExcludeNoneModel(BaseModel):
     update the defaults for everyone by just by updating the shared
     flow.
     """
+
     def dict(self, **kwargs):
         kwargs["exclude_none"] = True
         return super().dict(**kwargs)
 
     class Config:
-        extra=Extra.forbid
+        extra = Extra.forbid
 
 
 class QuotaConfig(ExcludeNoneModel):
-    enabled: bool = None
+    enabled: Optional[bool] = None
     interval: conint(gt=0) = None
     limit: conint(gt=0) = None
     timeunit: Literal["minute", "hour", "day", "week", "month"] = None
 
 
 class SpikeArrestConfig(ExcludeNoneModel):
-    enabled: bool = None
-    ratelimit: constr(regex=r"^[1-9][0-9]*(ps|pm)$") = None
+    enabled: Optional[bool] = None
+    ratelimit: Optional[constr(regex=r"^[1-9][0-9]*(ps|pm)$")] = None
 
 
 class RateLimitingConfig(ExcludeNoneModel):
-    quota: QuotaConfig = None
-    spikeArrest: SpikeArrestConfig = None
+    quota: Optional[QuotaConfig] = None
+    spikeArrest: Optional[SpikeArrestConfig] = None

ansible/collections/ansible_collections/nhsd/apigee/plugins/module_utils/models/manifest/meta.py

@@ -9,7 +9,6 @@
 _REGISTRY_DATA = {}
 
 
-
 class ManifestMetaApi(pydantic.BaseModel):
     name: pydantic.constr(regex=r"^[a-z][a-z0-9]*(-[a-z0-9]+)*$")
     id: typing.Optional[pydantic.UUID4] = pydantic.Field(
@@ -29,7 +28,7 @@ def dict(self, **kwargs):
             {
                 "guid": str(native["guid"]),
                 "spec_guids": [str(guid) for guid in spec_guids],
-             }
+            }
         )
         return native
 
@@ -58,7 +57,9 @@ def validate_guid(cls, guid, values):
             guid = _REGISTRY_DATA[name]["guid"]
         registered_guid = _REGISTRY_DATA[name]["guid"]
         if str(guid) != registered_guid:
-            raise ValueError(f"Supplied guid {guid} does not match registered guid {registered_guid}")
+            raise ValueError(
+                f"Supplied guid {guid} does not match registered guid {registered_guid}"
+            )
         return guid
 
     @pydantic.validator("spec_guids")
@@ -78,11 +79,12 @@ def validate_spec_guids(cls, spec_guids, values):
             if str(spec_guid) not in registered_spec_guids:
                 invalid.append(str(spec_guid))
         if len(invalid) > 0:
-            raise ValueError(f"Supplied spec_guids {invalid} do not match registered spec_guids {registered_spec_guids}")
+            raise ValueError(
+                f"Supplied spec_guids {invalid} do not match registered spec_guids {registered_spec_guids}"
+            )
         return spec_guids
 
 
-
 class ManifestMeta(pydantic.BaseModel):
     schema_version: pydantic.constr(regex=r"[1-9][0-9]*(\.[0-9]+){0,2}")
     api: ManifestMetaApi
@@ -91,7 +93,7 @@ class ManifestMeta(pydantic.BaseModel):
     def validate_schema_version(cls, schema_version):
         semantic_parts = schema_version.split(".")
 
-        MAJOR, MINOR, PATCH = [int(x) for x in SCHEMA_VERSION.split(".")]
+        MAJOR, _, _ = [int(x) for x in SCHEMA_VERSION.split(".")]
         major = int(semantic_parts[0])
 
         # Checking against minor/patch would not allow to us deploy

ansible/collections/ansible_collections/nhsd/apigee/scripts/update_schema.py

@@ -8,10 +8,11 @@
 import typing
 
 from ansible_collections.nhsd.apigee.plugins.module_utils.models.manifest import meta
-from ansible_collections.nhsd.apigee.plugins.module_utils.models.manifest.manifest import Manifest
+from ansible_collections.nhsd.apigee.plugins.module_utils.models.manifest.manifest import (
+    Manifest,
+)
 
-
-SCHEMA_VERSION_REGEX = re.compile(r"^([1-9][0-9]*)\.([0-9]+)\.([0-9]+)$")
+SCHEMA_VERSION_REGEX = re.compile(r"^([1-9]\d*)\.(\d+)\.(\d+)$")
 
 
 class SchemaString:
@@ -23,7 +24,9 @@ def __init__(self, schema_or_major: typing.Union[str, int], minor=None, patch=No
             match = re.match(SCHEMA_VERSION_REGEX, schema_version)
             if not match:
                 raise ValueError(f"Invalid SchemaString {schema_version}")
-            self.major, self.minor, self.patch = [int(match.group(x)) for x in range(1, 4)]
+            self.major, self.minor, self.patch = [
+                int(match.group(x)) for x in range(1, 4)
+            ]
         elif isinstance(schema_or_major, int):
             if not isinstance(minor, int) and isinstance(patch, int):
                 raise ValueError("SchemaString.__init__ requires major, minor, patch")
@@ -47,41 +50,51 @@ def __lt__(self, other):
         return (
             self.major < other.major
             or (self.major == other.major and self.minor < other.minor)
-            or (self.major == other.major and self.minor == other.minor and self.patch < other.patch)
+            or (
+                self.major == other.major
+                and self.minor == other.minor
+                and self.patch < other.patch
+            )
         )
 
     def __le__(self, other):
         return self < other or self == other
 
     def __gt__(self, other):
-        return not self <= other
+        return (self.major, self.minor, self.patch) > (
+            other.major,
+            other.minor,
+            other.patch,
+        )
 
     def __ge__(self, other):
-        return not self < other
+        return (self.major, self.minor, self.patch) >= (
+            other.major,
+            other.minor,
+            other.patch,
+        )
 
     def valid_increments(self):
         return [
             SchemaString(self.major + 1, 0, 0),
             SchemaString(self.major, self.minor + 1, 0),
-            SchemaString(self.major, self.minor, self.patch + 1)
+            SchemaString(self.major, self.minor, self.patch + 1),
         ]
 
 
 def main():
-    """
-
-    """
+    """ """
     UPDATED_SCHEMA_VERSION = SchemaString(meta.SCHEMA_VERSION)
 
     script_dir = pathlib.Path(__file__).parent
-    relative_schema_dir = script_dir.joinpath(f"../tests/unit/plugins/module_utils/models/manifest/schema_versions/")
-
-    # last_schema_file_name = pathlib.Path(f"v{}.json")
+    relative_schema_dir = script_dir.joinpath(
+        "../tests/unit/plugins/module_utils/models/manifest/schema_versions/"
+    )
 
     schema_dir_glob = relative_schema_dir.glob("v*.json")
 
     SCHEMA_VERSION = SchemaString("1.0.0")
-    SCHEMA_FILE_NAME_PATTERN = re.compile(r"v([1-9][0-9]*\.[0-9]+\.[0-9]+)\.json$")
+    SCHEMA_FILE_NAME_PATTERN = re.compile(r"v([1-9]\d*\.\d+\.\d+)\.json$")
     for schema_file in schema_dir_glob:
         match = re.match(SCHEMA_FILE_NAME_PATTERN, schema_file.name)
         schema_version = SchemaString(match.group(1))
@@ -100,17 +113,22 @@ def main():
         fromfile=str(SCHEMA_VERSION),
         tofile=str(UPDATED_SCHEMA_VERSION),
     )
-    deltas = [delta for delta in deltas]
+    deltas = list(deltas)
 
     if not deltas:
-        raise ValueError(f"No difference between proposed {UPDATED_SCHEMA_VERSION} schema and current {meta.SCHEMA_VERSION}")
+        raise ValueError(
+            f"No difference between proposed {UPDATED_SCHEMA_VERSION} schema and current {meta.SCHEMA_VERSION}"
+        )
 
     if UPDATED_SCHEMA_VERSION not in SCHEMA_VERSION.valid_increments():
-        raise ValueError(f"""{UPDATED_SCHEMA_VERSION} is invalid increment after current {SCHEMA_VERSION}.
+        raise ValueError(
+            f"""{UPDATED_SCHEMA_VERSION} is invalid increment after current {SCHEMA_VERSION}.
 Please increment major, minor or patch integer, e.g:
-""" + f"\n".join(str(x) for x in SCHEMA_VERSION.valid_increments()))
+"""
+            + "\n".join(str(x) for x in SCHEMA_VERSION.valid_increments())
+        )
 
-    print("-"*50)
+    print("-" * 50)
     for delta in deltas:
         if delta.startswith("+"):
             col = colorama.Fore.GREEN
@@ -122,7 +140,7 @@ def main():
         print(colorama.Fore.RESET)
 
     _input = None
-    print("-"*50)
+    print("-" * 50)
     print("Confirm spec changes? ", end="")
     while _input not in ["y", "n"]:
         if _input is not None:
@@ -139,8 +157,7 @@ def main():
 
     with open(new_schema_file, "w") as f:
         f.write(new_schema)
-    print(
-        f"""Wrote {new_schema_file}"
+    print(f"""Wrote {new_schema_file}"
 Validate this file by executing
 $ make test
 in {script_dir.parent}""")