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()