from __future__ import annotations
from pathlib import Path
import pytest
import yaml
from workflow_reader import Workflow, load_workflows, pull_request_closure
TRIGGER_SPELLINGS = [
"on: pull_request",
"'on': pull_request",
"on: [push, pull_request]",
"'on': [push, pull_request]",
"on:\n pull_request:\n branches: [main]",
"'on':\n pull_request:",
]
@pytest.mark.parametrize("trigger", TRIGGER_SPELLINGS)
def test_every_trigger_spelling_reads_as_a_pull_request_lane(trigger: str) -> None:
flow = Workflow.parse("x.yml", f"{trigger}\njobs: {{}}\n")
assert flow.on_pull_request, flow.triggers
@pytest.mark.parametrize(
"text",
[
"on: push\n'on': pull_request\njobs: {}\n",
"on: 3\njobs: {}\n",
"jobs: {}\n",
],
ids=["both keys", "a number", "no trigger"],
)
def test_an_unreadable_trigger_is_refused(text: str) -> None:
with pytest.raises(ValueError, match="trigger"):
Workflow.parse("x.yml", text)
def test_a_key_declared_twice_is_refused() -> None:
text = (
"on: push\njobs:\n a:\n runs-on: ubicloud-standard-2\n"
" runs-on: ubuntu-latest\n"
)
with pytest.raises(yaml.constructor.ConstructorError, match="duplicate key"):
Workflow.parse("x.yml", text)
@pytest.mark.parametrize(
("text", "error"),
[("on: [push\n", yaml.YAMLError), ("- on: push\n", TypeError)],
ids=["not YAML", "not a mapping"],
)
def test_a_file_that_is_not_a_workflow_is_refused(
text: str, error: type[Exception]
) -> None:
with pytest.raises(error):
Workflow.parse("x.yml", text)
def test_an_upper_case_suffix_is_loaded(tmp_path: Path) -> None:
directory = tmp_path / ".github" / "workflows"
directory.mkdir(parents=True)
(directory / "LANE.YML").write_text("on: pull_request\njobs: {}\n")
(directory / "notes.txt").write_text("not a workflow\n")
assert [flow.path for flow in load_workflows(tmp_path)] == [
".github/workflows/LANE.YML"
]
def call(path: str, target: str, trigger: str = "workflow_call") -> Workflow:
return Workflow.parse(path, f"on: {trigger}\njobs: {{x: {{uses: {target}}}}}\n")
def test_the_closure_follows_calls_transitively_and_stops_on_a_cycle() -> None:
flows = [
call(".github/workflows/a.yml", "./.github/workflows/b.yml", "pull_request"),
call(".github/workflows/b.yml", "$/.github/workflows/c.yml"),
call(".github/workflows/c.yml", ".github/workflows/b.yml"),
call(".github/workflows/d.yml", "./.github/workflows/a.yml"),
]
reached = [flow.path for flow in pull_request_closure(flows)]
assert reached == [f".github/workflows/{name}.yml" for name in "abc"]
def test_a_call_to_a_missing_workflow_is_refused() -> None:
flows = [
call(".github/workflows/a.yml", "./.github/workflows/gone.yml", "pull_request")
]
with pytest.raises(AssertionError, match="gone.yml"):
pull_request_closure(flows)
@pytest.mark.parametrize(
"uses",
["./.github/workflows/b.yml@main", "$/.github/workflows/b.yml@v1"],
)
def test_a_local_call_carrying_a_ref_is_refused(uses: str) -> None:
flow = call(".github/workflows/a.yml", uses, "pull_request")
with pytest.raises(ValueError, match="carries a ref"):
flow.callees()
def test_a_remote_workflow_is_not_a_local_call() -> None:
flow = call(
".github/workflows/a.yml",
"leynos/shared-actions/.github/workflows/x.yml@abc",
"pull_request",
)
assert flow.callees() == []