feat: add ability for top document in a YAML file to merge down

tavern/_core/dict_util.py

@@ -203,7 +203,6 @@ def deep_dict_merge(initial_dct: dict, merge_dct: Mapping) -> dict:
     dict_merge recurses down into dicts nested to an arbitrary depth
     and returns the merged dict. Keys values present in merge_dct take
     precedence over values in initial_dct.
-    Modified from: https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
 
     Params:
         initial_dct: dict onto which the merge is executed
@@ -212,15 +211,9 @@ def deep_dict_merge(initial_dct: dict, merge_dct: Mapping) -> dict:
     Returns:
         recursively merged dict
     """
-    dct = initial_dct.copy()
-
-    for k in merge_dct:
-        if k in dct and isinstance(dct[k], dict) and isinstance(merge_dct[k], Mapping):
-            dct[k] = deep_dict_merge(dct[k], merge_dct[k])
-        else:
-            dct[k] = merge_dct[k]
-
-    return dct
+    initial_box = Box(initial_dct)
+    initial_box.merge_update(merge_dct)
+    return dict(initial_box)
 
 
 _CanCheck = Sequence | Mapping | set | Collection

tavern/_core/plugins.py

@@ -172,7 +172,7 @@ def get_extra_sessions(test_spec: Mapping, test_block_config: TestConfig) -> dic
 
     sessions = {}
 
-    plugins = load_plugins(test_block_config)
+    plugins: list[_Plugin] = load_plugins(test_block_config)
 
     for p in plugins:
         if any(
@@ -182,8 +182,8 @@ def get_extra_sessions(test_spec: Mapping, test_block_config: TestConfig) -> dic
             logger.debug(
                 "Initialising session for %s (%s)", p.name, p.plugin.session_type
             )
-            session_spec = test_spec.get(p.name, {})
-            formatted = format_keys(session_spec, test_block_config.variables)
+            session_spec: dict = test_spec.get(p.name, {})
+            formatted: dict = format_keys(session_spec, test_block_config.variables)
             sessions[p.name] = p.plugin.session_type(**formatted)
 
     return sessions
@@ -229,6 +229,7 @@ def get_request_type(
 
     # We've validated that 1 and only 1 is there, so just loop until the first
     # one is found
+    request_class: type[BaseRequest] | None = None
     for p in plugins:
         try:
             request_args = stage[p.plugin.request_block_name]
@@ -242,6 +243,9 @@ def get_request_type(
             )
             break
 
+    if not request_class:
+        raise exceptions.MissingSettingsError("No request type found")
+
     request_maker = request_class(session, request_args, test_block_config)
 
     return request_maker

tavern/_core/pytest/file.py

@@ -9,7 +9,7 @@
 from typing import Any, Union
 
 import pytest
-import yaml
+import yaml.parser
 from box import Box
 from pytest import Mark
 
@@ -445,16 +445,55 @@ def collect(self) -> Iterator[YamlItem]:
         except yaml.parser.ParserError as e:
             raise exceptions.BadSchemaError from e
 
-        for test_spec in all_tests:
+        merge_down = None
+        # Iterate over yaml documents and tests
+        for document_idx, test_spec in enumerate(all_tests):
             if not test_spec:
                 logger.warning("Empty document in input file '%s'", self.path)
                 continue
 
+            # Check if this document has the explicit 'defaults' marker
+            has_defaults_marker: bool = test_spec.pop("is_defaults", False)
+
+            # Validate that 'defaults' marker is only used in the first document
+            if has_defaults_marker and document_idx > 0:
+                raise exceptions.BadSchemaError(
+                    f"'defaults' marker can only be used in the first YAML document, but found it in document {document_idx + 1} of '{self.path}'"
+                )
+
+            missing_stages_and_name = (
+                "stages" not in test_spec or "test_name" not in test_spec
+            )
+            if document_idx == 0:
+                if has_defaults_marker:
+                    logger.info(
+                        "Found explicit defaults marker in first document from %s",
+                        self.path,
+                    )
+                    merge_down = test_spec
+                    continue
+                elif missing_stages_and_name:
+                    # Has a name but no stages - this is an error
+                    raise exceptions.BadSchemaError(
+                        f"First document in '{self.path}' has a name but no stages. "
+                        f"If this is meant to be defaults for the file, add 'defaults: true'. "
+                        f"If this is meant to be a test, add a 'stages' section."
+                    )
+            elif missing_stages_and_name:
+                raise exceptions.BadSchemaError(
+                    f"Document {document_idx + 1} in '{self.path}' does not have a 'test_name' or 'stages' section"
+                )
+
+            if merge_down:
+                test_spec = deep_dict_merge(test_spec, merge_down)
+
             try:
                 for i in self._generate_items(test_spec):
                     i.initialise_fixture_attrs()
                     yield i
             except (TypeError, KeyError) as e:
+                # If there was one of these errors, we can probably figure out
+                # if the error is from a bad test layout by calling verify_tests
                 try:
                     verify_tests(test_spec, with_plugins=False)
                 except Exception as e2:

tavern/_core/pytest/item.py

@@ -25,6 +25,7 @@
 from tavern._core.report import attach_text
 from tavern._core.run import run_test
 from tavern._core.schema.files import verify_tests
+from tavern._core.stage_lines import start_mark
 
 from .config import TestConfig
 from .util import load_global_cfg
@@ -130,7 +131,7 @@ def initialise_fixture_attrs(self) -> None:
     def location(self):
         """get location in file"""
         location = super().location
-        location = (location[0], self.spec.start_mark.line, location[2])
+        location = (location[0], start_mark(self.spec).line, location[2])
         return location
 
     #     Hack to stop issue with pytest-rerunfailures

tavern/_core/schema/tests.jsonschema.yaml

@@ -434,15 +434,17 @@ definitions:
 
 type: object
 additionalProperties: false
-required:
-  - test_name
-  - stages
 
 properties:
   test_name:
     type: string
     description: Name of test
 
+  is_defaults:
+    type: boolean
+    description: Whether this document contains default values to be merged with subsequent test documents
+    default: false
+
   _xfail:
     oneOf:
       - type: string

tests/integration/test_merge_down.tavern.yaml

@@ -0,0 +1,27 @@
+---
+is_defaults: true
+
+includes:
+  - !include common.yaml
+---
+test_name: Test redirecting loops
+
+stages:
+  - name: Expect a 302 without setting the flag
+    max_retries: 2
+    request:
+      follow_redirects: true
+      url: "{host}/redirect/loop"
+    response:
+      status_code: 200
+---
+test_name: Test redirecting loops in another test
+
+stages:
+  - name: Expect a 302 without setting the flag
+    max_retries: 2
+    request:
+      follow_redirects: true
+      url: "{host}/redirect/loop"
+    response:
+      status_code: 200

tests/unit/test_files.py

@@ -0,0 +1,115 @@
+import contextlib
+import dataclasses
+import pathlib
+import tempfile
+from collections.abc import Callable, Generator
+from typing import Any
+from unittest.mock import Mock
+
+import pytest
+import yaml
+
+from tavern._core import exceptions
+from tavern._core.pytest.file import YamlFile
+from tavern._core.pytest.item import YamlItem
+
+
+@pytest.fixture(scope="function")
+def tavern_test_content():
+    """return some example tests"""
+
+    test_docs = [
+        {"test_name": "First test", "stages": [{"name": "stage 1"}]},
+        {"test_name": "Second test", "stages": [{"name": "stage 2"}]},
+        {"test_name": "Third test", "stages": [{"name": "stage 3"}]},
+    ]
+
+    return test_docs
+
+
+@contextlib.contextmanager
+def tavern_test_file(test_content: list[Any]) -> Generator[pathlib.Path, Any, None]:
+    """Create a temporary YAML file with multiple documents"""
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        file_path = pathlib.Path(tmpdir) / "test.yaml"
+
+        # Write the documents to the file
+        with file_path.open("w", encoding="utf-8") as f:
+            for doc in test_content:
+                yaml.dump(doc, f)
+                f.write("---\n")
+
+        yield file_path
+
+
+@dataclasses.dataclass
+class Opener:
+    """Simple mock for generating items because pytest makes it hard to wrap
+    their internal functionality"""
+
+    path: pathlib.Path
+    _generate_items: Callable[[dict], Any]
+
+
+class TestGenerateFiles:
+    @pytest.mark.parametrize("with_merge_down_test", (True, False))
+    def test_multiple_documents(self, tavern_test_content, with_merge_down_test):
+        """Verify that multiple documents in a YAML file result in multiple tests"""
+
+        # Collect all tests
+        if with_merge_down_test:
+            tavern_test_content.insert(0, {"includes": [], "is_defaults": True})
+
+        def generate_yamlitem(test_spec):
+            mock = Mock(spec=YamlItem)
+            mock.name = test_spec["test_name"]
+            yield mock
+
+        with tavern_test_file(tavern_test_content) as filename:
+            tests = list(
+                YamlFile.collect(
+                    Opener(
+                        path=filename,
+                        _generate_items=generate_yamlitem,
+                    )
+                )
+            )
+
+        assert len(tests) == 3
+
+        # Verify each test has the correct name
+        expected_names = ["First test", "Second test", "Third test"]
+        for test, expected_name in zip(tests, expected_names):
+            assert test.name == expected_name
+
+    @pytest.mark.parametrize(
+        "content, exception",
+        (
+            ({"kookdff": "?A?A??"}, exceptions.BadSchemaError),
+            ({"test_name": "name", "stages": [{"name": "lflfl"}]}, TypeError),
+        ),
+    )
+    def test_reraise_exception(
+        self, tavern_test_content, content: dict, exception: BaseException
+    ):
+        """Verify that exceptions are properly reraised when loading YAML test files.
+
+        Test that when an exception occurs during test generation, it is properly
+        reraised as a schema error if the schema is bad."""
+
+        def raise_error(test_spec):
+            raise TypeError
+
+        tavern_test_content.insert(0, content)
+
+        with tavern_test_file(tavern_test_content) as filename:
+            with pytest.raises(exception):
+                list(
+                    YamlFile.collect(
+                        Opener(
+                            path=filename,
+                            _generate_items=raise_error,
+                        )
+                    )
+                )