Skip to content

API Reference

This section contains the complete API reference for the baygon package. It is generated automatically from the source code using mkdocstrings and reflects the latest docstrings available in the project.

baygon

Baygon public API.

Classes

CaseModel dataclass

Leaf test case definition.

Source code in baygon/core/models.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@dataclass(frozen=True)
class CaseModel:
    """Leaf test case definition."""

    id: tuple[int, ...]
    name: str
    min_points: float | int
    points: float | int | None
    executable: str | None
    args: tuple[str, ...]
    env: Mapping[str, str]
    stdin: str | None
    stdout: tuple[ConditionModel, ...]
    stderr: tuple[ConditionModel, ...]
    repeat: int
    exit: int | str | None
    filters: Mapping[str, Any]
    eval: Mapping[str, Any] | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "env", _deep_freeze(self.env))
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

    @property
    def id_str(self) -> str:
        """Return the dotted identifier (e.g. '1.2.3')."""
        return ".".join(str(part) for part in self.id)
Attributes
id_str property 
id_str

Return the dotted identifier (e.g. '1.2.3').

ConditionModel dataclass

Matcher configuration applied to stdout/stderr.

Source code in baygon/core/models.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
@dataclass(frozen=True)
class ConditionModel:
    """Matcher configuration applied to stdout/stderr."""

    filters: Mapping[str, Any] = field(default_factory=dict)
    equals: str | None = None
    regex: str | None = None
    contains: str | None = None
    expected: str | None = None
    negated: tuple[NegatedConditionModel, ...] = field(default_factory=tuple)

    def __post_init__(self) -> None:
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

ExecutionResult dataclass

Lightweight execution summary for downstream reporting.

Source code in baygon/core/models.py 
142
143
144
145
146
147
148
149
150
@dataclass(frozen=True)
class ExecutionResult:
    """Lightweight execution summary for downstream reporting."""

    case: CaseModel
    status: str
    issues: tuple[Any, ...] = field(default_factory=tuple)
    duration_seconds: float | None = None
    telemetry: Mapping[str, Any] | None = None

GroupModel dataclass

Hierarchical test group definition.

Source code in baygon/core/models.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@dataclass(frozen=True)
class GroupModel:
    """Hierarchical test group definition."""

    id: tuple[int, ...]
    name: str
    min_points: float | int
    points: float | int | None
    executable: str | None
    filters: Mapping[str, Any]
    tests: tuple[TestNode, ...]
    eval: Mapping[str, Any] | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

    def iter_cases(self) -> Iterator[CaseModel]:
        """Iterate over every leaf case contained in this group."""
        for test in self.tests:
            if isinstance(test, CaseModel):
                yield test
            else:
                yield from test.iter_cases()
Functions
iter_cases 
iter_cases()

Iterate over every leaf case contained in this group.

Source code in baygon/core/models.py 
103
104
105
106
107
108
109
def iter_cases(self) -> Iterator[CaseModel]:
    """Iterate over every leaf case contained in this group."""
    for test in self.tests:
        if isinstance(test, CaseModel):
            yield test
        else:
            yield from test.iter_cases()

NegatedConditionModel dataclass

Single negated matcher condition.

Source code in baygon/core/models.py 
31
32
33
34
35
36
37
@dataclass(frozen=True)
class NegatedConditionModel:
    """Single negated matcher condition."""

    equals: str | None = None
    regex: str | None = None
    contains: str | None = None

SuiteModel dataclass

Top-level immutable suite description.

Source code in baygon/core/models.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@dataclass(frozen=True)
class SuiteModel:
    """Top-level immutable suite description."""

    name: str
    version: int
    min_points: float | int
    points: float | int | None
    executable: str | None
    filters: Mapping[str, Any]
    tests: tuple[TestNode, ...]
    eval: Mapping[str, Any] | None = None
    verbose: int | None = None
    report: str | None = None
    report_format: str | None = None
    table: bool = False
    compute_score: bool = False

    def __post_init__(self) -> None:
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

    def iter_cases(self) -> Iterator[CaseModel]:
        """Iterate over every test case in the suite."""
        for test in self.tests:
            if isinstance(test, CaseModel):
                yield test
            else:
                yield from test.iter_cases()
Functions
iter_cases 
iter_cases()

Iterate over every test case in the suite.

Source code in baygon/core/models.py 
133
134
135
136
137
138
139
def iter_cases(self) -> Iterator[CaseModel]:
    """Iterate over every test case in the suite."""
    for test in self.tests:
        if isinstance(test, CaseModel):
            yield test
        else:
            yield from test.iter_cases()

Executable

An executable program.

Convenient execution and access to program outputs such as:

- Exit status
- Standard output
- Standard error

For example:

>>> e = Executable("echo")
>>> e
Executable<echo>
>>> e("-n", "Hello World")
Outputs(exit_status=0, stdout='Hello World', stderr='')
>>> e("-n", "Hello World").stdout
'Hello World'
Source code in baygon/executable.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class Executable:
    """An executable program.

    Convenient execution and access to program outputs such as:

        - Exit status
        - Standard output
        - Standard error

    For example:

        >>> e = Executable("echo")
        >>> e
        Executable<echo>
        >>> e("-n", "Hello World")
        Outputs(exit_status=0, stdout='Hello World', stderr='')
        >>> e("-n", "Hello World").stdout
        'Hello World'
    """

    def __new__(cls, filename):
        if isinstance(filename, cls):
            return filename

        return super().__new__(cls) if filename else None

    def __init__(self, filename, encoding="utf-8"):
        """Create an executable object.

        :param filename: The path of the executable.
        :param encoding: The encoding to be used for the outputs, default is UTF-8.
        """
        if isinstance(filename, self.__class__):
            self.filename = filename.filename
            self.encoding = filename.encoding
        else:
            self.filename = filename
            self.encoding = encoding

        if not self._is_executable(self.filename):
            if "/" not in filename and shutil.which(filename) is not None:
                if filename in forbidden_binaries:
                    raise InvalidExecutableError(f"Program '{filename}' is forbidden!")
                filename = shutil.which(filename)
            else:
                raise InvalidExecutableError(
                    f"Program '{filename}' is not an executable!"
                )

    def run(self, *args, stdin=None, env=None, hook=None):
        """Run the program and grab all the outputs."""

        cmd = [self.filename, *[str(a) for a in args]]

        with subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stdin=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=env,
        ) as proc:
            if stdin is not None:
                stdin = stdin.encode(self.encoding)

            stdout, stderr = proc.communicate(input=stdin)

            if stdout is not None:
                stdout = stdout.decode(self.encoding)
            if stderr is not None:
                stderr = stderr.decode(self.encoding)

            if hook and callable(hook):
                hook(
                    cmd=cmd,
                    stdin=stdin,
                    stdout=stdout,
                    stderr=stderr,
                    exit_status=proc.returncode,
                )

            return Outputs(proc.returncode, stdout, stderr)

    def __call__(self, *args, **kwargs):
        return self.run(*args, **kwargs)

    def __repr__(self):
        return f"{self.__class__.__name__}<{self.filename}>"

    @staticmethod
    def _is_executable(filename):
        path = Path(filename)
        return path.is_file() and os.access(path, os.X_OK)
Functions
run 
run(*args, stdin=None, env=None, hook=None)

Run the program and grab all the outputs.

Source code in baygon/executable.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def run(self, *args, stdin=None, env=None, hook=None):
    """Run the program and grab all the outputs."""

    cmd = [self.filename, *[str(a) for a in args]]

    with subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stdin=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=env,
    ) as proc:
        if stdin is not None:
            stdin = stdin.encode(self.encoding)

        stdout, stderr = proc.communicate(input=stdin)

        if stdout is not None:
            stdout = stdout.decode(self.encoding)
        if stderr is not None:
            stderr = stderr.decode(self.encoding)

        if hook and callable(hook):
            hook(
                cmd=cmd,
                stdin=stdin,
                stdout=stdout,
                stderr=stderr,
                exit_status=proc.returncode,
            )

        return Outputs(proc.returncode, stdout, stderr)

BaygonRunner

Execute suites described by immutable models.

Source code in baygon/runtime/runner.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
class BaygonRunner:
    """Execute suites described by immutable models."""

    def __init__(
        self,
        suite: SuiteModel,
        *,
        base_dir: Path,
        executable: str | Path | None = None,
        executable_factory: Callable[[str], Executable] = Executable,
        clock: Callable[[], float] = time.perf_counter,
    ) -> None:
        self._suite = suite
        self._base_dir = base_dir
        self._clock = clock
        self._executable_factory = executable_factory
        self._executables: MutableMapping[str, Executable] = {}

        cli_executable = self._resolve_path(executable)
        suite_executable = self._resolve_path(suite.executable)
        if cli_executable and suite_executable:
            raise InvalidExecutableError("Executable can't be overridden")
        self._root_executable = cli_executable or suite_executable

    @property
    def suite(self) -> SuiteModel:
        """Return the suite model handled by the runner."""
        return self._suite

    def run(self, limit: int = -1) -> RunReport:
        """Run the test suite."""
        results: list[CaseResult] = []
        counters = defaultdict(int)
        points_total: float | int = 0
        points_earned: float | int = 0

        start = self._clock()
        root_context = _ExecutionContext(
            filters=_merge_filters(None, self._suite.filters),
            eval_filter=_resolve_eval(None, self._suite.eval),
            executable=self._root_executable,
        )

        stop_requested = False

        for case, context in self._iter_cases(root_context):
            if stop_requested:
                break

            case_points = case.points or 0
            points_total += case_points
            case_result = self._run_case(case, context)

            results.append(case_result)
            status = case_result.status
            counters[status] += 1
            if status == "passed":
                points_earned += case_result.points_earned or 0
            elif status == "failed" and limit > 0 and counters["failed"] > limit:
                stop_requested = True

        duration = round(self._clock() - start, 6)
        return RunReport(
            suite=self._suite,
            successes=counters["passed"],
            failures=counters["failed"],
            skipped=counters["skipped"],
            points_total=points_total,
            points_earned=points_earned,
            duration=duration,
            cases=tuple(results),
        )

    def _iter_cases(
        self, root: _ExecutionContext
    ) -> Iterator[tuple[CaseModel, _ExecutionContext]]:
        for test in self._suite.tests:
            yield from self._walk(test, root)

    def _walk(
        self,
        node: CaseModel | GroupModel,
        parent_context: _ExecutionContext,
    ) -> Iterator[tuple[CaseModel, _ExecutionContext]]:
        if isinstance(node, CaseModel):
            context = _ExecutionContext(
                filters=_merge_filters(parent_context.filters, node.filters),
                eval_filter=_resolve_eval(parent_context.eval_filter, node.eval),
                executable=_inherit_executable(
                    parent_context.executable, node.executable, self._base_dir
                ),
            )
            yield (node, context)
            return

        context = _ExecutionContext(
            filters=_merge_filters(parent_context.filters, node.filters),
            eval_filter=_resolve_eval(parent_context.eval_filter, node.eval),
            executable=_inherit_executable(
                parent_context.executable, node.executable, self._base_dir
            ),
        )
        for child in node.tests:
            yield from self._walk(child, context)

    def _run_case(self, case: CaseModel, context: _ExecutionContext) -> CaseResult:
        start = self._clock()
        issues: list[Any] = []
        command_logs: list[CommandLog] = []
        exec_path = context.executable
        if exec_path is None:
            raise InvalidExecutableError(
                f"Executable not provided for test '{case.name}' (id {case.id_str})."
            )

        exec_obj = self._get_executable(exec_path)
        filters = context.filters
        eval_filter = context.eval_filter

        for _ in range(case.repeat):
            filtered_args = tuple(_apply_eval(eval_filter, list(case.args)))
            filtered_stdin = _apply_eval(eval_filter, case.stdin)
            filtered_env = _apply_eval_env(eval_filter, case.env)
            expected_exit = (
                int(_apply_eval(eval_filter, str(case.exit)))
                if case.exit is not None
                else None
            )

            hook = _capture_hook(command_logs)
            output = exec_obj.run(
                *filtered_args,
                stdin=filtered_stdin if filtered_stdin is not None else None,
                env=get_env(filtered_env),
                hook=hook,
            )

            issues.extend(
                _match_streams(
                    case,
                    filters,
                    eval_filter,
                    output,
                    "stdout",
                    case.stdout,
                )
            )
            issues.extend(
                _match_streams(
                    case,
                    filters,
                    eval_filter,
                    output,
                    "stderr",
                    case.stderr,
                )
            )

            if expected_exit is not None and expected_exit != output.exit_status:
                issues.append(
                    InvalidExitStatus(
                        expected_exit,
                        output.exit_status,
                        on="exit",
                        test=case,
                    )
                )

        status = "failed" if issues else "passed"
        duration = round(self._clock() - start, 6)
        points = case.points or 0
        earned = points if status == "passed" else 0

        return CaseResult(
            case=case,
            status=status,
            issues=tuple(issues),
            commands=tuple(command_logs),
            duration=duration,
            points_earned=earned,
        )

    def _get_executable(self, path: str) -> Executable:
        if path not in self._executables:
            self._executables[path] = self._executable_factory(path)
        return self._executables[path]

    def _resolve_path(self, value: str | Path | None) -> str | None:
        if value is None:
            return None
        path = Path(value)
        if not path.is_absolute():
            path = (self._base_dir / path).resolve()
        return str(path)
Attributes
suite property 
suite

Return the suite model handled by the runner.

Functions
run 
run(limit=-1)

Run the test suite.

Source code in baygon/runtime/runner.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def run(self, limit: int = -1) -> RunReport:
    """Run the test suite."""
    results: list[CaseResult] = []
    counters = defaultdict(int)
    points_total: float | int = 0
    points_earned: float | int = 0

    start = self._clock()
    root_context = _ExecutionContext(
        filters=_merge_filters(None, self._suite.filters),
        eval_filter=_resolve_eval(None, self._suite.eval),
        executable=self._root_executable,
    )

    stop_requested = False

    for case, context in self._iter_cases(root_context):
        if stop_requested:
            break

        case_points = case.points or 0
        points_total += case_points
        case_result = self._run_case(case, context)

        results.append(case_result)
        status = case_result.status
        counters[status] += 1
        if status == "passed":
            points_earned += case_result.points_earned or 0
        elif status == "failed" and limit > 0 and counters["failed"] > limit:
            stop_requested = True

    duration = round(self._clock() - start, 6)
    return RunReport(
        suite=self._suite,
        successes=counters["passed"],
        failures=counters["failed"],
        skipped=counters["skipped"],
        points_total=points_total,
        points_earned=points_earned,
        duration=duration,
        cases=tuple(results),
    )

CaseResult dataclass

Individual case execution result.

Source code in baygon/runtime/runner.py 
30
31
32
33
34
35
36
37
38
39
@dataclass(frozen=True)
class CaseResult:
    """Individual case execution result."""

    case: CaseModel
    status: str
    issues: tuple[Any, ...]
    commands: tuple[CommandLog, ...]
    duration: float | None = None
    points_earned: float | int | None = None

CommandLog dataclass

Captured information about a single executed command.

Source code in baygon/runtime/runner.py 
19
20
21
22
23
24
25
26
27
@dataclass(frozen=True)
class CommandLog:
    """Captured information about a single executed command."""

    argv: tuple[str, ...]
    stdin: str | None
    stdout: str
    stderr: str
    exit_status: int

RunReport dataclass

Aggregated information after running a suite.

Source code in baygon/runtime/runner.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass(frozen=True)
class RunReport:
    """Aggregated information after running a suite."""

    suite: SuiteModel
    successes: int
    failures: int
    skipped: int
    points_total: float | int
    points_earned: float | int
    duration: float
    cases: tuple[CaseResult, ...]

    @property
    def total(self) -> int:
        return self.successes + self.failures + self.skipped

SuiteContext dataclass

Immutable bundle describing a suite ready to be executed.

Source code in baygon/suite.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@dataclass(frozen=True)
class SuiteContext:
    """Immutable bundle describing a suite ready to be executed."""

    config: Mapping[str, Any]
    model: SuiteModel
    base_dir: Path
    source_path: Path | None

    @property
    def name(self) -> str:
        return str(self.config.get("name", ""))

    @property
    def version(self) -> int | None:
        version = self.config.get("version")
        return int(version) if version is not None else None

    def create_runner(
        self,
        *,
        executable: str | Path | None = None,
        runner_factory: Callable[..., BaygonRunner] = BaygonRunner,
    ) -> BaygonRunner:
        """Return a runner configured for this suite."""
        return runner_factory(
            self.model,
            base_dir=self.base_dir,
            executable=executable,
        )
Functions
create_runner 
create_runner(
    *, executable=None, runner_factory=BaygonRunner
)

Return a runner configured for this suite.

Source code in baygon/suite.py 
53
54
55
56
57
58
59
60
61
62
63
64
def create_runner(
    self,
    *,
    executable: str | Path | None = None,
    runner_factory: Callable[..., BaygonRunner] = BaygonRunner,
) -> BaygonRunner:
    """Return a runner configured for this suite."""
    return runner_factory(
        self.model,
        base_dir=self.base_dir,
        executable=executable,
    )

SuiteExecutor

Execute suites described by SuiteContext.

Source code in baygon/suite.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class SuiteExecutor:
    """Execute suites described by `SuiteContext`."""

    def __init__(
        self,
        *,
        runner_factory: Callable[..., BaygonRunner] = BaygonRunner,
    ) -> None:
        self._runner_factory = runner_factory

    def run(
        self,
        context: SuiteContext,
        *,
        executable: str | Path | None = None,
        limit: int = -1,
    ) -> RunReport:
        """Run the suite described by the provided context."""
        runner = context.create_runner(
            executable=executable,
            runner_factory=self._runner_factory,
        )
        return runner.run(limit=limit)
Functions
run 
run(context, *, executable=None, limit=-1)

Run the suite described by the provided context.

Source code in baygon/suite.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
def run(
    self,
    context: SuiteContext,
    *,
    executable: str | Path | None = None,
    limit: int = -1,
) -> RunReport:
    """Run the suite described by the provided context."""
    runner = context.create_runner(
        executable=executable,
        runner_factory=self._runner_factory,
    )
    return runner.run(limit=limit)

SuiteLoader

Load suite configurations from files or raw mappings.

Source code in baygon/suite.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class SuiteLoader:
    """Load suite configurations from files or raw mappings."""

    def __init__(
        self,
        *,
        schema_loader: Callable[[Any], MutableMapping[str, Any]] = Schema,
        builder: SuiteBuilder | None = None,
    ) -> None:
        self._schema_loader = schema_loader
        self._builder = builder or SuiteBuilder()

    def from_mapping(
        self,
        data: Mapping[str, Any] | MutableMapping[str, Any],
        *,
        cwd: str | Path | None = None,
    ) -> SuiteContext:
        """Validate a raw mapping and build its execution context."""
        validated = self._schema_loader(data)
        compute_points(validated)

        base_dir = Path(cwd) if cwd is not None else Path.cwd()
        model = self._builder.build(validated)

        return SuiteContext(
            config=validated,
            model=model,
            base_dir=base_dir.resolve(),
            source_path=None,
        )

    def from_path(
        self,
        path: str | Path | None,
    ) -> SuiteContext:
        """Load configuration from disk and build its execution context."""
        config_path = discover_config(path)
        config = load_config_dict(config_path)
        model = load_config_model(config_path)
        base_dir = config_path.parent

        return SuiteContext(
            config=config,
            model=model,
            base_dir=base_dir.resolve(),
            source_path=config_path,
        )

    def load(
        self,
        *,
        data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
        path: str | Path | None = None,
        cwd: str | Path | None = None,
    ) -> SuiteContext:
        """Load a suite configuration from a mapping or a path."""
        if data is not None and path is not None:
            raise ValueError("Provide either 'data' or 'path', not both.")

        if data is not None:
            return self.from_mapping(data, cwd=cwd)

        return self.from_path(path)
Functions
from_mapping 
from_mapping(data, *, cwd=None)

Validate a raw mapping and build its execution context.

Source code in baygon/suite.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def from_mapping(
    self,
    data: Mapping[str, Any] | MutableMapping[str, Any],
    *,
    cwd: str | Path | None = None,
) -> SuiteContext:
    """Validate a raw mapping and build its execution context."""
    validated = self._schema_loader(data)
    compute_points(validated)

    base_dir = Path(cwd) if cwd is not None else Path.cwd()
    model = self._builder.build(validated)

    return SuiteContext(
        config=validated,
        model=model,
        base_dir=base_dir.resolve(),
        source_path=None,
    )
from_path 
from_path(path)

Load configuration from disk and build its execution context.

Source code in baygon/suite.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def from_path(
    self,
    path: str | Path | None,
) -> SuiteContext:
    """Load configuration from disk and build its execution context."""
    config_path = discover_config(path)
    config = load_config_dict(config_path)
    model = load_config_model(config_path)
    base_dir = config_path.parent

    return SuiteContext(
        config=config,
        model=model,
        base_dir=base_dir.resolve(),
        source_path=config_path,
    )
load 
load(*, data=None, path=None, cwd=None)

Load a suite configuration from a mapping or a path.

Source code in baygon/suite.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def load(
    self,
    *,
    data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
    path: str | Path | None = None,
    cwd: str | Path | None = None,
) -> SuiteContext:
    """Load a suite configuration from a mapping or a path."""
    if data is not None and path is not None:
        raise ValueError("Provide either 'data' or 'path', not both.")

    if data is not None:
        return self.from_mapping(data, cwd=cwd)

    return self.from_path(path)

SuiteService

Facade coordinating loading and execution of Baygon suites.

Source code in baygon/suite.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class SuiteService:
    """Facade coordinating loading and execution of Baygon suites."""

    def __init__(
        self,
        loader: SuiteLoader | None = None,
        executor: SuiteExecutor | None = None,
    ) -> None:
        self._loader = loader or SuiteLoader()
        self._executor = executor or SuiteExecutor()

    def load(
        self,
        *,
        data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
        path: str | Path | None = None,
        cwd: str | Path | None = None,
    ) -> SuiteContext:
        """Return a validated suite context without running it."""
        return self._loader.load(data=data, path=path, cwd=cwd)

    def run(
        self,
        *,
        data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
        path: str | Path | None = None,
        cwd: str | Path | None = None,
        executable: str | Path | None = None,
        limit: int = -1,
    ) -> RunReport:
        """Load and execute a suite in one call."""
        context = self._loader.load(data=data, path=path, cwd=cwd)
        return self._executor.run(
            context,
            executable=executable,
            limit=limit,
        )
Functions
load 
load(*, data=None, path=None, cwd=None)

Return a validated suite context without running it.

Source code in baygon/suite.py 
184
185
186
187
188
189
190
191
192
def load(
    self,
    *,
    data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
    path: str | Path | None = None,
    cwd: str | Path | None = None,
) -> SuiteContext:
    """Return a validated suite context without running it."""
    return self._loader.load(data=data, path=path, cwd=cwd)
run 
run(
    *,
    data=None,
    path=None,
    cwd=None,
    executable=None,
    limit=-1
)

Load and execute a suite in one call.

Source code in baygon/suite.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def run(
    self,
    *,
    data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
    path: str | Path | None = None,
    cwd: str | Path | None = None,
    executable: str | Path | None = None,
    limit: int = -1,
) -> RunReport:
    """Load and execute a suite in one call."""
    context = self._loader.load(data=data, path=path, cwd=cwd)
    return self._executor.run(
        context,
        executable=executable,
        limit=limit,
    )

Functions

discover_config 

discover_config(path)

Locate a configuration file starting from the given path or CWD.

Source code in baygon/config/loader.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def discover_config(path: str | Path | None) -> Path:
    """Locate a configuration file starting from the given path or CWD."""
    start = _resolve_start(path)

    if start.is_file():
        if start.suffix.lower() not in _SUPPORTED_EXTENSIONS:
            raise ConfigError(f"Unknown file extension '{start.suffix}' for '{start}'.")
        return start

    current = start
    while True:
        found = _scan_directory(current)
        if found is not None:
            return found
        if current.parent == current:
            break
        current = current.parent

    raise ConfigError("Couldn't find configuration file.")

load_config 

load_config(path)

Load a configuration file and build a SuiteModel.

Source code in baygon/config/loader.py 
39
40
41
42
43
def load_config(path: str | Path | None) -> SuiteModel:
    """Load a configuration file and build a SuiteModel."""
    config_path = discover_config(path)
    config_dict = load_config_dict(config_path)
    return build_suite_model(config_dict)

load_config_dict 

load_config_dict(path)

Load a configuration file and return the validated dictionary form.

Source code in baygon/config/loader.py 
46
47
48
49
50
51
def load_config_dict(path: str | Path | None) -> dict[str, Any]:
    """Load a configuration file and return the validated dictionary form."""
    config_path = discover_config(path)
    data = _read_config_mapping(config_path)
    compute_points(data)
    return data

build_suite_model 

build_suite_model(config)

Build an immutable SuiteModel from validated schema data.

Parameters:

Name Type Description Default
config Mapping[str, Any]

Mapping returned by baygon.schema.Schema.

 required

Returns:

Name Type Description
SuiteModel SuiteModel

frozen domain representation suitable as a SSOT.
Source code in baygon/core/models.py 
153
154
155
156
157
158
159
160
161
162
163
164
def build_suite_model(config: Mapping[str, Any]) -> SuiteModel:
    """Build an immutable `SuiteModel` from validated schema data.

    Args:
        config: Mapping returned by `baygon.schema.Schema`.

    Returns:
        SuiteModel: frozen domain representation suitable as a SSOT.
    """
    prepared = deepcopy(config)
    compute_points(prepared)
    return _build_suite(prepared)

Schema 

Schema(data, humanize=False)

Validate the given data against the Baygon schema.

Source code in baygon/schema.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def Schema(data: Any, humanize: bool = False):  # noqa: N802
    """Validate the given data against the Baygon schema."""

    if isinstance(data, str):
        data = _load_yaml(data)
    elif hasattr(data, "read"):
        data = _load_yaml(data.read())

    if not isinstance(data, Mapping):
        raise ConfigError("Schema expects a mapping or YAML string")

    include_eval = "eval" in data
    raw_eval_value = data.get("eval") if include_eval else None

    try:
        config = BaygonConfig.model_validate(data)
    except ValidationError as exc:  # pragma: no cover - exercised indirectly
        if humanize:
            raise ConfigError(_humanize_errors(exc)) from exc
        raise

    _assign_test_ids(config.tests)
    return _dump_config(
        config, include_eval=include_eval, raw_eval_value=raw_eval_value
    )

Modules

config

Configuration loading utilities.

Functions
discover_config 
discover_config(path)

Locate a configuration file starting from the given path or CWD.

Source code in baygon/config/loader.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def discover_config(path: str | Path | None) -> Path:
    """Locate a configuration file starting from the given path or CWD."""
    start = _resolve_start(path)

    if start.is_file():
        if start.suffix.lower() not in _SUPPORTED_EXTENSIONS:
            raise ConfigError(f"Unknown file extension '{start.suffix}' for '{start}'.")
        return start

    current = start
    while True:
        found = _scan_directory(current)
        if found is not None:
            return found
        if current.parent == current:
            break
        current = current.parent

    raise ConfigError("Couldn't find configuration file.")
load_config 
load_config(path)

Load a configuration file and build a SuiteModel.

Source code in baygon/config/loader.py 
39
40
41
42
43
def load_config(path: str | Path | None) -> SuiteModel:
    """Load a configuration file and build a SuiteModel."""
    config_path = discover_config(path)
    config_dict = load_config_dict(config_path)
    return build_suite_model(config_dict)
load_config_dict 
load_config_dict(path)

Load a configuration file and return the validated dictionary form.

Source code in baygon/config/loader.py 
46
47
48
49
50
51
def load_config_dict(path: str | Path | None) -> dict[str, Any]:
    """Load a configuration file and return the validated dictionary form."""
    config_path = discover_config(path)
    data = _read_config_mapping(config_path)
    compute_points(data)
    return data
Modules
loader

Load Baygon configuration files and return immutable suite models.

Classes Functions
discover_config 
discover_config(path)

Locate a configuration file starting from the given path or CWD.

Source code in baygon/config/loader.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def discover_config(path: str | Path | None) -> Path:
    """Locate a configuration file starting from the given path or CWD."""
    start = _resolve_start(path)

    if start.is_file():
        if start.suffix.lower() not in _SUPPORTED_EXTENSIONS:
            raise ConfigError(f"Unknown file extension '{start.suffix}' for '{start}'.")
        return start

    current = start
    while True:
        found = _scan_directory(current)
        if found is not None:
            return found
        if current.parent == current:
            break
        current = current.parent

    raise ConfigError("Couldn't find configuration file.")
load_config 
load_config(path)

Load a configuration file and build a SuiteModel.

Source code in baygon/config/loader.py 
39
40
41
42
43
def load_config(path: str | Path | None) -> SuiteModel:
    """Load a configuration file and build a SuiteModel."""
    config_path = discover_config(path)
    config_dict = load_config_dict(config_path)
    return build_suite_model(config_dict)
load_config_dict 
load_config_dict(path)

Load a configuration file and return the validated dictionary form.

Source code in baygon/config/loader.py 
46
47
48
49
50
51
def load_config_dict(path: str | Path | None) -> dict[str, Any]:
    """Load a configuration file and return the validated dictionary form."""
    config_path = discover_config(path)
    data = _read_config_mapping(config_path)
    compute_points(data)
    return data

core

Domain-layer models exposed by Baygon.

Classes
CaseModel dataclass

Leaf test case definition.

Source code in baygon/core/models.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@dataclass(frozen=True)
class CaseModel:
    """Leaf test case definition."""

    id: tuple[int, ...]
    name: str
    min_points: float | int
    points: float | int | None
    executable: str | None
    args: tuple[str, ...]
    env: Mapping[str, str]
    stdin: str | None
    stdout: tuple[ConditionModel, ...]
    stderr: tuple[ConditionModel, ...]
    repeat: int
    exit: int | str | None
    filters: Mapping[str, Any]
    eval: Mapping[str, Any] | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "env", _deep_freeze(self.env))
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

    @property
    def id_str(self) -> str:
        """Return the dotted identifier (e.g. '1.2.3')."""
        return ".".join(str(part) for part in self.id)
Attributes
id_str property 
id_str

Return the dotted identifier (e.g. '1.2.3').

ConditionModel dataclass

Matcher configuration applied to stdout/stderr.

Source code in baygon/core/models.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
@dataclass(frozen=True)
class ConditionModel:
    """Matcher configuration applied to stdout/stderr."""

    filters: Mapping[str, Any] = field(default_factory=dict)
    equals: str | None = None
    regex: str | None = None
    contains: str | None = None
    expected: str | None = None
    negated: tuple[NegatedConditionModel, ...] = field(default_factory=tuple)

    def __post_init__(self) -> None:
        object.__setattr__(self, "filters", _deep_freeze(self.filters))
ExecutionResult dataclass

Lightweight execution summary for downstream reporting.

Source code in baygon/core/models.py 
142
143
144
145
146
147
148
149
150
@dataclass(frozen=True)
class ExecutionResult:
    """Lightweight execution summary for downstream reporting."""

    case: CaseModel
    status: str
    issues: tuple[Any, ...] = field(default_factory=tuple)
    duration_seconds: float | None = None
    telemetry: Mapping[str, Any] | None = None
GroupModel dataclass

Hierarchical test group definition.

Source code in baygon/core/models.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@dataclass(frozen=True)
class GroupModel:
    """Hierarchical test group definition."""

    id: tuple[int, ...]
    name: str
    min_points: float | int
    points: float | int | None
    executable: str | None
    filters: Mapping[str, Any]
    tests: tuple[TestNode, ...]
    eval: Mapping[str, Any] | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

    def iter_cases(self) -> Iterator[CaseModel]:
        """Iterate over every leaf case contained in this group."""
        for test in self.tests:
            if isinstance(test, CaseModel):
                yield test
            else:
                yield from test.iter_cases()
Functions
iter_cases 
iter_cases()

Iterate over every leaf case contained in this group.

Source code in baygon/core/models.py 
103
104
105
106
107
108
109
def iter_cases(self) -> Iterator[CaseModel]:
    """Iterate over every leaf case contained in this group."""
    for test in self.tests:
        if isinstance(test, CaseModel):
            yield test
        else:
            yield from test.iter_cases()
NegatedConditionModel dataclass

Single negated matcher condition.

Source code in baygon/core/models.py 
31
32
33
34
35
36
37
@dataclass(frozen=True)
class NegatedConditionModel:
    """Single negated matcher condition."""

    equals: str | None = None
    regex: str | None = None
    contains: str | None = None
SuiteModel dataclass

Top-level immutable suite description.

Source code in baygon/core/models.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@dataclass(frozen=True)
class SuiteModel:
    """Top-level immutable suite description."""

    name: str
    version: int
    min_points: float | int
    points: float | int | None
    executable: str | None
    filters: Mapping[str, Any]
    tests: tuple[TestNode, ...]
    eval: Mapping[str, Any] | None = None
    verbose: int | None = None
    report: str | None = None
    report_format: str | None = None
    table: bool = False
    compute_score: bool = False

    def __post_init__(self) -> None:
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

    def iter_cases(self) -> Iterator[CaseModel]:
        """Iterate over every test case in the suite."""
        for test in self.tests:
            if isinstance(test, CaseModel):
                yield test
            else:
                yield from test.iter_cases()
Functions
iter_cases 
iter_cases()

Iterate over every test case in the suite.

Source code in baygon/core/models.py 
133
134
135
136
137
138
139
def iter_cases(self) -> Iterator[CaseModel]:
    """Iterate over every test case in the suite."""
    for test in self.tests:
        if isinstance(test, CaseModel):
            yield test
        else:
            yield from test.iter_cases()
Functions
build_suite_model 
build_suite_model(config)

Build an immutable SuiteModel from validated schema data.

Parameters:

Name Type Description Default
config Mapping[str, Any]

Mapping returned by baygon.schema.Schema.

 required

Returns:

Name Type Description
SuiteModel SuiteModel

frozen domain representation suitable as a SSOT.
Source code in baygon/core/models.py 
153
154
155
156
157
158
159
160
161
162
163
164
def build_suite_model(config: Mapping[str, Any]) -> SuiteModel:
    """Build an immutable `SuiteModel` from validated schema data.

    Args:
        config: Mapping returned by `baygon.schema.Schema`.

    Returns:
        SuiteModel: frozen domain representation suitable as a SSOT.
    """
    prepared = deepcopy(config)
    compute_points(prepared)
    return _build_suite(prepared)
Modules
models

Immutable domain models for Baygon configurations and execution state.

Classes
NegatedConditionModel dataclass

Single negated matcher condition.

Source code in baygon/core/models.py 
31
32
33
34
35
36
37
@dataclass(frozen=True)
class NegatedConditionModel:
    """Single negated matcher condition."""

    equals: str | None = None
    regex: str | None = None
    contains: str | None = None
ConditionModel dataclass

Matcher configuration applied to stdout/stderr.

Source code in baygon/core/models.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
@dataclass(frozen=True)
class ConditionModel:
    """Matcher configuration applied to stdout/stderr."""

    filters: Mapping[str, Any] = field(default_factory=dict)
    equals: str | None = None
    regex: str | None = None
    contains: str | None = None
    expected: str | None = None
    negated: tuple[NegatedConditionModel, ...] = field(default_factory=tuple)

    def __post_init__(self) -> None:
        object.__setattr__(self, "filters", _deep_freeze(self.filters))
CaseModel dataclass

Leaf test case definition.

Source code in baygon/core/models.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@dataclass(frozen=True)
class CaseModel:
    """Leaf test case definition."""

    id: tuple[int, ...]
    name: str
    min_points: float | int
    points: float | int | None
    executable: str | None
    args: tuple[str, ...]
    env: Mapping[str, str]
    stdin: str | None
    stdout: tuple[ConditionModel, ...]
    stderr: tuple[ConditionModel, ...]
    repeat: int
    exit: int | str | None
    filters: Mapping[str, Any]
    eval: Mapping[str, Any] | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "env", _deep_freeze(self.env))
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

    @property
    def id_str(self) -> str:
        """Return the dotted identifier (e.g. '1.2.3')."""
        return ".".join(str(part) for part in self.id)
Attributes
id_str property 
id_str

Return the dotted identifier (e.g. '1.2.3').

GroupModel dataclass

Hierarchical test group definition.

Source code in baygon/core/models.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@dataclass(frozen=True)
class GroupModel:
    """Hierarchical test group definition."""

    id: tuple[int, ...]
    name: str
    min_points: float | int
    points: float | int | None
    executable: str | None
    filters: Mapping[str, Any]
    tests: tuple[TestNode, ...]
    eval: Mapping[str, Any] | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

    def iter_cases(self) -> Iterator[CaseModel]:
        """Iterate over every leaf case contained in this group."""
        for test in self.tests:
            if isinstance(test, CaseModel):
                yield test
            else:
                yield from test.iter_cases()
Functions
iter_cases 
iter_cases()

Iterate over every leaf case contained in this group.

Source code in baygon/core/models.py 
103
104
105
106
107
108
109
def iter_cases(self) -> Iterator[CaseModel]:
    """Iterate over every leaf case contained in this group."""
    for test in self.tests:
        if isinstance(test, CaseModel):
            yield test
        else:
            yield from test.iter_cases()
SuiteModel dataclass

Top-level immutable suite description.

Source code in baygon/core/models.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@dataclass(frozen=True)
class SuiteModel:
    """Top-level immutable suite description."""

    name: str
    version: int
    min_points: float | int
    points: float | int | None
    executable: str | None
    filters: Mapping[str, Any]
    tests: tuple[TestNode, ...]
    eval: Mapping[str, Any] | None = None
    verbose: int | None = None
    report: str | None = None
    report_format: str | None = None
    table: bool = False
    compute_score: bool = False

    def __post_init__(self) -> None:
        object.__setattr__(self, "filters", _deep_freeze(self.filters))

    def iter_cases(self) -> Iterator[CaseModel]:
        """Iterate over every test case in the suite."""
        for test in self.tests:
            if isinstance(test, CaseModel):
                yield test
            else:
                yield from test.iter_cases()
Functions
iter_cases 
iter_cases()

Iterate over every test case in the suite.

Source code in baygon/core/models.py 
133
134
135
136
137
138
139
def iter_cases(self) -> Iterator[CaseModel]:
    """Iterate over every test case in the suite."""
    for test in self.tests:
        if isinstance(test, CaseModel):
            yield test
        else:
            yield from test.iter_cases()
ExecutionResult dataclass

Lightweight execution summary for downstream reporting.

Source code in baygon/core/models.py 
142
143
144
145
146
147
148
149
150
@dataclass(frozen=True)
class ExecutionResult:
    """Lightweight execution summary for downstream reporting."""

    case: CaseModel
    status: str
    issues: tuple[Any, ...] = field(default_factory=tuple)
    duration_seconds: float | None = None
    telemetry: Mapping[str, Any] | None = None
Functions
build_suite_model 
build_suite_model(config)

Build an immutable SuiteModel from validated schema data.

Parameters:

Name Type Description Default
config Mapping[str, Any]

Mapping returned by baygon.schema.Schema.

 required

Returns:

Name Type Description
SuiteModel SuiteModel

frozen domain representation suitable as a SSOT.
Source code in baygon/core/models.py 
153
154
155
156
157
158
159
160
161
162
163
164
def build_suite_model(config: Mapping[str, Any]) -> SuiteModel:
    """Build an immutable `SuiteModel` from validated schema data.

    Args:
        config: Mapping returned by `baygon.schema.Schema`.

    Returns:
        SuiteModel: frozen domain representation suitable as a SSOT.
    """
    prepared = deepcopy(config)
    compute_points(prepared)
    return _build_suite(prepared)

error

Errors for Baygon

Classes
BaygonError

Bases: Exception

Base class for Baygon errors

Source code in baygon/error.py 
6
7
class BaygonError(Exception):
    """Base class for Baygon errors"""
ConfigError

Bases: BaygonError

Raised when a config value is not valid

Source code in baygon/error.py 
10
11
class ConfigError(BaygonError):
    """Raised when a config value is not valid"""
InvalidExecutableError

Bases: BaygonError

Raised when an executable is not found

Source code in baygon/error.py 
14
15
class InvalidExecutableError(BaygonError):
    """Raised when an executable is not found"""
InvalidFilterError

Bases: BaygonError

Raised when a filter is not found

Source code in baygon/error.py 
18
19
class InvalidFilterError(BaygonError):
    """Raised when a filter is not found"""
ConfigSyntaxError

Bases: ConfigError

Raised when the configuration file cannot be parsed.

Source code in baygon/error.py 
22
23
24
25
26
27
28
29
30
31
32
class ConfigSyntaxError(ConfigError):
    """Raised when the configuration file cannot be parsed."""

    def __init__(
        self, message: str, *, line: int | None = None, column: int | None = None
    ):
        if line is not None and column is not None:
            message = f"{message} (line {line}, column {column})"
        super().__init__(message)
        self.line = line
        self.column = column

eval

Helper functions used in eval filter. These functions are injected into the kernel and made available from the mustaches templates.

Examples:

>>> reset()
>>> iter()
0
>>> iter()
1
>>> iter()
2
>>> iter(100, 10)
100
>>> iter(100, 10)
110
>>> iter(ctx="foo")
0
>>> iter(ctx="foo")
1
>>> reset()
>>> iter()
0
>>> iter(100, 10)
100
Functions
reset 
reset()

Reset the context.

Source code in baygon/eval.py 
31
32
33
def reset():
    """Reset the context."""
    _context.clear()
iter 
iter(start=0, step=1, ctx=None)

Custom iterator for eval input filter.

Source code in baygon/eval.py 
36
37
38
39
40
def iter(start=0, step=1, ctx=None):  # noqa: A001 - intentionally shadow built-in
    """Custom iterator for eval input filter."""
    ctx = (start, step, ctx)
    _context[ctx] = _context.get(ctx, start - step) + step
    return _context[ctx]

executable

Executable class. To be used with the Test class.

Classes
Executable

An executable program.

Convenient execution and access to program outputs such as:

- Exit status
- Standard output
- Standard error

For example:

>>> e = Executable("echo")
>>> e
Executable<echo>
>>> e("-n", "Hello World")
Outputs(exit_status=0, stdout='Hello World', stderr='')
>>> e("-n", "Hello World").stdout
'Hello World'
Source code in baygon/executable.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class Executable:
    """An executable program.

    Convenient execution and access to program outputs such as:

        - Exit status
        - Standard output
        - Standard error

    For example:

        >>> e = Executable("echo")
        >>> e
        Executable<echo>
        >>> e("-n", "Hello World")
        Outputs(exit_status=0, stdout='Hello World', stderr='')
        >>> e("-n", "Hello World").stdout
        'Hello World'
    """

    def __new__(cls, filename):
        if isinstance(filename, cls):
            return filename

        return super().__new__(cls) if filename else None

    def __init__(self, filename, encoding="utf-8"):
        """Create an executable object.

        :param filename: The path of the executable.
        :param encoding: The encoding to be used for the outputs, default is UTF-8.
        """
        if isinstance(filename, self.__class__):
            self.filename = filename.filename
            self.encoding = filename.encoding
        else:
            self.filename = filename
            self.encoding = encoding

        if not self._is_executable(self.filename):
            if "/" not in filename and shutil.which(filename) is not None:
                if filename in forbidden_binaries:
                    raise InvalidExecutableError(f"Program '{filename}' is forbidden!")
                filename = shutil.which(filename)
            else:
                raise InvalidExecutableError(
                    f"Program '{filename}' is not an executable!"
                )

    def run(self, *args, stdin=None, env=None, hook=None):
        """Run the program and grab all the outputs."""

        cmd = [self.filename, *[str(a) for a in args]]

        with subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stdin=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=env,
        ) as proc:
            if stdin is not None:
                stdin = stdin.encode(self.encoding)

            stdout, stderr = proc.communicate(input=stdin)

            if stdout is not None:
                stdout = stdout.decode(self.encoding)
            if stderr is not None:
                stderr = stderr.decode(self.encoding)

            if hook and callable(hook):
                hook(
                    cmd=cmd,
                    stdin=stdin,
                    stdout=stdout,
                    stderr=stderr,
                    exit_status=proc.returncode,
                )

            return Outputs(proc.returncode, stdout, stderr)

    def __call__(self, *args, **kwargs):
        return self.run(*args, **kwargs)

    def __repr__(self):
        return f"{self.__class__.__name__}<{self.filename}>"

    @staticmethod
    def _is_executable(filename):
        path = Path(filename)
        return path.is_file() and os.access(path, os.X_OK)
Functions
run 
run(*args, stdin=None, env=None, hook=None)

Run the program and grab all the outputs.

Source code in baygon/executable.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def run(self, *args, stdin=None, env=None, hook=None):
    """Run the program and grab all the outputs."""

    cmd = [self.filename, *[str(a) for a in args]]

    with subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stdin=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=env,
    ) as proc:
        if stdin is not None:
            stdin = stdin.encode(self.encoding)

        stdout, stderr = proc.communicate(input=stdin)

        if stdout is not None:
            stdout = stdout.decode(self.encoding)
        if stderr is not None:
            stderr = stderr.decode(self.encoding)

        if hook and callable(hook):
            hook(
                cmd=cmd,
                stdin=stdin,
                stdout=stdout,
                stderr=stderr,
                exit_status=proc.returncode,
            )

        return Outputs(proc.returncode, stdout, stderr)
Functions
get_env 
get_env(env=None)

Get the environment variables to be used for the subprocess.

Source code in baygon/executable.py 
20
21
22
def get_env(env: typing.Optional[str] = None) -> dict:
    """Get the environment variables to be used for the subprocess."""
    return {**os.environ, **(env or {})}

filters

String filters for Baygon. Each filter is a class that implements a filter method. A filter is used to modify stdout and stderr before they are tested.

Classes
Filter

Bases: ABC

Base class for filters.

Source code in baygon/filters.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class Filter(ABC):
    """Base class for filters."""

    @abstractmethod
    def apply(self, value: str) -> str:
        """Apply the filter to a value."""
        return value

    def filter(self, value: str) -> str:
        """Apply the filter to a value."""
        return self.apply(value)

    def __init__(self, *args, **kwargs):
        """Initialize the filter.

        Args:
            *args: Positional arguments forwarded by subclasses.
            **kwargs: Keyword arguments, notably `input` to mark filters for stdin.
        """
        self.input = kwargs.get("input", False)

    def __repr__(self):
        return f"{self.__class__.__name__}"

    def __call__(self, value: str) -> str:
        return self.filter(value)

    @classmethod
    def name(cls):
        """Return the name of the filter."""
        return cls.__name__.split("Filter", maxsplit=1)[1].lower()
Functions
apply abstractmethod 
apply(value)

Apply the filter to a value.

Source code in baygon/filters.py 
23
24
25
26
@abstractmethod
def apply(self, value: str) -> str:
    """Apply the filter to a value."""
    return value
filter 
filter(value)

Apply the filter to a value.

Source code in baygon/filters.py 
28
29
30
def filter(self, value: str) -> str:
    """Apply the filter to a value."""
    return self.apply(value)
name classmethod 
name()

Return the name of the filter.

Source code in baygon/filters.py 
47
48
49
50
@classmethod
def name(cls):
    """Return the name of the filter."""
    return cls.__name__.split("Filter", maxsplit=1)[1].lower()
FilterNone

Bases: Filter

Filter that does nothing.

Source code in baygon/filters.py 
90
91
92
93
94
95
@register_filter
class FilterNone(Filter):
    """Filter that does nothing."""

    def apply(self, value: str) -> str:
        return value
FilterUppercase

Bases: Filter

Filter for uppercase strings.

f = FilterUppercase() f("hello") 'HELLO'

Source code in baygon/filters.py 
 98
 99
100
101
102
103
104
105
106
107
@register_filter
class FilterUppercase(Filter):
    """Filter for uppercase strings.
    >>> f = FilterUppercase()
    >>> f("hello")
    'HELLO'
    """

    def apply(self, value: str) -> str:
        return value.upper()
FilterLowercase

Bases: Filter

Filter for lowercase strings.

f = FilterLowercase() f("HELLO") 'hello'

Source code in baygon/filters.py 
110
111
112
113
114
115
116
117
118
119
@register_filter
class FilterLowercase(Filter):
    """Filter for lowercase strings.
    >>> f = FilterLowercase()
    >>> f("HELLO")
    'hello'
    """

    def apply(self, value: str) -> str:
        return value.lower()
FilterTrim

Bases: Filter

Filter for trimmed strings.

f = FilterTrim() f(" hello ") 'hello'

Source code in baygon/filters.py 
122
123
124
125
126
127
128
129
130
131
132
133
@register_filter
class FilterTrim(Filter):
    """Filter for trimmed strings.
    >>> f = FilterTrim()
    >>> f(" hello   ")
    'hello'
    """

    __output__ = True

    def apply(self, value: str) -> str:
        return value.strip()
FilterIgnoreSpaces

Bases: Filter

Filter for strings with no spaces.

f = FilterIgnoreSpaces() f("hello world") 'helloworld'

Source code in baygon/filters.py 
136
137
138
139
140
141
142
143
144
145
@register_filter
class FilterIgnoreSpaces(Filter):
    """Filter for strings with no spaces.
    >>> f = FilterIgnoreSpaces()
    >>> f("hello   world")
    'helloworld'
    """

    def apply(self, value: str) -> str:
        return value.replace(" ", "")
FilterReplace

Bases: Filter

Filter for strings with simple replacements.

f = FilterReplace("hello", "world") f("hello world") 'world world'

Source code in baygon/filters.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
@register_filter
class FilterReplace(Filter):
    """Filter for strings with simple replacements.
    >>> f = FilterReplace("hello", "world")
    >>> f("hello world")
    'world world'
    """

    def __init__(self, pattern: str, replacement: str):
        super().__init__()
        self.pattern = pattern
        self.replacement = replacement

    def apply(self, value: str) -> str:
        return value.replace(self.pattern, self.replacement)
FilterRegex

Bases: Filter

Filter for strings using regular expressions.

f = FilterRegex("[aeiou]", "-") f("hello world") 'h-ll- w-rld'

Source code in baygon/filters.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
@register_filter
class FilterRegex(Filter):
    """Filter for strings using regular expressions.
    >>> f = FilterRegex("[aeiou]", "-")
    >>> f("hello world")
    'h-ll- w-rld'
    """

    def __init__(self, pattern: str, replacement: str):
        super().__init__()
        self.pattern = pattern
        self.replacement = replacement
        self.regex = re.compile(pattern)

    def apply(self, value: str) -> str:
        return self.regex.sub(self.replacement, value)
FilterEval

Bases: Filter

Filter for evaluating mustaches in strings.

Source code in baygon/filters.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@register_filter
class FilterEval(Filter):
    """Filter for evaluating mustaches in strings."""

    def __init__(
        self, start: str = "{{", end: str = "}}", init: list[str] | None = None
    ):
        super().__init__()
        self._mustache = re.compile(f"{start}(.*?){end}")
        self._kernel = TinyKernel()

        seed = list(init) if init is not None else []
        seed += [
            "from math import *",
            "from random import *",
            "from statistics import *",
            "from baygon.eval import iter",
        ]

        for item in seed:
            self._kernel(item)

    def apply(self, value: str) -> str:
        """Evaluate mustaches in a string."""
        pos = 0
        ret = ""
        for match in self._mustache.finditer(value):
            ret += value[pos : match.start()]
            ret += str(self.exec(match.group(1)))
            pos = match.end()
        ret += value[pos:]
        return ret

    def exec(self, code: str):
        """Execute code in the kernel."""

        # Inject context to custom functions
        code = re.sub(r"((?<=\b)iter\(.*?)(\))", f"\\1,ctx={hash(code)}\\2", code)

        # Workaround to get the value of assignments
        try:
            self._kernel("_ = " + code)
            return self._kernel.glb["_"]
        except SyntaxError:
            return self._kernel(code)

    def __repr__(self):
        return f"{self.__class__.__name__}({self._mustache.pattern})"
Functions
apply 
apply(value)

Evaluate mustaches in a string.

Source code in baygon/filters.py 
205
206
207
208
209
210
211
212
213
214
def apply(self, value: str) -> str:
    """Evaluate mustaches in a string."""
    pos = 0
    ret = ""
    for match in self._mustache.finditer(value):
        ret += value[pos : match.start()]
        ret += str(self.exec(match.group(1)))
        pos = match.end()
    ret += value[pos:]
    return ret
exec 
exec(code)

Execute code in the kernel.

Source code in baygon/filters.py 
216
217
218
219
220
221
222
223
224
225
226
227
def exec(self, code: str):
    """Execute code in the kernel."""

    # Inject context to custom functions
    code = re.sub(r"((?<=\b)iter\(.*?)(\))", f"\\1,ctx={hash(code)}\\2", code)

    # Workaround to get the value of assignments
    try:
        self._kernel("_ = " + code)
        return self._kernel.glb["_"]
    except SyntaxError:
        return self._kernel(code)
Filters

Bases: Filter, Sequence

A sequence of filters.

Source code in baygon/filters.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
class Filters(Filter, Sequence):
    """A sequence of filters."""

    def __init__(self, filters=None):
        super().__init__()
        self._filters = self._parse_filter(filters)

    def _parse_filter(self, filters):
        if filters is None:
            return []
        if isinstance(filters, Filter):
            return [filters]
        if isinstance(filters, Filters):
            return list(filters)
        if isinstance(filters, dict):
            instances = []
            for name, args in filters.items():
                if not isinstance(args, list):
                    args = [args]
                instances.append(FilterFactory(name, *args))
            return instances

        raise InvalidFilterError(f"Invalid type for filters: {type(filters)}")

    def __getitem__(self, index):
        return self._filters[index]

    def __len__(self):
        return len(self._filters)

    def extend(self, filters):
        """Extend the filters with another Filters object."""
        self._filters.extend(self._parse_filter(filters))
        return self

    def apply(self, value: str) -> str:
        for filter_ in self._filters:
            value = filter_.filter(value)
        return value

    def __repr__(self):
        return f"{self.__class__.__name__}<{self._filters}>"
Functions
extend 
extend(filters)

Extend the filters with another Filters object.

Source code in baygon/filters.py 
263
264
265
266
def extend(self, filters):
    """Extend the filters with another Filters object."""
    self._filters.extend(self._parse_filter(filters))
    return self
FilterFactory

Factory for filters.

Source code in baygon/filters.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
class FilterFactory:
    """Factory for filters."""

    @staticmethod
    def _get_filter_class(name: str) -> type[Filter]:
        key = name.lower()
        try:
            return _FILTER_REGISTRY[key]
        except KeyError as exc:
            raise ValueError(f"Unknown filter: {name}") from exc

    def __new__(cls, name, *args, **kwargs) -> Filter:
        filter_cls = cls._get_filter_class(name)
        return filter_cls(*args, **kwargs)
Functions
register_filter 
register_filter(name=None)

Register a filter so it can be referenced from configuration.

Usage examples:

@register_filter ... class FilterFoo(Filter): ...

@register_filter("bar") ... class CustomFilter(Filter): ...

Source code in baygon/filters.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def register_filter(
    name: str | None = None,
) -> Callable[[type[FilterType]], type[FilterType]] | type[FilterType]:
    """Register a filter so it can be referenced from configuration.

    Usage examples:

    >>> @register_filter
    ... class FilterFoo(Filter): ...
    >>>
    >>> @register_filter("bar")
    ... class CustomFilter(Filter): ...
    """

    def decorator(cls: type[FilterType]) -> type[FilterType]:
        key = (name or cls.name()).lower()
        existing = _FILTER_REGISTRY.get(key)
        if existing is not None and existing is not cls:
            raise ValueError(f"Filter '{key}' is already registered.")
        _FILTER_REGISTRY[key] = cls
        return cls

    if isinstance(name, type):
        cls = name
        name = None
        return decorator(cls)
    return decorator
get_registered_filters 
get_registered_filters()

Return a copy of the registered filters mapping.

Source code in baygon/filters.py 
85
86
87
def get_registered_filters() -> dict[str, type[Filter]]:
    """Return a copy of the registered filters mapping."""
    return dict(_FILTER_REGISTRY)

helpers

Classes
GreppableString

Bases: str

A string that can be parsed with regular expressions.

Source code in baygon/helpers.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class GreppableString(str):
    """A string that can be parsed with regular expressions."""

    def grep(self, pattern: str, *args) -> bool:
        """Return True if the pattern is found in the string.

        >>> GreppableString("hello world").grep("w.{3}d")
        ['world']
        >>> GreppableString("hello world").grep(r"\b[a-z]{4}\b")
        []
        """
        return re.findall(pattern, self, *args)

    def contains(self, value: str) -> bool:
        """Return True if the value is found in the string.

        >>> GreppableString("hello world").contains("world")
        True
        >>> GreppableString("hello world").contains("earth")
        False
        """
        return value in self
Functions
grep 
grep(pattern, *args)

Return True if the pattern is found in the string.

GreppableString("hello world").grep("w.{3}d") ['world'] GreppableString("hello world").grep(r"[a-z]{4}") []

Source code in baygon/helpers.py 
31
32
33
34
35
36
37
38
39
def grep(self, pattern: str, *args) -> bool:
    """Return True if the pattern is found in the string.

    >>> GreppableString("hello world").grep("w.{3}d")
    ['world']
    >>> GreppableString("hello world").grep(r"\b[a-z]{4}\b")
    []
    """
    return re.findall(pattern, self, *args)
contains 
contains(value)

Return True if the value is found in the string.

GreppableString("hello world").contains("world") True GreppableString("hello world").contains("earth") False

Source code in baygon/helpers.py 
41
42
43
44
45
46
47
48
49
def contains(self, value: str) -> bool:
    """Return True if the value is found in the string.

    >>> GreppableString("hello world").contains("world")
    True
    >>> GreppableString("hello world").contains("earth")
    False
    """
    return value in self
Functions
escape_argument 
escape_argument(arg)

Escape a command line argument.

print(escape_argument("hello")) hello print(escape_argument("hello world")) 'hello world' print(escape_argument("hello'world'")) 'hello'"'"'world'"'"''

Source code in baygon/helpers.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
def escape_argument(arg):
    """Escape a command line argument.

    >>> print(escape_argument("hello"))
    hello
    >>> print(escape_argument("hello world"))
    'hello world'
    >>> print(escape_argument("hello'world'"))
    'hello'"'"'world'"'"''
    """
    return shlex.quote(arg)
create_command_line 
create_command_line(args)

Create a command line from a list of arguments.

print(create_command_line(["echo", "hello world"])) echo 'hello world'

Source code in baygon/helpers.py 
18
19
20
21
22
23
24
25
def create_command_line(args):
    """Create a command line from a list of arguments.

    >>> print(create_command_line(["echo", "hello world"]))
    echo 'hello world'
    """
    escaped_args = [escape_argument(str(arg)) for arg in args]
    return " ".join(escaped_args)

id

Hierarchical Id class to identify nested sequences

Classes
Id

Bases: Sequence

Test identifier. Helper class to number tests.

Example:

i = Id() i = i.down() i Id(1.1) i += 1 i Id(1.2) i += 2 i Id(1.4) str(i) '1.4' i.up() Id(1) ((i + 1).down() + 1).down() Id(1.5.2.1) tuple(Id().down().next().down().next().down().next()) (1, 2, 2, 2)

Source code in baygon/id.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class Id(Sequence):
    """Test identifier. Helper class to number tests.

    Example:
    >>> i = Id()
    >>> i = i.down()
    >>> i
    Id(1.1)
    >>> i += 1
    >>> i
    Id(1.2)
    >>> i += 2
    >>> i
    Id(1.4)
    >>> str(i)
    '1.4'
    >>> i.up()
    Id(1)
    >>> ((i + 1).down() + 1).down()
    Id(1.5.2.1)
    >>> tuple(Id().down().next().down().next().down().next())
    (1, 2, 2, 2)
    """

    def __init__(self, ids: list[int] | int | Id | str | None = None):
        if isinstance(ids, int):
            ids = [ids]
        if isinstance(ids, Id):
            ids = ids.ids
        if isinstance(ids, str) and re.match(r"^\d+(\.\d+)*$", ids):
            ids = [int(i) for i in ids.split(".")]
        if ids is not None and not isinstance(ids, list):
            raise ValueError(f"Invalid type for Id: {type(ids)}")

        self.ids = ids or [1]

    def next(self):
        """Return a new Id with the last id incremented."""
        return self + 1

    def down(self, base: int = 1):
        """Return a new Id with the given id appended."""
        return Id([*self.ids, base])

    def up(self):
        """Return a new Id with the given id appended."""
        return Id(self.ids[:-1])

    def __str__(self):
        return ".".join([str(i) for i in self.ids])

    def __repr__(self):
        return f"Id({self!s})"

    def __add__(self, other):
        return Id([*self.ids[:-1], self.ids[-1] + other])

    def __list__(self):
        return self.ids

    def __iter__(self):
        yield from self.ids

    def __len__(self):
        return len(self.ids)

    def __hash__(self):
        return hash(tuple(self))

    def __getitem__(self, item):
        return self.ids[item]

    def pad(self, length="  "):
        """Return id with initial padding."""
        return length * (len(self) - 1)
Functions
next 
next()

Return a new Id with the last id incremented.

Source code in baygon/id.py 
45
46
47
def next(self):
    """Return a new Id with the last id incremented."""
    return self + 1
down 
down(base=1)

Return a new Id with the given id appended.

Source code in baygon/id.py 
49
50
51
def down(self, base: int = 1):
    """Return a new Id with the given id appended."""
    return Id([*self.ids, base])
up 
up()

Return a new Id with the given id appended.

Source code in baygon/id.py 
53
54
55
def up(self):
    """Return a new Id with the given id appended."""
    return Id(self.ids[:-1])
pad 
pad(length='  ')

Return id with initial padding.

Source code in baygon/id.py 
81
82
83
def pad(self, length="  "):
    """Return id with initial padding."""
    return length * (len(self) - 1)
TrackId

Keep the id of the test.

Source code in baygon/id.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class TrackId:
    """Keep the id of the test."""

    def __init__(self):
        self._id = Id()

    def reset(self):
        """Reset the id."""

        def _(v=None):
            self._id = Id()
            return v

        return _

    def down(self):
        """Return a new Id with the given id appended."""

        def _(v=None):
            self._id = self._id.down()
            return v

        return _

    def up(self):
        """Return a new Id with the given id appended."""

        def _(v=None):
            self._id = self._id.up()
            return v

        return _

    def next(self):
        """Return a new Id with the last id incremented."""

        def _(v: dict):
            v["test_id"] = list(self._id)
            self._id = self._id.next()
            return v

        return _
Functions
reset 
reset()

Reset the id.

Source code in baygon/id.py 
92
93
94
95
96
97
98
99
def reset(self):
    """Reset the id."""

    def _(v=None):
        self._id = Id()
        return v

    return _
down 
down()

Return a new Id with the given id appended.

Source code in baygon/id.py 
101
102
103
104
105
106
107
108
def down(self):
    """Return a new Id with the given id appended."""

    def _(v=None):
        self._id = self._id.down()
        return v

    return _
up 
up()

Return a new Id with the given id appended.

Source code in baygon/id.py 
110
111
112
113
114
115
116
117
def up(self):
    """Return a new Id with the given id appended."""

    def _(v=None):
        self._id = self._id.up()
        return v

    return _
next 
next()

Return a new Id with the last id incremented.

Source code in baygon/id.py 
119
120
121
122
123
124
125
126
127
def next(self):
    """Return a new Id with the last id incremented."""

    def _(v: dict):
        v["test_id"] = list(self._id)
        self._id = self._id.next()
        return v

    return _

matchers

Output matchers used to assert command results.

Classes
InvalidCondition

Invalid test case condition.

Source code in baygon/matchers.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class InvalidCondition:
    """Invalid test case condition."""

    def __init__(self, value, expected, on=None, test=None, **kwargs):
        self.on = on
        self.test = test
        self.value = value
        self.expected = expected

    def __str__(self):
        return (
            f'Expected value "{self.expected}" on {self.on}, '
            f'but got "{self.value}" instead.'
        )

    def __repr__(self):
        return f"{self.__class__.__name__}<{self!s}>"
InvalidExitStatus

Bases: InvalidCondition

Invalid exit status error.

Source code in baygon/matchers.py 
31
32
33
34
35
class InvalidExitStatus(InvalidCondition):
    """Invalid exit status error."""

    def __str__(self):
        return f"Invalid exit status: {self.value} != {self.expected}."
InvalidContains

Bases: InvalidCondition

Invalid contains error.

Source code in baygon/matchers.py 
38
39
40
41
42
43
44
45
class InvalidContains(InvalidCondition):
    """Invalid contains error."""

    def __str__(self):
        return (
            f"Output {self.on} does not contain {self.expected}. "
            f'Found "{self.value}" instead.'
        )
InvalidRegex

Bases: InvalidCondition

Invalid regex error.

Source code in baygon/matchers.py 
48
49
50
51
52
53
54
class InvalidRegex(InvalidCondition):
    """Invalid regex error."""

    def __str__(self):
        return (
            f"Output '{self.on}' does not match /{self.expected}/ on \"{self.value}\"."
        )
InvalidEquals

Bases: InvalidCondition

Invalid equals error.

Source code in baygon/matchers.py 
57
58
59
60
61
62
class InvalidEquals(InvalidCondition):
    """Invalid equals error."""

    def __str__(self):
        value = "(empty)" if not self.value else f"'{self.value}'"
        return f"Output {value} does not equal '{self.expected}' on {self.on}."
MatchBase

Bases: ABC

Base class for all matchers.

Source code in baygon/matchers.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class MatchBase(ABC):
    """Base class for all matchers."""

    def __init__(self, inverse=False, **kwargs):
        """Initialize the matcher."""
        self.inverse = inverse

    @classmethod
    def name(cls):
        """Return the registration name of the matcher."""
        return cls.__name__.split("Match", maxsplit=1)[1].lower()

    @abstractmethod
    def __call__(self, value, **kwargs):
        """Match the value against the condition."""
        raise NotImplementedError
Functions
name classmethod 
name()

Return the registration name of the matcher.

Source code in baygon/matchers.py 
72
73
74
75
@classmethod
def name(cls):
    """Return the registration name of the matcher."""
    return cls.__name__.split("Match", maxsplit=1)[1].lower()
MatchRegex

Bases: MatchBase

Match a regex.

Source code in baygon/matchers.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
@register_matcher
class MatchRegex(MatchBase):
    """Match a regex."""

    def __init__(self, pattern, **kwargs):
        """Initialize the matcher."""
        self.pattern = re.compile(pattern)
        super().__init__(**kwargs)

    def __call__(self, value, **kwargs):
        if (not self.pattern.findall(value)) ^ self.inverse:
            return InvalidRegex(value, self.pattern.pattern, **kwargs)
        return None
MatchContains

Bases: MatchBase

Match if a string contains a specific value.

Source code in baygon/matchers.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
@register_matcher
class MatchContains(MatchBase):
    """Match if a string contains a specific value."""

    def __init__(self, contains, **kwargs):
        """Initialize the matcher."""
        self.contains = contains
        super().__init__(**kwargs)

    def __call__(self, value, **kwargs):
        if (self.contains not in value) ^ self.inverse:
            return InvalidContains(value, self.contains, **kwargs)
        return None
MatchEquals

Bases: MatchBase

Match if a string contains a specific value.

Source code in baygon/matchers.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
@register_matcher
class MatchEquals(MatchBase):
    """Match if a string contains a specific value."""

    def __init__(self, equal, **kwargs):
        """Initialize the matcher."""
        self.equal = equal
        super().__init__(**kwargs)

    def __call__(self, value, **kwargs):
        if (self.equal != value) ^ self.inverse:
            return InvalidEquals(value, self.equal, **kwargs)
        return None
MatcherFactory

Factory for matchers.

Source code in baygon/matchers.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class MatcherFactory:
    """Factory for matchers."""

    @staticmethod
    def _get_matcher_class(name: str) -> type[MatchBase]:
        key = name.lower()
        try:
            return _MATCHER_REGISTRY[key]
        except KeyError as exc:
            raise ValueError(f"Unknown matcher: {name}") from exc

    def __new__(cls, name, *args, **kwargs) -> MatchBase:
        matcher_cls = cls._get_matcher_class(name)
        return matcher_cls(*args, **kwargs)
Functions
register_matcher 
register_matcher(name=None)

Register a matcher class that can be referenced from configuration.

Source code in baygon/matchers.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def register_matcher(
    name: str | None = None,
) -> Callable[[type[MatchType]], type[MatchType]] | type[MatchType]:
    """Register a matcher class that can be referenced from configuration."""

    def decorator(cls: type[MatchType]) -> type[MatchType]:
        key = (name or cls.name()).lower()
        existing = _MATCHER_REGISTRY.get(key)
        if existing is not None and existing is not cls:
            raise ValueError(f"Matcher '{key}' is already registered.")
        _MATCHER_REGISTRY[key] = cls
        return cls

    if isinstance(name, type):
        cls = name
        name = None
        return decorator(cls)
    return decorator
get_registered_matchers 
get_registered_matchers()

Return the registered matchers mapping.

Source code in baygon/matchers.py 
106
107
108
def get_registered_matchers() -> dict[str, type[MatchBase]]:
    """Return the registered matchers mapping."""
    return dict(_MATCHER_REGISTRY)

presentation

Presentation helpers for Baygon reports.

Modules
rich

Rich presentation utilities for Baygon.

Classes Functions
render_summary_table 
render_summary_table(report, *, console)

Render a summary table for the given report.

Source code in baygon/presentation/rich.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def render_summary_table(report: RunReport, *, console: Console) -> None:
    """Render a summary table for the given report."""
    grey_line_style = Style(color="grey37")
    title_style = Style(bold=False, color="white")
    table = Table(
        title="Test Summary",
        title_style=title_style,
        border_style=grey_line_style,
        box=SQUARE_DOUBLE_HEAD,
    )

    table.add_column("ID", justify="left")
    table.add_column("Test Name", justify="left", style=grey_line_style)
    table.add_column("Points", justify="center", style=grey_line_style)
    table.add_column("Status", justify="center", style=grey_line_style)

    for result in report.cases:
        status = _format_status(result.status)
        points_value = ""
        if result.points_earned is not None and result.case.points is not None:
            points_value = f"{result.points_earned}/{result.case.points}"
        table.add_row(
            result.case.id_str,
            f"{result.case.name}",
            points_value,
            status,
        )

    console.print(table)
render_pretty_failures 
render_pretty_failures(report, *, console)

Render rich failure panels for failing cases.

Source code in baygon/presentation/rich.py 
50
51
52
53
54
55
def render_pretty_failures(report: RunReport, *, console: Console) -> None:
    """Render rich failure panels for failing cases."""
    for result in report.cases:
        if result.status != "failed":
            continue
        console.print(_build_pretty_failure_panel(result))
render_command_panels 
render_command_panels(
    result, *, console, hide_empty_streams
)

Render command telemetry panels for a case.

Source code in baygon/presentation/rich.py 
58
59
60
61
62
63
64
65
66
67
def render_command_panels(
    result: CaseResult,
    *,
    console: Console,
    hide_empty_streams: bool,
) -> None:
    """Render command telemetry panels for a case."""
    panels = _command_panels(result.commands, hide_empty=hide_empty_streams)
    for panel in panels:
        console.print(panel)
rich_presenter

Rich presentation utilities for Baygon.

Classes Functions
render_summary_table 
render_summary_table(report, *, console)

Render a summary table for the given report.

Source code in baygon/presentation/rich.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def render_summary_table(report: RunReport, *, console: Console) -> None:
    """Render a summary table for the given report."""
    grey_line_style = Style(color="grey37")
    title_style = Style(bold=False, color="white")
    table = Table(
        title="Test Summary",
        title_style=title_style,
        border_style=grey_line_style,
        box=SQUARE_DOUBLE_HEAD,
    )

    table.add_column("ID", justify="left")
    table.add_column("Test Name", justify="left", style=grey_line_style)
    table.add_column("Points", justify="center", style=grey_line_style)
    table.add_column("Status", justify="center", style=grey_line_style)

    for result in report.cases:
        status = _format_status(result.status)
        points_value = ""
        if result.points_earned is not None and result.case.points is not None:
            points_value = f"{result.points_earned}/{result.case.points}"
        table.add_row(
            result.case.id_str,
            f"{result.case.name}",
            points_value,
            status,
        )

    console.print(table)
render_pretty_failures 
render_pretty_failures(report, *, console)

Render rich failure panels for failing cases.

Source code in baygon/presentation/rich.py 
50
51
52
53
54
55
def render_pretty_failures(report: RunReport, *, console: Console) -> None:
    """Render rich failure panels for failing cases."""
    for result in report.cases:
        if result.status != "failed":
            continue
        console.print(_build_pretty_failure_panel(result))
render_command_panels 
render_command_panels(
    result, *, console, hide_empty_streams
)

Render command telemetry panels for a case.

Source code in baygon/presentation/rich.py 
58
59
60
61
62
63
64
65
66
67
def render_command_panels(
    result: CaseResult,
    *,
    console: Console,
    hide_empty_streams: bool,
) -> None:
    """Render command telemetry panels for a case."""
    panels = _command_panels(result.commands, hide_empty=hide_empty_streams)
    for panel in panels:
        console.print(panel)
text

Plain-text presentation of Baygon run reports.

Classes Functions
render_case_results 
render_case_results(
    report, *, write, verbose=0, include_issues=True
)

Render individual case outcomes.

Source code in baygon/presentation/text.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def render_case_results(
    report: RunReport,
    *,
    write: Writer,
    verbose: int = 0,
    include_issues: bool = True,
) -> None:
    """Render individual case outcomes."""

    del verbose  # Reserved for future verbosity handling.

    for result in report.cases:
        header = f"Test {result.case.id_str}: {result.case.name}"
        status = result.status.lower()
        if status == "passed":
            write(f"{header} PASSED")
        elif status == "failed":
            write(f"{header} FAILED")
            if include_issues and result.issues:
                for issue in result.issues:
                    write(str(issue))
        elif status == "skipped":
            write(f"{header} SKIPPED")
        else:
            write(f"{header} {result.status}")
render_summary 
render_summary(report, *, write)

Render the global summary for a run.

Source code in baygon/presentation/text.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def render_summary(report: RunReport, *, write: Writer) -> None:
    """Render the global summary for a run."""

    write("")
    write(f"Ran {report.total} tests in {report.duration:.2f} s.")

    if report.points_total:
        write(f"Points: {report.points_earned}/{report.points_total}")

    if report.failures > 0:
        denominator = max(report.failures + report.successes, 1)
        ratio = 100 - round((report.failures / denominator) * 100, 2)
        write(f"{report.failures} failed, {report.successes} passed ({ratio}% ok).")
        write("fail.")
    else:
        write("ok.")

    if report.skipped > 0:
        write(f"{report.skipped} test(s) skipped, some executables may be missing.")
text_presenter

Plain-text presentation of Baygon run reports.

Classes Functions
render_case_results 
render_case_results(
    report, *, write, verbose=0, include_issues=True
)

Render individual case outcomes.

Source code in baygon/presentation/text.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def render_case_results(
    report: RunReport,
    *,
    write: Writer,
    verbose: int = 0,
    include_issues: bool = True,
) -> None:
    """Render individual case outcomes."""

    del verbose  # Reserved for future verbosity handling.

    for result in report.cases:
        header = f"Test {result.case.id_str}: {result.case.name}"
        status = result.status.lower()
        if status == "passed":
            write(f"{header} PASSED")
        elif status == "failed":
            write(f"{header} FAILED")
            if include_issues and result.issues:
                for issue in result.issues:
                    write(str(issue))
        elif status == "skipped":
            write(f"{header} SKIPPED")
        else:
            write(f"{header} {result.status}")
render_summary 
render_summary(report, *, write)

Render the global summary for a run.

Source code in baygon/presentation/text.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def render_summary(report: RunReport, *, write: Writer) -> None:
    """Render the global summary for a run."""

    write("")
    write(f"Ran {report.total} tests in {report.duration:.2f} s.")

    if report.points_total:
        write(f"Points: {report.points_earned}/{report.points_total}")

    if report.failures > 0:
        denominator = max(report.failures + report.successes, 1)
        ratio = 100 - round((report.failures / denominator) * 100, 2)
        write(f"{report.failures} failed, {report.successes} passed ({ratio}% ok).")
        write("fail.")
    else:
        write("ok.")

    if report.skipped > 0:
        write(f"{report.skipped} test(s) skipped, some executables may be missing.")

runtime

Runtime execution services.

Classes
BaygonRunner

Execute suites described by immutable models.

Source code in baygon/runtime/runner.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
class BaygonRunner:
    """Execute suites described by immutable models."""

    def __init__(
        self,
        suite: SuiteModel,
        *,
        base_dir: Path,
        executable: str | Path | None = None,
        executable_factory: Callable[[str], Executable] = Executable,
        clock: Callable[[], float] = time.perf_counter,
    ) -> None:
        self._suite = suite
        self._base_dir = base_dir
        self._clock = clock
        self._executable_factory = executable_factory
        self._executables: MutableMapping[str, Executable] = {}

        cli_executable = self._resolve_path(executable)
        suite_executable = self._resolve_path(suite.executable)
        if cli_executable and suite_executable:
            raise InvalidExecutableError("Executable can't be overridden")
        self._root_executable = cli_executable or suite_executable

    @property
    def suite(self) -> SuiteModel:
        """Return the suite model handled by the runner."""
        return self._suite

    def run(self, limit: int = -1) -> RunReport:
        """Run the test suite."""
        results: list[CaseResult] = []
        counters = defaultdict(int)
        points_total: float | int = 0
        points_earned: float | int = 0

        start = self._clock()
        root_context = _ExecutionContext(
            filters=_merge_filters(None, self._suite.filters),
            eval_filter=_resolve_eval(None, self._suite.eval),
            executable=self._root_executable,
        )

        stop_requested = False

        for case, context in self._iter_cases(root_context):
            if stop_requested:
                break

            case_points = case.points or 0
            points_total += case_points
            case_result = self._run_case(case, context)

            results.append(case_result)
            status = case_result.status
            counters[status] += 1
            if status == "passed":
                points_earned += case_result.points_earned or 0
            elif status == "failed" and limit > 0 and counters["failed"] > limit:
                stop_requested = True

        duration = round(self._clock() - start, 6)
        return RunReport(
            suite=self._suite,
            successes=counters["passed"],
            failures=counters["failed"],
            skipped=counters["skipped"],
            points_total=points_total,
            points_earned=points_earned,
            duration=duration,
            cases=tuple(results),
        )

    def _iter_cases(
        self, root: _ExecutionContext
    ) -> Iterator[tuple[CaseModel, _ExecutionContext]]:
        for test in self._suite.tests:
            yield from self._walk(test, root)

    def _walk(
        self,
        node: CaseModel | GroupModel,
        parent_context: _ExecutionContext,
    ) -> Iterator[tuple[CaseModel, _ExecutionContext]]:
        if isinstance(node, CaseModel):
            context = _ExecutionContext(
                filters=_merge_filters(parent_context.filters, node.filters),
                eval_filter=_resolve_eval(parent_context.eval_filter, node.eval),
                executable=_inherit_executable(
                    parent_context.executable, node.executable, self._base_dir
                ),
            )
            yield (node, context)
            return

        context = _ExecutionContext(
            filters=_merge_filters(parent_context.filters, node.filters),
            eval_filter=_resolve_eval(parent_context.eval_filter, node.eval),
            executable=_inherit_executable(
                parent_context.executable, node.executable, self._base_dir
            ),
        )
        for child in node.tests:
            yield from self._walk(child, context)

    def _run_case(self, case: CaseModel, context: _ExecutionContext) -> CaseResult:
        start = self._clock()
        issues: list[Any] = []
        command_logs: list[CommandLog] = []
        exec_path = context.executable
        if exec_path is None:
            raise InvalidExecutableError(
                f"Executable not provided for test '{case.name}' (id {case.id_str})."
            )

        exec_obj = self._get_executable(exec_path)
        filters = context.filters
        eval_filter = context.eval_filter

        for _ in range(case.repeat):
            filtered_args = tuple(_apply_eval(eval_filter, list(case.args)))
            filtered_stdin = _apply_eval(eval_filter, case.stdin)
            filtered_env = _apply_eval_env(eval_filter, case.env)
            expected_exit = (
                int(_apply_eval(eval_filter, str(case.exit)))
                if case.exit is not None
                else None
            )

            hook = _capture_hook(command_logs)
            output = exec_obj.run(
                *filtered_args,
                stdin=filtered_stdin if filtered_stdin is not None else None,
                env=get_env(filtered_env),
                hook=hook,
            )

            issues.extend(
                _match_streams(
                    case,
                    filters,
                    eval_filter,
                    output,
                    "stdout",
                    case.stdout,
                )
            )
            issues.extend(
                _match_streams(
                    case,
                    filters,
                    eval_filter,
                    output,
                    "stderr",
                    case.stderr,
                )
            )

            if expected_exit is not None and expected_exit != output.exit_status:
                issues.append(
                    InvalidExitStatus(
                        expected_exit,
                        output.exit_status,
                        on="exit",
                        test=case,
                    )
                )

        status = "failed" if issues else "passed"
        duration = round(self._clock() - start, 6)
        points = case.points or 0
        earned = points if status == "passed" else 0

        return CaseResult(
            case=case,
            status=status,
            issues=tuple(issues),
            commands=tuple(command_logs),
            duration=duration,
            points_earned=earned,
        )

    def _get_executable(self, path: str) -> Executable:
        if path not in self._executables:
            self._executables[path] = self._executable_factory(path)
        return self._executables[path]

    def _resolve_path(self, value: str | Path | None) -> str | None:
        if value is None:
            return None
        path = Path(value)
        if not path.is_absolute():
            path = (self._base_dir / path).resolve()
        return str(path)
Attributes
suite property 
suite

Return the suite model handled by the runner.

Functions
run 
run(limit=-1)

Run the test suite.

Source code in baygon/runtime/runner.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def run(self, limit: int = -1) -> RunReport:
    """Run the test suite."""
    results: list[CaseResult] = []
    counters = defaultdict(int)
    points_total: float | int = 0
    points_earned: float | int = 0

    start = self._clock()
    root_context = _ExecutionContext(
        filters=_merge_filters(None, self._suite.filters),
        eval_filter=_resolve_eval(None, self._suite.eval),
        executable=self._root_executable,
    )

    stop_requested = False

    for case, context in self._iter_cases(root_context):
        if stop_requested:
            break

        case_points = case.points or 0
        points_total += case_points
        case_result = self._run_case(case, context)

        results.append(case_result)
        status = case_result.status
        counters[status] += 1
        if status == "passed":
            points_earned += case_result.points_earned or 0
        elif status == "failed" and limit > 0 and counters["failed"] > limit:
            stop_requested = True

    duration = round(self._clock() - start, 6)
    return RunReport(
        suite=self._suite,
        successes=counters["passed"],
        failures=counters["failed"],
        skipped=counters["skipped"],
        points_total=points_total,
        points_earned=points_earned,
        duration=duration,
        cases=tuple(results),
    )
CaseResult dataclass

Individual case execution result.

Source code in baygon/runtime/runner.py 
30
31
32
33
34
35
36
37
38
39
@dataclass(frozen=True)
class CaseResult:
    """Individual case execution result."""

    case: CaseModel
    status: str
    issues: tuple[Any, ...]
    commands: tuple[CommandLog, ...]
    duration: float | None = None
    points_earned: float | int | None = None
CommandLog dataclass

Captured information about a single executed command.

Source code in baygon/runtime/runner.py 
19
20
21
22
23
24
25
26
27
@dataclass(frozen=True)
class CommandLog:
    """Captured information about a single executed command."""

    argv: tuple[str, ...]
    stdin: str | None
    stdout: str
    stderr: str
    exit_status: int
RunReport dataclass

Aggregated information after running a suite.

Source code in baygon/runtime/runner.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass(frozen=True)
class RunReport:
    """Aggregated information after running a suite."""

    suite: SuiteModel
    successes: int
    failures: int
    skipped: int
    points_total: float | int
    points_earned: float | int
    duration: float
    cases: tuple[CaseResult, ...]

    @property
    def total(self) -> int:
        return self.successes + self.failures + self.skipped
Modules
runner

Pure runner built on top of Baygon domain models.

Classes
CommandLog dataclass

Captured information about a single executed command.

Source code in baygon/runtime/runner.py 
19
20
21
22
23
24
25
26
27
@dataclass(frozen=True)
class CommandLog:
    """Captured information about a single executed command."""

    argv: tuple[str, ...]
    stdin: str | None
    stdout: str
    stderr: str
    exit_status: int
CaseResult dataclass

Individual case execution result.

Source code in baygon/runtime/runner.py 
30
31
32
33
34
35
36
37
38
39
@dataclass(frozen=True)
class CaseResult:
    """Individual case execution result."""

    case: CaseModel
    status: str
    issues: tuple[Any, ...]
    commands: tuple[CommandLog, ...]
    duration: float | None = None
    points_earned: float | int | None = None
RunReport dataclass

Aggregated information after running a suite.

Source code in baygon/runtime/runner.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass(frozen=True)
class RunReport:
    """Aggregated information after running a suite."""

    suite: SuiteModel
    successes: int
    failures: int
    skipped: int
    points_total: float | int
    points_earned: float | int
    duration: float
    cases: tuple[CaseResult, ...]

    @property
    def total(self) -> int:
        return self.successes + self.failures + self.skipped
BaygonRunner

Execute suites described by immutable models.

Source code in baygon/runtime/runner.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
class BaygonRunner:
    """Execute suites described by immutable models."""

    def __init__(
        self,
        suite: SuiteModel,
        *,
        base_dir: Path,
        executable: str | Path | None = None,
        executable_factory: Callable[[str], Executable] = Executable,
        clock: Callable[[], float] = time.perf_counter,
    ) -> None:
        self._suite = suite
        self._base_dir = base_dir
        self._clock = clock
        self._executable_factory = executable_factory
        self._executables: MutableMapping[str, Executable] = {}

        cli_executable = self._resolve_path(executable)
        suite_executable = self._resolve_path(suite.executable)
        if cli_executable and suite_executable:
            raise InvalidExecutableError("Executable can't be overridden")
        self._root_executable = cli_executable or suite_executable

    @property
    def suite(self) -> SuiteModel:
        """Return the suite model handled by the runner."""
        return self._suite

    def run(self, limit: int = -1) -> RunReport:
        """Run the test suite."""
        results: list[CaseResult] = []
        counters = defaultdict(int)
        points_total: float | int = 0
        points_earned: float | int = 0

        start = self._clock()
        root_context = _ExecutionContext(
            filters=_merge_filters(None, self._suite.filters),
            eval_filter=_resolve_eval(None, self._suite.eval),
            executable=self._root_executable,
        )

        stop_requested = False

        for case, context in self._iter_cases(root_context):
            if stop_requested:
                break

            case_points = case.points or 0
            points_total += case_points
            case_result = self._run_case(case, context)

            results.append(case_result)
            status = case_result.status
            counters[status] += 1
            if status == "passed":
                points_earned += case_result.points_earned or 0
            elif status == "failed" and limit > 0 and counters["failed"] > limit:
                stop_requested = True

        duration = round(self._clock() - start, 6)
        return RunReport(
            suite=self._suite,
            successes=counters["passed"],
            failures=counters["failed"],
            skipped=counters["skipped"],
            points_total=points_total,
            points_earned=points_earned,
            duration=duration,
            cases=tuple(results),
        )

    def _iter_cases(
        self, root: _ExecutionContext
    ) -> Iterator[tuple[CaseModel, _ExecutionContext]]:
        for test in self._suite.tests:
            yield from self._walk(test, root)

    def _walk(
        self,
        node: CaseModel | GroupModel,
        parent_context: _ExecutionContext,
    ) -> Iterator[tuple[CaseModel, _ExecutionContext]]:
        if isinstance(node, CaseModel):
            context = _ExecutionContext(
                filters=_merge_filters(parent_context.filters, node.filters),
                eval_filter=_resolve_eval(parent_context.eval_filter, node.eval),
                executable=_inherit_executable(
                    parent_context.executable, node.executable, self._base_dir
                ),
            )
            yield (node, context)
            return

        context = _ExecutionContext(
            filters=_merge_filters(parent_context.filters, node.filters),
            eval_filter=_resolve_eval(parent_context.eval_filter, node.eval),
            executable=_inherit_executable(
                parent_context.executable, node.executable, self._base_dir
            ),
        )
        for child in node.tests:
            yield from self._walk(child, context)

    def _run_case(self, case: CaseModel, context: _ExecutionContext) -> CaseResult:
        start = self._clock()
        issues: list[Any] = []
        command_logs: list[CommandLog] = []
        exec_path = context.executable
        if exec_path is None:
            raise InvalidExecutableError(
                f"Executable not provided for test '{case.name}' (id {case.id_str})."
            )

        exec_obj = self._get_executable(exec_path)
        filters = context.filters
        eval_filter = context.eval_filter

        for _ in range(case.repeat):
            filtered_args = tuple(_apply_eval(eval_filter, list(case.args)))
            filtered_stdin = _apply_eval(eval_filter, case.stdin)
            filtered_env = _apply_eval_env(eval_filter, case.env)
            expected_exit = (
                int(_apply_eval(eval_filter, str(case.exit)))
                if case.exit is not None
                else None
            )

            hook = _capture_hook(command_logs)
            output = exec_obj.run(
                *filtered_args,
                stdin=filtered_stdin if filtered_stdin is not None else None,
                env=get_env(filtered_env),
                hook=hook,
            )

            issues.extend(
                _match_streams(
                    case,
                    filters,
                    eval_filter,
                    output,
                    "stdout",
                    case.stdout,
                )
            )
            issues.extend(
                _match_streams(
                    case,
                    filters,
                    eval_filter,
                    output,
                    "stderr",
                    case.stderr,
                )
            )

            if expected_exit is not None and expected_exit != output.exit_status:
                issues.append(
                    InvalidExitStatus(
                        expected_exit,
                        output.exit_status,
                        on="exit",
                        test=case,
                    )
                )

        status = "failed" if issues else "passed"
        duration = round(self._clock() - start, 6)
        points = case.points or 0
        earned = points if status == "passed" else 0

        return CaseResult(
            case=case,
            status=status,
            issues=tuple(issues),
            commands=tuple(command_logs),
            duration=duration,
            points_earned=earned,
        )

    def _get_executable(self, path: str) -> Executable:
        if path not in self._executables:
            self._executables[path] = self._executable_factory(path)
        return self._executables[path]

    def _resolve_path(self, value: str | Path | None) -> str | None:
        if value is None:
            return None
        path = Path(value)
        if not path.is_absolute():
            path = (self._base_dir / path).resolve()
        return str(path)
Attributes
suite property 
suite

Return the suite model handled by the runner.

Functions
run 
run(limit=-1)

Run the test suite.

Source code in baygon/runtime/runner.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def run(self, limit: int = -1) -> RunReport:
    """Run the test suite."""
    results: list[CaseResult] = []
    counters = defaultdict(int)
    points_total: float | int = 0
    points_earned: float | int = 0

    start = self._clock()
    root_context = _ExecutionContext(
        filters=_merge_filters(None, self._suite.filters),
        eval_filter=_resolve_eval(None, self._suite.eval),
        executable=self._root_executable,
    )

    stop_requested = False

    for case, context in self._iter_cases(root_context):
        if stop_requested:
            break

        case_points = case.points or 0
        points_total += case_points
        case_result = self._run_case(case, context)

        results.append(case_result)
        status = case_result.status
        counters[status] += 1
        if status == "passed":
            points_earned += case_result.points_earned or 0
        elif status == "failed" and limit > 0 and counters["failed"] > limit:
            stop_requested = True

    duration = round(self._clock() - start, 6)
    return RunReport(
        suite=self._suite,
        successes=counters["passed"],
        failures=counters["failed"],
        skipped=counters["skipped"],
        points_total=points_total,
        points_earned=points_earned,
        duration=duration,
        cases=tuple(results),
    )
Functions

schema

Schema definition and validation for Baygon configurations.

Classes
FiltersConfig

Bases: BaseModel

Filters available at the configuration or case level.

Source code in baygon/schema.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class FiltersConfig(BaseModel):
    """Filters available at the configuration or case level."""

    model_config = ConfigDict(extra="forbid", populate_by_name=True)

    uppercase: bool | None = None
    lowercase: bool | None = None
    trim: bool | None = None
    ignore_spaces: bool | None = Field(
        default=None,
        validation_alias=AliasChoices("ignorespaces", "ignore-spaces"),
        serialization_alias="ignorespaces",
    )
    regex: list[str] | None = None
    replace: list[str] | None = None

    @field_validator("regex", "replace", mode="before")
    @classmethod
    def _validate_pairs(cls, value: Any):
        if value is None:
            return None
        if (
            isinstance(value, Sequence)
            and not isinstance(value, str)
            and len(value) == 2
        ):
            return [_coerce_value(value[0]), _coerce_value(value[1])]
        raise ValueError("must be a sequence of exactly two values")
EvalConfig

Bases: BaseModel

Configuration of the Jinja-like evaluation filters.

Source code in baygon/schema.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class EvalConfig(BaseModel):
    """Configuration of the Jinja-like evaluation filters."""

    model_config = ConfigDict(extra="forbid")

    start: str = "{{"
    end: str = "}}"
    init: list[str] = Field(default_factory=list)

    @field_validator("init", mode="before")
    @classmethod
    def _validate_init(cls, value: Any):
        if value is None:
            return []
        if isinstance(value, str):
            return [value]
        if isinstance(value, Sequence) and not isinstance(value, str):
            return [_coerce_value(v) for v in value]
        raise TypeError("eval.init must be a string or a list of strings")
NegatedCondition

Bases: BaseModel

Negative matcher definition.

Source code in baygon/schema.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class NegatedCondition(BaseModel):
    """Negative matcher definition."""

    model_config = ConfigDict(extra="forbid")

    equals: str | None = None
    regex: str | None = None
    contains: str | None = None

    @field_validator("equals", "regex", "contains", mode="before")
    @classmethod
    def _convert_values(cls, value: Any):
        if value is None:
            return None
        return _coerce_value(value)

    @model_validator(mode="after")
    def _ensure_single_matcher(self):
        provided = [self.equals, self.regex, self.contains]
        if sum(value is not None for value in provided) != 1:
            raise ValueError("A negated condition must define exactly one matcher")
        return self
CaseCondition

Bases: BaseModel

Matchers applied to stdout/stderr values.

Source code in baygon/schema.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class CaseCondition(BaseModel):
    """Matchers applied to stdout/stderr values."""

    model_config = ConfigDict(extra="forbid", populate_by_name=True)

    filters: FiltersConfig = Field(default_factory=FiltersConfig)
    equals: str | None = None
    regex: str | None = None
    contains: str | None = None
    expected: str | None = None
    not_conditions: list[NegatedCondition] | None = Field(default=None, alias="not")

    @field_validator("equals", "regex", "contains", "expected", mode="before")
    @classmethod
    def _convert_values(cls, value: Any):
        if value is None:
            return None
        return _coerce_value(value)

    @model_validator(mode="after")
    def _ensure_matcher(self):
        if not any([self.equals, self.regex, self.contains, self.not_conditions]):
            raise ValueError("A condition must define at least one matcher")
        return self
CommonSettings

Bases: BaseModel

Shared configuration items for suites, groups and test cases.

Source code in baygon/schema.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class CommonSettings(BaseModel):
    """Shared configuration items for suites, groups and test cases."""

    model_config = ConfigDict(extra="forbid", populate_by_name=True)

    name: str = ""
    executable: str | None = None
    points: float | int | None = None
    weight: float | int | None = None
    min_points: float | int = Field(0.1, alias="min-points")

    @model_validator(mode="after")
    def _check_points_weight(self):
        if self.points is not None and self.weight is not None:
            raise ValueError("'points' and 'weight' cannot be used together")
        return self
TestCaseModel

Bases: CommonSettings

Single executable test case.

Source code in baygon/schema.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class TestCaseModel(CommonSettings):
    """Single executable test case."""

    model_config = ConfigDict(extra="forbid", populate_by_name=True)

    args: list[str] = Field(default_factory=list)
    env: dict[str, str] = Field(default_factory=dict)
    stdin: str | None = ""
    stdout: list[CaseCondition] = Field(default_factory=list)
    stderr: list[CaseCondition] = Field(default_factory=list)
    repeat: int = 1
    exit: int | str | bool | None = None
    test_id: list[int] = Field(default_factory=list, alias="test_id")

    @field_validator("args", mode="before")
    @classmethod
    def _convert_args(cls, value: Any):
        if value is None:
            return []
        if not isinstance(value, Sequence) or isinstance(value, str):
            raise TypeError("args must be a list")
        return [_coerce_value(item) for item in value]

    @field_validator("env", mode="before")
    @classmethod
    def _convert_env(cls, value: Any):
        if value is None:
            return {}
        if not isinstance(value, Mapping):
            raise TypeError("env must be a mapping")
        return {str(key): _coerce_value(val) for key, val in value.items()}

    @field_validator("stdin", mode="before")
    @classmethod
    def _convert_stdin(cls, value: Any):
        if value is None:
            return None
        return _coerce_value(value)

    @field_validator("stdout", "stderr", mode="before")
    @classmethod
    def _convert_matches(cls, value: Any):
        return _coerce_match_list(value)
TestGroupModel

Bases: CommonSettings

Group of tests that share settings.

Source code in baygon/schema.py 
216
217
218
219
220
221
222
class TestGroupModel(CommonSettings):
    """Group of tests that share settings."""

    model_config = ConfigDict(extra="forbid", populate_by_name=True)

    tests: list[BaygonTest]
    test_id: list[int] = Field(default_factory=list, alias="test_id")
BaygonConfig

Bases: CommonSettings

Top-level Baygon configuration.

Source code in baygon/schema.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
class BaygonConfig(CommonSettings):
    """Top-level Baygon configuration."""

    model_config = ConfigDict(extra="forbid", populate_by_name=True)

    version: int = 2
    filters: FiltersConfig = Field(default_factory=FiltersConfig)
    tests: list[BaygonTest]
    eval: EvalConfig | None = None
    verbose: int | None = None
    report: str | None = None
    format: Literal["json", "yaml"] | None = None
    table: bool = False

    @field_validator("version", mode="before")
    @classmethod
    def _validate_version(cls, value: Any):
        if value is None:
            return 2
        if value not in {1, 2}:
            raise ValueError("version must be 1 or 2")
        return value

    @field_validator("eval", mode="before")
    @classmethod
    def _convert_eval(cls, value: Any):
        if isinstance(value, bool):
            return {"init": []}
        return value
Functions
Schema 
Schema(data, humanize=False)

Validate the given data against the Baygon schema.

Source code in baygon/schema.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def Schema(data: Any, humanize: bool = False):  # noqa: N802
    """Validate the given data against the Baygon schema."""

    if isinstance(data, str):
        data = _load_yaml(data)
    elif hasattr(data, "read"):
        data = _load_yaml(data.read())

    if not isinstance(data, Mapping):
        raise ConfigError("Schema expects a mapping or YAML string")

    include_eval = "eval" in data
    raw_eval_value = data.get("eval") if include_eval else None

    try:
        config = BaygonConfig.model_validate(data)
    except ValidationError as exc:  # pragma: no cover - exercised indirectly
        if humanize:
            raise ConfigError(_humanize_errors(exc)) from exc
        raise

    _assign_test_ids(config.tests)
    return _dump_config(
        config, include_eval=include_eval, raw_eval_value=raw_eval_value
    )

score

Module used to compute the score of a test case. Used in academic.

Functions
float_or_int 
float_or_int(value)

Return a float or an integer.

float_or_int(1.0) 1 float_or_int(1.1) 1.1

Source code in baygon/score.py 
 8
 9
10
11
12
13
14
15
16
17
def float_or_int(value):
    """Return a float or an integer.
    >>> float_or_int(1.0)
    1
    >>> float_or_int(1.1)
    1.1
    """
    if value == int(value):
        return int(value)
    return float(value)
distribute 
distribute(values, total, min_value)

Distrubute the values given using a minimum step to ensure the total sum is respected.

distribute([1, 2, 3, 4], 10, 0.001) [1, 2, 3, 4] distribute([1, 1, 1, 1], 100, 0.01) [25, 25, 25, 25] distribute([12.5, 19.7, 42.1, 8.9], 100, 0.2) [15, 23.7, 50.6, 10.7] distribute([100, 100, 200, 200], 50, 1) [8, 8, 17, 17]

Source code in baygon/score.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def distribute(values, total, min_value):
    """Distrubute the values given using a minimum step to ensure
    the total sum is respected.

    >>> distribute([1, 2, 3, 4], 10, 0.001)
    [1, 2, 3, 4]
    >>> distribute([1, 1, 1, 1], 100, 0.01)
    [25, 25, 25, 25]
    >>> distribute([12.5, 19.7, 42.1, 8.9], 100, 0.2)
    [15, 23.7, 50.6, 10.7]
    >>> distribute([100, 100, 200, 200], 50, 1)
    [8, 8, 17, 17]
    """
    getcontext().prec = 28
    total_weight = sum(values)
    total = Decimal(str(total))
    min_value = Decimal(str(min_value))
    decimal_values = [Decimal(str(v)) for v in values]
    allocations = [v / Decimal(str(total_weight)) * total for v in decimal_values]

    # Round down to the nearest multiple of min_value
    quantizer = min_value
    allocations_rounded = [
        a.quantize(quantizer, rounding=ROUND_HALF_UP) for a in allocations
    ]

    # Adjust the allocations to match the total
    total_allocated = sum(allocations_rounded)
    difference = total - total_allocated

    # If the difference is not zero, adjust the allocations
    if difference != Decimal("0"):
        # Compute the number of units to adjust
        units_to_adjust = int(
            (difference / min_value).to_integral_value(rounding=ROUND_HALF_UP)
        )
        # Sort the allocations by their remainders
        remainders = [
            a - a.quantize(quantizer, rounding=ROUND_DOWN) for a in allocations
        ]
        if units_to_adjust > 0:
            indices = sorted(
                range(len(values)), key=lambda i: remainders[i], reverse=True
            )
            adjustment = min_value
        else:
            indices = sorted(range(len(values)), key=lambda i: remainders[i])
            adjustment = -min_value
            units_to_adjust = abs(units_to_adjust)

        for i in range(units_to_adjust):
            idx = indices[i % len(indices)]
            allocations_rounded[idx] += adjustment

    return [float_or_int(a) for a in allocations_rounded]
assign_points 
assign_points(test, parent=None)

Assign points recursively to each test in the structure.

Source code in baygon/score.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def assign_points(test, parent=None):
    """Assign points recursively to each test in the structure."""
    min_point = test.get("min-points", parent.get("min-points", 1) if parent else 1)

    # Case 6: Default points if there are weights or points somewhere
    default_point = 1 if has_weights_or_points(test) else 0

    # Set default weight to 10 if not specified
    if "weight" not in test and "points" not in test and "tests" in test:
        test["weight"] = 10

    # If 'points' is not in test, compute it based on parent's points and weights
    if "points" not in test:
        if "weight" in test and parent and "points" in parent:
            total_siblings_weight = parent.get("_total_weights", 0)
            if total_siblings_weight == 0:
                total_siblings_weight = sum(
                    t.get("weight", 10) for t in parent.get("tests", [])
                )
                parent["_total_weights"] = total_siblings_weight
            test["points"] = test["weight"] / total_siblings_weight * parent["points"]
            test["points"] = float_or_int(
                Decimal(str(test["points"])).quantize(
                    Decimal(str(min_point)), rounding=ROUND_HALF_UP
                )
            )
        else:
            test["points"] = default_point

    # If there are subtests, assign points to them
    if "tests" in test:
        fixed_points = sum(
            subtest.get("points", 0) for subtest in test["tests"] if "points" in subtest
        )
        weights = []
        subtests_to_distribute = []
        for subtest in test["tests"]:
            if "points" in subtest:
                continue  # Skip subtests that already have points
            elif "weight" in subtest:
                weights.append(subtest["weight"])
                subtests_to_distribute.append(subtest)
            else:
                # Case 2 and 6: Default weight is 10 if not specified
                subtest["weight"] = 10
                weights.append(subtest["weight"])
                subtests_to_distribute.append(subtest)

        points_to_distribute = test["points"] - fixed_points

        if weights and points_to_distribute > 0:
            allocated_points = distribute(weights, points_to_distribute, min_point)
            for subtest, points in zip(subtests_to_distribute, allocated_points):
                subtest["points"] = points
        elif not weights and points_to_distribute > 0:
            # Case 3: No weights but points are given to all tests
            equally_divided_point = points_to_distribute / len(test["tests"])
            for subtest in test["tests"]:
                if "points" not in subtest:
                    subtest["points"] = float_or_int(
                        Decimal(str(equally_divided_point)).quantize(
                            Decimal(str(min_point)), rounding=ROUND_HALF_UP
                        )
                    )

    # Recursive call for subtests
    if "tests" in test:
        for subtest in test["tests"]:
            assign_points(subtest, parent=test)

    # Clean up temporary keys
    if "_total_weights" in test:
        del test["_total_weights"]
has_weights_or_points 
has_weights_or_points(test)

Check if there are weights or points in the test or its subtests.

has_weights_or_points({"weight": 10}) True

Source code in baygon/score.py 
152
153
154
155
156
157
158
159
160
161
def has_weights_or_points(test):
    """Check if there are weights or points in the test or its subtests.
    >>> has_weights_or_points({"weight": 10})
    True
    """
    if "weight" in test or "points" in test:
        return True
    if "tests" in test:
        return any(has_weights_or_points(subtest) for subtest in test["tests"])
    return False
compute_points 
compute_points(data)

Compute points for the entire structure.

Source code in baygon/score.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def compute_points(data):
    """Compute points for the entire structure."""
    # Case 4: If no weights or points exist anywhere, do nothing
    if not has_weights_or_points(data):
        data["compute-score"] = False
        return data

    # Case 5: Set compute-score to true
    data["compute-score"] = True

    # Case 7: If total points are given at root but no other info, set default weight
    if "points" in data and "tests" in data:
        for test in data["tests"]:
            if "weight" not in test and "points" not in test:
                test["weight"] = 10

    assign_points(data)
    return data

suite

High-level services for loading and running Baygon suites.

Classes
SuiteContext dataclass

Immutable bundle describing a suite ready to be executed.

Source code in baygon/suite.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@dataclass(frozen=True)
class SuiteContext:
    """Immutable bundle describing a suite ready to be executed."""

    config: Mapping[str, Any]
    model: SuiteModel
    base_dir: Path
    source_path: Path | None

    @property
    def name(self) -> str:
        return str(self.config.get("name", ""))

    @property
    def version(self) -> int | None:
        version = self.config.get("version")
        return int(version) if version is not None else None

    def create_runner(
        self,
        *,
        executable: str | Path | None = None,
        runner_factory: Callable[..., BaygonRunner] = BaygonRunner,
    ) -> BaygonRunner:
        """Return a runner configured for this suite."""
        return runner_factory(
            self.model,
            base_dir=self.base_dir,
            executable=executable,
        )
Functions
create_runner 
create_runner(
    *, executable=None, runner_factory=BaygonRunner
)

Return a runner configured for this suite.

Source code in baygon/suite.py 
53
54
55
56
57
58
59
60
61
62
63
64
def create_runner(
    self,
    *,
    executable: str | Path | None = None,
    runner_factory: Callable[..., BaygonRunner] = BaygonRunner,
) -> BaygonRunner:
    """Return a runner configured for this suite."""
    return runner_factory(
        self.model,
        base_dir=self.base_dir,
        executable=executable,
    )
SuiteBuilder

Build immutable suite models from validated configuration mappings.

Source code in baygon/suite.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
class SuiteBuilder:
    """Build immutable suite models from validated configuration mappings."""

    def __init__(
        self,
        *,
        model_factory: Callable[[Mapping[str, Any]], SuiteModel] = build_suite_model,
    ) -> None:
        self._model_factory = model_factory

    def build(self, config: Mapping[str, Any]) -> SuiteModel:
        """Return a SuiteModel derived from the given configuration."""
        return self._model_factory(config)
Functions
build 
build(config)

Return a SuiteModel derived from the given configuration.

Source code in baygon/suite.py 
77
78
79
def build(self, config: Mapping[str, Any]) -> SuiteModel:
    """Return a SuiteModel derived from the given configuration."""
    return self._model_factory(config)
SuiteLoader

Load suite configurations from files or raw mappings.

Source code in baygon/suite.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class SuiteLoader:
    """Load suite configurations from files or raw mappings."""

    def __init__(
        self,
        *,
        schema_loader: Callable[[Any], MutableMapping[str, Any]] = Schema,
        builder: SuiteBuilder | None = None,
    ) -> None:
        self._schema_loader = schema_loader
        self._builder = builder or SuiteBuilder()

    def from_mapping(
        self,
        data: Mapping[str, Any] | MutableMapping[str, Any],
        *,
        cwd: str | Path | None = None,
    ) -> SuiteContext:
        """Validate a raw mapping and build its execution context."""
        validated = self._schema_loader(data)
        compute_points(validated)

        base_dir = Path(cwd) if cwd is not None else Path.cwd()
        model = self._builder.build(validated)

        return SuiteContext(
            config=validated,
            model=model,
            base_dir=base_dir.resolve(),
            source_path=None,
        )

    def from_path(
        self,
        path: str | Path | None,
    ) -> SuiteContext:
        """Load configuration from disk and build its execution context."""
        config_path = discover_config(path)
        config = load_config_dict(config_path)
        model = load_config_model(config_path)
        base_dir = config_path.parent

        return SuiteContext(
            config=config,
            model=model,
            base_dir=base_dir.resolve(),
            source_path=config_path,
        )

    def load(
        self,
        *,
        data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
        path: str | Path | None = None,
        cwd: str | Path | None = None,
    ) -> SuiteContext:
        """Load a suite configuration from a mapping or a path."""
        if data is not None and path is not None:
            raise ValueError("Provide either 'data' or 'path', not both.")

        if data is not None:
            return self.from_mapping(data, cwd=cwd)

        return self.from_path(path)
Functions
from_mapping 
from_mapping(data, *, cwd=None)

Validate a raw mapping and build its execution context.

Source code in baygon/suite.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def from_mapping(
    self,
    data: Mapping[str, Any] | MutableMapping[str, Any],
    *,
    cwd: str | Path | None = None,
) -> SuiteContext:
    """Validate a raw mapping and build its execution context."""
    validated = self._schema_loader(data)
    compute_points(validated)

    base_dir = Path(cwd) if cwd is not None else Path.cwd()
    model = self._builder.build(validated)

    return SuiteContext(
        config=validated,
        model=model,
        base_dir=base_dir.resolve(),
        source_path=None,
    )
from_path 
from_path(path)

Load configuration from disk and build its execution context.

Source code in baygon/suite.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def from_path(
    self,
    path: str | Path | None,
) -> SuiteContext:
    """Load configuration from disk and build its execution context."""
    config_path = discover_config(path)
    config = load_config_dict(config_path)
    model = load_config_model(config_path)
    base_dir = config_path.parent

    return SuiteContext(
        config=config,
        model=model,
        base_dir=base_dir.resolve(),
        source_path=config_path,
    )
load 
load(*, data=None, path=None, cwd=None)

Load a suite configuration from a mapping or a path.

Source code in baygon/suite.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def load(
    self,
    *,
    data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
    path: str | Path | None = None,
    cwd: str | Path | None = None,
) -> SuiteContext:
    """Load a suite configuration from a mapping or a path."""
    if data is not None and path is not None:
        raise ValueError("Provide either 'data' or 'path', not both.")

    if data is not None:
        return self.from_mapping(data, cwd=cwd)

    return self.from_path(path)
SuiteExecutor

Execute suites described by SuiteContext.

Source code in baygon/suite.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class SuiteExecutor:
    """Execute suites described by `SuiteContext`."""

    def __init__(
        self,
        *,
        runner_factory: Callable[..., BaygonRunner] = BaygonRunner,
    ) -> None:
        self._runner_factory = runner_factory

    def run(
        self,
        context: SuiteContext,
        *,
        executable: str | Path | None = None,
        limit: int = -1,
    ) -> RunReport:
        """Run the suite described by the provided context."""
        runner = context.create_runner(
            executable=executable,
            runner_factory=self._runner_factory,
        )
        return runner.run(limit=limit)
Functions
run 
run(context, *, executable=None, limit=-1)

Run the suite described by the provided context.

Source code in baygon/suite.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
def run(
    self,
    context: SuiteContext,
    *,
    executable: str | Path | None = None,
    limit: int = -1,
) -> RunReport:
    """Run the suite described by the provided context."""
    runner = context.create_runner(
        executable=executable,
        runner_factory=self._runner_factory,
    )
    return runner.run(limit=limit)
SuiteService

Facade coordinating loading and execution of Baygon suites.

Source code in baygon/suite.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class SuiteService:
    """Facade coordinating loading and execution of Baygon suites."""

    def __init__(
        self,
        loader: SuiteLoader | None = None,
        executor: SuiteExecutor | None = None,
    ) -> None:
        self._loader = loader or SuiteLoader()
        self._executor = executor or SuiteExecutor()

    def load(
        self,
        *,
        data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
        path: str | Path | None = None,
        cwd: str | Path | None = None,
    ) -> SuiteContext:
        """Return a validated suite context without running it."""
        return self._loader.load(data=data, path=path, cwd=cwd)

    def run(
        self,
        *,
        data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
        path: str | Path | None = None,
        cwd: str | Path | None = None,
        executable: str | Path | None = None,
        limit: int = -1,
    ) -> RunReport:
        """Load and execute a suite in one call."""
        context = self._loader.load(data=data, path=path, cwd=cwd)
        return self._executor.run(
            context,
            executable=executable,
            limit=limit,
        )
Functions
load 
load(*, data=None, path=None, cwd=None)

Return a validated suite context without running it.

Source code in baygon/suite.py 
184
185
186
187
188
189
190
191
192
def load(
    self,
    *,
    data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
    path: str | Path | None = None,
    cwd: str | Path | None = None,
) -> SuiteContext:
    """Return a validated suite context without running it."""
    return self._loader.load(data=data, path=path, cwd=cwd)
run 
run(
    *,
    data=None,
    path=None,
    cwd=None,
    executable=None,
    limit=-1
)

Load and execute a suite in one call.

Source code in baygon/suite.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def run(
    self,
    *,
    data: Mapping[str, Any] | MutableMapping[str, Any] | None = None,
    path: str | Path | None = None,
    cwd: str | Path | None = None,
    executable: str | Path | None = None,
    limit: int = -1,
) -> RunReport:
    """Load and execute a suite in one call."""
    context = self._loader.load(data=data, path=path, cwd=cwd)
    return self._executor.run(
        context,
        executable=executable,
        limit=limit,
    )
Functions
find_testfile 
find_testfile(path=None)

Return the path to the first Baygon configuration file found.

Source code in baygon/suite.py 
22
23
24
25
26
27
def find_testfile(path: str | Path | None = None) -> Path | None:
    """Return the path to the first Baygon configuration file found."""
    try:
        return discover_config(path)
    except ConfigError:
        return None
load_config 
load_config(path=None)

Load and validate a configuration file (YAML or JSON).

Source code in baygon/suite.py 
30
31
32
def load_config(path: str | Path | None = None) -> Mapping[str, Any]:
    """Load and validate a configuration file (YAML or JSON)."""
    return load_config_dict(path)