The generator class#
The generator is responsible for two things:
identifying which files it can interpret; and
turning those files into runnable test specs.
In the YAML plugin, this is implemented by YAMLTestGenerator.
File matching via file_patterns#
The generator advertises which filenames it recognizes using file_patterns:
class YAMLTestGenerator(canary.AbstractTestGenerator):
"""Define a YAML defined test case with the following schema:
.. code-block:: yaml
tests:
str:
description: str
script: list[str]
keywords: list[str]
parameters: dict[str, list[float | int | str | None]]
"""
file_patterns: ClassVar[tuple[str, ...]] = ("test_*.yaml",)
def lock(self, on_options: list[str] | None = None) -> list[canary.JobSpec]:
"""Take the cartesian product of parameters and from each combination create a test case."""
with open(self.file, "r") as fh:
fd = yaml.safe_load(fh)
fd = yaml_schema.validate(fd)
specs: list[canary.ResolvedSpec] = []
Only files matching these patterns are considered YAML test files by this generator.
Generating cases with lock()#
The core method is lock(). It reads the YAML, validates it, expands any
parameter combinations, and returns a list of ResolvedSpec objects (one per runnable
test case):
def lock(self, on_options: list[str] | None = None) -> list[canary.JobSpec]:
"""Take the cartesian product of parameters and from each combination create a test case."""
with open(self.file, "r") as fh:
fd = yaml.safe_load(fh)
fd = yaml_schema.validate(fd)
specs: list[canary.ResolvedSpec] = []
for name, details in fd["tests"].items():
kwds: dict[str, Any] = dict(
file_root=Path(self.root),
file_path=Path(self.path),
family=name,
keywords=details.get("keywords", []),
attributes={"description": details.get("description")},
)
script = details["script"]
sh = canary.filesystem.which("sh", required=True)
if parameters := details.get("parameters"):
keys = list(parameters.keys())
for values in product(*parameters.values()):
p = kwds["parameters"] = dict(zip(keys, values))
shell_cmds: list[str] = [Template(_).safe_substitute(**p) for _ in script]
kwds["command"] = [sh, "-c", "set -e\n" + "\n".join(shell_cmds)]
spec = canary.JobSpec(**kwds)
specs.append(spec)
else:
kwds["command"] = [sh, "-c", "set -e\n" + "\n".join(script)]
spec = canary.JobSpec(**kwds)
specs.append(spec)
return specs
Implementing describe()#
A generator can optionally implement describe() to provide human-readable
output for canary describe FILE. This is extremely useful when developing a new generator
because it exercises parsing and case generation without running anything.
def describe(self, on_options: list[str] | None = None) -> str:
cases = self.lock(on_options=on_options)
file = io.StringIO()
file.write(f"--- {self.name} ------------\n")
file.write(f"File: {self.file}\n")
file.write(f"{len(cases)} test cases:\n")
canary.graph.print(cases, file=file)
return file.getvalue()