feat: render common steps into every test case

CHANGELOG.md

@@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- Common test steps: files in a `shared` directory next to the test templates (i.e.
+  `<template_dir>/shared`), or an explicit `--common_dir`, are rendered into every generated test
+  case in addition to the test's own steps. This lets shared steps such as a teardown live in a single
+  place instead of being copied into each test. No-op unless the directory exists.
+
 ### Changed
 
 - Update registry references to oci ([#36]).

README.md

@@ -61,6 +61,18 @@ rm -rf tests/_work && beku
 cd tests/_work && kubectl kuttl test
 ```
 
+### Common steps
+
+Files placed in a `shared` directory next to the test templates (i.e.
+`tests/templates/kuttl/shared`) are rendered into *every* generated test case, in addition to that
+test's own steps. This lets shared steps — for example a teardown that deletes the product custom
+resources before the namespace is removed — live in a single place instead of being copied into each
+test. The directory is templated the same way as regular steps (`.j2`/`.jinja2` files are rendered,
+others copied; `NAMESPACE` and `lookup` are available). It is skipped if it does not exist, and its
+location can be overridden with `--common_dir`. On a name collision the test's own file wins — a
+shared file that would overwrite a file the test already provides is skipped (with a warning) — so a
+shared step never silently clobbers a test-specific one.
+
 Also see the `examples` folder.
 
 ## Release a new version

src/beku/kuttl.py

@@ -44,12 +44,16 @@ class TestFile:
     source_dir: str
     file_name: str
 
-    def build_destination(self) -> str:
+    def build_destination(self, skip_existing: bool = False) -> Optional[str]:
         """Copies the file name to the destination directory.
-        Returns the destination file name.
+        Returns the destination file name, or None if it was skipped because it already exists and
+        `skip_existing` is set (used for common/shared files so a test's own file always wins).
         """
         source = path.join(self.source_dir, self.file_name)
         dest = path.join(self.dest_dir, self.file_name)
+        if skip_existing and path.exists(dest):
+            logging.warning("Skipping common file %s: test case already provides %s", source, dest)
+            return None
         logging.debug("Copy file %s to %s", source, dest)
         copy2(source, dest)
         logging.debug("Update file mode for %s", dest)
@@ -68,13 +72,17 @@ class TestTemplate:
     env: Environment
     values: Dict[str, str]
 
-    def build_destination(self) -> str:
+    def build_destination(self, skip_existing: bool = False) -> Optional[str]:
         """Renders the template to file in the destination directory. The resulting file has the same name as the
         template but with the .j2 or .jinja2 ending removed.
-        Returns the rendered file name.
+        Returns the rendered file name, or None if it was skipped because it already exists and
+        `skip_existing` is set (used for common/shared files so a test's own file always wins).
         """
         source = path.join(self.source_dir, self.file_name)
         dest = path.join(self.dest_dir, re.sub(PATTERN_EXTENSION_JINJA, "", self.file_name))
+        if skip_existing and path.exists(dest):
+            logging.warning("Skipping common template %s: test case already provides %s", source, dest)
+            return None
         logging.debug("Render template %s to %s", source, dest)
         template = self.env.get_template(self.file_name)
         with open(dest, encoding="utf8", mode="w") as stream:
@@ -140,28 +148,50 @@ def tid(self) -> str:
         else:
             return name
 
-    def expand(self, template_dir: str, target_dir: str, namespace: str) -> None:
-        """Expand test case This will create the target folder, copy files and render render templates."""
+    def expand(self, template_dir: str, target_dir: str, namespace: str, common_dir: Optional[str] = None) -> None:
+        """Expand test case. This will create the target folder, copy files and render templates.
+
+        The test's own steps (from ``template_dir/<name>``) are rendered first. If ``common_dir`` is
+        given and exists, its files are then rendered into the SAME test folder, so shared steps (e.g.
+        a teardown) can live in a single place instead of being copied into every test. On a name
+        collision the test's OWN file wins: a common file that would overwrite a file the test already
+        provides is skipped (with a warning), so a shared file can never silently clobber a
+        test-specific one.
+        """
         logging.info("Expanding test case id [%s]", self.tid)
-        td_root = path.join(template_dir, self.name)
         tc_root = path.join(target_dir, self.name, self.tid)
         _mkdir_ignore_exists(tc_root)
-        test_env = Environment(loader=FileSystemLoader(path.join(template_dir, self.name)), trim_blocks=True)
-        test_env.globals["lookup"] = ansible_lookup
-        test_env.globals["NAMESPACE"] = determine_namespace(self.tid, namespace)
+        namespace = determine_namespace(self.tid, namespace)
+        self._render_dir(path.join(template_dir, self.name), tc_root, namespace)
+        if common_dir:
+            if path.isdir(common_dir):
+                logging.debug("Rendering common steps from [%s] into [%s]", common_dir, tc_root)
+                self._render_dir(common_dir, tc_root, namespace, skip_existing=True)
+            else:
+                logging.warning("Common steps directory [%s] does not exist, skipping", common_dir)
+
+    def _render_dir(self, source_root: str, tc_root: str, namespace: str, skip_existing: bool = False) -> None:
+        """Render/copy every file under ``source_root`` into ``tc_root``, preserving sub-directories.
+
+        When ``skip_existing`` is set, files whose destination already exists are skipped (used for
+        common/shared files so a test's own file always takes precedence).
+        """
+        env = Environment(loader=FileSystemLoader(source_root), trim_blocks=True)
+        env.globals["lookup"] = ansible_lookup
+        env.globals["NAMESPACE"] = namespace
         sub_level: int = 0
-        for root, dirs, files in walk(td_root):
+        for root, dirs, files in walk(source_root):
             sub_level += 1
             if sub_level == 8:
                 # Sanity check
                 raise ValueError("Maximum recursive level (8) reached.")
             for dir_name in dirs:
-                _mkdir_ignore_exists(path.join(tc_root, root[len(td_root) + 1 :], dir_name))
+                _mkdir_ignore_exists(path.join(tc_root, root[len(source_root) + 1 :], dir_name))
             for file_name in files:
                 test_source = make_test_source_with_context(
-                    file_name, root, path.join(tc_root, root[len(td_root) + 1 :]), test_env, self.values
+                    file_name, root, path.join(tc_root, root[len(source_root) + 1 :]), env, self.values
                 )
-                test_source.build_destination()
+                test_source.build_destination(skip_existing=skip_existing)
 
 
 @dataclass(frozen=True, eq=True)
@@ -336,6 +366,7 @@ def expand(
     output_dir: str,
     kuttl_tests: str,
     namespace: str,
+    common_dir: Optional[str] = None,
 ) -> int:
     """Expand test suite."""
     try:
@@ -344,7 +375,7 @@ def expand(
         _mkdir_ignore_exists(output_dir)
         _expand_kuttl_tests(ets.test_cases, output_dir, kuttl_tests)
         for test_case in ets.test_cases:
-            test_case.expand(template_dir, output_dir, namespace)
+            test_case.expand(template_dir, output_dir, namespace, common_dir)
     except StopIteration as exc:
         raise ValueError(f"Cannot expand test suite [{suite}] because cannot find it in [{kuttl_tests}]") from exc
     return 0

src/beku/main.py

@@ -60,6 +60,19 @@ def parse_cli_args() -> Namespace:
         default="tests/kuttl-test.yaml.jinja2",
     )
 
+    parser.add_argument(
+        "-c",
+        "--common_dir",
+        help="Folder with common test step templates/files that are rendered into EVERY generated "
+        "test case (in addition to the test's own steps). Lets shared steps such as a teardown live "
+        "in a single place instead of being copied into each test. Defaults to a 'shared' folder "
+        "next to the test templates (i.e. <template_dir>/shared); skipped if that folder does not "
+        "exist, so this is a no-op unless you create it.",
+        type=str,
+        required=False,
+        default=None,
+    )
+
     parser.add_argument(
         "-s",
         "--suite",
@@ -94,11 +107,17 @@ def main() -> int:
     rmtree(path=cli_args.output_dir, ignore_errors=True)
     # Compatibility warning: add 'tests' to output_dir
     output_dir = path.join(cli_args.output_dir, "tests")
+    # Default the common steps directory to a 'shared' folder next to the test templates. It is
+    # rendered into every test case if present, and silently skipped otherwise (so this stays a no-op
+    # for repositories that don't opt in by creating it). Note: 'commons' is deliberately NOT used, as
+    # some operators already keep unrelated helper scripts there.
+    common_dir = cli_args.common_dir if cli_args.common_dir is not None else path.join(cli_args.template_dir, "shared")
     return expand(
         cli_args.suite,
         effective_test_suites,
         cli_args.template_dir,
         output_dir,
         cli_args.kuttl_test,
         cli_args.namespace,
+        common_dir,
     )

src/beku/test/test_common_steps.py

@@ -0,0 +1,85 @@
+"""Tests for the common-steps feature: files in a common directory are rendered into every test case."""
+
+import tempfile
+import unittest
+from os import makedirs, path
+
+from beku.kuttl import TestCase
+
+
+def _write(file_path: str, content: str) -> None:
+    makedirs(path.dirname(file_path), exist_ok=True)
+    with open(file_path, encoding="utf8", mode="w") as stream:
+        stream.write(content)
+
+
+class TestCommonSteps(unittest.TestCase):
+    def test_common_dir_rendered_into_test_case(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            template_dir = path.join(tmp, "templates")
+            common_dir = path.join(tmp, "commons")
+            target_dir = path.join(tmp, "_work")
+
+            # A test with its own step, and a plain + templated common file.
+            _write(path.join(template_dir, "mytest", "00-install.yaml"), "step\n")
+            _write(path.join(common_dir, "99-teardown.yaml"), "teardown\n")
+            _write(path.join(common_dir, "50-note.txt.j2"), "ns={{ NAMESPACE }}\n")
+
+            TestCase(name="mytest", values={}).expand(template_dir, target_dir, "kuttl-fixed", common_dir)
+
+            tc_dir = path.join(target_dir, "mytest", "mytest")
+            # the test's own step is present
+            self.assertTrue(path.isfile(path.join(tc_dir, "00-install.yaml")))
+            # the common plain file is present
+            self.assertTrue(path.isfile(path.join(tc_dir, "99-teardown.yaml")))
+            # the common template is rendered (suffix stripped, NAMESPACE substituted)
+            rendered = path.join(tc_dir, "50-note.txt")
+            self.assertTrue(path.isfile(rendered))
+            with open(rendered, encoding="utf8") as stream:
+                self.assertIn("ns=kuttl-fixed", stream.read())
+
+    def test_test_own_file_wins_on_collision(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            template_dir = path.join(tmp, "templates")
+            common_dir = path.join(tmp, "shared")
+            target_dir = path.join(tmp, "_work")
+
+            # Same file name in the test and in the shared dir, with different content.
+            _write(path.join(template_dir, "mytest", "10-check.yaml"), "TEST-OWN\n")
+            _write(path.join(common_dir, "10-check.yaml"), "SHARED\n")
+
+            TestCase(name="mytest", values={}).expand(template_dir, target_dir, "kuttl-fixed", common_dir)
+
+            # The test's own file must win; the shared file must not clobber it.
+            with open(path.join(target_dir, "mytest", "mytest", "10-check.yaml"), encoding="utf8") as stream:
+                self.assertEqual("TEST-OWN\n", stream.read())
+
+    def test_missing_common_dir_is_skipped(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            template_dir = path.join(tmp, "templates")
+            target_dir = path.join(tmp, "_work")
+            _write(path.join(template_dir, "mytest", "00-install.yaml"), "step\n")
+
+            # Non-existent common dir must not raise and must not add anything.
+            TestCase(name="mytest", values={}).expand(
+                template_dir, target_dir, "kuttl-fixed", path.join(tmp, "does-not-exist")
+            )
+
+            tc_dir = path.join(target_dir, "mytest", "mytest")
+            self.assertTrue(path.isfile(path.join(tc_dir, "00-install.yaml")))
+            self.assertFalse(path.isfile(path.join(tc_dir, "99-teardown.yaml")))
+
+    def test_no_common_dir_argument_is_backwards_compatible(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            template_dir = path.join(tmp, "templates")
+            target_dir = path.join(tmp, "_work")
+            _write(path.join(template_dir, "mytest", "00-install.yaml"), "step\n")
+
+            # Old call signature (no common_dir) keeps working.
+            TestCase(name="mytest", values={}).expand(template_dir, target_dir, "kuttl-fixed")
+
+            self.assertTrue(path.isfile(path.join(target_dir, "mytest", "mytest", "00-install.yaml")))
+
+
+if __name__ == "__main__":
+    unittest.main()