import pathlib
from fractions import Fraction
import pytest
from coverage_lanes import (
WorkflowReadError,
WorkflowValueError,
coverage_jobs_in,
load_workflow_documents,
)
from nextest_budgets import (
global_timeout,
grace_period,
largest_test_allowance,
termination_allowance,
)
from timeout_budgets import (
CEILING_MARGIN_SECONDS,
COVERAGE_ACTION,
NEXTEST_DEFAULT_GRACE_PERIOD_SECONDS,
TERMINATION_SAFETY_MARGIN_SECONDS,
NextestConfigurationError,
UnboundedTestError,
required_ceiling,
seconds,
)
def _profile(*lines: str) -> str:
return "\n".join(("[profile.default]", *lines)) + "\n"
@pytest.mark.parametrize(
("config_text", "expected"),
[
pytest.param(
_profile('slow-timeout = { period = "180s", terminate-after = 1 }'),
Fraction(180),
id="a-single-period",
),
pytest.param(
_profile('slow-timeout = { period = "60s", terminate-after = 5 }'),
Fraction(300),
id="five-warning-periods",
),
pytest.param(
_profile('slow-timeout = { period = "2m", terminate-after = 3 }'),
Fraction(360),
id="minutes-times-three",
),
pytest.param(
_profile(
'slow-timeout = { period = "30s", terminate-after = 2, '
'grace-period = "5s" }',
"",
"[[profile.default.overrides]]",
'filter = "binary(=ui)"',
'slow-timeout = { period = "60s", terminate-after = 1 }',
),
Fraction(60),
id="the-largest-of-several",
),
pytest.param(
_profile(
'slow-timeout = { period = "30s", terminate-after = 1 }',
"",
"[[profile.default.overrides]]",
'filter = "binary(=slow_timeout_probe)"',
'slow-timeout = { period = "45s", terminate-after = 1 }',
),
Fraction(45),
id="a-filter-naming-the-key-is-not-a-budget",
),
],
)
def test_the_largest_per_test_allowance_counts_the_multiplier(
config_text: str, expected: Fraction
) -> None:
assert largest_test_allowance(config_text) == expected, (
f"{config_text!r} must yield a {float(expected):.0f}s largest per-test "
f"allowance; terminate-after scales the period"
)
@pytest.mark.parametrize(
"config_text",
[
pytest.param(_profile('slow-timeout = "2m"'), id="a-bare-duration"),
pytest.param(
_profile('slow-timeout = { period = "2m" }'),
id="a-table-without-terminate-after",
),
pytest.param(
_profile('slow-timeout = { period = "2m", grace-period = "5s" }'),
id="a-table-with-only-a-grace-period",
),
],
)
def test_a_slow_timeout_that_never_terminates_is_refused(config_text: str) -> None:
with pytest.raises(UnboundedTestError, match=r"terminate-after"):
largest_test_allowance(config_text)
def test_a_commented_out_slow_timeout_is_not_a_budget() -> None:
config_text = _profile(
'# slow-timeout = { period = "30s", terminate-after = 1 }',
'slow-timeout = { period = "180s", terminate-after = 1 }',
)
assert largest_test_allowance(config_text) == Fraction(180), (
"a commented-out slow-timeout was read as a live one"
)
with pytest.raises(NextestConfigurationError, match=r"no slow-timeout"):
largest_test_allowance(
_profile('# slow-timeout = { period = "30s", terminate-after = 1 }')
)
def test_a_commented_out_grace_period_is_not_in_force() -> None:
config_text = _profile(
'# slow-timeout = { period = "30s", terminate-after = 1, '
'grace-period = "30m" }',
'slow-timeout = { period = "180s", terminate-after = 1, grace-period = "5s" }',
)
assert grace_period(config_text) == Fraction(5), (
"a commented-out grace period was read as the one in force"
)
assert (
termination_allowance(config_text)
== Fraction(5) + TERMINATION_SAFETY_MARGIN_SECONDS
)
def test_a_commented_out_global_timeout_is_absent() -> None:
assert global_timeout(_profile('# global-timeout = "10m"')) is None, (
"a commented-out global-timeout was read as the budget in force"
)
assert global_timeout(_profile('global-timeout = "10m"')) == Fraction(600)
def test_the_whole_run_budget_is_read_from_the_default_profile() -> None:
config_text = _profile(
'slow-timeout = { period = "30s", terminate-after = 1 }',
"",
"[[profile.default.overrides]]",
'filter = "binary(=ui)"',
'global-timeout = "90m"',
)
assert global_timeout(config_text) is None
def test_a_grace_period_is_not_read_as_a_per_test_budget() -> None:
config_text = _profile(
'slow-timeout = { period = "30s", terminate-after = 1, grace-period = "30m" }'
)
assert largest_test_allowance(config_text) == Fraction(30), (
"the per-test ceiling read a grace period as a slow-timeout"
)
def test_the_termination_allowance_is_the_grace_period_plus_the_margin() -> None:
unset = termination_allowance(
_profile('slow-timeout = { period = "30s", terminate-after = 1 }')
)
assert unset == (
NEXTEST_DEFAULT_GRACE_PERIOD_SECONDS + TERMINATION_SAFETY_MARGIN_SECONDS
), "an unset grace period must fall back to nextest's own default"
configured = termination_allowance(
_profile(
'slow-timeout = { period = "30s", terminate-after = 1, '
'grace-period = "5s" }'
)
)
assert configured == Fraction(5) + TERMINATION_SAFETY_MARGIN_SECONDS, (
"a grace period below the margin must still raise the allowance; "
"a maximum over the two terms would have discarded it"
)
largest = termination_allowance(
_profile(
'slow-timeout = { period = "30s", terminate-after = 1, '
'grace-period = "5s" }',
"",
"[[profile.default.overrides]]",
'filter = "binary(=ui)"',
'slow-timeout = { period = "30s", terminate-after = 1, '
'grace-period = "45s" }',
)
)
assert largest == Fraction(45) + TERMINATION_SAFETY_MARGIN_SECONDS, (
"the largest configured grace period governs the allowance"
)
def test_the_required_ceiling_carries_all_three_terms() -> None:
assert required_ceiling(
[Fraction(1800), Fraction(2700)], Fraction(1200)
) == Fraction(4500) + Fraction(1200) + CEILING_MARGIN_SECONDS, (
"two watchdogs, the allowance and the margin are all added"
)
assert required_ceiling([Fraction(1800)], Fraction(0)) == (
Fraction(1800) + CEILING_MARGIN_SECONDS
), "the margin applies even when nothing runs outside the watchdog"
assert required_ceiling([], Fraction(0)) == CEILING_MARGIN_SECONDS, (
"the margin is a term of its own, not a fraction of the others"
)
@pytest.mark.parametrize(
("duration", "expected"),
[
pytest.param("120s", Fraction(120), id="one-component"),
pytest.param("1m 30s", Fraction(90), id="two-components-spaced"),
pytest.param("1m30s", Fraction(90), id="two-components-joined"),
pytest.param("1h 30m 15s", Fraction(5415), id="three-components"),
pytest.param("1day", Fraction(86400), id="an-extended-unit"),
pytest.param("1w", Fraction(604800), id="a-week"),
pytest.param("15min", Fraction(900), id="a-long-unit-spelling"),
pytest.param("500ms", Fraction(1, 2), id="milliseconds"),
pytest.param("1.5m", Fraction(90), id="a-fractional-value"),
pytest.param("1 . 5 m", Fraction(90), id="a-fractional-value-spaced"),
pytest.param("2wk", Fraction(1209600), id="an-abbreviated-week"),
pytest.param("1yr", Fraction(31557600), id="an-abbreviated-year"),
pytest.param("500\u00b5s", Fraction(1, 2000), id="the-micro-sign"),
pytest.param("1 0s", Fraction(10), id="whitespace-inside-a-number"),
pytest.param("0", Fraction(0), id="a-bare-zero"),
pytest.param("0.5s 0.5s", Fraction(1), id="two-halves-carry-to-one-second"),
pytest.param(
"18446744073709551615s 999999999ns",
Fraction(18446744073709551615) + Fraction(999999999, 1_000_000_000),
id="one-nanosecond-short-of-the-ceiling",
),
pytest.param(
"18446744073709551615ns 1ns",
Fraction(18446744073709551616, 1_000_000_000),
id="nanoseconds-normalized-between-components",
),
],
)
def test_the_duration_grammar_matches_the_one_nextest_reads(
duration: str, expected: Fraction
) -> None:
assert seconds(duration) == expected, (
f"{duration!r} must read as {expected} seconds"
)
@pytest.mark.parametrize(
"duration",
[
pytest.param("120", id="no-unit"),
pytest.param("s", id="no-value"),
pytest.param("-30s", id="negative"),
pytest.param("120 fortnights", id="an-unknown-unit"),
pytest.param("", id="empty"),
pytest.param(".5s", id="a-leading-point"),
pytest.param("1.s", id="a-trailing-point"),
pytest.param("1.2.3s", id="a-second-point"),
pytest.param("0.0000000002s", id="finer-than-a-nanosecond"),
pytest.param("1.5ns", id="a-fraction-of-a-nanosecond"),
pytest.param("18446744073709551616s", id="past-the-range-humantime-holds"),
pytest.param("600000000000y", id="a-value-that-overflows-its-unit"),
pytest.param(
"18446744073709551615s 500ms 500ms",
id="a-carry-that-passes-the-ceiling",
),
pytest.param(
"18446744073709551615ns 18446744073709551615ns",
id="a-nanosecond-part-that-overflows-before-it-carries",
),
pytest.param(
"1.00000000000000000000s",
id="a-denominator-past-the-range-humantime-holds",
),
pytest.param("\u0661s", id="an-arabic-indic-digit"),
pytest.param(" 0 ", id="a-padded-bare-zero"),
pytest.param("1\x1cs", id="a-file-separator-inside-a-number"),
pytest.param("\x1c45m", id="a-file-separator-leading"),
pytest.param("45m\x1f", id="a-unit-separator-trailing"),
pytest.param("1\x1d0s", id="a-group-separator-between-digits"),
],
)
def test_a_duration_nextest_would_refuse_is_refused_here(duration: str) -> None:
with pytest.raises(NextestConfigurationError):
seconds(duration)
def test_minutes_and_months_are_told_apart() -> None:
assert seconds("10m") == Fraction(600), "m is minutes"
assert seconds("10M") == Fraction(10 * 2630016), "M is months"
@pytest.mark.parametrize(
("name", "contents"),
[
pytest.param("broken.yml", b"jobs: [unclosed", id="not-yaml"),
pytest.param("stray.yml", b"jobs:\n build:\n name: \xff\n", id="not-utf-8"),
],
)
def test_an_unreadable_workflow_is_reported_as_one(
tmp_path: pathlib.Path, name: str, contents: bytes
) -> None:
(tmp_path / name).write_bytes(contents)
with pytest.raises(WorkflowReadError):
load_workflow_documents(tmp_path)
@pytest.mark.parametrize(
("case", "prepare"),
[
pytest.param("absent", lambda root: root / "gone", id="no-directory"),
pytest.param("empty", lambda root: root, id="no-documents"),
],
)
def test_a_directory_with_no_workflows_is_refused_rather_than_read_as_none(
tmp_path: pathlib.Path,
case: str,
prepare: object,
) -> None:
root = prepare(tmp_path) if case == "empty":
(root / "notes.txt").write_text("not a workflow", encoding="utf-8")
with pytest.raises(WorkflowReadError):
load_workflow_documents(root)
@pytest.mark.parametrize(
("field", "document"),
[
pytest.param(
"timeout-minutes",
{
"jobs": {
"build": {
"timeout-minutes": "${{ inputs.ceiling }}",
"env": {"RUN_RUST_CARGO_WAIT_TIMEOUT": "1800"},
"steps": [{"uses": f"{COVERAGE_ACTION}@0000000"}],
}
}
},
id="job-ceiling",
),
pytest.param(
"RUN_RUST_CARGO_WAIT_TIMEOUT",
{
"jobs": {
"build": {
"timeout-minutes": 66,
"env": {"RUN_RUST_CARGO_WAIT_TIMEOUT": "30m"},
"steps": [{"uses": f"{COVERAGE_ACTION}@0000000"}],
}
}
},
id="watchdog",
),
],
)
def test_a_duration_that_is_not_a_number_is_refused_by_name(
field: str, document: dict[str, object]
) -> None:
with pytest.raises(WorkflowValueError) as caught:
coverage_jobs_in({"ci.yml": document})
assert caught.value.field == field, "the refusal must name the field"
assert caught.value.location == "ci.yml:build", (
"the refusal must name the workflow and the job"
)
def test_budgets_beyond_the_exact_float_range_stay_distinguishable() -> None:
larger = seconds("9007199254740993s")
smaller = seconds("9007199254740992s")
assert larger != smaller, "budgets one second apart must not compare equal"
assert larger > smaller, "the larger budget must order above the smaller"
assert float(larger) == float(smaller), (
"the float collapse this guards against must still be real; if these "
"differ, the case no longer exercises what it was written for"
)
def test_a_fractional_budget_is_exact_rather_than_rounded() -> None:
assert seconds("0.1s") == Fraction(1, 10), "a tenth of a second is exact"
assert seconds("0.1s") * 3 == Fraction(3, 10), (
"exact budgets stay exact under the arithmetic the tiers apply"
)
def test_the_terminate_after_product_is_exact() -> None:
config_text = _profile('slow-timeout = { period = "0.1s", terminate-after = 3 }')
assert largest_test_allowance(config_text) == Fraction(3, 10), (
"a tenth of a second taken three times is three tenths exactly"
)
assert float(Fraction(1, 10)) * 3.0 != 0.3, (
"the floating-point product this guards against must still be wrong; "
"if it is not, the case no longer exercises what it was written for"
)
def test_a_table_omitting_the_grace_period_takes_nextest_s_own_default() -> None:
omitted_second = _profile(
'slow-timeout = { period = "30s", terminate-after = 1, grace-period = "5s" }',
"",
"[[profile.default.overrides]]",
'filter = "binary(=ui)"',
'slow-timeout = { period = "60s", terminate-after = 1 }',
)
omitted_first = _profile(
'slow-timeout = { period = "30s", terminate-after = 1 }',
"",
"[[profile.default.overrides]]",
'filter = "binary(=ui)"',
'slow-timeout = { period = "60s", terminate-after = 1, '
'grace-period = "5s" }',
)
for config_text in (omitted_second, omitted_first):
assert grace_period(config_text) == NEXTEST_DEFAULT_GRACE_PERIOD_SECONDS, (
"a table omitting grace-period allows nextest's ten-second "
"default, which is longer than the five seconds the other "
"table names"
)
assert NEXTEST_DEFAULT_GRACE_PERIOD_SECONDS > Fraction(5), (
"this case only discriminates while the default exceeds the "
"configured five seconds; if it no longer does, the pair no "
"longer exercises what it was written for"
)
def test_a_non_string_grace_period_is_refused_rather_than_defaulted() -> None:
invalid = _profile(
'slow-timeout = { period = "30s", terminate-after = 1, grace-period = 10 }'
)
with pytest.raises(NextestConfigurationError) as caught:
grace_period(invalid)
assert caught.value.field == "grace-period", (
"the refusal must name the field that is wrong"
)
assert caught.value.value == 10, (
"the refusal must carry the value it refused"
)
omitted = _profile('slow-timeout = { period = "30s", terminate-after = 1 }')
assert grace_period(omitted) == NEXTEST_DEFAULT_GRACE_PERIOD_SECONDS, (
"an omitted grace-period is still nextest's own default; only a "
"present non-string is a configuration error"
)
def test_a_neighbouring_action_is_not_read_as_a_coverage_lane() -> None:
ref = "@" + "0" * 40
documents = {
"ci.yml": {
"jobs": {
"neighbour": {
"timeout-minutes": 66,
"steps": [{"uses": f"{COVERAGE_ACTION}-old{ref}"}],
},
"coverage": {
"timeout-minutes": 66,
"steps": [{"uses": f"{COVERAGE_ACTION}{ref}"}],
},
}
}
}
jobs = coverage_jobs_in(documents)
assert [job.job for job in jobs] == ["coverage"], (
"only the job using the coverage action itself is a coverage lane"
)