Merge pull request #40015 from BerriAI/litellm_fix_check_run_name_collisions

fix(ci): stop the auto-close duplicates job colliding with the required test check
This commit is contained in:
Mateo Wang 2026-09-07 10:55:34 -07:00 committed by GitHub
commit 9275cf42ed
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 1408 additions and 1 deletions

View file

@ -22,7 +22,7 @@ on:
permissions: {}
jobs:
test:
sweep-tests:
if: github.event_name == 'pull_request'
runs-on: ubuntu-latest
timeout-minutes: 5

View file

@ -74,6 +74,12 @@ jobs:
- name: check_workflow_startup_safety
run: uv run --no-sync python ./tests/code_coverage_tests/check_workflow_startup_safety.py
- name: check_workflow_job_name_collisions
run: uv run --no-sync python ./tests/code_coverage_tests/check_workflow_job_name_collisions.py
- name: test_workflow_job_name_collisions
run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_workflow_job_name_collisions.py
- name: test_e2e_changed_gate
run: uv run --no-sync pytest -q --noconftest -p no:cacheprovider -c /dev/null tests/code_coverage_tests/test_e2e_changed_gate.py

View file

@ -0,0 +1,521 @@
"""Catch workflow jobs that publish check runs under the same name.
A ruleset's required status check names a check run and GitHub matches it by that
name alone. When two jobs publish the same name the required context stops
mapping to the job that proves it: the commit carries two check runs under one
name and nothing says which one the ruleset required. Both being green hides the
clash completely, so the context quietly stops meaning what the ruleset intended.
One job lands in the same place when its `name:` holds no matrix value, since
every combination it runs then reports under that one name.
`.github/workflows/auto-close-duplicates.yml` shipped a job id `test` while
`.github/workflows/test-mcp.yml` already published the required `test` context,
and commit ed5761daef4ae17152446d182c860630c38b7268 carried both check runs.
This invariant has to be enforced here because CI cannot enforce it on itself.
A job publishes its `name:` when it sets one, and otherwise its job id plus the
values of the combination it runs, the way GitHub writes `build (3.12)`. A name
carrying `${{ ... }}` publishes one check run per combination the matrix
produces: `exclude` rows drop combinations before `include` rows fold into the
survivors, and each `include` row's values stay together rather than crossing
with the other rows', so two shard lists that overlap collide even though their
templates read differently. Each expression is evaluated per combination over the
pieces a job name can hold: string literals, `matrix.<key>`, `format()`, `==` and
`!=`, and the `<cond> && <a> || <b>` idiom, which is how the shards reach their
real `<shard> / Run tests` names rather than staying opaque.
Whatever the sweep cannot work out is left out of the comparison and reported
instead of guessed, because a guess that lands wrong fails a workflow GitHub
would have published perfectly well. A name still holding an expression once the
combination is filled in is usually one GitHub resolves per job, so it is one of
those: guessing that two jobs sharing such a template clash would fail workflows
over a context this sweep cannot read. The exception is a name whose leftover
expressions all read a `github.` property other than `github.job`, which one run
fills in the same way for every job in it, so those are compared against the
other jobs of their own workflow and stay out of the comparison across files,
where two workflows can run on different events. A matrix that is itself an
expression or that lists values which are not scalars, an `include` or `exclude`
row shaped the same way, a whole `strategy:` that comes from an expression, and a
call this sweep cannot follow, go in the same bucket. The cost is that a real clash hiding behind
one of them goes unseen, which leaves a merge no worse off than before this check
existed, where the opposite direction would block work that was fine.
A job calling a local reusable workflow publishes one check run per job of the
callee, named `<caller> / <callee>` and chained through however many levels of
local calls it takes, which is why a caller's name never collides with a plain
job that happens to match it. A file under `.github/workflows/` that does not
read as one workflow at all is reported rather than skipped, since skipping it
silently would hide every job it holds.
"""
import itertools
import operator
import re
import sys
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Final
import yaml
from pydantic import BaseModel, Field, ValidationError
REPO_ROOT: Final = Path(__file__).resolve().parent.parent.parent
WORKFLOWS_DIR: Final = REPO_ROOT / ".github" / "workflows"
EXPRESSION: Final = re.compile(r"\$\{\{(?P<body>.*?)\}\}", re.DOTALL)
MATRIX_REF: Final = re.compile(r"^matrix\.(?P<key>[\w-]+)$")
LITERAL: Final = re.compile(r"^'(?P<text>[^']*)'$")
FORMAT_CALL: Final = re.compile(r"^format\((?P<args>.*)\)$", re.DOTALL)
COMPARISON: Final = re.compile(r"^(?P<left>.+?)\s*(?P<operator>==|!=)\s*(?P<right>.+)$", re.DOTALL)
RUN_WIDE: Final = re.compile(r"^github\.(?!job\b)[\w.]+$")
GITHUB_PLACEHOLDER: Final = re.compile(r"\{\{|\}\}|\{\d+\}")
NO_MATRIX: Final[Mapping[str, str]] = MappingProxyType({})
NO_CALLERS: Final[frozenset[str]] = frozenset()
SCALAR: Final = (str, int, float)
MATRIX_DIRECTIVES: Final = frozenset({"include", "exclude"})
LOCAL_CALL_PREFIX: Final = "./"
@dataclass(frozen=True, slots=True)
class Unreadable:
reason: str
@dataclass(frozen=True, slots=True)
class Opaque:
reason: str
@dataclass(frozen=True, slots=True)
class Names:
"""The check-run names a job publishes, beside the reasons the rest of them stay unknown."""
known: tuple[str, ...] = ()
unknown: tuple[str, ...] = ()
local: tuple[str, ...] = ()
class Job(BaseModel):
name: object = None
uses: str | None = None
strategy: object = Field(default_factory=dict)
class Workflow(BaseModel):
jobs: Mapping[str, Job] = Field(default_factory=dict)
def scalar_text(value: object) -> str:
"""A YAML scalar the way GitHub renders it, so `true` never reaches a name as `True`."""
return str(value).lower() if isinstance(value, bool) else str(value)
def parse(source: str) -> tuple[Workflow, object] | Unreadable:
"""The workflow plus its raw `on:` value, or why the file does not read as one."""
try:
parsed: Final = yaml.safe_load(source)
except yaml.YAMLError:
return Unreadable("it does not read as one YAML document")
if not isinstance(parsed, dict):
return Unreadable("its top level is not a mapping of workflow keys")
try:
return Workflow.model_validate(parsed), parsed.get(True, parsed.get("on"))
except ValidationError as error:
return Unreadable(f"{error.error_count()} of its job definitions have a shape GitHub would reject")
def events(raw_on: object) -> frozenset[str]:
if isinstance(raw_on, Mapping):
return frozenset(str(key) for key in raw_on)
if isinstance(raw_on, str):
return frozenset({raw_on})
if isinstance(raw_on, Sequence):
return frozenset(str(event) for event in raw_on)
return frozenset()
def publishes_check_runs(raw_on: object) -> bool:
"""A `workflow_call`-only workflow posts its check runs through callers, never itself."""
return events(raw_on) != frozenset({"workflow_call"})
def scalar_list(value: object) -> tuple[str, ...] | Opaque:
"""One matrix key's values, or why the combinations it produces cannot be worked out."""
if not isinstance(value, Sequence) or isinstance(value, str):
return Opaque("a matrix key holds something other than a list of values")
if any(not isinstance(item, SCALAR) for item in value):
return Opaque("a matrix key lists values that are not plain scalars")
return tuple(scalar_text(item) for item in value)
def listed_values(matrix: Mapping[str, object]) -> tuple[tuple[str, tuple[str, ...]], ...] | Opaque:
listed: Final = tuple(
(str(key), scalar_list(values)) for key, values in matrix.items() if str(key) not in MATRIX_DIRECTIVES
)
opaque: Final = next((values for _, values in listed if isinstance(values, Opaque)), None)
if opaque is not None:
return opaque
return tuple((key, values) for key, values in listed if not isinstance(values, Opaque))
def directive_rows(matrix: Mapping[str, object], directive: str) -> tuple[Mapping[str, str], ...] | Opaque:
"""One `include` or `exclude` row, or why the combinations they shape cannot be worked out."""
rows: Final = matrix.get(directive)
if rows is None:
return ()
if not isinstance(rows, Sequence) or isinstance(rows, str):
return Opaque(f"a matrix `{directive}` is itself an expression rather than a list of rows")
mappings: Final = tuple(row for row in rows if isinstance(row, Mapping))
if len(mappings) != len(rows):
return Opaque(f"a matrix `{directive}` row is not a mapping of values")
if any(not isinstance(value, SCALAR) for row in mappings for value in row.values()):
return Opaque(f"a matrix `{directive}` row holds a value that is not a plain scalar")
return tuple(MappingProxyType({str(key): scalar_text(value) for key, value in row.items()}) for row in mappings)
def drops(row: Mapping[str, str], combination: Mapping[str, str]) -> bool:
"""GitHub removes a combination that carries every value one `exclude` row names."""
return all(combination.get(key) == value for key, value in row.items())
def extends(row: Mapping[str, str], combination: Mapping[str, str]) -> bool:
"""GitHub folds an `include` row into a combination only where it overwrites no listed value."""
return all(combination[key] == value for key, value in row.items() if key in combination)
def extended(combination: Mapping[str, str], rows: Sequence[Mapping[str, str]]) -> Mapping[str, str]:
additions: Final = {key: value for row in rows if extends(row, combination) for key, value in row.items()}
return MappingProxyType({**combination, **additions})
def crossed_values(listed: Sequence[tuple[str, tuple[str, ...]]]) -> tuple[Mapping[str, str], ...]:
if not listed:
return ()
return tuple(
MappingProxyType(dict(zip((key for key, _ in listed), values)))
for values in itertools.product(*(values for _, values in listed))
)
def matrix_combinations(job: Job) -> tuple[Mapping[str, str], ...] | Opaque:
"""One mapping per job the matrix produces, `exclude` applied before `include` as GitHub does."""
if not isinstance(job.strategy, Mapping):
return Opaque("its whole `strategy` comes from an expression")
matrix: Final = job.strategy.get("matrix")
if matrix is None:
return ()
if not isinstance(matrix, Mapping):
return Opaque("the matrix itself comes from an expression")
listed: Final = listed_values(matrix)
if isinstance(listed, Opaque):
return listed
rows: Final = directive_rows(matrix, "include")
if isinstance(rows, Opaque):
return rows
dropped: Final = directive_rows(matrix, "exclude")
if isinstance(dropped, Opaque):
return dropped
kept: Final = tuple(
combination for combination in crossed_values(listed) if not any(drops(row, combination) for row in dropped)
)
standalone: Final = tuple(row for row in rows if not any(extends(row, combination) for combination in kept))
return (*(extended(combination, rows) for combination in kept), *standalone)
def scanned(state: tuple[int, bool], char: str) -> tuple[int, bool]:
depth, quoted = state
if char == "'":
return depth, not quoted
if quoted:
return depth, quoted
return depth + int(char == "(") - int(char == ")"), quoted
def split_outside(text: str, token: str) -> tuple[str, ...]:
"""`text` cut on every `token` that sits outside quotes and parentheses."""
states: Final = tuple(itertools.accumulate(text, scanned, initial=(0, False)))
cuts: Final = tuple(
index
for index in range(len(text) - len(token) + 1)
if text.startswith(token, index) and states[index] == (0, False)
)
starts: Final = (0, *(cut + len(token) for cut in cuts))
return tuple(text[start:end] for start, end in zip(starts, (*cuts, len(text))))
def formatted(template: str, arguments: Sequence[str]) -> str | None:
"""GitHub's `format()` fills `{0}`-style holes and escapes braces, so anything richer resolves to nothing."""
residue: Final = GITHUB_PLACEHOLDER.sub("", template)
if "{" in residue or "}" in residue:
return None
try:
return template.format(*arguments)
except (IndexError, KeyError, ValueError):
return None
def value_of(text: str, values: Mapping[str, str]) -> str | None:
expression: Final = text.strip()
literal: Final = LITERAL.match(expression)
if literal is not None:
return literal.group("text")
reference: Final = MATRIX_REF.match(expression)
if reference is not None:
return values.get(reference.group("key"))
call: Final = FORMAT_CALL.match(expression)
if call is None:
return None
arguments: Final = tuple(value_of(part, values) for part in split_outside(call.group("args"), ","))
resolved: Final = tuple(argument for argument in arguments if argument is not None)
if not resolved or len(resolved) != len(arguments):
return None
return formatted(resolved[0], resolved[1:])
def holds(condition: str, values: Mapping[str, str]) -> bool | None:
comparison: Final = COMPARISON.match(condition.strip())
if comparison is None:
return None
left: Final = value_of(comparison.group("left"), values)
right: Final = value_of(comparison.group("right"), values)
if left is None or right is None:
return None
return (left == right) == (comparison.group("operator") == "==")
def evaluate(body: str, values: Mapping[str, str]) -> str | None:
"""The single string this expression yields, or None when its shape is not understood."""
branches: Final = tuple(split_outside(alternative, "&&") for alternative in split_outside(body, "||"))
outcomes: Final = tuple(tuple(holds(part, values) for part in branch[:-1]) for branch in branches)
if any(outcome is None for branch in outcomes for outcome in branch):
return None
taken: Final = next((branch[-1] for branch, outcome in zip(branches, outcomes) if all(outcome)), None)
return None if taken is None else value_of(taken, values)
def resolved_span(span: re.Match[str], values: Mapping[str, str]) -> str:
substitution: Final = evaluate(span.group("body"), values)
return span.group(0) if substitution is None else substitution
def rendered(template: str, values: Mapping[str, str]) -> str:
return EXPRESSION.sub(lambda span: resolved_span(span, values), template)
def comparable(name: str) -> bool:
"""A name still holding an expression is one GitHub resolves per job, so it is nothing to compare."""
return EXPRESSION.search(name) is None
def run_wide(name: str) -> bool:
"""A name whose leftover expressions one workflow run fills in the same way for every job in it."""
return all(RUN_WIDE.match(span.group("body").strip()) is not None for span in EXPRESSION.finditer(name))
def settled(names: Sequence[str]) -> Names:
unresolved: Final = tuple(name for name in names if not comparable(name))
return Names(
tuple(name for name in names if comparable(name)),
tuple(f"its name stays `{name}`" for name in unresolved if not run_wide(name)),
tuple(name for name in unresolved if run_wide(name)),
)
def expand(template: str, job: Job) -> Names:
combinations: Final = matrix_combinations(job)
if isinstance(combinations, Opaque):
return Names((), (combinations.reason,))
over: Final = combinations or (NO_MATRIX,)
return settled(tuple(rendered(template, values) for values in over))
def suffixed(job_id: str, combination: Mapping[str, str]) -> str:
"""The name GitHub gives a job with no `name:`, its id plus the combination it runs."""
return f"{job_id} ({', '.join(combination.values())})" if combination else job_id
def published_names(job_id: str, job: Job) -> Names:
if job.name is not None:
return expand(scalar_text(job.name), job)
combinations: Final = matrix_combinations(job)
if isinstance(combinations, Opaque):
return Names((), (combinations.reason,))
suffixes: Final = tuple(dict.fromkeys(suffixed(job_id, values) for values in combinations))
return Names(suffixes or (job_id,))
def callee_path(job: Job) -> str | None:
if job.uses is None or not job.uses.startswith(LOCAL_CALL_PREFIX):
return None
return job.uses[len(LOCAL_CALL_PREFIX) :].split("@")[0]
def joined(groups: Sequence[Names]) -> Names:
return Names(
tuple(name for group in groups for name in group.known),
tuple(reason for group in groups for reason in group.unknown),
tuple(name for group in groups for name in group.local),
)
def tagged(names: Names) -> tuple[tuple[str, bool], ...]:
"""Each name a job publishes beside whether only its own workflow's run settles it."""
return (*((name, False) for name in names.known), *((name, True) for name in names.local))
def call_blocker(job: Job, workflows: Mapping[str, Workflow], callers: frozenset[str]) -> str | None:
path: Final = callee_path(job)
if path is None:
return "it calls a reusable workflow outside this repository"
if path in callers:
return f"its call to {path} loops back on itself"
return None if path in workflows else f"it calls {path}, which this checkout does not hold"
def job_names(job_id: str, job: Job, workflows: Mapping[str, Workflow], callers: frozenset[str] = NO_CALLERS) -> Names:
prefixes: Final = published_names(job_id, job)
if job.uses is None:
return prefixes
blocker: Final = call_blocker(job, workflows, callers)
if blocker is not None:
return Names((), (*prefixes.unknown, blocker))
path: Final = callee_path(job) or ""
suffixes: Final = joined(
tuple(
job_names(callee_id, callee_job, workflows, callers | {path})
for callee_id, callee_job in workflows[path].jobs.items()
)
)
composed: Final = tuple(
(f"{prefix} / {suffix}", prefix_local or suffix_local)
for prefix, prefix_local in tagged(prefixes)
for suffix, suffix_local in tagged(suffixes)
)
return Names(
tuple(name for name, is_local in composed if not is_local),
(*prefixes.unknown, *suffixes.unknown),
tuple(name for name, is_local in composed if is_local),
)
def readable(sources: Mapping[str, str]) -> Mapping[str, tuple[Workflow, object]]:
parsed: Final = {rel: parse(source) for rel, source in sources.items()}
return MappingProxyType({rel: entry for rel, entry in parsed.items() if not isinstance(entry, Unreadable)})
def unreadable(sources: Mapping[str, str]) -> tuple[str, ...]:
parsed: Final = {rel: parse(source) for rel, source in sources.items()}
return tuple(
f"{rel} sits in the workflows directory but {entry.reason}, so none of its jobs were checked."
for rel, entry in sorted(parsed.items())
if isinstance(entry, Unreadable)
)
def scanned_jobs(sources: Mapping[str, str]) -> Iterator[tuple[str, str, Names]]:
parsed: Final = readable(sources)
workflows: Final = {rel: workflow for rel, (workflow, _) in parsed.items()}
for rel, (workflow, raw_on) in parsed.items():
if not publishes_check_runs(raw_on):
continue
for job_id, job in workflow.jobs.items():
yield rel, job_id, job_names(job_id, job, workflows)
def published(sources: Mapping[str, str]) -> Iterator[tuple[str, str]]:
for rel, job_id, names in scanned_jobs(sources):
for name in names.known:
yield name, f"{rel} job `{job_id}`"
def blind_spots(sources: Mapping[str, str]) -> tuple[str, ...]:
"""Jobs whose published names GitHub decides at run time, which no offline sweep can compare."""
return tuple(
f"{rel} job `{job_id}` publishes a name this check cannot work out because {reason}."
for rel, job_id, names in scanned_jobs(sources)
for reason in sorted(names.unknown)
)
def owners_by_name(sources: Mapping[str, str]) -> Iterator[tuple[str, tuple[str, ...]]]:
for name, pairs in itertools.groupby(sorted(published(sources)), key=operator.itemgetter(0)):
yield name, tuple(owner for _, owner in pairs)
def clash(name: str, owners: Sequence[str]) -> str | None:
"""Why one name is ambiguous, whether two jobs carry it or one job repeats it over its matrix."""
jobs: Final = tuple(dict.fromkeys(owners))
if len(jobs) > 1:
return (
f"`{name}` is published by {len(jobs)} jobs: {', '.join(jobs)}. A required status check matching "
f"that name cannot say which job proves it; give one of them a distinct `name:` or job id."
)
if len(owners) > 1:
return (
f"`{name}` is published {len(owners)} times by {jobs[0]}, once per matrix combination. A required "
f"status check matching that name cannot say which run proves it; put a matrix value in its `name:`."
)
return None
def local_published(sources: Mapping[str, str]) -> Iterator[tuple[tuple[str, str], str]]:
"""Names their own workflow's run settles, keyed by the file whose run settles them."""
for rel, job_id, names in scanned_jobs(sources):
for name in names.local:
yield (rel, name), f"job `{job_id}`"
def local_clash(rel: str, name: str, owners: Sequence[str]) -> str | None:
"""Why one workflow's own run lands several of its jobs on one check run."""
if len(owners) < 2:
return None
jobs: Final = tuple(dict.fromkeys(owners))
return (
f"`{name}` is published {len(owners)} times inside {rel}, by {', '.join(jobs)}. One run fills that "
f"expression in the same way throughout, so they all land on one check run; make the names differ."
)
def local_clashes(sources: Mapping[str, str]) -> tuple[str, ...]:
grouped: Final = itertools.groupby(sorted(local_published(sources)), key=operator.itemgetter(0))
found: Final = tuple(local_clash(rel, name, tuple(owner for _, owner in pairs)) for (rel, name), pairs in grouped)
return tuple(message for message in found if message is not None)
def collisions(sources: Mapping[str, str]) -> tuple[str, ...]:
found: Final = tuple(clash(name, owners) for name, owners in owners_by_name(sources))
return (*(message for message in found if message is not None), *local_clashes(sources))
def workflow_sources() -> Mapping[str, str]:
"""Repo-relative posix paths to text, the keys `uses: ./...` resolves against."""
return {path.relative_to(REPO_ROOT).as_posix(): path.read_text() for path in sorted(WORKFLOWS_DIR.glob("*.y*ml"))}
def report(header: str, problems: Sequence[str]) -> None:
if problems:
print(f"ERROR: {header}:\n - " + "\n - ".join(problems), file=sys.stderr)
def exit_code(sources: Mapping[str, str]) -> int:
unread: Final = unreadable(sources)
found: Final = collisions(sources)
blind: Final = blind_spots(sources)
if blind:
print("NOTE: names left out of the comparison:\n - " + "\n - ".join(blind))
report("Some workflows could not be read", unread)
report("Check-run names are not unique", found)
if unread or found:
return 1
print(f"Check-run names are unique across {len(sources)} workflows")
return 0
def main() -> int:
return exit_code(workflow_sources())
if __name__ == "__main__":
sys.exit(main())

View file

@ -0,0 +1,880 @@
from typing import Final
from check_workflow_job_name_collisions import (
Unreadable,
blind_spots,
callee_path,
collisions,
exit_code,
parse,
published,
unreadable,
workflow_sources,
)
REUSABLE_BASE: Final = """on:
workflow_call:
jobs:
run:
name: >-
${{ matrix.python-version == '3.12' && 'Run tests'
|| format('Run tests (Python {0})', matrix.python-version) }}
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.12", "3.13"]
"""
SHARD_CALLER: Final = """on: pull_request
jobs:
unit:
name: ${{ matrix.shard }}
uses: ./.github/workflows/base.yml
strategy:
matrix:
include:
- shard: core-utils
"""
CORRELATED_ROWS: Final = """on: pull_request
jobs:
unit:
name: ${{ matrix.shard }} on ${{ matrix.test-path }}
runs-on: ubuntu-latest
strategy:
matrix:
include:
- shard: core-utils
test-path: tests/core
- shard: proxy
test-path: tests/proxy
"""
LISTED_PLUS_ROW: Final = """on: pull_request
jobs:
unit:
name: ${{ matrix.python-version }} ${{ matrix.label }}
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.12", "3.13"]
include:
- label: fast
"""
NAMELESS_MATRIX: Final = """on: pull_request
jobs:
build:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.12", "3.13"]
"""
EXCLUDED_PAIR: Final = """on: pull_request
jobs:
unit:
name: ${{ matrix.os }}-${{ matrix.python-version }}
runs-on: ubuntu-latest
strategy:
matrix:
os: [ubuntu, macos]
python-version: ["3.12", "3.13"]
exclude:
- os: macos
python-version: "3.13"
"""
EXCLUDED_KEY: Final = """on: pull_request
jobs:
unit:
name: ${{ matrix.os }}-${{ matrix.python-version }}
runs-on: ubuntu-latest
strategy:
matrix:
os: [ubuntu, macos]
python-version: ["3.12", "3.13"]
exclude:
- os: macos
"""
BOOLEAN_MATRIX: Final = """on: pull_request
jobs:
unit:
name: cache ${{ matrix.cached }}
runs-on: ubuntu-latest
strategy:
matrix:
cached: [true, false]
"""
UNFILLABLE_FORMAT: Final = """on: pull_request
jobs:
unit:
name: ${{ format('{0} {1}', matrix.shard) }}
runs-on: ubuntu-latest
strategy:
matrix:
shard: [core]
"""
def test_every_workflow_in_the_repo_publishes_a_unique_check_run_name() -> None:
assert collisions(workflow_sources()) == ()
def test_every_workflow_in_the_repo_parses_into_jobs() -> None:
unparsed: Final = tuple(
rel
for rel, source in workflow_sources().items()
if isinstance(entry := parse(source), Unreadable) or not entry[0].jobs
)
assert unparsed == ()
def test_every_local_reusable_call_in_the_repo_resolves_to_a_workflow() -> None:
sources: Final = workflow_sources()
parsed: Final = tuple(entry for text in sources.values() if not isinstance(entry := parse(text), Unreadable))
unresolved: Final = tuple(
job.uses
for workflow, _ in parsed
for job in workflow.jobs.values()
if (callee := callee_path(job)) is not None and callee not in sources
)
assert unresolved == ()
def test_two_jobs_falling_back_to_the_same_job_id_collide() -> None:
sources: Final = {
"a.yml": "on: pull_request\njobs:\n test:\n runs-on: ubuntu-latest\n",
"b.yml": "on: pull_request\njobs:\n test:\n runs-on: ubuntu-latest\n",
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`test` is published by 2 jobs" in found[0]
assert "a.yml job `test`" in found[0] and "b.yml job `test`" in found[0]
def test_an_explicit_name_overrides_the_job_id_and_clears_the_collision() -> None:
sources: Final = {
"a.yml": "on: pull_request\njobs:\n test:\n name: Sweep tests\n runs-on: ubuntu-latest\n",
"b.yml": "on: pull_request\njobs:\n test:\n runs-on: ubuntu-latest\n",
}
assert collisions(sources) == ()
def test_an_explicit_name_matching_another_job_id_collides() -> None:
sources: Final = {
"a.yml": "on: pull_request\njobs:\n sweep:\n name: test\n runs-on: ubuntu-latest\n",
"b.yml": "on: pull_request\njobs:\n test:\n runs-on: ubuntu-latest\n",
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`test` is published by 2 jobs" in found[0]
def test_two_callers_of_one_reusable_workflow_collide_on_a_shared_matrix_value() -> None:
base: Final = "on:\n workflow_call:\njobs:\n run:\n name: Run tests\n runs-on: ubuntu-latest\n"
caller: Final = (
"on: pull_request\n"
"jobs:\n"
" {job}:\n"
" name: ${{{{ matrix.shard }}}}\n"
" uses: ./.github/workflows/base.yml\n"
" strategy:\n"
" matrix:\n"
" include:\n"
" - shard: {shard}\n"
)
sources: Final = {
".github/workflows/base.yml": base,
"unit.yml": caller.format(job="unit", shard="proxy-auth"),
"proxy-db.yml": caller.format(job="proxy-db", shard="proxy-auth"),
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`proxy-auth / Run tests` is published by 2 jobs" in found[0]
def test_distinct_matrix_values_through_one_reusable_workflow_do_not_collide() -> None:
base: Final = "on:\n workflow_call:\njobs:\n run:\n name: Run tests\n runs-on: ubuntu-latest\n"
caller: Final = (
"on: pull_request\n"
"jobs:\n"
" {job}:\n"
" name: ${{{{ matrix.shard }}}}\n"
" uses: ./.github/workflows/base.yml\n"
" strategy:\n"
" matrix:\n"
" include:\n"
" - shard: {shard}\n"
)
sources: Final = {
".github/workflows/base.yml": base,
"unit.yml": caller.format(job="unit", shard="proxy-auth"),
"proxy-db.yml": caller.format(job="proxy-db", shard="budgets"),
}
assert collisions(sources) == ()
def test_a_reusable_caller_does_not_collide_with_a_plain_job_of_the_same_name() -> None:
sources: Final = {
".github/workflows/base.yml": (
"on:\n workflow_call:\njobs:\n run:\n name: Run tests\n runs-on: ubuntu-latest\n"
),
"unit.yml": (
"on: pull_request\n"
"jobs:\n"
" unit:\n"
" name: ${{ matrix.shard }}\n"
" uses: ./.github/workflows/base.yml\n"
" strategy:\n"
" matrix:\n"
" include:\n"
" - shard: proxy-behavior\n"
),
"postgres.yml": (
"on: pull_request\n"
"jobs:\n"
" postgres:\n"
" name: ${{ matrix.shard }}\n"
" runs-on: ubuntu-latest\n"
" strategy:\n"
" matrix:\n"
" include:\n"
" - shard: proxy-behavior\n"
),
}
assert collisions(sources) == ()
def test_a_workflow_call_only_workflow_publishes_nothing_of_its_own() -> None:
sources: Final = {
"base.yml": "on:\n workflow_call:\njobs:\n run:\n runs-on: ubuntu-latest\n",
"other.yml": "on:\n workflow_call:\njobs:\n run:\n runs-on: ubuntu-latest\n",
}
assert collisions(sources) == ()
def test_a_workflow_call_workflow_that_also_runs_on_pull_request_still_publishes() -> None:
sources: Final = {
"base.yml": "on:\n workflow_call:\n pull_request:\njobs:\n run:\n runs-on: ubuntu-latest\n",
"other.yml": "on: pull_request\njobs:\n run:\n runs-on: ubuntu-latest\n",
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`run` is published by 2 jobs" in found[0]
def test_a_matrix_list_supplies_values_the_same_way_include_rows_do() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\n"
"jobs:\n"
" build:\n"
" name: Analyze (${{ matrix.language }})\n"
" runs-on: ubuntu-latest\n"
" strategy:\n"
" matrix:\n"
" language: [python, go]\n"
),
"b.yml": "on: pull_request\njobs:\n go:\n name: Analyze (go)\n runs-on: ubuntu-latest\n",
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`Analyze (go)` is published by 2 jobs" in found[0]
def test_two_workflows_sharing_a_run_wide_template_are_not_called_a_collision() -> None:
template: Final = (
"on: pull_request\njobs:\n {job}:\n name: ${{{{ github.event_name }}}}-build\n runs-on: ubuntu-latest\n"
)
sources: Final = {
"a.yml": template.format(job="one"),
"b.yml": template.format(job="two"),
}
assert collisions(sources) == ()
assert blind_spots(sources) == ()
def test_two_jobs_of_one_workflow_sharing_a_run_wide_template_are_a_collision() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n"
" one:\n name: ${{ github.event_name }}-build\n runs-on: ubuntu-latest\n"
" two:\n name: ${{ github.event_name }}-build\n runs-on: ubuntu-latest\n"
),
}
found: Final = collisions(sources)
assert len(found) == 1
assert "is published 2 times inside a.yml, by job `one`, job `two`" in found[0]
assert exit_code(sources) == 1
def test_a_run_wide_template_carrying_a_matrix_value_does_not_collide_inside_one_workflow() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n"
" one:\n name: ${{ github.event_name }}-${{ matrix.shard }}\n runs-on: ubuntu-latest\n"
" strategy:\n matrix:\n shard: [core, extras]\n"
),
}
assert collisions(sources) == ()
assert blind_spots(sources) == ()
def test_a_run_wide_name_repeated_over_a_matrix_by_one_job_is_a_collision() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n"
" one:\n name: ${{ github.event_name }}-build\n runs-on: ubuntu-latest\n"
" strategy:\n matrix:\n shard: [core, extras]\n"
),
}
found: Final = collisions(sources)
assert len(found) == 1
assert "is published 2 times inside a.yml, by job `one`" in found[0]
def test_a_name_reading_the_job_it_sits_in_stays_out_of_the_comparison() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n"
" one:\n name: ${{ github.job }}-build\n runs-on: ubuntu-latest\n"
" two:\n name: ${{ github.job }}-build\n runs-on: ubuntu-latest\n"
),
}
assert collisions(sources) == ()
assert len(blind_spots(sources)) == 2
def test_a_run_wide_caller_name_collides_through_the_workflow_it_calls() -> None:
sources: Final = {
".github/workflows/a.yml": (
"on: pull_request\njobs:\n"
" one:\n name: ${{ github.event_name }}\n uses: ./.github/workflows/c.yml\n"
" two:\n name: ${{ github.event_name }}\n uses: ./.github/workflows/c.yml\n"
),
".github/workflows/c.yml": "on:\n workflow_call:\njobs:\n build:\n runs-on: ubuntu-latest\n",
}
found: Final = collisions(sources)
assert len(found) == 1
assert "github.event_name }} / build` is published 2 times inside .github/workflows/a.yml" in found[0]
def test_a_run_wide_name_inside_a_called_workflow_collides_under_the_caller() -> None:
sources: Final = {
".github/workflows/a.yml": ("on: pull_request\njobs:\n one:\n uses: ./.github/workflows/c.yml\n"),
".github/workflows/c.yml": (
"on:\n workflow_call:\njobs:\n"
" build:\n name: ${{ github.event_name }}\n runs-on: ubuntu-latest\n"
" lint:\n name: ${{ github.event_name }}\n runs-on: ubuntu-latest\n"
),
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`one / ${{ github.event_name }}` is published 2 times inside .github/workflows/a.yml" in found[0]
def test_a_name_reading_the_workflow_it_sits_in_is_not_called_a_collision() -> None:
template: Final = (
"on: pull_request\njobs:\n {job}:\n name: ${{{{ github.workflow }}}} / build\n runs-on: ubuntu-latest\n"
)
sources: Final = {
"a.yml": template.format(job="one"),
"b.yml": template.format(job="two"),
}
assert collisions(sources) == ()
assert exit_code(sources) == 0
def test_a_format_call_python_accepts_but_github_does_not_publishes_nothing_to_compare() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n one:\n name: ${{ format('{0.real}', matrix.shard) }}\n"
" runs-on: ubuntu-latest\n strategy:\n matrix:\n shard: [core]\n"
),
}
assert collisions(sources) == ()
assert blind_spots(sources) != ()
def test_a_format_call_padding_its_argument_publishes_nothing_to_compare() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n one:\n name: ${{ format('{0:>8}', matrix.shard) }}\n"
" runs-on: ubuntu-latest\n strategy:\n matrix:\n shard: [core]\n"
),
"b.yml": "on: pull_request\njobs:\n two:\n name: ' core'\n runs-on: ubuntu-latest\n",
}
assert collisions(sources) == ()
assert blind_spots(sources) != ()
def test_an_exclude_row_that_is_not_a_mapping_is_reported_rather_than_skipped() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n build:\n runs-on: ubuntu-latest\n"
" strategy:\n matrix:\n v: [1, 2]\n exclude:\n - oops\n"
),
"b.yml": "on: pull_request\njobs:\n other:\n name: build (1)\n runs-on: ubuntu-latest\n",
}
assert collisions(sources) == ()
assert blind_spots(sources) != ()
def test_an_exclude_row_holding_a_non_scalar_never_drops_every_combination() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n build:\n runs-on: ubuntu-latest\n"
" strategy:\n matrix:\n v: [1, 2]\n exclude:\n - cfg: {k: 1}\n"
),
"b.yml": "on: pull_request\njobs:\n build:\n runs-on: ubuntu-latest\n",
}
assert collisions(sources) == ()
assert blind_spots(sources) != ()
def test_two_jobs_sharing_a_template_that_reads_per_job_are_not_called_a_collision() -> None:
template: Final = (
"on: pull_request\njobs:\n {job}:\n name: ${{{{ matrix.shard }}}}\n runs-on: ubuntu-latest\n"
)
sources: Final = {
"a.yml": template.format(job="one"),
"b.yml": template.format(job="two"),
}
assert collisions(sources) == ()
assert len(blind_spots(sources)) == 2
def test_a_file_that_is_not_a_workflow_is_reported_rather_than_skipped() -> None:
sources: Final = {
"notes.yml": "just a string\n",
"a.yml": "on: pull_request\njobs:\n test:\n runs-on: ubuntu-latest\n",
}
found: Final = unreadable(sources)
assert len(found) == 1
assert "notes.yml" in found[0]
assert collisions(sources) == ()
def test_a_workflow_holding_a_job_shape_github_would_reject_is_reported() -> None:
sources: Final = {"a.yml": "on: pull_request\njobs:\n test:\n uses: [not, a, string]\n"}
found: Final = unreadable(sources)
assert len(found) == 1
assert "a.yml" in found[0]
def test_a_conditional_name_expands_to_the_branch_each_matrix_value_takes() -> None:
names: Final = frozenset(
name for name, _ in published({".github/workflows/base.yml": REUSABLE_BASE, "unit.yml": SHARD_CALLER})
)
assert names == frozenset({"core-utils / Run tests", "core-utils / Run tests (Python 3.13)"})
def test_a_conditional_name_never_publishes_the_branch_its_condition_rules_out() -> None:
names: Final = frozenset(
name for name, _ in published({".github/workflows/base.yml": REUSABLE_BASE, "unit.yml": SHARD_CALLER})
)
assert "core-utils / Run tests (Python 3.12)" not in names
def test_a_conditional_reusable_name_collides_with_a_plain_job_publishing_the_same_name() -> None:
sources: Final = {
".github/workflows/base.yml": REUSABLE_BASE,
"unit.yml": SHARD_CALLER,
"postgres.yml": ("on: pull_request\njobs:\n legacy:\n name: core-utils / Run tests\n"),
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`core-utils / Run tests` is published by 2 jobs" in found[0]
def test_a_name_reading_two_matrix_keys_publishes_only_the_pairs_each_include_row_holds() -> None:
names: Final = frozenset(name for name, _ in published({"unit.yml": CORRELATED_ROWS}))
assert names == frozenset({"core-utils on tests/core", "proxy on tests/proxy"})
def test_a_name_reading_two_matrix_keys_never_publishes_a_pair_no_include_row_holds() -> None:
names: Final = frozenset(name for name, _ in published({"unit.yml": CORRELATED_ROWS}))
assert "core-utils on tests/proxy" not in names
assert "proxy on tests/core" not in names
def test_an_include_row_carrying_no_listed_key_extends_every_listed_combination() -> None:
names: Final = frozenset(name for name, _ in published({"unit.yml": LISTED_PLUS_ROW}))
assert names == frozenset({"3.12 fast", "3.13 fast"})
def test_every_workflow_in_the_repo_resolves_every_expression_in_its_job_names() -> None:
unresolved: Final = tuple(f"{owner}: {name}" for name, owner in published(workflow_sources()) if "${{" in name)
assert unresolved == ()
def test_a_matrix_job_with_no_name_publishes_the_id_and_values_github_appends() -> None:
names: Final = frozenset(name for name, _ in published({"a.yml": NAMELESS_MATRIX}))
assert names == frozenset({"build (3.12)", "build (3.13)"})
def test_a_matrix_job_with_no_name_does_not_collide_with_a_plain_job_carrying_its_id() -> None:
sources: Final = {
"a.yml": NAMELESS_MATRIX,
"b.yml": "on: pull_request\njobs:\n build:\n runs-on: ubuntu-latest\n",
}
assert collisions(sources) == ()
def test_a_matrix_job_with_no_name_collides_with_the_suffixed_name_github_writes() -> None:
sources: Final = {
"a.yml": NAMELESS_MATRIX,
"b.yml": "on: pull_request\njobs:\n legacy:\n name: build (3.13)\n runs-on: ubuntu-latest\n",
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`build (3.13)` is published by 2 jobs" in found[0]
def test_an_excluded_combination_publishes_no_check_run() -> None:
names: Final = frozenset(name for name, _ in published({"unit.yml": EXCLUDED_PAIR}))
assert names == frozenset({"ubuntu-3.12", "ubuntu-3.13", "macos-3.12"})
def test_an_exclude_row_naming_one_key_drops_every_combination_carrying_it() -> None:
names: Final = frozenset(name for name, _ in published({"unit.yml": EXCLUDED_KEY}))
assert names == frozenset({"ubuntu-3.12", "ubuntu-3.13"})
def test_a_boolean_matrix_value_renders_the_way_github_writes_it() -> None:
names: Final = frozenset(name for name, _ in published({"unit.yml": BOOLEAN_MATRIX}))
assert names == frozenset({"cache true", "cache false"})
def test_a_format_call_its_arguments_cannot_fill_publishes_nothing_to_compare() -> None:
sources: Final = {"unit.yml": UNFILLABLE_FORMAT}
assert frozenset(name for name, _ in published(sources)) == frozenset()
assert "its name stays" in blind_spots(sources)[0]
def test_a_call_to_a_workflow_outside_the_repo_is_reported_rather_than_guessed() -> None:
sources: Final = {
"a.yml": "on: pull_request\njobs:\n unit:\n uses: BerriAI/other/.github/workflows/base.yml@main\n",
"b.yml": "on: pull_request\njobs:\n unit:\n runs-on: ubuntu-latest\n",
}
assert frozenset(name for name, _ in published(sources)) == frozenset({"unit"})
assert collisions(sources) == ()
assert "outside this repository" in blind_spots(sources)[0]
def test_a_chain_of_local_reusable_calls_publishes_every_level_of_the_chain() -> None:
sources: Final = {
".github/workflows/leaf.yml": (
"on:\n workflow_call:\njobs:\n run:\n name: Leaf\n runs-on: ubuntu-latest\n"
),
".github/workflows/mid.yml": (
"on:\n workflow_call:\njobs:\n call:\n name: Mid\n uses: ./.github/workflows/leaf.yml\n"
),
"top.yml": "on: pull_request\njobs:\n top:\n name: Top\n uses: ./.github/workflows/mid.yml\n",
}
names: Final = frozenset(name for name, _ in published(sources))
assert names == frozenset({"Top / Mid / Leaf"})
def test_a_job_name_that_is_not_a_string_still_publishes_the_value_github_renders() -> None:
sources: Final = {
"a.yml": "on: pull_request\njobs:\n sweep:\n name: 2024\n runs-on: ubuntu-latest\n",
"b.yml": 'on: pull_request\njobs:\n other:\n name: "2024"\n runs-on: ubuntu-latest\n',
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`2024` is published by 2 jobs" in found[0]
def test_the_check_fails_when_a_file_in_the_workflows_directory_cannot_be_read() -> None:
assert exit_code({"notes.yml": "just a string\n"}) == 1
def test_the_check_fails_when_two_jobs_publish_one_check_run_name() -> None:
plain: Final = "on: pull_request\njobs:\n test:\n runs-on: ubuntu-latest\n"
assert exit_code({"a.yml": plain, "b.yml": plain}) == 1
def test_the_check_passes_when_every_file_reads_and_every_name_is_unique() -> None:
sources: Final = {
"a.yml": "on: pull_request\njobs:\n test:\n runs-on: ubuntu-latest\n",
"b.yml": "on: pull_request\njobs:\n sweep:\n runs-on: ubuntu-latest\n",
}
assert exit_code(sources) == 0
def test_two_callers_of_one_reusable_workflow_named_from_its_inputs_do_not_collide() -> None:
sources: Final = {
".github/workflows/callee.yml": (
"on:\n workflow_call:\njobs:\n run:\n name: ${{ inputs.suite }}\n runs-on: ubuntu-latest\n"
),
"caller.yml": (
"on: pull_request\njobs:\n"
" alpha:\n name: A\n uses: ./.github/workflows/callee.yml\n with:\n suite: alpha\n"
" beta:\n name: A\n uses: ./.github/workflows/callee.yml\n with:\n suite: beta\n"
),
}
assert collisions(sources) == ()
assert len(blind_spots(sources)) == 2
def test_a_matrix_that_is_itself_an_expression_never_collapses_onto_the_bare_job_id() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n build:\n strategy:\n"
" matrix: ${{ fromJson(needs.plan.outputs.matrix) }}\n runs-on: ubuntu-latest\n"
),
"b.yml": "on: pull_request\njobs:\n build:\n runs-on: ubuntu-latest\n",
}
assert frozenset(name for name, _ in published(sources)) == frozenset({"build"})
assert collisions(sources) == ()
assert "the matrix itself comes from an expression" in blind_spots(sources)[0]
def test_a_matrix_listing_objects_never_collapses_onto_the_bare_job_id() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n build:\n strategy:\n matrix:\n target:\n"
" - os: ubuntu\n - os: windows\n runs-on: ubuntu-latest\n"
),
"b.yml": "on: pull_request\njobs:\n build:\n runs-on: ubuntu-latest\n",
}
assert frozenset(name for name, _ in published(sources)) == frozenset({"build"})
assert collisions(sources) == ()
assert "not plain scalars" in blind_spots(sources)[0]
def test_a_call_to_a_workflow_file_the_checkout_does_not_hold_is_reported() -> None:
sources: Final = {"a.yml": "on: pull_request\njobs:\n unit:\n uses: ./.github/workflows/gone.yml\n"}
assert collisions(sources) == ()
assert "which this checkout does not hold" in blind_spots(sources)[0]
def test_reusable_workflows_calling_each_other_in_a_loop_are_reported_not_followed() -> None:
sources: Final = {
".github/workflows/a.yml": (
"on:\n workflow_call:\njobs:\n call:\n name: A\n uses: ./.github/workflows/b.yml\n"
),
".github/workflows/b.yml": (
"on:\n workflow_call:\njobs:\n call:\n name: B\n uses: ./.github/workflows/a.yml\n"
),
"top.yml": "on: pull_request\njobs:\n top:\n name: Top\n uses: ./.github/workflows/a.yml\n",
}
assert collisions(sources) == ()
assert any("loops back on itself" in spot for spot in blind_spots(sources))
def test_a_caller_still_publishes_the_callee_jobs_it_can_read() -> None:
sources: Final = {
".github/workflows/callee.yml": (
"on:\n workflow_call:\njobs:\n"
" lint:\n name: Lint\n runs-on: ubuntu-latest\n"
" suite:\n name: ${{ inputs.suite }}\n runs-on: ubuntu-latest\n"
),
"caller.yml": "on: pull_request\njobs:\n call:\n name: A\n uses: ./.github/workflows/callee.yml\n",
}
assert frozenset(name for name, _ in published(sources)) == frozenset({"A / Lint"})
assert len(blind_spots(sources)) == 1
def test_a_name_the_check_cannot_work_out_is_reported_without_failing_the_check() -> None:
sources: Final = {
"a.yml": "on: pull_request\njobs:\n unit:\n uses: BerriAI/other/.github/workflows/base.yml@main\n",
}
assert blind_spots(sources) != ()
assert exit_code(sources) == 0
def test_a_caller_whose_own_name_is_unreadable_publishes_none_of_its_callee_names() -> None:
sources: Final = {
".github/workflows/callee.yml": (
"on:\n workflow_call:\njobs:\n lint:\n name: Lint\n runs-on: ubuntu-latest\n"
),
"caller.yml": (
"on: pull_request\njobs:\n call:\n name: ${{ matrix.suite }}\n"
" uses: ./.github/workflows/callee.yml\n"
),
"other.yml": "on: pull_request\njobs:\n plain:\n name: Lint\n runs-on: ubuntu-latest\n",
}
assert frozenset(name for name, _ in published(sources)) == frozenset({"Lint"})
assert collisions(sources) == ()
assert "its name stays" in blind_spots(sources)[0]
def test_an_include_row_naming_a_listed_key_extends_only_the_combinations_it_matches() -> None:
sources: Final = {
"unit.yml": (
"on: pull_request\njobs:\n unit:\n"
" name: ${{ matrix.python-version }} ${{ matrix.label }}\n"
" runs-on: ubuntu-latest\n strategy:\n matrix:\n"
' python-version: ["3.12", "3.13"]\n'
" include:\n"
' - python-version: "3.12"\n'
" label: fast\n"
)
}
assert frozenset(name for name, _ in published(sources)) == frozenset({"3.12 fast"})
assert len(blind_spots(sources)) == 1
def test_a_job_whose_whole_strategy_is_an_expression_is_reported_rather_than_rejecting_the_file() -> None:
sources: Final = {
"plan.yml": (
"on: pull_request\njobs:\n plan:\n name: Plan\n runs-on: ubuntu-latest\n"
" fan:\n strategy: ${{ fromJSON(needs.plan.outputs.strategy) }}\n runs-on: ubuntu-latest\n"
)
}
assert unreadable(sources) == ()
assert frozenset(name for name, _ in published(sources)) == frozenset({"Plan"})
assert "`strategy` comes from an expression" in blind_spots(sources)[0]
assert exit_code(sources) == 0
def test_one_job_publishing_one_name_for_every_matrix_combination_is_a_collision() -> None:
sources: Final = {
"unit.yml": (
"on: pull_request\njobs:\n build:\n name: Run tests\n runs-on: ubuntu-latest\n"
' strategy:\n matrix:\n python-version: ["3.12", "3.13"]\n'
)
}
found: Final = collisions(sources)
assert len(found) == 1
assert "`Run tests` is published 2 times by unit.yml job `build`" in found[0]
assert exit_code(sources) == 1
def test_a_name_carrying_a_matrix_value_publishes_one_name_per_combination_without_colliding() -> None:
sources: Final = {
"unit.yml": (
"on: pull_request\njobs:\n build:\n name: Run tests ${{ matrix.python-version }}\n"
' runs-on: ubuntu-latest\n strategy:\n matrix:\n python-version: ["3.12", "3.13"]\n'
)
}
assert frozenset(name for name, _ in published(sources)) == frozenset({"Run tests 3.12", "Run tests 3.13"})
assert collisions(sources) == ()
assert exit_code(sources) == 0
def test_a_file_that_is_not_valid_yaml_is_reported_rather_than_raising() -> None:
sources: Final = {"broken.yml": "jobs:\n build: [\n"}
assert unreadable(sources) == (
"broken.yml sits in the workflows directory but it does not read as one YAML "
"document, so none of its jobs were checked.",
)
assert exit_code(sources) == 1
def test_a_file_holding_two_yaml_documents_is_reported_rather_than_raising() -> None:
sources: Final = {"two.yml": "on: pull_request\n---\non: push\n"}
assert len(unreadable(sources)) == 1
assert exit_code(sources) == 1
def test_an_exclude_that_is_itself_an_expression_is_reported_rather_than_ignored() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n one:\n name: build\n runs-on: ubuntu-latest\n"
" strategy:\n matrix:\n python: ['3.11', '3.12']\n"
" exclude: ${{ fromJson(vars.SKIP) }}\n"
),
}
found: Final = blind_spots(sources)
assert collisions(sources) == ()
assert len(found) == 1
assert "a matrix `exclude` is itself an expression" in found[0]
def test_an_include_that_is_itself_an_expression_is_reported_rather_than_ignored() -> None:
sources: Final = {
"a.yml": (
"on: pull_request\njobs:\n one:\n name: build-${{ matrix.python }}\n runs-on: ubuntu-latest\n"
" strategy:\n matrix:\n python: ['3.11']\n"
" include: ${{ fromJson(vars.EXTRA) }}\n"
),
}
found: Final = blind_spots(sources)
assert collisions(sources) == ()
assert len(found) == 1
assert "a matrix `include` is itself an expression" in found[0]