Support customResourceReference at Scenario level and ability to override it at Step level

Author: Kumar Gaurav Sharma

test/declarative_test_fwk/__init__.py

@@ -0,0 +1,12 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You may
+# not use this file except in compliance with the License. A copy of the
+# License is located at
+#
+#	 http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is distributed
+# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+# express or implied. See the License for the specific language governing
+# permissions and limitations under the License.

test/declarative_test_fwk/helper.py

@@ -81,9 +81,11 @@ def delete(self, reference: k8s.CustomResourceReference):
         sleep(self.DEFAULT_WAIT_SECS)
         self.wait_for_delete(reference) # throws exception if wait fails
 
-    def assert_expectations(self, expectations: dict, reference: k8s.CustomResourceReference):
+    def assert_expectations(self, verb: str, input_data: dict, expectations: dict, reference: k8s.CustomResourceReference):
         """
         Asserts custom resource reference against supplied expectations
+        :param verb: expectations after performing the verb (apply, patch, delete)
+        :param input_data: input data to verb
         :param expectations: expectations to assert
         :param reference: custom resource reference
         :return: None

test/declarative_test_fwk/loader.py

@@ -14,7 +14,7 @@
 """Test Scenarios loader for Declarative tests framework for custom resources
 """
 
-from declarative_test_fwk import model, helper
+from declarative_test_fwk import model
 import pytest
 import os
 from typing import Iterable
@@ -34,11 +34,12 @@ def list_scenarios(scenarios_directory: Path) -> Iterable[Path]:
         scenario_file_full_path = join(scenarios_directory, scenario_file)
         if not isfile(scenario_file_full_path) or not scenario_file.endswith(".yaml"):
             continue
-        scenarios_list.append(Path(scenario_file_full_path))
+        scenario = load_scenario(Path(scenario_file_full_path))
+        scenarios_list.append(pytest.param(Path(scenario_file_full_path),marks=marks(scenario)))
     return scenarios_list
 
 
-def load_scenario(scenario_file: Path, replacements: dict = {}) -> Iterable:
+def load_scenario(scenario_file: Path, replacements: dict = {}) -> model.Scenario:
     """
     Loads scenario from given scenario_file
     :param scenario_file: yaml file containing scenarios
@@ -50,7 +51,7 @@ def load_scenario(scenario_file: Path, replacements: dict = {}) -> Iterable:
     replacements["RANDOM_SUFFIX"] = random_suffix_name("", 32)
     scenario = model.Scenario(load_resource_file(
         scenario_file.parent, scenario_name, additional_replacements=replacements), replacements)
-    yield pytest.param(scenario, marks=marks(scenario))
+    return scenario
 
 
 def idfn(scenario_file_full_path: Path) -> str:

test/declarative_test_fwk/model.py

@@ -21,38 +21,56 @@ class Step:
     """
     Represents a declarative test step
     """
-    indent = "\t\t"
 
-    def __init__(self, config: dict, replacements: dict = {}):
+    def __init__(self, config: dict, custom_resource_details: dict, replacements: dict = {}):
         self.config = config
+        self.custom_resource_details = custom_resource_details
         self.replacements = replacements
 
         self.verb = None
-        self.input_data = None
+        self.input_data = {}
         self.expectations = None
-        self.resource_helper = None
 
         # (k8s.CustomResourceReference, ko) to teardown
         self.teardown_list = []
 
-        supported_verbs=["create", "patch", "delete"]
+        supported_verbs = ["create", "patch", "delete"]
         for verb in supported_verbs:
             if verb not in self.config:
                 continue
             self.verb = verb
             self.input_data = self.config.get(verb)
+            if type(self.input_data) is str:
+                # consider the input as resource name
+                # confirm that self.custom_resource_details must be provided with same name
+                if self.custom_resource_details["metadata"]["name"] != self.input_data:
+                    raise ValueError(f"Unable to determine input data for '{self.verb}' at step: {self.id()}")
+                # self.custom_resource_details will be mixed in into self.input_data
+                self.input_data = {}
             break
 
-        if self.input_data:
-            self.expectations = self.config.get("expect")
-            self.resource_helper = helper.get_resource_helper(self.input_data.get("kind"))
+        if len(self.input_data) == 0 and not self.custom_resource_details:
+            raise ValueError(f"Unable to determine custom resource at step: {self.id()}")
+
+        if self.custom_resource_details:
+            self.input_data = {**self.custom_resource_details, **self.input_data}
+        self.expectations = self.config.get("expect")
 
     def id(self) -> str:
         return self.config.get("id", "")
 
     def description(self) -> str:
         return self.config.get("description", "")
 
+    def resource_kind(self) -> str:
+        return self.input_data.get("kind")
+
+    def __str__(self) -> str:
+        return f"Step(id='{self.id()}')"
+
+    def __repr__(self) -> str:
+        return str(self)
+
 
 class Scenario:
     """
@@ -63,8 +81,9 @@ def __init__(self, config: dict, replacements: dict = {}):
         self.config = config
         self.test_steps = []
         self.replacements = replacements
+        custom_resource_details = self.config.get("customResourceReference", {})
         for step in self.config.get("steps", []):
-            self.test_steps.append(Step(step, replacements))
+            self.test_steps.append(Step(step, custom_resource_details.copy(), replacements))
 
     def id(self) -> str:
         return self.config.get("id", "")
@@ -80,3 +99,9 @@ def marks(self) -> list:
 
     def steps(self):
         return self.test_steps
+
+    def __str__(self) -> str:
+        return f"Scenario(id='{self.id()}')"
+
+    def __repr__(self) -> str:
+        return str(self)

test/declarative_test_fwk/runner.py

@@ -14,7 +14,7 @@
 """Runner for Declarative tests framework scenarios for custom resources
 """
 
-from declarative_test_fwk import model
+from declarative_test_fwk import model, helper
 import pytest
 import sys
 import logging
@@ -28,7 +28,7 @@ def run(scenario: model.Scenario) -> None:
     if not scenario:
         return
 
-    logging.info(f"Execute: Scenario: {scenario.id()}")
+    logging.info(f"Execute: {scenario}")
     for step in scenario.steps():
         run_step(step)
 
@@ -40,20 +40,20 @@ def teardown(scenario: model.Scenario) -> None:
     if not scenario:
         return
 
-    logging.info(f"Teardown: Scenario: {scenario.id()}")
+    logging.info(f"Teardown: {scenario}")
     teardown_failures = []
     # tear down steps in reverse order
     for step in reversed(scenario.steps()):
         try:
             teardown_step(step)
         except:
-            error = f"Failed to teardown Step: {step.id()}. " \
+            error = f"Failed to teardown: {step}. " \
                     f"Unexpected error: {sys.exc_info()[0]}"
             teardown_failures.append(error)
             logging.debug(error)
 
     if len(teardown_failures) != 0:
-        teardown_failures.insert(0, f"Failures during teardown. Scenario: {scenario.id()}")
+        teardown_failures.insert(0, f"Failures during teardown: {scenario}")
         failures = "\n\t- ".join(teardown_failures)
         logging.error(failures)
         pytest.fail(failures)
@@ -68,54 +68,61 @@ def run_step(step: model.Step) -> None:
 
     if not step.verb:
         logging.warning(
-            f"skipped: Step: {step.id()}. No matching verb found."
+            f"skipped: {step}. No matching verb found."
             f" Supported verbs: create, patch, delete.")
         return
 
+    logging.info(f"Execute: {step}")
     if step.verb == "create":
         create_resource(step)
     elif step.verb == "patch":
         patch_resource(step)
     elif step.verb == "delete":
-        pass
+        delete_resource(step)
     assert_expectations(step)
 
 
 def create_resource(step: model.Step) -> None:
-    logging.debug(f"create:  Step: {step.id()}")
+    logging.debug(f"create:  {step}")
     if not step.input_data:
         return
-
-    (reference, ko) = step.resource_helper.create(step.input_data, step.replacements)
+    resource_helper = helper.get_resource_helper(step.resource_kind())
+    (reference, ko) = resource_helper.create(step.input_data, step.replacements)
     # track created reference to teardown later
     step.teardown_list.append((reference, ko))
 
 
 def patch_resource(step: model.Step) -> None:
-    logging.debug(f"patch:   Step: {step.id()}")
+    logging.debug(f"patch:   {step}")
     if not step.input_data:
         return
 
-    (reference, ko) = step.resource_helper.patch(step.input_data, step.replacements)
+    resource_helper = helper.get_resource_helper(step.resource_kind())
+    (reference, ko) = resource_helper.patch(step.input_data, step.replacements)
     # no need to teardown patched reference, its creator should tear it down.
 
 
 def delete_resource(step: model.Step, reference: k8s.CustomResourceReference = None) -> None:
+    resource_helper = helper.get_resource_helper(step.resource_kind())
     if not reference:
-        logging.debug(f"delete:  Step: {step.id()}")
-        reference = step.resource_helper.custom_resource_reference(step.input_data, step.replacements)
+        logging.debug(f"delete:  {step}")
+        reference = resource_helper.custom_resource_reference(step.input_data, step.replacements)
 
-    step.resource_helper.delete(reference)
+    resource_helper.delete(reference)
 
 
 def assert_expectations(step: model.Step) -> None:
-    logging.debug(f"assert:  Step: {step.id()}")
+    logging.info(f"assert:  {step}")
     if not step.expectations:
         return
 
-    resource_helper = step.resource_helper
+    resource_helper = helper.get_resource_helper(step.resource_kind())
     reference = resource_helper.custom_resource_reference(step.input_data, step.replacements)
-    resource_helper.assert_expectations(step.expectations, reference)
+    try:
+        resource_helper.assert_expectations(step.verb, step.input_data, step.expectations, reference)
+    except AssertionError as ae:
+        logging.error(f"AssertionError at {step}")
+        raise ae
 
 
 def teardown_step(step: model.Step) -> None:
@@ -125,7 +132,7 @@ def teardown_step(step: model.Step) -> None:
     if not step or len(step.teardown_list) == 0:
         return
 
-    logging.info(f"teardown: Step: {step.id()}")
+    logging.info(f"teardown: {step}")
 
     for (reference, _) in step.teardown_list:
         if reference:

test/e2e/scenarios/scenario1.yaml renamed to test/e2e/scenarios/cmd_basic_create_update.yaml

@@ -6,14 +6,15 @@ usecases:
 #marks:
 #  - slow
 #  - blocked
+customResourceReference:
+  apiVersion: $CRD_GROUP/$CRD_VERSION
+  kind: ReplicationGroup
+  metadata:
+    name: test$RANDOM_SUFFIX
 steps:
   - id: "create_CMD_replication_group"
     description: "Initial config"
     create:
-      apiVersion: $CRD_GROUP/$CRD_VERSION
-      kind: ReplicationGroup
-      metadata:
-        name: test$RANDOM_SUFFIX
       spec:
         engine: redis
         replicationGroupID: test$RANDOM_SUFFIX
@@ -28,10 +29,6 @@ steps:
   - id: "invalid_cacheNodeType_on_CMD_replication_group_causes_Terminal_condition"
     description: "Negative test include transitEncryptionEnabled"
     patch:
-      apiVersion: $CRD_GROUP/$CRD_VERSION
-      kind: ReplicationGroup
-      metadata:
-        name: test$RANDOM_SUFFIX
       spec:
         cacheNodeType: cache.micro # invalid value
     expect:
@@ -40,8 +37,4 @@ steps:
           ACK.Terminal: "True"
   - id: "delete_CMD_RG"
     description: "Delete cluster mode disabled replication group"
-    delete:
-      apiVersion: $CRD_GROUP/$CRD_VERSION
-      kind: ReplicationGroup
-      metadata:
-        name: test$RANDOM_SUFFIX
+    delete: test$RANDOM_SUFFIX

test/e2e/declarative_tests.py renamed to test/e2e/tests/test_scenarios.py

@@ -29,6 +29,14 @@
 
 @helper.resource_helper("ReplicationGroup")
 class ReplicationGroupHelper(helper.ResourceHelper):
+    def assert_expectations(self, verb: str, input_data: dict, expectations: dict,
+                            reference: k8s.CustomResourceReference):
+        # default assertions
+        super().assert_expectations(verb, input_data, expectations, reference)
+
+        # perform custom server side checks based on:
+        # verb, input data to verb, expectations for given resource
+
     """
     Helper for replication group scenarios
     """
@@ -83,7 +91,7 @@ def scenario(request, input_replacements):
 
 
 @service_marker
-class TestSuite:
+class TestScenarios:
     """
     Declarative scenarios based test suite
     """