Use fixture for service_bootstrap resources and input replacements

Author: Kumar Gaurav Sharma

test/declarative_test_fwk/helper.py

@@ -18,21 +18,9 @@
 from time import sleep
 from acktest.k8s import resource as k8s
 
-input_replacements_dict = dict()
 TEST_HELPERS = dict()
 
 
-def input_replacements(provider):
-    """
-    Decorator to discover input replacements
-    :param provider: function that returns input replacements
-    :return: provider
-    """
-    global input_replacements_dict
-    input_replacements_dict = provider()
-    return provider
-
-
 def resource_helper(resource_kind: str):
     """
     Decorator to discover Custom Resource Helper
@@ -51,25 +39,27 @@ class ResourceHelper:
     """
     DEFAULT_WAIT_SECS = 30
 
-    def create(self, input_data: dict):
+    def create(self, input_data: dict, input_replacements: dict = {}):
         """
         Creates custom resource
         :param input_data: resource details
+        :param input_replacements: input replacements
         :return: k8s.CustomResourceReference, created custom resource
         """
-        reference = self.custom_resource_reference(input_data)
+        reference = self.custom_resource_reference(input_data, input_replacements)
         _ = k8s.create_custom_resource(reference, input_data)
         resource = k8s.wait_resource_consumed_by_controller(reference, wait_periods=10)
         assert resource is not None
         return (reference, resource)
 
-    def patch(self, input_data: dict):
+    def patch(self, input_data: dict, input_replacements: dict = {}):
         """
         Patches custom resource
         :param input_data: resource patch details
+        :param input_replacements: input replacements
         :return: k8s.CustomResourceReference, created custom resource
         """
-        reference = self.custom_resource_reference(input_data)
+        reference = self.custom_resource_reference(input_data, input_replacements)
         _ = k8s.patch_custom_resource(reference, input_data)
         sleep(self.DEFAULT_WAIT_SECS)  # required as controller has likely not placed the resource in modifying
         resource = k8s.wait_resource_consumed_by_controller(reference, wait_periods=10)
@@ -134,16 +124,17 @@ def assert_items(self, expectations: dict, state: dict):
                 continue
             assert (property, value) == (property, state.get(property))
 
-    def custom_resource_reference(self, input_data: dict) -> k8s.CustomResourceReference:
+    def custom_resource_reference(self, input_data: dict, input_replacements: dict = {}) -> k8s.CustomResourceReference:
         """
         Helper method to provide k8s.CustomResourceReference for supplied input
         :param input_data: custom resource input data
+        :param input_replacements: input replacements
         :return: k8s.CustomResourceReference
         """
         resource_plural = self.resource_plural(input_data.get("kind"))
         resource_name = input_data.get("metadata").get("name")
-        crd_group = input_replacements_dict.get("CRD_GROUP")
-        crd_version = input_replacements_dict.get("CRD_VERSION")
+        crd_group = input_replacements.get("CRD_GROUP")
+        crd_version = input_replacements.get("CRD_VERSION")
 
         reference = k8s.CustomResourceReference(
             crd_group, crd_version, resource_plural, resource_name, namespace="default")

test/declarative_test_fwk/loader.py

@@ -19,33 +19,47 @@
 import os
 from typing import Iterable
 from pathlib import Path
+from os.path import isfile, join
 from acktest.resources import load_resource_file, random_suffix_name
 
 
-def scenarios(scenarios_directory: Path) -> Iterable:
+def list_scenarios(scenarios_directory: Path) -> Iterable[Path]:
     """
-    Loads scenarios from given directory
+    Lists test scenarios from given directory
     :param scenarios_directory: directory containing scenarios yaml files
     :return: Iterable scenarios
     """
+    scenarios_list = []
     for scenario_file in os.listdir(scenarios_directory):
-        if not scenario_file.endswith(".yaml"):
+        scenario_file_full_path = join(scenarios_directory, scenario_file)
+        if not isfile(scenario_file_full_path) or not scenario_file.endswith(".yaml"):
             continue
-        scenario_name = scenario_file.split(".yaml")[0]
-        replacements = helper.input_replacements_dict.copy()
-        replacements["RANDOM_SUFFIX"] = random_suffix_name("", 32)
-        scenario = model.Scenario(load_resource_file(
-            scenarios_directory, scenario_name, additional_replacements=replacements))
-        yield pytest.param(scenario, marks=marks(scenario))
+        scenarios_list.append(Path(scenario_file_full_path))
+    return scenarios_list
 
 
-def idfn(scenario: model.Scenario) -> str:
+def load_scenario(scenario_file: Path, replacements: dict = {}) -> Iterable:
     """
-    Provides scenario test id
-    :param scenario: test scenario
+    Loads scenario from given scenario_file
+    :param scenario_file: yaml file containing scenarios
+    :param replacements: input replacements
+    :return: Iterable scenarios
+    """
+    scenario_name = scenario_file.name.split(".yaml")[0]
+    replacements = replacements.copy()
+    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))
+
+
+def idfn(scenario_file_full_path: Path) -> str:
+    """
+    Provides scenario file name as scenario test id
+    :param scenario: test scenario file path
     :return: scenario test id string
     """
-    return scenario.id()
+    return scenario_file_full_path.name
 
 
 def marks(scenario: model.Scenario) -> list:

test/declarative_test_fwk/model.py

@@ -23,8 +23,9 @@ class Step:
     """
     indent = "\t\t"
 
-    def __init__(self, config: dict):
+    def __init__(self, config: dict, replacements: dict = {}):
         self.config = config
+        self.replacements = replacements
 
         self.verb = None
         self.input_data = None
@@ -58,11 +59,12 @@ class Scenario:
     Represents a declarative test scenario with steps
     """
 
-    def __init__(self, config: dict):
+    def __init__(self, config: dict, replacements: dict = {}):
         self.config = config
         self.test_steps = []
+        self.replacements = replacements
         for step in self.config.get("steps", []):
-            self.test_steps.append(Step(step))
+            self.test_steps.append(Step(step, replacements))
 
     def id(self) -> str:
         return self.config.get("id", "")

test/declarative_test_fwk/runner.py

@@ -86,7 +86,7 @@ def create_resource(step: model.Step) -> None:
     if not step.input_data:
         return
 
-    (reference, ko) = step.resource_helper.create(step.input_data)
+    (reference, ko) = step.resource_helper.create(step.input_data, step.replacements)
     # track created reference to teardown later
     step.teardown_list.append((reference, ko))
 
@@ -96,14 +96,14 @@ def patch_resource(step: model.Step) -> None:
     if not step.input_data:
         return
 
-    (reference, ko) = step.resource_helper.patch(step.input_data)
+    (reference, ko) = step.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:
     if not reference:
         logging.debug(f"delete:  Step: {step.id()}")
-        reference = step.resource_helper.custom_resource_reference(step.input_data)
+        reference = step.resource_helper.custom_resource_reference(step.input_data, step.replacements)
 
     step.resource_helper.delete(reference)
 
@@ -114,7 +114,7 @@ def assert_expectations(step: model.Step) -> None:
         return
 
     resource_helper = step.resource_helper
-    reference = resource_helper.custom_resource_reference(step.input_data)
+    reference = resource_helper.custom_resource_reference(step.input_data, step.replacements)
     resource_helper.assert_expectations(step.expectations, reference)
 
 

test/e2e/bootstrap_resources.py

@@ -29,17 +29,6 @@ class TestBootstrapResources:
     NonDefaultUser: str
     CWLogGroup: str
 
-    def replacement_dict(self):
-        return {
-            "SNS_TOPIC_ARN": self.SnsTopicARN,
-            "SG_ID": self.SecurityGroupID,
-            "USERGROUP_ID": self.UserGroupID,
-            "KMS_KEY_ID": self.KmsKeyID,
-            "SNAPSHOT_NAME": self.SnapshotName,
-            "NON_DEFAULT_USER": self.SnapshotName
-        }
-
-
 _bootstrap_resources = None
 
 

test/e2e/declarative_tests.py

@@ -24,23 +24,9 @@
 import boto3
 import logging
 
-from e2e import service_marker, scenarios_directory, CRD_VERSION, CRD_GROUP, SERVICE_NAME
-from e2e.bootstrap_resources import get_bootstrap_resources
+from e2e import service_bootstrap, service_cleanup, service_marker, scenarios_directory, CRD_VERSION, CRD_GROUP, SERVICE_NAME
 from acktest.k8s import resource as k8s
 
-
-@helper.input_replacements
-def input_replacements():
-    """
-    Input replacements for test scenarios
-    """
-    replacements = get_bootstrap_resources().replacement_dict()
-    replacements["CRD_VERSION"] = CRD_VERSION
-    replacements["CRD_GROUP"] = CRD_GROUP
-    replacements["SERVICE_NAME"] = SERVICE_NAME
-    return replacements
-
-
 @helper.resource_helper("ReplicationGroup")
 class ReplicationGroupHelper(helper.ResourceHelper):
     """
@@ -54,16 +40,46 @@ def wait_for_delete(self, reference: k8s.CustomResourceReference):
         waiter.wait(ReplicationGroupId=reference.name)
 
 
-@pytest.fixture(params=loader.scenarios(scenarios_directory), ids=loader.idfn)
-def scenario(request):
+@pytest.fixture(scope="session")
+def input_replacements():
+    """
+    Session scoped fixture to bootstrap service resources and teardown
+    provides input replacements for test scenarios.
+    Eliminates the need for:
+     - bootstrap.yaml and
+     - call to <service_controller_test_dir>/service_bootstrap.py from test-infra
+     - call to <service_controller_test_dir>/service_cleanup.py from test-infra
+    """
+
+    resources_dict = service_bootstrap.service_bootstrap()
+    replacements = {
+        "CRD_VERSION": CRD_VERSION,
+        "CRD_GROUP": CRD_GROUP,
+        "SERVICE_NAME": SERVICE_NAME,
+        "SNS_TOPIC_ARN": resources_dict.get("SnsTopicARN"),
+        "SG_ID": resources_dict.get("SecurityGroupID"),
+        "USERGROUP_ID": resources_dict.get("UserGroupID"),
+        "KMS_KEY_ID": resources_dict.get("KmsKeyID"),
+        "SNAPSHOT_NAME": resources_dict.get("SnapshotName"),
+        "NON_DEFAULT_USER": resources_dict.get("SnapshotName")
+    }
+
+    yield replacements
+    # teardown
+    service_cleanup.service_cleanup(resources_dict)
+
+
+@pytest.fixture(params=loader.list_scenarios(scenarios_directory), ids=loader.idfn)
+def scenario(request, input_replacements):
     """
     Parameterized fixture
     Provides scenarios to execute
     Supports parallel execution of scenarios
     """
-    s = request.param
-    yield s
-    runner.teardown(s)
+    scenario_file_path = request.param
+    scenario = loader.load_scenario(scenario_file_path, input_replacements)
+    yield scenario
+    runner.teardown(scenario)
 
 
 @service_marker