Extend test collector to include yaml files under tests/schemas/ and fast data files (#2489)

* Extend test collector to include yaml files in tests/schemas/

* Silence linter

* Simplify yaml schema test names

* Vaidate FAST data files schema

modules/vpc-sc/schemas/access-level.schema.json

@@ -74,7 +74,8 @@
           "members": {
             "type": "array",
             "items": {
-              "type": "string"
+              "type": "string",
+              "pattern": "^(?:serviceAccount:|user:)"
             }
           },
           "negate": {
@@ -96,4 +97,4 @@
       }
     }
   }
-}
+}

tests/collectors.py

@@ -1,4 +1,4 @@
-# Copyright 2023 Google LLC
+# Copyright 2024 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -19,11 +19,20 @@ See FabricTestFile for details on the file structure.
 
 """
 
+import fnmatch
+import json
+import re
+from pathlib import Path
+
+import jsonschema
 import pytest
 import yaml
 
+from .examples.utils import get_tftest_directive
 from .fixtures import plan_summary, plan_validator
 
+_REPO_ROOT = Path(__file__).parents[1]
+
 
 class FabricTestFile(pytest.File):
 
@@ -88,20 +97,102 @@ class FabricTestItem(pytest.Item):
 
   def runtest(self):
     try:
-      summary = plan_validator(self.module, self.inventory, self.parent.path.parent,
-                             self.tf_var_files, self.extra_files)
+      summary = plan_validator(self.module, self.inventory,
+                               self.parent.path.parent, self.tf_var_files,
+                               self.extra_files)
     except AssertionError:
+
       def full_paths(x):
-        return [str(self.parent.path.parent / x ) for x in x]
+        return [str(self.parent.path.parent / x) for x in x]
+
       print(f'Error in inventory file: {" ".join(full_paths(self.inventory))}')
-      print(f'To regenerate inventory run: python tools/plan_summary.py {self.module} {" ".join(full_paths(self.tf_var_files))}')
+      print(
+          f'To regenerate inventory run: python tools/plan_summary.py {self.module} {" ".join(full_paths(self.tf_var_files))}'
+      )
       raise
 
   def reportinfo(self):
     return self.path, None, self.name
 
 
+class FabricSchemaTestFile(pytest.File):
+
+  def collect(self):
+    try:
+      raw = self.path.read_text()
+      content = yaml.safe_load(raw)
+    except (IOError, OSError, yaml.YAMLError) as e:
+      raise Exception(f'Cannot read test spec {self.path}: {e}')
+    directive = get_tftest_directive(raw)
+    if not directive or directive.name != 'tftest':
+      raise Exception(f'Schema test file without tftest directive: {self.path}')
+
+    if 'schema' not in directive.kwargs:
+      raise Exception(
+          f'Schema test file does not declare schema: {self.parent.path}')
+    schema_path = _REPO_ROOT / directive.kwargs['schema']
+    schema = json.load(schema_path.open())
+
+    yield SchemaTestItem.from_parent(self, name='schema', schema=schema,
+                                     content=content, directive=directive)
+
+
+class FastDataFile(pytest.File):
+
+  @staticmethod
+  def _get_yaml_schema(code):
+    regexp = rf"^ *# *yaml-language-server: *\$schema=([\S]*) *$"
+    if match := re.search(regexp, code, re.M):
+      return match.group(1)
+
+  def collect(self):
+    try:
+      raw = self.path.read_text()
+      content = yaml.safe_load(raw)
+    except (IOError, OSError, yaml.YAMLError) as e:
+      raise Exception(f'Cannot read test spec {self.path}: {e}')
+
+    schema_path = self._get_yaml_schema(raw)
+    if schema_path is None:
+      return
+
+    print(f"---- {schema_path}")
+    schema_path = (self.path.parent / schema_path).resolve()
+    print(f"---- {schema_path}")
+    schema = json.load(schema_path.open())
+
+    yield SchemaTestItem.from_parent(self, name='schema', schema=schema,
+                                     content=content)
+
+
+class SchemaTestItem(pytest.Item):
+
+  def __init__(self, name, parent, schema, content, directive=None):
+    super().__init__(name, parent)
+    self.schema = schema
+    self.content = content
+    self.directive = directive
+
+  def runtest(self):
+    if self.directive and 'fail' in self.directive.args:
+      with pytest.raises(jsonschema.exceptions.ValidationError):
+        jsonschema.validate(instance=self.content, schema=self.schema)
+    else:
+      jsonschema.validate(instance=self.content, schema=self.schema)
+
+  def reportinfo(self):
+    return self.path, None, self.name
+
+
 def pytest_collect_file(parent, file_path):
   'Collect tftest*.yaml files and run plan_validator from them.'
   if file_path.suffix == '.yaml' and file_path.name.startswith('tftest'):
     return FabricTestFile.from_parent(parent, path=file_path)
+
+  # use fnmatch because Path.match doesn't support **
+  file_path_relative = str(file_path.relative_to(_REPO_ROOT))
+  if fnmatch.fnmatch(file_path_relative, "tests/schemas/**.yaml"):
+    return FabricSchemaTestFile.from_parent(parent, path=file_path)
+
+  if fnmatch.fnmatch(file_path_relative, "fast/stages/*/data/**.yaml"):
+    return FastDataFile.from_parent(parent, path=file_path)

tests/schemas/vpc-sc/fail-access-level-invalid-member-prefix.yaml

@@ -0,0 +1,6 @@
+# skip boilerplate check
+# tftest schema=modules/vpc-sc/schemas/access-level.schema.json fail
+# fails because members must be prefixed with serviceAccount: or user:
+conditions:
+  - members:
+      - "group:group@example.com"

tests/schemas/vpc-sc/fail-access-level-no-member-prefix.yaml

@@ -0,0 +1,6 @@
+# skip boilerplate check
+# tftest schema=modules/vpc-sc/schemas/access-level.schema.json fail
+# fails because members must be prefixed with serviceAccount: or user:
+conditions:
+  - members:
+      - "user@example.com"

tests/schemas/vpc-sc/pass-access-level-member-prefix.yaml

@@ -0,0 +1,6 @@
+# skip boilerplate check
+# tftest schema=modules/vpc-sc/schemas/access-level.schema.json
+conditions:
+  - members:
+      - "user:user@example.com"
+      - "serviceAccount:group@example.com"