Check presence and match of fields in CRD and CSV in bundle

- Check if the following fields in CRD matches that in CSV in bundle - `CRD.spec.names.plural`.`CRD.spec.group` == `CSV.spec.crd.owned.name` - `CRD.spec.names.kind` == `CSV.spec.crd.owned.kind` - `CRD.spec.version` == `CSV.spec.crd.owned.version` - Check the presence of all above fields in the bundle
operator-framework · Mar 15, 2019 · 3f168f9 · 3f168f9
1 parent 4628bfa
commit 3f168f9
Show file tree

Hide file tree

Showing 11 changed files with 2,840 additions and 31 deletions.
diff --git a/operatorcourier/validate.py b/operatorcourier/validate.py
@@ -18,6 +18,14 @@ class ValidateCmd():
     crdKey = "customResourceDefinitions"
     csvKey = "clusterServiceVersions"
     pkgsKey = "packages"
+    crd_required_nested_fields = [
+        'metadata.name',
+        'apiVersion',
+        'spec.names.kind',
+        'spec.names.plural',
+        'spec.group',
+        'spec.version',
+    ]
 
     def __init__(self, ui_validate_io=False):
         self.ui_validate_io = ui_validate_io
@@ -85,30 +93,27 @@ def _bundle_validation(self, bundle):
 
     def _crd_validation(self, bundleData):
         logger.info("Validating custom resource definitions.")
-        valid = True
-
         crds = bundleData[self.crdKey]
 
+        valid = True
         for crd in crds:
-            if "metadata" in crd:
-                if "name" in crd["metadata"]:
-                    logger.info("Evaluating crd %s", crd["metadata"]["name"])
-                else:
-                    self._log_error("crd metadata.name not defined.")
-                    valid = False
-            else:
-                self._log_error("crd metadata not defined.")
-                valid = False
-
-            if "apiVersion" not in crd:
-                self._log_error("crd apiVersion not defined.")
-                valid = False
+            for crd_required_nested_field in self.crd_required_nested_fields:
+                valid &= self._crd_fields_validation(crd_required_nested_field.split('.'),
+                                                     crd, "")
+        return valid
 
-            if "spec" not in crd:
-                self._log_error("crd spec not defined.")
-                valid = False
+    def _crd_fields_validation(self, fields, yaml_dict, prefix):
+        if len(fields) == 0:
+            return True
+        root_field, subfields = fields[0], fields[1:]
+        field_string = root_field if prefix == "" else prefix + "." + root_field
 
-        return valid
+        if root_field == 'metadata' and 'name' in yaml_dict['metadata']:
+            logging.info("Evaluating crd %s", yaml_dict['metadata']['name'])
+        if root_field not in yaml_dict:
+            self._log_error("crd field %s not defined.", field_string)
+            return False
+        return self._crd_fields_validation(subfields, yaml_dict[root_field], field_string)
 
     def _csv_validation(self, bundleData):
         valid = True
@@ -167,20 +172,55 @@ def _csv_spec_validation(self, spec, bundleData):
                     pass
 
             if "owned" not in customresourcedefinitions:
-                self._log_error("spec.customresourcedefinitions.owned"
+                self._log_error("spec.customresourcedefinitions.owned "
                                 "not defined for csv")
-                valid = False
-            else:
-                for crd in customresourcedefinitions["owned"]:
-                    if "name" not in crd:
-                        self._log_error("name not defined for item in "
-                                        "spec.customresourcedefinitions.")
-                        valid = False
-                    else:
-                        if crd["name"] not in crdList:
-                            self._log_error("custom resource definition referenced "
-                                            "in csv not defined in root list of crds")
+                return False
+
+            for csvOwnedCrd in customresourcedefinitions["owned"]:
+                if "name" not in csvOwnedCrd:
+                    self._log_error("name not defined for item in "
+                                    "spec.customresourcedefinitions.")
+                    valid = False
+                elif csvOwnedCrd["name"] not in crdList:
+                    self._log_error("custom resource definition %s referenced in csv "
+                                    "not defined in root list of crds",
+                                    csvOwnedCrd["name"])
+                    valid = False
 
+                if "kind" not in csvOwnedCrd:
+                    self._log_error("kind not defined for item in "
+                                    "spec.customresourcedefinitions.")
+                    valid = False
+                if "version" not in csvOwnedCrd:
+                    self._log_error("version not defined for item in "
+                                    "spec.customresourcedefinitions.")
+                    valid = False
+
+                for crd in bundleData[self.crdKey]:
+                    if 'name' not in csvOwnedCrd or not \
+                        self._crd_fields_validation(['metadata', 'name'], crd, '') \
+                            or csvOwnedCrd['name'] != crd['metadata']['name']:
+                        continue
+                    if 'kind' in csvOwnedCrd and \
+                        self._crd_fields_validation(['spec', 'names', 'kind'], crd, '') \
+                            and csvOwnedCrd['kind'] != crd['spec']['names']['kind']:
+                        self._log_error('CRD.spec.names.kind does not match '
+                                        'CSV.spec.crd.owned.kind')
+                        valid = False
+                    if 'version' in csvOwnedCrd and \
+                        self._crd_fields_validation(['spec', 'version'], crd, '') and \
+                            csvOwnedCrd['version'] != crd['spec']['version']:
+                        self._log_error('CRD.spec.version does not match '
+                                        'CSV.spec.crd.owned.version')
+                        valid = False
+                    if 'name' in csvOwnedCrd and \
+                        self._crd_fields_validation(['spec', 'names', 'plural'], crd, '')\
+                            and self._crd_fields_validation(['spec', 'group'], crd, '') \
+                            and csvOwnedCrd['name'] != \
+                            crd['spec']['names']['plural'] + '.' + crd['spec']['group']:
+                        self._log_error("`CRD.spec.names.plural`.`CRD.spec.group` "
+                                        "does not match CSV.spec.crd.owned.name")
+                        valid = False
         return valid
 
     def _csv_metadata_validation(self, metadata):