hyera.testing

Helpers for testing a :class:~hyera.backends.Backend: a hook runs the way the engine runs it, from a unit test and without a hiera.yaml.

Nothing here imports pytest or an optional dependency.

NOT_FOUND = _NotFoundType() module-attribute

BackendContract

Checks every backend must pass, as plain test_* methods.

A test module subclasses it, names the class under test in :attr:backend and implements :meth:write_source; a runner such as pytest then collects the subclass. Nothing here imports pytest: a check asks for the tmp_path fixture by parameter name and fails with assert. Each check runs for every hook kind the backend implements (data_hash, lookup_key, data_dig); a kind the backend does not implement is left out.

backend class-attribute

data = {'contract_text': 'value', 'contract_number': 7, 'contract_flag': True, 'contract_list': ['a', 'b'], 'contract_nested': {'key': 'v'}} class-attribute

test_a_hook_returns_the_values_of_present_keys(tmp_path)

Each hook returns the source's value for every key it holds.

Parameters:

Name	Type	Description	Default
tmp_path	Any	a fresh directory.	required

Raises:

Type	Description
AssertionError	a value differs.

Source code in src/hyera/testing.py

def test_a_hook_returns_the_values_of_present_keys(self, tmp_path: _ty.Any) -> None:
    """Each hook returns the source's value for every key it holds.

    :param tmp_path: a fresh directory.
    :raises AssertionError: a value differs.
    """
    for kind in self._kinds():
        options, path = self._source(tmp_path / kind, self.data)
        got = self._values(kind, self.backend(), options, path)
        for key, value in self.data.items():
            assert got.get(key) == value, "{}: {!r} gave {!r}, not {!r}".format(
                kind, key, got.get(key), value
            )

test_a_lookup_through_hiera_returns_the_same_values(tmp_path)

A :class:~hyera.Hiera whose hierarchy names the backend's function returns the source's values. Applies to a backend that registers a plain function name.

Parameters:

Name	Type	Description	Default
tmp_path	Any	a fresh directory.	required

Raises:

Type	Description
AssertionError	a lookup differs from the source.

Source code in src/hyera/testing.py

def test_a_lookup_through_hiera_returns_the_same_values(
    self, tmp_path: _ty.Any
) -> None:
    """A :class:`~hyera.Hiera` whose hierarchy names the backend's function
    returns the source's values. Applies to a backend that registers a plain
    ``function`` name.

    :param tmp_path: a fresh directory.
    :raises AssertionError: a lookup differs from the source.
    """
    function_names = self.backend.NAMES.get("function", ())
    names = [name for name in function_names if isinstance(name, str)]
    if not names:
        return
    for kind in self._kinds():
        root = tmp_path / kind
        options = dict(self._write(root / "data", self.data))
        level: _ty.Dict[str, _ty.Any] = {"name": "contract", kind: names[0]}
        if "path" in options:
            level["path"] = options.pop("path")
        if options:
            level["options"] = options
        config = {
            "version": 5,
            "defaults": {"datadir": "data"},
            "hierarchy": [level],
        }
        h = Hiera(config, base_path=str(root), backends=default_backends())
        for key, value in self.data.items():
            assert h.lookup(key) == value, "{}: lookup of {!r} differs".format(
                kind, key
            )

test_a_malformed_source_raises_backend_error(tmp_path)

A source the backend cannot read raises :class:~hyera.BackendError and no other exception type.

Parameters:

Name	Type	Description	Default
tmp_path	Any	a fresh directory.	required

Raises:

Type	Description
AssertionError	another exception type, or none, was raised.

Source code in src/hyera/testing.py

def test_a_malformed_source_raises_backend_error(self, tmp_path: _ty.Any) -> None:
    """A source the backend cannot read raises :class:`~hyera.BackendError`
    and no other exception type.

    :param tmp_path: a fresh directory.
    :raises AssertionError: another exception type, or none, was raised.
    """
    for kind in self._kinds():
        options = self.write_malformed_source(tmp_path / kind)
        if options is None:
            continue
        options = dict(options)
        path = os.path.join(str(tmp_path / kind), options["path"])
        options["path"] = path
        self._expect_backend_error(kind, options, path)

test_a_missing_key_is_not_found(tmp_path)

A key the source does not hold is absent from a data_hash result, and a lookup_key or data_dig hook calls context.not_found() for it.

Parameters:

Name	Type	Description	Default
tmp_path	Any	a fresh directory.	required

Raises:

Type	Description
AssertionError	a missing key is found.

Source code in src/hyera/testing.py

def test_a_missing_key_is_not_found(self, tmp_path: _ty.Any) -> None:
    """A key the source does not hold is absent from a ``data_hash`` result,
    and a ``lookup_key`` or ``data_dig`` hook calls ``context.not_found()``
    for it.

    :param tmp_path: a fresh directory.
    :raises AssertionError: a missing key is found.
    """
    for kind in self._kinds():
        options, path = self._source(tmp_path / kind, self.data)
        backend = self.backend()
        if kind == "data_hash":
            found = _MISSING_KEY in _call_hook(backend, kind, "", options, path)
        else:
            got = _call_hook(backend, kind, _MISSING_KEY, options, path)
            found = got is not _Miss
        assert not found, "{}: a missing key was found".format(kind)

test_an_unreadable_source_raises_backend_error(tmp_path)

A source that has been removed raises :class:~hyera.BackendError and no other exception type.

Parameters:

Name	Type	Description	Default
tmp_path	Any	a fresh directory.	required

Raises:

Type	Description
AssertionError	another exception type, or none, was raised.

Source code in src/hyera/testing.py

def test_an_unreadable_source_raises_backend_error(self, tmp_path: _ty.Any) -> None:
    """A source that has been removed raises :class:`~hyera.BackendError` and
    no other exception type.

    :param tmp_path: a fresh directory.
    :raises AssertionError: another exception type, or none, was raised.
    """
    for kind in self._kinds():
        options, path = self._source(tmp_path / kind, self.data)
        if path is None:
            continue
        os.remove(path)
        self._expect_backend_error(kind, options, path)

test_check_available_returns_or_raises_backend_error()

check_available() returns, or raises :class:~hyera.BackendError.

Raises:

Type	Description
AssertionError	it raised another type.

Source code in src/hyera/testing.py

def test_check_available_returns_or_raises_backend_error(self) -> None:
    """``check_available()`` returns, or raises :class:`~hyera.BackendError`.

    :raises AssertionError: it raised another type.
    """
    try:
        self.backend.check_available()
    except BackendError:
        pass

test_every_name_resolves_to_the_class()

Every plain name in NAMES is lowercase and the registry resolves it to the class.

Raises:

Type	Description
AssertionError	a name is not lowercase or resolves elsewhere.

Source code in src/hyera/testing.py

def test_every_name_resolves_to_the_class(self) -> None:
    """Every plain name in ``NAMES`` is lowercase and the registry resolves it
    to the class.

    :raises AssertionError: a name is not lowercase or resolves elsewhere.
    """
    for kind, names in self.backend.NAMES.items():
        for name in names:
            if isinstance(name, str):
                assert name == name.lower(), "{!r} is not lowercase".format(name)
                assert (
                    Backend.find(name, kind) is self.backend
                ), "{!r} ({} kind) does not resolve to {}".format(
                    name, kind, self.backend.__name__
                )

test_every_returned_value_is_puppet_data(tmp_path)

Whatever a hook returns for the source is of a type Puppet data allows (a tuple counts as a list), the engine's own rule.

Parameters:

Name	Type	Description	Default
tmp_path	Any	a fresh directory.	required

Raises:

Type	Description
AssertionError	a value is outside Puppet's data types.

Source code in src/hyera/testing.py

def test_every_returned_value_is_puppet_data(self, tmp_path: _ty.Any) -> None:
    """Whatever a hook returns for the source is of a type Puppet data allows
    (a tuple counts as a list), the engine's own rule.

    :param tmp_path: a fresh directory.
    :raises AssertionError: a value is outside Puppet's data types.
    """
    accepts = _lookup_value_type().instance
    for kind in self._kinds():
        options, path = self._source(tmp_path / kind, self.data)
        got = self._values(kind, self.backend(), options, path)
        for key, value in got.items():
            assert value is _Miss or accepts(value), "{}: {!r} gave a {}".format(
                kind, key, type(value).__name__
            )

test_implements_agrees_with_what_the_class_overrides()

implements() answers from the methods the class overrides, and at least one hook kind is implemented.

Raises:

Type	Description
AssertionError	implements() disagrees with the overrides, or no hook kind is implemented.

Source code in src/hyera/testing.py

def test_implements_agrees_with_what_the_class_overrides(self) -> None:
    """``implements()`` answers from the methods the class overrides, and at
    least one hook kind is implemented.

    :raises AssertionError: ``implements()`` disagrees with the overrides, or
        no hook kind is implemented.
    """
    cls = self.backend
    expected = {
        "lookup_key": cls.lookup_key is not Backend.lookup_key,
        "data_dig": cls.data_dig is not Backend.data_dig,
        "data_hash": cls.data_hash is not Backend.data_hash
        or cls.loads is not Backend.loads,
    }
    for kind, overridden in expected.items():
        assert (
            cls.implements(kind) == overridden
        ), "implements({!r}) is {} but the class {} it".format(
            kind,
            cls.implements(kind),
            "overrides" if overridden else "does not override",
        )
    assert any(expected.values()), "no hook kind is implemented"

write_malformed_source(directory)

Write a source the backend cannot read.

The default writes bytes that are not a document in any text format over the file :meth:write_source made. A backend whose source is not a file the default can spoil overrides it.

Parameters:

Name	Type	Description	Default
directory	Any	the directory to write into.	required

Returns:

Type	Description
Optional[Mapping[str, Any]]	the level options for the source, or None when the source is not a file.

Source code in src/hyera/testing.py

def write_malformed_source(
    self, directory: _ty.Any
) -> _ty.Optional[_ty.Mapping[str, _ty.Any]]:
    """Write a source the backend cannot read.

    The default writes bytes that are not a document in any text format over
    the file :meth:`write_source` made. A backend whose source is not a file
    the default can spoil overrides it.

    :param directory: the directory to write into.
    :returns: the level options for the source, or ``None`` when the source is
        not a file.
    """
    options = dict(self._write(directory, self.data))
    if "path" not in options:
        return None
    with open(os.path.join(str(directory), options["path"]), "wb") as fh:
        fh.write(_MALFORMED)
    return options

write_source(directory, data)

Write a source holding data under directory.

Parameters:

Name	Type	Description	Default
directory	Any	the directory to write into; it exists.	required
data	Mapping[str, Any]	the key/value pairs the source must hold.	required

Returns:

Type	Description
Mapping[str, Any]	the level options for the source. A path is relative to directory; the checks make it absolute, as the engine does.

Raises:

Type	Description
NotImplementedError	the subclass did not implement it.

Source code in src/hyera/testing.py

def write_source(
    self, directory: _ty.Any, data: _ty.Mapping[str, _ty.Any]
) -> _ty.Mapping[str, _ty.Any]:
    """Write a source holding ``data`` under ``directory``.

    :param directory: the directory to write into; it exists.
    :param data: the key/value pairs the source must hold.
    :returns: the level options for the source. A ``path`` is relative to
        ``directory``; the checks make it absolute, as the engine does.
    :raises NotImplementedError: the subclass did not implement it.
    """
    raise NotImplementedError(
        "{} must implement write_source()".format(type(self).__name__)
    )

data_dig(backend, segments, options=None, *, context=None)

Call backend's data_dig hook as the engine does.

Parameters:

Name	Type	Description	Default
backend	Union[Backend, Type[Backend]]	a backend instance, or a class to instantiate with no arguments.	required
segments	Sequence[str]	the already-split key the hook is asked for.	required
options	Optional[Mapping[str, Any]]	the hierarchy entry's options; none when omitted.	None
context	Optional[LookupContext]	the context passed to the hook; :meth:LookupContext.for_testing() <hyera.LookupContext.for_testing> when omitted.	None

Returns:

Type	Description
Any	the hook's value, or :data:NOT_FOUND when it called context.not_found().

Raises:

Type	Description
ConfigError	the backend does not implement data_dig.
BackendError	the hook returned a value outside Puppet's data types.

Source code in src/hyera/testing.py

def data_dig(
    backend: _ty.Union[Backend, _ty.Type[Backend]],
    segments: _ty.Sequence[str],
    options: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    context: _ty.Optional[LookupContext] = None,
) -> _ty.Any:
    """Call ``backend``'s ``data_dig`` hook as the engine does.

    :param backend: a backend instance, or a class to instantiate with no
        arguments.
    :param segments: the already-split key the hook is asked for.
    :param options: the hierarchy entry's ``options``; none when omitted.
    :param context: the context passed to the hook;
        :meth:`LookupContext.for_testing() <hyera.LookupContext.for_testing>`
        when omitted.
    :returns: the hook's value, or :data:`NOT_FOUND` when it called
        ``context.not_found()``.
    :raises ConfigError: the backend does not implement ``data_dig``.
    :raises BackendError: the hook returned a value outside Puppet's data types.
    """
    backend = _instance(backend)
    _check_kind_implemented(backend, "data_dig")
    if context is None:
        context = LookupContext.for_testing()
    try:
        value = backend.data_dig(list(segments), dict(options or {}), context)
    except _NotFound:
        return NOT_FOUND
    except HieraError:
        raise
    except Exception as e:
        raise _hook_error(e, "data_dig", backend.name, None) from e
    return _validate_provider_value(value, "data_dig", backend.name, None)

data_hash(backend, path=None, options=None, *, context=None)

Call backend's data_hash hook as the engine does.

The hook must return a hash whose values are Puppet data.

Parameters:

Name	Type	Description	Default
backend	Union[Backend, Type[Backend]]	a backend instance, or a class to instantiate with no arguments.	required
path	Optional[str]	the location's file path, or None for a location-less entry.	None
options	Optional[Mapping[str, Any]]	the hierarchy entry's options; none when omitted. The hook receives them with path added, as the engine does for a located entry.	None
context	Optional[LookupContext]	the context passed to the hook; :meth:LookupContext.for_testing() <hyera.LookupContext.for_testing> when omitted.	None

Returns:

Type	Description
Dict[str, Any]	the hash, with tuples read as lists.

Raises:

Type	Description
ConfigError	the backend does not implement data_hash.
BackendError	the hook returned something other than a hash, or called context.not_found().
HieraLookupError	a value of the hash is outside Puppet's data types.

Source code in src/hyera/testing.py

def data_hash(
    backend: _ty.Union[Backend, _ty.Type[Backend]],
    path: _ty.Optional[str] = None,
    options: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    context: _ty.Optional[LookupContext] = None,
) -> _ty.Dict[str, _ty.Any]:
    """Call ``backend``'s ``data_hash`` hook as the engine does.

    The hook must return a hash whose values are Puppet data.

    :param backend: a backend instance, or a class to instantiate with no
        arguments.
    :param path: the location's file path, or ``None`` for a location-less
        entry.
    :param options: the hierarchy entry's ``options``; none when omitted. The
        hook receives them with ``path`` added, as the engine does for a located
        entry.
    :param context: the context passed to the hook;
        :meth:`LookupContext.for_testing() <hyera.LookupContext.for_testing>`
        when omitted.
    :returns: the hash, with tuples read as lists.
    :raises ConfigError: the backend does not implement ``data_hash``.
    :raises BackendError: the hook returned something other than a hash, or
        called ``context.not_found()``.
    :raises HieraLookupError: a value of the hash is outside Puppet's data
        types.
    """
    backend = _instance(backend)
    _check_kind_implemented(backend, "data_hash")
    if context is None:
        context = LookupContext.for_testing()
    label = None if path is None else str(path)
    merged = dict(options or {})
    if label is not None:
        merged["path"] = label
    try:
        data = backend.data_hash(path, merged, context)
    except _NotFound:
        raise _data_hash_not_found(backend.name, label) from None
    except HieraError:
        raise
    except Exception as e:
        raise _hook_error(e, "data_hash", backend.name, label) from e
    _validate_data_hash(data, backend.name, label)
    for key, value in data.items():
        validate_data_value(value, backend.name, label, key)
    return {key: _tuples_to_lists(value) for key, value in data.items()}

lookup_key(backend, key, options=None, *, context=None)

Call backend's lookup_key hook as the engine does.

The hook must be implemented and its value must be one Hiera accepts; the value comes back with tuples read as lists.

Parameters:

Name	Type	Description	Default
backend	Union[Backend, Type[Backend]]	a backend instance, or a class to instantiate with no arguments.	required
key	str	the key the hook is asked for.	required
options	Optional[Mapping[str, Any]]	the hierarchy entry's options; none when omitted.	None
context	Optional[LookupContext]	the context passed to the hook; :meth:LookupContext.for_testing() <hyera.LookupContext.for_testing> when omitted.	None

Returns:

Type	Description
Any	the hook's value, or :data:NOT_FOUND when it called context.not_found().

Raises:

Type	Description
ConfigError	the backend does not implement lookup_key.
BackendError	the hook returned a value outside Puppet's data types.

Source code in src/hyera/testing.py

def lookup_key(
    backend: _ty.Union[Backend, _ty.Type[Backend]],
    key: str,
    options: _ty.Optional[_ty.Mapping[str, _ty.Any]] = None,
    *,
    context: _ty.Optional[LookupContext] = None,
) -> _ty.Any:
    """Call ``backend``'s ``lookup_key`` hook as the engine does.

    The hook must be implemented and its value must be one Hiera accepts; the
    value comes back with tuples read as lists.

    :param backend: a backend instance, or a class to instantiate with no
        arguments.
    :param key: the key the hook is asked for.
    :param options: the hierarchy entry's ``options``; none when omitted.
    :param context: the context passed to the hook;
        :meth:`LookupContext.for_testing() <hyera.LookupContext.for_testing>`
        when omitted.
    :returns: the hook's value, or :data:`NOT_FOUND` when it called
        ``context.not_found()``.
    :raises ConfigError: the backend does not implement ``lookup_key``.
    :raises BackendError: the hook returned a value outside Puppet's data types.
    """
    backend = _instance(backend)
    _check_kind_implemented(backend, "lookup_key")
    if context is None:
        context = LookupContext.for_testing()
    try:
        value = backend.lookup_key(key, dict(options or {}), context)
    except _NotFound:
        return NOT_FOUND
    except HieraError:
        raise
    except Exception as e:
        raise _hook_error(e, "lookup_key", backend.name, None) from e
    return _validate_provider_value(value, "lookup_key", backend.name, None)