mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-13 06:06:38 +00:00
refactor(layout): sort shipped scripts by language and split the unit suite
Every script a role ships now lives under `files/<language>/`, and the unit
suite is split the same way: `tests/unit/python/` beside new sibling trees for
the other languages. `tests/lint/filesystem/test_role_files_layout.py` keeps it
that way; directories whose shape is dictated from outside carry a `.nocheck`
naming the constraint (Docker build contexts, WordPress mu-plugins, the Joomla
extension, Odoo addons, Playwright suites, the `files/test/test.sh` harness
that `test-e2e-cli` resolves generically).
This cannot be applied in halves. Moving a file and repointing what names it
are the same change: 14 references resolve only at deploy time, so the tree is
green either before or after, never between.
Those 14 were found by static resolution over every reference class, not by the
test suite -- a full `make quality-high` was green while all of them were
broken:
* 7 `src:` values built from a role variable holding a bare filename
(svc-bkp-remote-2-local, decidim, espocrm, friendica, joomla, pixelfed x2).
The variable doubles as the destination name inside the container, so the
language segment belongs at the call site, not in the variable.
* `sys-service/tasks/04_base.yml` resolved every consumer's systemd script
through `first_found(..., errors='strict')` over `files/script.{sh,py}`.
18 roles ship exactly that and have no template fallback: the play would
have died on each.
* `sys-front-inj-javascript` fell back to `<role>/files/javascript.js`; the
interface now reads `files/javascript/javascript.js`.
* `sys-front-inj-logout/templates/head_sub.j2` built the cache-busting mtime
path through `path_absolute(... 'files', ...)`.
* 4 `ansible.builtin.script:` invocations naming a bare command
(pull.py, prune.sh, set_postgres_superuser_password.sh, ganesha-serve.sh).
* `dev-python/files/shell/install.sh` pointed at a sibling role's script.
Depth-dependent paths inside the moved trees were bumped accordingly: 59
`parents[N]` computations, 12 relative climbs, and the per-role recover CLI,
whose path `cli/administration/recover/recoverers.py` still assembled from the
pre-move location.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
6eff9d2ca1
commit
575a5a87ac
1076 files changed
+2141
-550
No files matched your search
@@ -0,0 +1,3 @@
|
||||
from pathlib import Path
|
||||
|
||||
PROJECT_ROOT: Path = Path(__file__).resolve().parents[5]
|
||||
Whitespace-only changes.
@@ -0,0 +1,495 @@
|
||||
"""Unit tests for :mod:`utils.github.playwright_summary`.
|
||||
|
||||
Synthetic JUnit XML fixtures are produced under a tempdir so the real
|
||||
project tree is never touched. Coverage spans the small pure helpers
|
||||
(``_first_line``, ``_md_escape_cell``), the parser (``_extract_records``,
|
||||
``_summarize_file``), the renderer (``_render``), and the CLI entry
|
||||
point (``main``), including the three early-exit branches for missing
|
||||
directories / no junit files / unparseable junit files.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import io
|
||||
import os
|
||||
import unittest
|
||||
import xml.etree.ElementTree as ET
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
from unittest.mock import patch
|
||||
|
||||
from utils.github import playwright_summary as ps
|
||||
|
||||
_TWO_SUITES_FIXTURE = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<testsuites tests="0" failures="0" skipped="0" time="0">
|
||||
<testsuite name="suite-a" tests="2" failures="1" skipped="0" time="3.5" timestamp="2026-05-17T12:00:00Z">
|
||||
<testcase name="passes" time="1.0"/>
|
||||
<testcase name="fails on assertion" time="2.5">
|
||||
<failure message="expected 1 got 0">stack-line-1\nstack-line-2</failure>
|
||||
</testcase>
|
||||
</testsuite>
|
||||
<testsuite name="suite-b" tests="2" failures="0" skipped="1" time="0.5" timestamp="2026-05-17T12:05:00Z">
|
||||
<testcase name="another pass" time="0.5"/>
|
||||
<testcase name="skipped one" time="0">
|
||||
<skipped message="persona blocked by env"/>
|
||||
</testcase>
|
||||
</testsuite>
|
||||
</testsuites>"""
|
||||
|
||||
_ERROR_INSTEAD_OF_FAILURE_FIXTURE = """<?xml version="1.0" encoding="UTF-8"?>
|
||||
<testsuites>
|
||||
<testsuite name="suite-c" time="1.0">
|
||||
<testcase name="crashes" time="1.0">
|
||||
<error message="ENOENT">no such file</error>
|
||||
</testcase>
|
||||
</testsuite>
|
||||
</testsuites>"""
|
||||
|
||||
|
||||
def _write(path: Path, content: str) -> Path:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(content, encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
class TestToIsoZ(unittest.TestCase):
|
||||
def test_utc_datetime_formats_with_z_suffix(self) -> None:
|
||||
self.assertEqual(
|
||||
ps._to_iso_z(
|
||||
datetime.datetime(2026, 5, 17, 12, 34, 56, tzinfo=datetime.UTC)
|
||||
),
|
||||
"2026-05-17T12:34:56Z",
|
||||
)
|
||||
|
||||
|
||||
class TestParseIsoTimestamp(unittest.TestCase):
|
||||
def test_empty_or_missing_returns_none(self) -> None:
|
||||
self.assertIsNone(ps._parse_iso_timestamp(""))
|
||||
self.assertIsNone(ps._parse_iso_timestamp(None))
|
||||
|
||||
def test_unparseable_returns_none(self) -> None:
|
||||
self.assertIsNone(ps._parse_iso_timestamp("bogus"))
|
||||
|
||||
def test_iso_with_z_suffix(self) -> None:
|
||||
result = ps._parse_iso_timestamp("2026-05-17T12:00:00Z")
|
||||
self.assertEqual(ps._to_iso_z(result), "2026-05-17T12:00:00Z")
|
||||
|
||||
def test_naive_iso_treated_as_utc(self) -> None:
|
||||
result = ps._parse_iso_timestamp("2026-05-17T12:00:00")
|
||||
self.assertEqual(ps._to_iso_z(result), "2026-05-17T12:00:00Z")
|
||||
|
||||
|
||||
class TestFormatDuration(unittest.TestCase):
|
||||
def test_sub_second_keeps_decimal(self) -> None:
|
||||
self.assertEqual(ps._format_duration(0.0), "0.0s")
|
||||
self.assertEqual(ps._format_duration(0.3), "0.3s")
|
||||
self.assertEqual(ps._format_duration(0.04), "0.0s")
|
||||
|
||||
def test_sub_minute_keeps_decimal(self) -> None:
|
||||
self.assertEqual(ps._format_duration(12.5), "12.5s")
|
||||
self.assertEqual(ps._format_duration(59.9), "59.9s")
|
||||
|
||||
def test_minute_boundary_rounds_seconds(self) -> None:
|
||||
self.assertEqual(ps._format_duration(60), "1m 0s")
|
||||
self.assertEqual(ps._format_duration(83.4), "1m 23s")
|
||||
self.assertEqual(ps._format_duration(3600), "60m 0s")
|
||||
|
||||
|
||||
class TestFirstLine(unittest.TestCase):
|
||||
def test_none_returns_empty(self) -> None:
|
||||
self.assertEqual(ps._first_line(None), "")
|
||||
|
||||
def test_whitespace_only_returns_empty(self) -> None:
|
||||
self.assertEqual(ps._first_line(" \n\t "), "")
|
||||
|
||||
def test_extracts_first_logical_line(self) -> None:
|
||||
self.assertEqual(
|
||||
ps._first_line("expected 1 got 0\nat /spec/foo.js:42"),
|
||||
"expected 1 got 0",
|
||||
)
|
||||
self.assertEqual(ps._first_line(" padded line "), "padded line")
|
||||
|
||||
|
||||
class TestMdEscapeCell(unittest.TestCase):
|
||||
def test_pipe_is_escaped(self) -> None:
|
||||
self.assertEqual(ps._md_escape_cell("a|b|c"), "a\\|b\\|c")
|
||||
|
||||
def test_newlines_collapse_to_spaces(self) -> None:
|
||||
self.assertEqual(ps._md_escape_cell("line1\nline2"), "line1 line2")
|
||||
|
||||
|
||||
class TestExtractRecords(unittest.TestCase):
|
||||
def test_mixed_pass_fail_skip(self) -> None:
|
||||
cases = list(
|
||||
ET.fromstring(_TWO_SUITES_FIXTURE).findall(".//testcase") # noqa: S314 — fixture is hard-coded above
|
||||
)
|
||||
records = ps._extract_records(cases)
|
||||
self.assertEqual(
|
||||
records,
|
||||
[
|
||||
("passes", "passed", "", 1.0),
|
||||
("fails on assertion", "failed", "expected 1 got 0", 2.5),
|
||||
("another pass", "passed", "", 0.5),
|
||||
("skipped one", "skipped", "persona blocked by env", 0.0),
|
||||
],
|
||||
)
|
||||
|
||||
def test_error_child_counts_as_failed(self) -> None:
|
||||
cases = list(
|
||||
ET.fromstring(_ERROR_INSTEAD_OF_FAILURE_FIXTURE).findall(".//testcase") # noqa: S314 — fixture is hard-coded above
|
||||
)
|
||||
records = ps._extract_records(cases)
|
||||
self.assertEqual(records, [("crashes", "failed", "ENOENT", 1.0)])
|
||||
|
||||
def test_failure_with_only_text_uses_text(self) -> None:
|
||||
case = ET.fromstring( # noqa: S314 — fixture literal, not user input
|
||||
'<testcase name="t"><failure>boom\nstack</failure></testcase>'
|
||||
)
|
||||
records = ps._extract_records([case])
|
||||
self.assertEqual(records, [("t", "failed", "boom", 0.0)])
|
||||
|
||||
def test_skipped_without_message_yields_empty_string(self) -> None:
|
||||
case = ET.fromstring( # noqa: S314 — fixture literal, not user input
|
||||
'<testcase name="anonymous-skip"><skipped/></testcase>'
|
||||
)
|
||||
records = ps._extract_records([case])
|
||||
self.assertEqual(records, [("anonymous-skip", "skipped", "", 0.0)])
|
||||
|
||||
def test_malformed_time_falls_back_to_zero(self) -> None:
|
||||
case = ET.fromstring( # noqa: S314 — fixture literal, not user input
|
||||
'<testcase name="bad-time" time="bogus"/>'
|
||||
)
|
||||
records = ps._extract_records([case])
|
||||
self.assertEqual(records, [("bad-time", "passed", "", 0.0)])
|
||||
|
||||
|
||||
class TestPathInfo(unittest.TestCase):
|
||||
def test_full_variant_pass_layout(self) -> None:
|
||||
self.assertEqual(
|
||||
ps._path_info(
|
||||
Path("/tmp/root/web-app-foo/variant-3/async/playwright-junit.xml")
|
||||
),
|
||||
("web-app-foo", "3", "async"),
|
||||
)
|
||||
self.assertEqual(
|
||||
ps._path_info(
|
||||
Path("/tmp/root/web-app-foo/variant-0/sync/playwright-junit.xml")
|
||||
),
|
||||
("web-app-foo", "0", "sync"),
|
||||
)
|
||||
|
||||
def test_variant_only_layout_leaves_pass_empty(self) -> None:
|
||||
self.assertEqual(
|
||||
ps._path_info(Path("/tmp/root/web-app-foo/variant-3/playwright-junit.xml")),
|
||||
("web-app-foo", "3", ""),
|
||||
)
|
||||
|
||||
def test_plain_layout_leaves_variant_and_pass_empty(self) -> None:
|
||||
self.assertEqual(
|
||||
ps._path_info(Path("/tmp/root/web-app-foo/playwright-junit.xml")),
|
||||
("web-app-foo", "", ""),
|
||||
)
|
||||
|
||||
def test_non_variant_subdir_falls_back_to_parent_only(self) -> None:
|
||||
self.assertEqual(
|
||||
ps._path_info(Path("/tmp/root/web-app-foo/leftover/playwright-junit.xml")),
|
||||
("leftover", "", ""),
|
||||
)
|
||||
|
||||
def test_pass_dir_without_variant_parent_is_not_recognized(self) -> None:
|
||||
self.assertEqual(
|
||||
ps._path_info(Path("/tmp/root/web-app-foo/sync/playwright-junit.xml")),
|
||||
("sync", "", ""),
|
||||
)
|
||||
|
||||
|
||||
class TestSummarizeFile(unittest.TestCase):
|
||||
def test_plain_layout_fills_only_app(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
xml_path = _write(
|
||||
Path(tmp) / "web-app-foo" / "playwright-junit.xml", _TWO_SUITES_FIXTURE
|
||||
)
|
||||
summary = ps._summarize_file(xml_path)
|
||||
assert summary is not None
|
||||
self.assertEqual(summary["app"], "web-app-foo")
|
||||
self.assertEqual(summary["variant"], "")
|
||||
self.assertEqual(summary["pass"], "")
|
||||
self.assertEqual(len(summary["records"]), 4)
|
||||
self.assertEqual(summary["records"][1][2], "failed")
|
||||
self.assertEqual(summary["records"][3][2], "skipped")
|
||||
self.assertAlmostEqual(summary["records"][0][4], 1.0)
|
||||
self.assertAlmostEqual(summary["records"][1][4], 2.5)
|
||||
|
||||
def test_variant_and_pass_layout_populates_all_three_keys(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
xml_path = _write(
|
||||
Path(tmp)
|
||||
/ "web-app-foo"
|
||||
/ "variant-2"
|
||||
/ "async"
|
||||
/ "playwright-junit.xml",
|
||||
_TWO_SUITES_FIXTURE,
|
||||
)
|
||||
summary = ps._summarize_file(xml_path)
|
||||
assert summary is not None
|
||||
self.assertEqual(summary["app"], "web-app-foo")
|
||||
self.assertEqual(summary["variant"], "2")
|
||||
self.assertEqual(summary["pass"], "async")
|
||||
|
||||
def test_per_case_timestamp_uses_suite_timestamp_plus_offset(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
xml_path = _write(
|
||||
Path(tmp) / "web-app-foo" / "playwright-junit.xml",
|
||||
_TWO_SUITES_FIXTURE,
|
||||
)
|
||||
summary = ps._summarize_file(xml_path)
|
||||
assert summary is not None
|
||||
records = summary["records"]
|
||||
self.assertEqual(records[0][0], "2026-05-17T12:00:00Z")
|
||||
self.assertEqual(records[1][0], "2026-05-17T12:00:01Z")
|
||||
self.assertEqual(records[2][0], "2026-05-17T12:05:00Z")
|
||||
self.assertEqual(records[3][0], "2026-05-17T12:05:00Z")
|
||||
|
||||
def test_missing_timestamp_falls_back_to_file_mtime(self) -> None:
|
||||
no_timestamp_fixture = (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>'
|
||||
'<testsuites><testsuite name="t" time="1.0">'
|
||||
'<testcase name="ok" time="1.0"/>'
|
||||
"</testsuite></testsuites>"
|
||||
)
|
||||
with TemporaryDirectory() as tmp:
|
||||
xml_path = _write(
|
||||
Path(tmp) / "web-app-foo" / "playwright-junit.xml",
|
||||
no_timestamp_fixture,
|
||||
)
|
||||
target = datetime.datetime(
|
||||
2025, 5, 17, 5, 34, 56, tzinfo=datetime.UTC
|
||||
).timestamp()
|
||||
os.utime(xml_path, (target, target))
|
||||
summary = ps._summarize_file(xml_path)
|
||||
assert summary is not None
|
||||
self.assertEqual(summary["records"][0][0], "2025-05-17T05:34:56Z")
|
||||
|
||||
def test_unparseable_returns_none(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
xml_path = _write(Path(tmp) / "x" / "playwright-junit.xml", "<broken")
|
||||
self.assertIsNone(ps._summarize_file(xml_path))
|
||||
|
||||
def test_empty_testsuites_returns_none(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
xml_path = _write(
|
||||
Path(tmp) / "x" / "playwright-junit.xml",
|
||||
"<testsuites></testsuites>",
|
||||
)
|
||||
self.assertIsNone(ps._summarize_file(xml_path))
|
||||
|
||||
|
||||
_T0 = "2026-05-17T12:00:00Z"
|
||||
|
||||
|
||||
class TestRender(unittest.TestCase):
|
||||
def _summary(
|
||||
self,
|
||||
app: str,
|
||||
records: list[tuple[str, str, str, str, float]],
|
||||
*,
|
||||
variant: str = "",
|
||||
pass_: str = "",
|
||||
) -> dict:
|
||||
return {"app": app, "variant": variant, "pass": pass_, "records": records}
|
||||
|
||||
def test_emits_one_row_per_test_with_status_emoji(self) -> None:
|
||||
out = ps._render(
|
||||
[
|
||||
self._summary(
|
||||
"web-app-foo",
|
||||
[
|
||||
(_T0, "test pass", "passed", "", 0.3),
|
||||
(_T0, "test fail", "failed", "boom", 12.4),
|
||||
(_T0, "test skip", "skipped", "blocked", 0.0),
|
||||
],
|
||||
variant="variant-1",
|
||||
pass_="sync",
|
||||
),
|
||||
],
|
||||
"web-app-foo",
|
||||
)
|
||||
self.assertIn("### 🎭 Playwright — web-app-foo", out)
|
||||
self.assertIn(
|
||||
"| Time | App | Variant | Pass | Test | Status | Duration | Message |",
|
||||
out,
|
||||
)
|
||||
self.assertIn(
|
||||
f"| `{_T0}` | `web-app-foo` | variant-1 | sync | test pass | 🟢 | 0.3s | |",
|
||||
out,
|
||||
)
|
||||
self.assertIn(
|
||||
f"| `{_T0}` | `web-app-foo` | variant-1 | sync | test fail | 🔴 | 12.4s | boom |",
|
||||
out,
|
||||
)
|
||||
self.assertIn(
|
||||
f"| `{_T0}` | `web-app-foo` | variant-1 | sync | test skip | 🔵 | 0.0s | blocked |",
|
||||
out,
|
||||
)
|
||||
|
||||
def test_status_legend_is_emitted(self) -> None:
|
||||
out = ps._render([self._summary("foo", [(_T0, "t", "passed", "", 1.0)])], "ctx")
|
||||
self.assertIn("🟢 passed", out)
|
||||
self.assertIn("🔴 failed", out)
|
||||
self.assertIn("🔵 skipped", out)
|
||||
|
||||
def test_multiple_apps_grouped_in_single_table(self) -> None:
|
||||
out = ps._render(
|
||||
[
|
||||
self._summary(
|
||||
"a",
|
||||
[(_T0, "t1", "passed", "", 0.1)],
|
||||
variant="variant-0",
|
||||
pass_="sync",
|
||||
),
|
||||
self._summary(
|
||||
"b",
|
||||
[(_T0, "t2", "failed", "x", 65.0)],
|
||||
variant="variant-0",
|
||||
pass_="async",
|
||||
),
|
||||
],
|
||||
"",
|
||||
)
|
||||
self.assertTrue(out.startswith("### 🎭 Playwright\n"))
|
||||
self.assertIn(f"| `{_T0}` | `a` | variant-0 | sync | t1 | 🟢 | 0.1s | |", out)
|
||||
self.assertIn(
|
||||
f"| `{_T0}` | `b` | variant-0 | async | t2 | 🔴 | 1m 5s | x |", out
|
||||
)
|
||||
self.assertEqual(
|
||||
out.count(
|
||||
"| Time | App | Variant | Pass | Test | Status | Duration | Message |"
|
||||
),
|
||||
1,
|
||||
)
|
||||
|
||||
def test_rows_are_sorted_ascending_by_time(self) -> None:
|
||||
out = ps._render(
|
||||
[
|
||||
self._summary(
|
||||
"late",
|
||||
[
|
||||
("2026-05-17T12:00:05Z", "late-1", "passed", "", 0.1),
|
||||
("2026-05-17T12:00:09Z", "late-2", "passed", "", 0.1),
|
||||
],
|
||||
),
|
||||
self._summary(
|
||||
"early",
|
||||
[
|
||||
("2026-05-17T12:00:01Z", "early-1", "passed", "", 0.1),
|
||||
("2026-05-17T12:00:07Z", "early-2", "passed", "", 0.1),
|
||||
],
|
||||
),
|
||||
],
|
||||
"ctx",
|
||||
)
|
||||
body = out.split("|---|---|---|---|---|:---:|---:|---|\n", 1)[1]
|
||||
order = [line.split("|")[5].strip() for line in body.splitlines() if line]
|
||||
self.assertEqual(order, ["early-1", "late-1", "early-2", "late-2"])
|
||||
|
||||
def test_plain_layout_leaves_variant_and_pass_cells_empty(self) -> None:
|
||||
out = ps._render(
|
||||
[self._summary("foo", [(_T0, "t1", "passed", "", 1.0)])],
|
||||
"ctx",
|
||||
)
|
||||
self.assertIn(f"| `{_T0}` | `foo` | | | t1 | 🟢 | 1.0s | |", out)
|
||||
|
||||
def test_message_is_truncated(self) -> None:
|
||||
long_msg = "x" * (ps._MAX_MSG_LEN + 50)
|
||||
out = ps._render(
|
||||
[self._summary("foo", [(_T0, "name", "failed", long_msg, 1.0)])],
|
||||
"ctx",
|
||||
)
|
||||
truncated = "x" * ps._MAX_MSG_LEN + "…"
|
||||
self.assertIn(truncated, out)
|
||||
|
||||
def test_pipes_in_name_and_message_are_escaped(self) -> None:
|
||||
out = ps._render(
|
||||
[
|
||||
self._summary(
|
||||
"foo",
|
||||
[(_T0, 'up{job="x|y"} == 1', "failed", "got 0|1", 1.0)],
|
||||
)
|
||||
],
|
||||
"ctx",
|
||||
)
|
||||
self.assertIn('up{job="x\\|y"} == 1', out)
|
||||
self.assertIn("got 0\\|1", out)
|
||||
|
||||
|
||||
class TestMain(unittest.TestCase):
|
||||
def _run(self, *argv: str) -> tuple[int, str, str]:
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
with patch.object(ps, "sys") as fake_sys:
|
||||
fake_sys.argv = ["playwright_summary.py", *argv]
|
||||
fake_sys.stdout = stdout
|
||||
fake_sys.stderr = stderr
|
||||
rc = ps.main()
|
||||
return rc, stdout.getvalue(), stderr.getvalue()
|
||||
|
||||
def test_missing_argv_returns_two(self) -> None:
|
||||
rc, out, err = self._run()
|
||||
self.assertEqual(rc, 2)
|
||||
self.assertIn("Usage", err)
|
||||
self.assertEqual(out, "")
|
||||
|
||||
def test_nonexistent_dir_emits_friendly_note(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
missing = Path(tmp) / "absent"
|
||||
rc, out, _err = self._run(str(missing), "web-app-foo")
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertIn("does not exist", out)
|
||||
self.assertIn("web-app-foo", out)
|
||||
|
||||
def test_empty_dir_emits_no_junit_note(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
rc, out, _err = self._run(tmp, "web-app-foo")
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertIn("No `playwright-junit.xml` found", out)
|
||||
|
||||
def test_all_unparseable_emits_failure_note(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
_write(Path(tmp) / "x" / "playwright-junit.xml", "<broken")
|
||||
rc, out, _err = self._run(tmp, "web-app-foo")
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertIn("failed to parse", out)
|
||||
|
||||
def test_happy_path_renders_table(self) -> None:
|
||||
with TemporaryDirectory() as tmp:
|
||||
_write(
|
||||
Path(tmp)
|
||||
/ "web-app-foo"
|
||||
/ "variant-1"
|
||||
/ "sync"
|
||||
/ "playwright-junit.xml",
|
||||
_TWO_SUITES_FIXTURE,
|
||||
)
|
||||
rc, out, _err = self._run(tmp, "web-app-foo")
|
||||
self.assertEqual(rc, 0)
|
||||
self.assertIn("### 🎭 Playwright — web-app-foo", out)
|
||||
self.assertIn(
|
||||
"| `2026-05-17T12:00:00Z` | `web-app-foo` | 1 | sync | passes | "
|
||||
"🟢 | 1.0s | |",
|
||||
out,
|
||||
)
|
||||
self.assertIn(
|
||||
"| `2026-05-17T12:00:01Z` | `web-app-foo` | 1 | sync | "
|
||||
"fails on assertion | 🔴 | 2.5s | expected 1 got 0 |",
|
||||
out,
|
||||
)
|
||||
self.assertIn(
|
||||
"| `2026-05-17T12:05:00Z` | `web-app-foo` | 1 | sync | "
|
||||
"skipped one | 🔵 | 0.0s | persona blocked by env |",
|
||||
out,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
unittest.main()
|
||||
@@ -0,0 +1,51 @@
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from utils.github import pytest_summary as ps
|
||||
|
||||
_XML = """<?xml version="1.0" encoding="utf-8"?>
|
||||
<testsuites>
|
||||
<testsuite name="pytest" tests="4" failures="1" skipped="1" time="3.5">
|
||||
<testcase classname="tests.unit.test_a.TestA" name="test_fast" time="0.10"/>
|
||||
<testcase classname="tests.unit.test_a.TestA" name="test_slow" time="2.50"/>
|
||||
<testcase classname="tests.unit.test_b.TestB" name="test_broken" time="0.40">
|
||||
<failure message="boom">trace</failure>
|
||||
</testcase>
|
||||
<testcase classname="tests.unit.test_b.TestB" name="test_off" time="0.00">
|
||||
<skipped message="disabled"/>
|
||||
</testcase>
|
||||
</testsuite>
|
||||
</testsuites>
|
||||
"""
|
||||
|
||||
|
||||
class TestRenderReport(unittest.TestCase):
|
||||
def _render(self, xml: str, name: str = "unit.xml") -> str:
|
||||
with TemporaryDirectory() as tmp:
|
||||
path = Path(tmp) / name
|
||||
path.write_text(xml, encoding="utf-8")
|
||||
return ps._render_report(path)
|
||||
|
||||
def test_headline_counts_and_time(self) -> None:
|
||||
out = self._render(_XML)
|
||||
self.assertIn("### 🧪 unit — 🟢 2 · 🔴 1 · 🔵 1 · 3.0s", out)
|
||||
|
||||
def test_failures_render_open_and_all_tests_collapsed(self) -> None:
|
||||
out = self._render(_XML)
|
||||
self.assertIn("| `tests.unit.test_b.TestB::test_broken` | 0.40s |", out)
|
||||
self.assertIn("<summary>All 4 tests (slowest first)</summary>", out)
|
||||
self.assertIn("| `tests.unit.test_a.TestA::test_slow` | 🟢 | 2.50s |", out)
|
||||
self.assertIn("| `tests.unit.test_b.TestB::test_off` | 🔵 | 0.00s |", out)
|
||||
|
||||
def test_slowest_first_ordering(self) -> None:
|
||||
out = self._render(_XML)
|
||||
self.assertLess(out.index("test_slow"), out.index("test_fast"))
|
||||
|
||||
def test_unparseable_xml_degrades_gracefully(self) -> None:
|
||||
out = self._render("not xml at all")
|
||||
self.assertIn("_Unparseable junit XML:", out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Unit tests for utils.github.run_name: the emoji head each input's segment
|
||||
opens with, the openings that terminate a value, and the round trip through
|
||||
the declared run name."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from tests.utils.ci.run_name import render
|
||||
from utils.github import run_name
|
||||
|
||||
TPL = (
|
||||
"🕹️ "
|
||||
"${{ inputs.sequencing != 'auto'"
|
||||
" && format('🐛{0} ', inputs.sequencing == 'serial' && '序' || '并') || '' }}"
|
||||
"${{ inputs.mode_fail_fast && '🛑 ' || '' }}"
|
||||
"${{ inputs.workspace != 'auto'"
|
||||
" && format('🧑{0} ', inputs.workspace == 'true' && '是' || '否') || '' }}"
|
||||
"${{ inputs.distros != '' && format('🐧{0} ', inputs.distros) || '' }}"
|
||||
"${{ inputs.priority != '' && format('⭐{0} ', inputs.priority) || '' }}"
|
||||
"${{ inputs.whitelist != ''"
|
||||
" && format('🎯{0}', inputs.whitelist) || '🔀 diff (origin/main)' }}"
|
||||
)
|
||||
|
||||
|
||||
class SegmentTests(unittest.TestCase):
|
||||
def test_value_segments_map_their_input_to_a_head(self) -> None:
|
||||
self.assertEqual(
|
||||
run_name.heads(TPL),
|
||||
{
|
||||
"sequencing": "🐛",
|
||||
"workspace": "🧑",
|
||||
"distros": "🐧",
|
||||
"priority": "⭐",
|
||||
"whitelist": "🎯",
|
||||
},
|
||||
)
|
||||
|
||||
def test_a_valueless_segment_is_a_marker(self) -> None:
|
||||
self.assertEqual(run_name.markers(TPL), {"mode_fail_fast": "🛑"})
|
||||
|
||||
def test_an_input_without_a_value_segment_is_rejected(self) -> None:
|
||||
with self.assertRaises(ValueError):
|
||||
run_name.value_from_title("🕹️ 🐧arch", "lifecycles", TPL)
|
||||
|
||||
|
||||
class HeadlessSegmentTests(unittest.TestCase):
|
||||
"""An input whose ``format()`` opens with nothing carries no value.
|
||||
|
||||
``str.find('')`` matches at position 0, so an empty head would hand back
|
||||
the title's own prefix as the value. Such an input renders a glyph for the
|
||||
reader and is recovered from the job log instead."""
|
||||
|
||||
TPL = (
|
||||
"🕹️ "
|
||||
"${{ inputs.tor != 'auto' && format('{0} ',"
|
||||
" inputs.tor == 'disabled' && '🌐'"
|
||||
" || inputs.tor == 'enforced' && '🧅强制' || '🧅独占') || '' }}"
|
||||
"${{ inputs.distros != '' && format('🐧{0} ', inputs.distros) || '' }}"
|
||||
)
|
||||
|
||||
def test_it_yields_no_segment(self) -> None:
|
||||
self.assertNotIn("tor", run_name.heads(self.TPL))
|
||||
self.assertNotIn("tor", run_name.markers(self.TPL))
|
||||
|
||||
def test_its_glyphs_still_terminate_the_value_before_them(self) -> None:
|
||||
found = run_name.openings(self.TPL)
|
||||
self.assertIn("🌐", found)
|
||||
self.assertIn("🧅强制", found)
|
||||
self.assertIn("🧅独占", found)
|
||||
|
||||
def test_a_neighbour_value_survives_the_glyph(self) -> None:
|
||||
self.assertEqual(
|
||||
run_name.value_from_title("🕹️ 🧅独占 🐧arch debian", "distros", self.TPL),
|
||||
"arch debian",
|
||||
)
|
||||
|
||||
def test_the_title_reports_no_value_for_it(self) -> None:
|
||||
self.assertNotIn("tor", run_name.values_from_title("🕹️ 🧅强制 🐧arch", self.TPL))
|
||||
|
||||
|
||||
class OpeningTests(unittest.TestCase):
|
||||
def test_compared_operands_and_value_glyphs_are_not_openings(self) -> None:
|
||||
found = run_name.openings(TPL)
|
||||
self.assertNotIn("auto", found)
|
||||
self.assertNotIn("true", found)
|
||||
self.assertNotIn("是", found)
|
||||
self.assertNotIn("否", found)
|
||||
self.assertNotIn("序", found)
|
||||
|
||||
def test_the_diff_fallback_is_an_opening(self) -> None:
|
||||
self.assertIn("🔀 diff (origin/main)", run_name.openings(TPL))
|
||||
|
||||
|
||||
class ValueTests(unittest.TestCase):
|
||||
def test_a_following_segment_ends_the_value(self) -> None:
|
||||
title = "🕹️ 🐧debian ubuntu ⭐web-app-x 🎯__ALL__"
|
||||
self.assertEqual(
|
||||
run_name.value_from_title(title, "distros", TPL), "debian ubuntu"
|
||||
)
|
||||
|
||||
def test_the_diff_fallback_ends_the_value(self) -> None:
|
||||
title = "🕹️ 🐧arch 🔀 diff (origin/main)"
|
||||
self.assertEqual(run_name.value_from_title(title, "distros", TPL), "arch")
|
||||
|
||||
def test_a_glyph_reads_back_as_its_value(self) -> None:
|
||||
title = "🕹️ 🧑是 🐧arch"
|
||||
self.assertEqual(run_name.value_from_title(title, "workspace", TPL), "true")
|
||||
|
||||
def test_a_marker_reads_back_by_its_presence(self) -> None:
|
||||
self.assertEqual(
|
||||
run_name.values_from_title("🕹️ 🛑 🐧arch", TPL)["mode_fail_fast"], "true"
|
||||
)
|
||||
self.assertEqual(
|
||||
run_name.values_from_title("🕹️ 🐧arch", TPL)["mode_fail_fast"], "false"
|
||||
)
|
||||
|
||||
def test_a_default_input_records_nothing(self) -> None:
|
||||
title = "🕹️ 🐧arch 🔀 diff (origin/main)"
|
||||
self.assertEqual(run_name.value_from_title(title, "sequencing", TPL), "")
|
||||
|
||||
def test_a_foreign_title_yields_nothing(self) -> None:
|
||||
self.assertEqual(run_name.values_from_title("CI: Pull Request", TPL), {})
|
||||
|
||||
|
||||
class DeclaredTemplateTests(unittest.TestCase):
|
||||
def test_the_declared_run_name_round_trips(self) -> None:
|
||||
dispatched = {
|
||||
"chunk_gate": "true",
|
||||
"workspace": "true",
|
||||
"instructions": "false",
|
||||
"filesystem": "ext4",
|
||||
"distros": "debian arch",
|
||||
"lifecycles": "stable",
|
||||
"mode": "swarm",
|
||||
}
|
||||
self.assertEqual(run_name.values_from_title(render(dispatched)), dispatched)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Whitespace-only changes.
@@ -0,0 +1,619 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
import unittest.mock as mock
|
||||
|
||||
from utils.github.variant import axes, pools, tor
|
||||
from utils.roles.display import display_names
|
||||
from utils.symbol_glossary import to_emoji
|
||||
|
||||
_VARIANTS = {
|
||||
"web-app-a": [
|
||||
{"services": {"tor": {"enabled": True}}},
|
||||
{"services": {"tor": {"enabled": False}}},
|
||||
],
|
||||
"web-app-b": [{"services": {}}],
|
||||
}
|
||||
|
||||
|
||||
def _row(name: str, variant: int, modes: tuple[str, ...], **extra) -> dict:
|
||||
return {"name": name, "variant": variant, "modes": modes, **extra}
|
||||
|
||||
|
||||
def _assign(rows, **kwargs) -> list[dict[str, str]]:
|
||||
""":func:`axes.assign` on the full declared pools, which is what an
|
||||
unnarrowed run hands it."""
|
||||
kwargs.setdefault("distros", axes.DISTROS)
|
||||
kwargs.setdefault("filesystems", axes.FILESYSTEMS)
|
||||
return axes.assign(rows, **kwargs)
|
||||
|
||||
|
||||
class TestResolveTorMode(unittest.TestCase):
|
||||
def test_known_modes_pass_through(self) -> None:
|
||||
for mode in tor.TOR_MODES:
|
||||
self.assertEqual(tor.resolve_tor_mode(mode), mode)
|
||||
|
||||
def test_unknown_and_empty_fall_back_to_auto(self) -> None:
|
||||
self.assertEqual(tor.resolve_tor_mode("nonsense"), "auto")
|
||||
self.assertEqual(tor.resolve_tor_mode(""), "auto")
|
||||
|
||||
|
||||
class TestResolveSweep(unittest.TestCase):
|
||||
def test_a_number_is_read(self) -> None:
|
||||
self.assertEqual(axes.resolve_sweep("7"), 7)
|
||||
|
||||
def test_garbage_reads_as_zero(self) -> None:
|
||||
self.assertEqual(axes.resolve_sweep("x"), 0)
|
||||
self.assertEqual(axes.resolve_sweep(""), 0)
|
||||
|
||||
|
||||
class TestTorCapable(unittest.TestCase):
|
||||
def test_a_variant_pinning_the_gate_false_is_incapable(self) -> None:
|
||||
self.assertFalse(axes.tor_capable("web-app-a", 1, _VARIANTS))
|
||||
|
||||
def test_a_variant_pinning_the_gate_true_is_capable(self) -> None:
|
||||
self.assertTrue(axes.tor_capable("web-app-a", 0, _VARIANTS))
|
||||
|
||||
def test_an_unset_gate_counts_as_capable(self) -> None:
|
||||
self.assertTrue(axes.tor_capable("web-app-b", 0, _VARIANTS))
|
||||
|
||||
|
||||
class TestPickMode(unittest.TestCase):
|
||||
def test_a_single_offer_is_always_taken(self) -> None:
|
||||
for position in range(4):
|
||||
self.assertEqual(axes.pick_mode(("host",), position, 0), "host")
|
||||
|
||||
def test_two_offers_alternate_by_position(self) -> None:
|
||||
offered = ("compose", "swarm")
|
||||
picks = [axes.pick_mode(offered, position, 0) for position in range(4)]
|
||||
self.assertEqual(picks, ["compose", "swarm", "compose", "swarm"])
|
||||
|
||||
def test_the_sweep_flips_which_offer_leads(self) -> None:
|
||||
offered = ("compose", "swarm")
|
||||
self.assertEqual(axes.pick_mode(offered, 0, 0), "compose")
|
||||
self.assertEqual(axes.pick_mode(offered, 0, 1), "swarm")
|
||||
|
||||
def test_an_empty_offer_is_a_bug_not_a_fallback(self) -> None:
|
||||
with self.assertRaises(ValueError):
|
||||
axes.pick_mode((), 0, 0)
|
||||
|
||||
|
||||
class TestAxesDecouple(unittest.TestCase):
|
||||
def test_a_row_walks_all_four_combinations_in_four_sweeps(self) -> None:
|
||||
offered = ("compose", "swarm")
|
||||
seen = {
|
||||
(axes.pick_mode(offered, 0, sweep), axes.wants_tor(0, sweep))
|
||||
for sweep in range(4)
|
||||
}
|
||||
self.assertEqual(len(seen), 4)
|
||||
|
||||
|
||||
class TestGlyphBinding(unittest.TestCase):
|
||||
def test_the_local_glyph_is_the_house_symbol(self) -> None:
|
||||
self.assertEqual("🏠", axes.LOCAL_GLYPH)
|
||||
|
||||
|
||||
class TestArtifactSlug(unittest.TestCase):
|
||||
def test_the_entry_carries_the_slug_the_reporter_looks_for(self) -> None:
|
||||
from cli.meta.ci.report_failures import Failure, artifact_name
|
||||
|
||||
rows = [_row("web-app-b", 0, ("compose", "swarm"), priority=True)]
|
||||
entries = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
for entry in entries:
|
||||
with self.subTest(entry["label"]):
|
||||
self.assertEqual(
|
||||
f"rescue-diagnostics-{entry['artifact']}",
|
||||
artifact_name(
|
||||
entry["apps"],
|
||||
Failure(
|
||||
entry["mode"],
|
||||
entry["variant"],
|
||||
entry["tor"] == "true",
|
||||
entry["distro"],
|
||||
entry["filesystem"],
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
def test_the_onion_state_keeps_two_runs_of_one_variant_apart(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose",), priority=True)]
|
||||
entries = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
self.assertEqual(len({e["artifact"] for e in entries}), len(entries))
|
||||
|
||||
def test_a_variantless_row_gets_no_dangling_separator(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.artifact_slug("host", "sys-front-proxy", "", False),
|
||||
"host-sys-front-proxy",
|
||||
)
|
||||
|
||||
|
||||
class TestTorStates(unittest.TestCase):
|
||||
def test_a_capable_tor_mode_covers_both_states_under_auto(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.tor_states("compose", capable=True, tor_mode="auto"), [True, False]
|
||||
)
|
||||
|
||||
def test_an_incapable_row_only_runs_clearnet(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.tor_states("compose", capable=False, tor_mode="auto"), [False]
|
||||
)
|
||||
|
||||
def test_host_carries_no_onion_axis(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.tor_states("host", capable=True, tor_mode="auto"), [False]
|
||||
)
|
||||
|
||||
def test_an_explicit_narrowing_wins_over_full_coverage(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.tor_states("compose", capable=True, tor_mode="enforced"), [True]
|
||||
)
|
||||
self.assertEqual(
|
||||
axes.tor_states("compose", capable=True, tor_mode="disabled"), [False]
|
||||
)
|
||||
|
||||
def test_exclusive_drops_an_incapable_row_entirely(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.tor_states("compose", capable=False, tor_mode="exclusive"), []
|
||||
)
|
||||
|
||||
|
||||
class TestCombinations(unittest.TestCase):
|
||||
def test_two_modes_on_the_onion_axis_yield_four_runs(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.combinations(("compose", "swarm"), capable=True, tor_mode="auto"),
|
||||
[
|
||||
("compose", True),
|
||||
("compose", False),
|
||||
("swarm", True),
|
||||
("swarm", False),
|
||||
],
|
||||
)
|
||||
|
||||
def test_a_stackless_role_yields_compose_pair_plus_one_host(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.combinations(("compose", "host"), capable=True, tor_mode="auto"),
|
||||
[("compose", True), ("compose", False), ("host", False)],
|
||||
)
|
||||
|
||||
def test_an_incapable_variant_halves_the_cross_product(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.combinations(("compose", "swarm"), capable=False, tor_mode="auto"),
|
||||
[("compose", False), ("swarm", False)],
|
||||
)
|
||||
|
||||
|
||||
class TestPriorityCoverage(unittest.TestCase):
|
||||
def test_a_priority_row_runs_every_combination_in_one_sweep(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose", "swarm"), priority=True)]
|
||||
entries = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
self.assertEqual(
|
||||
{(e["mode"], e["tor"]) for e in entries},
|
||||
{
|
||||
("compose", "true"),
|
||||
("compose", "false"),
|
||||
("swarm", "true"),
|
||||
("swarm", "false"),
|
||||
},
|
||||
)
|
||||
|
||||
def test_a_regular_row_still_takes_exactly_one_combination(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose", "swarm"))]
|
||||
entries = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
self.assertEqual(len(entries), 1)
|
||||
|
||||
def test_priority_coverage_does_not_move_with_the_sweep(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose", "swarm"), priority=True)]
|
||||
shapes = {
|
||||
frozenset(
|
||||
(e["mode"], e["tor"])
|
||||
for e in _assign(
|
||||
rows, sweep=sweep, tor_mode="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
)
|
||||
for sweep in range(4)
|
||||
}
|
||||
self.assertEqual(len(shapes), 1)
|
||||
|
||||
def test_every_priority_job_gets_a_distinct_label(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose", "swarm"), priority=True)]
|
||||
entries = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
self.assertEqual(len({e["label"] for e in entries}), len(entries))
|
||||
|
||||
def test_an_incapable_priority_variant_skips_its_onion_runs(self) -> None:
|
||||
rows = [_row("web-app-a", 1, ("compose", "swarm"), priority=True)]
|
||||
entries = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
self.assertEqual([e["tor"] for e in entries], ["false", "false"])
|
||||
|
||||
|
||||
class TestAssign(unittest.TestCase):
|
||||
def test_every_row_becomes_one_entry(self) -> None:
|
||||
rows = [
|
||||
_row("web-app-a", 0, ("compose", "swarm")),
|
||||
_row("web-app-a", 1, ("compose", "swarm")),
|
||||
]
|
||||
entries = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
self.assertEqual([e["variant"] for e in entries], ["0", "1"])
|
||||
|
||||
def test_the_label_opens_with_the_mode_glyph(self) -> None:
|
||||
rows = [_row("web-app-a", 0, ("compose",))]
|
||||
entry = _assign(rows, sweep=0, tor_mode="disabled", variants_per_app=_VARIANTS)[
|
||||
0
|
||||
]
|
||||
self.assertTrue(entry["label"].startswith(to_emoji("compose")))
|
||||
|
||||
def test_a_host_row_carries_no_onion_glyph(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("host",))]
|
||||
entry = _assign(rows, sweep=0, tor_mode="enforced", variants_per_app=_VARIANTS)[
|
||||
0
|
||||
]
|
||||
self.assertEqual(entry["tor"], "false")
|
||||
self.assertNotIn(to_emoji("tor"), entry["label"])
|
||||
self.assertNotIn(to_emoji("clearnet"), entry["label"])
|
||||
|
||||
def test_a_priority_row_wears_the_star(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose",), priority=True)]
|
||||
entry = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)[0]
|
||||
self.assertTrue(entry["label"].endswith(to_emoji("priority")))
|
||||
self.assertEqual(entry["priority"], "true")
|
||||
|
||||
def test_enforced_onions_every_capable_row(self) -> None:
|
||||
rows = [
|
||||
_row("web-app-a", 0, ("compose",)),
|
||||
_row("web-app-a", 1, ("compose",)),
|
||||
]
|
||||
entries = _assign(
|
||||
rows, sweep=0, tor_mode="enforced", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual([e["tor"] for e in entries], ["true", "false"])
|
||||
|
||||
def test_disabled_onions_nothing(self) -> None:
|
||||
rows = [_row("web-app-a", 0, ("compose",))]
|
||||
entries = _assign(
|
||||
rows, sweep=0, tor_mode="disabled", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual(entries[0]["tor"], "false")
|
||||
self.assertEqual(entries[0]["disable"], "tor")
|
||||
|
||||
def test_exclusive_drops_the_rows_that_cannot_take_an_onion(self) -> None:
|
||||
rows = [
|
||||
_row("web-app-a", 0, ("compose",)),
|
||||
_row("web-app-a", 1, ("compose",)),
|
||||
]
|
||||
entries = _assign(
|
||||
rows, sweep=0, tor_mode="exclusive", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual([e["variant"] for e in entries], ["0"])
|
||||
|
||||
def test_a_row_without_tor_disables_the_provider(self) -> None:
|
||||
rows = [_row("web-app-a", 1, ("compose",))]
|
||||
entry = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)[0]
|
||||
self.assertEqual(entry["disable"], "tor")
|
||||
|
||||
def test_a_host_row_carries_the_local_glyph_where_tor_rows_carry_theirs(
|
||||
self,
|
||||
) -> None:
|
||||
entries = _assign(
|
||||
[_row("web-app-a", 0, ("host",)), _row("web-app-a", 0, ("compose",))],
|
||||
sweep=0,
|
||||
tor_mode="enforced",
|
||||
variants_per_app=_VARIANTS,
|
||||
)
|
||||
host, compose = entries
|
||||
self.assertTrue(host["label"].startswith(to_emoji("host") + axes.LOCAL_GLYPH))
|
||||
self.assertTrue(
|
||||
compose["label"].startswith(to_emoji("compose") + to_emoji("tor"))
|
||||
)
|
||||
self.assertEqual(axes.parse_label(host["label"]).mode, "host")
|
||||
|
||||
def test_the_provider_row_never_takes_the_clearnet_state(self) -> None:
|
||||
provider = axes.tor_provider()
|
||||
self.assertIsNotNone(provider)
|
||||
for sweep in range(4):
|
||||
for tor_mode in tor.TOR_MODES:
|
||||
with self.subTest(sweep=sweep, tor_mode=tor_mode):
|
||||
entries = _assign(
|
||||
[_row(provider, 0, ("compose", "swarm"), priority=True)],
|
||||
sweep=sweep,
|
||||
tor_mode=tor_mode,
|
||||
)
|
||||
self.assertEqual(
|
||||
[e["disable"] for e in entries], [""] * len(entries)
|
||||
)
|
||||
self.assertNotIn("false", [e["tor"] for e in entries])
|
||||
|
||||
|
||||
class TestPinnedAxes(unittest.TestCase):
|
||||
def _entries(self, row: dict) -> list[dict[str, str]]:
|
||||
return _assign([row], sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
|
||||
def test_a_pinned_mode_replaces_the_rotation(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose", "swarm"), pin_mode="swarm")
|
||||
self.assertEqual([e["mode"] for e in self._entries(row)], ["swarm"])
|
||||
|
||||
def test_a_pinned_onion_state_replaces_the_rotation(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose",), pin_tor=True)
|
||||
self.assertEqual([e["tor"] for e in self._entries(row)], ["true"])
|
||||
|
||||
def test_an_open_axis_still_rotates(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose", "swarm"), pin_tor=False)
|
||||
picks = {
|
||||
_assign([row], sweep=sweep, tor_mode="auto", variants_per_app=_VARIANTS)[0][
|
||||
"mode"
|
||||
]
|
||||
for sweep in range(2)
|
||||
}
|
||||
self.assertEqual(picks, {"compose", "swarm"})
|
||||
|
||||
def test_pinning_the_onion_keeps_the_rotation_off_host(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose", "host"), pin_tor=True)
|
||||
for sweep in range(4):
|
||||
with self.subTest(sweep=sweep):
|
||||
entries = _assign(
|
||||
[row], sweep=sweep, tor_mode="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual([e["mode"] for e in entries], ["compose"])
|
||||
|
||||
def test_a_pin_narrows_the_priority_cross_product(self) -> None:
|
||||
row = _row(
|
||||
"web-app-b", 0, ("compose", "swarm"), priority=True, pin_mode="compose"
|
||||
)
|
||||
entries = self._entries(row)
|
||||
self.assertEqual(
|
||||
{(e["mode"], e["tor"]) for e in entries},
|
||||
{("compose", "true"), ("compose", "false")},
|
||||
)
|
||||
|
||||
def test_a_fully_pinned_priority_row_runs_exactly_once(self) -> None:
|
||||
row = _row(
|
||||
"web-app-b",
|
||||
0,
|
||||
("compose", "swarm"),
|
||||
priority=True,
|
||||
pin_mode="swarm",
|
||||
pin_tor=False,
|
||||
)
|
||||
self.assertEqual(len(self._entries(row)), 1)
|
||||
|
||||
def test_an_unoffered_mode_aborts_the_matrix(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose",), pin_mode="swarm")
|
||||
with self.assertRaises(SystemExit):
|
||||
self._entries(row)
|
||||
|
||||
def test_an_impossible_onion_state_aborts_the_matrix(self) -> None:
|
||||
row = _row("web-app-a", 1, ("compose",), pin_tor=True)
|
||||
with self.assertRaises(SystemExit):
|
||||
self._entries(row)
|
||||
|
||||
def test_a_pin_fighting_the_runs_tor_axis_aborts(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose",), pin_tor=True)
|
||||
with self.assertRaises(SystemExit):
|
||||
_assign([row], sweep=0, tor_mode="disabled", variants_per_app=_VARIANTS)
|
||||
|
||||
|
||||
class TestResolvePool(unittest.TestCase):
|
||||
def test_an_empty_input_opens_the_whole_declared_set(self) -> None:
|
||||
self.assertEqual(
|
||||
pools.resolve_pool("", axes.DISTROS, "distro"), tuple(axes.DISTROS)
|
||||
)
|
||||
self.assertEqual(
|
||||
pools.resolve_pool(None, axes.FILESYSTEMS, "filesystem"), axes.FILESYSTEMS
|
||||
)
|
||||
|
||||
def test_a_named_subset_keeps_the_declaration_order(self) -> None:
|
||||
self.assertEqual(
|
||||
pools.resolve_pool("ext4 zfs", axes.FILESYSTEMS, "filesystem"),
|
||||
("zfs", "ext4"),
|
||||
)
|
||||
|
||||
def test_a_typo_aborts_instead_of_narrowing_to_nothing(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
pools.resolve_pool("debain", axes.DISTROS, "distro")
|
||||
|
||||
|
||||
class TestDistroAndFilesystemAxes(unittest.TestCase):
|
||||
def _rows(self, count: int) -> list[dict]:
|
||||
return [_row("web-app-b", 0, ("compose",)) for _ in range(count)]
|
||||
|
||||
def test_consecutive_rows_spread_over_the_pool(self) -> None:
|
||||
entries = _assign(
|
||||
self._rows(len(axes.DISTROS)),
|
||||
sweep=0,
|
||||
tor_mode="auto",
|
||||
variants_per_app=_VARIANTS,
|
||||
)
|
||||
self.assertEqual([e["distro"] for e in entries], list(axes.DISTROS))
|
||||
|
||||
def test_the_sweep_moves_every_row_on_to_the_next_distro(self) -> None:
|
||||
picks = [
|
||||
_assign(
|
||||
self._rows(1), sweep=sweep, tor_mode="auto", variants_per_app=_VARIANTS
|
||||
)[0]["distro"]
|
||||
for sweep in range(len(axes.DISTROS))
|
||||
]
|
||||
self.assertEqual(set(picks), set(axes.DISTROS))
|
||||
|
||||
def test_a_narrowed_pool_is_the_only_thing_drawn_from(self) -> None:
|
||||
entries = _assign(
|
||||
self._rows(4),
|
||||
sweep=0,
|
||||
tor_mode="auto",
|
||||
distros=("debian",),
|
||||
filesystems=("btrfs",),
|
||||
variants_per_app=_VARIANTS,
|
||||
)
|
||||
self.assertEqual({e["distro"] for e in entries}, {"debian"})
|
||||
self.assertEqual({e["filesystem"] for e in entries}, {"btrfs"})
|
||||
|
||||
def test_a_priority_row_spreads_its_combinations_over_the_pool(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose", "swarm"), priority=True)]
|
||||
entries = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
self.assertEqual(len({e["distro"] for e in entries}), len(entries))
|
||||
|
||||
def test_a_pinned_distro_replaces_the_rotation(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose",), pin_distro="fedora")]
|
||||
entries = _assign(rows, sweep=3, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
self.assertEqual(entries[0]["distro"], "fedora")
|
||||
|
||||
def test_a_pinned_filesystem_replaces_the_rotation(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose",), pin_filesystem="ext4")]
|
||||
entries = _assign(rows, sweep=1, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
self.assertEqual(entries[0]["filesystem"], "ext4")
|
||||
|
||||
def test_collapsing_two_tokens_keeps_the_stronger_filesystem_claim(self) -> None:
|
||||
"""The token that named the kind may be the one collapsing into a token
|
||||
that did not; losing its demand would let the deploy fall back to a
|
||||
filesystem the operator named against."""
|
||||
rows = [
|
||||
_row("web-app-b", 0, ("compose",), pin_distro="debian"),
|
||||
_row("web-app-b", 0, ("compose",), pin_filesystem="zfs"),
|
||||
]
|
||||
entries = _assign(
|
||||
rows,
|
||||
sweep=0,
|
||||
tor_mode="disabled",
|
||||
distros=("debian",),
|
||||
filesystems=("zfs",),
|
||||
variants_per_app=_VARIANTS,
|
||||
)
|
||||
self.assertEqual(len(entries), 1)
|
||||
self.assertEqual(entries[0]["enforce_filesystem"], "true")
|
||||
|
||||
def test_a_pin_outside_the_runs_pool_aborts_the_matrix(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose",), pin_distro="arch")]
|
||||
with self.assertRaises(SystemExit):
|
||||
_assign(
|
||||
rows,
|
||||
sweep=0,
|
||||
tor_mode="auto",
|
||||
distros=("debian",),
|
||||
variants_per_app=_VARIANTS,
|
||||
)
|
||||
|
||||
def test_the_glyphs_follow_the_onion_slot_in_the_label(self) -> None:
|
||||
entry = _assign(
|
||||
[_row("web-app-b", 0, ("compose",))],
|
||||
sweep=0,
|
||||
tor_mode="disabled",
|
||||
distros=("debian",),
|
||||
filesystems=("zfs",),
|
||||
variants_per_app=_VARIANTS,
|
||||
)[0]
|
||||
self.assertTrue(
|
||||
entry["label"].startswith(
|
||||
to_emoji("compose")
|
||||
+ to_emoji("clearnet")
|
||||
+ to_emoji("debian")
|
||||
+ to_emoji("zfs")
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class TestSortKey(unittest.TestCase):
|
||||
def _entries(self) -> list[dict[str, str]]:
|
||||
rows = [
|
||||
_row("web-app-b", 0, ("swarm",)),
|
||||
_row("web-app-a", 1, ("compose",)),
|
||||
_row("web-app-a", 0, ("compose",)),
|
||||
]
|
||||
return _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
|
||||
def test_rows_sort_by_name_then_variant(self) -> None:
|
||||
ordered = sorted(self._entries(), key=axes.sort_key)
|
||||
self.assertEqual(
|
||||
[(e["apps"], e["variant"]) for e in ordered],
|
||||
[("web-app-a", "0"), ("web-app-a", "1"), ("web-app-b", "0")],
|
||||
)
|
||||
|
||||
def test_the_mode_sorts_in_deploy_order_not_alphabetically(self) -> None:
|
||||
entries = [
|
||||
{"apps": "web-app-a", "variant": "0", "mode": mode, "tor": "false"}
|
||||
for mode in ("host", "swarm", "compose")
|
||||
]
|
||||
ordered = sorted(entries, key=axes.sort_key)
|
||||
self.assertEqual([e["mode"] for e in ordered], list(axes.MODES))
|
||||
|
||||
def test_clearnet_sorts_ahead_of_the_onion(self) -> None:
|
||||
entries = [
|
||||
{"apps": "web-app-a", "variant": "0", "mode": "compose", "tor": tor}
|
||||
for tor in ("true", "false")
|
||||
]
|
||||
ordered = sorted(entries, key=axes.sort_key)
|
||||
self.assertEqual([e["tor"] for e in ordered], ["false", "true"])
|
||||
|
||||
def test_a_variantless_row_sorts_ahead_of_variant_zero(self) -> None:
|
||||
entries = [
|
||||
{"apps": "web-app-a", "variant": "0", "mode": "compose", "tor": "false"},
|
||||
{"apps": "web-app-a", "variant": "", "mode": "compose", "tor": "false"},
|
||||
]
|
||||
ordered = sorted(entries, key=axes.sort_key)
|
||||
self.assertEqual([e["variant"] for e in ordered], ["", "0"])
|
||||
|
||||
|
||||
class TestParseLabel(unittest.TestCase):
|
||||
def _title(self, mode: str, app: str, variant: str, **kw) -> str:
|
||||
rows = [_row(app, int(variant), (mode,), **kw)]
|
||||
return _assign(rows, sweep=0, tor_mode="enforced", variants_per_app=_VARIANTS)[
|
||||
0
|
||||
]["label"]
|
||||
|
||||
def test_a_label_round_trips_through_the_parser(self) -> None:
|
||||
for mode in ("compose", "swarm", "host"):
|
||||
with self.subTest(mode):
|
||||
title = self._title(mode, "web-app-a", "0")
|
||||
label = axes.parse_label(title)
|
||||
self.assertIsNotNone(label)
|
||||
self.assertEqual(label.mode, mode)
|
||||
self.assertEqual(label.variant, "0")
|
||||
self.assertEqual(display_names().decode(label.name), "web-app-a")
|
||||
|
||||
def test_the_onion_state_survives_the_round_trip(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose",), priority=True)]
|
||||
entries = _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
parsed = {axes.parse_label(e["label"]).tor for e in entries}
|
||||
self.assertEqual(parsed, {True, False})
|
||||
|
||||
def test_a_host_label_reads_back_as_clearnet(self) -> None:
|
||||
title = self._title("host", "web-app-a", "0")
|
||||
self.assertFalse(axes.parse_label(title).tor)
|
||||
|
||||
def test_the_priority_star_does_not_bleed_into_the_name(self) -> None:
|
||||
title = self._title("compose", "web-app-a", "0", priority=True)
|
||||
label = axes.parse_label(title)
|
||||
self.assertEqual(display_names().decode(label.name), "web-app-a")
|
||||
|
||||
def test_a_reusable_workflow_prefix_is_tolerated(self) -> None:
|
||||
title = "🎶 Orchestrate CI / test-deploy-chunk-1 / " + self._title(
|
||||
"swarm", "web-app-a", "0"
|
||||
)
|
||||
self.assertEqual(axes.parse_label(title).mode, "swarm")
|
||||
|
||||
def test_the_distro_and_filesystem_survive_the_round_trip(self) -> None:
|
||||
entry = _assign(
|
||||
[_row("web-app-a", 0, ("swarm",))],
|
||||
sweep=0,
|
||||
tor_mode="enforced",
|
||||
distros=("centos",),
|
||||
filesystems=("btrfs",),
|
||||
variants_per_app=_VARIANTS,
|
||||
)[0]
|
||||
label = axes.parse_label(entry["label"])
|
||||
self.assertEqual((label.distro, label.filesystem), ("centos", "btrfs"))
|
||||
self.assertEqual(display_names().decode(label.name), "web-app-a")
|
||||
|
||||
def test_a_non_deploy_job_yields_nothing(self) -> None:
|
||||
self.assertIsNone(axes.parse_label("🎲 Pick distro(s)"))
|
||||
self.assertIsNone(axes.parse_label("🧹 Lint"))
|
||||
|
||||
|
||||
class TestEnvironmentReads(unittest.TestCase):
|
||||
def test_the_sweep_comes_from_the_environment(self) -> None:
|
||||
with mock.patch.dict("os.environ", {"INFINITO_CI_SWEEP": "3"}):
|
||||
self.assertEqual(axes.resolve_sweep(), 3)
|
||||
|
||||
def test_the_tor_mode_comes_from_the_environment(self) -> None:
|
||||
with mock.patch.dict("os.environ", {"INFINITO_TOR": "exclusive"}):
|
||||
self.assertEqual(tor.resolve_tor_mode(), "exclusive")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,204 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from utils.github.variant import selection
|
||||
from utils.symbol_glossary import to_emoji
|
||||
|
||||
|
||||
def _row(name: str, variant: int) -> dict:
|
||||
return {"name": name, "variant": variant}
|
||||
|
||||
|
||||
class TestParseAscii(unittest.TestCase):
|
||||
def test_a_bare_role_pins_nothing(self) -> None:
|
||||
pin = selection.parse("web-app-a")
|
||||
self.assertEqual(pin, selection.Pin("web-app-a"))
|
||||
self.assertFalse(pin.pinned)
|
||||
|
||||
def test_every_axis_is_read(self) -> None:
|
||||
self.assertEqual(
|
||||
selection.parse("web-app-a#0,2@swarm+tor%debian/zfs"),
|
||||
selection.Pin("web-app-a", (0, 2), "swarm", True, "debian", "zfs"),
|
||||
)
|
||||
|
||||
def test_the_distro_and_filesystem_are_optional_like_the_rest(self) -> None:
|
||||
pin = selection.parse("web-app-a@swarm")
|
||||
self.assertIsNone(pin.distro)
|
||||
self.assertIsNone(pin.filesystem)
|
||||
|
||||
def test_a_filesystem_alone_still_counts_as_a_narrowing(self) -> None:
|
||||
pin = selection.parse("web-app-a/ext4")
|
||||
self.assertEqual(pin.filesystem, "ext4")
|
||||
self.assertTrue(pin.pinned)
|
||||
|
||||
def test_an_unknown_distro_aborts(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.parse("web-app-a%debain")
|
||||
|
||||
def test_an_unknown_filesystem_aborts(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.parse("web-app-a/xfs")
|
||||
|
||||
def test_the_token_round_trips_through_describe(self) -> None:
|
||||
token = "web-app-a#0,2@swarm+tor%debian/zfs"
|
||||
self.assertEqual(selection.describe(selection.parse(token)), token)
|
||||
|
||||
def test_clearnet_is_the_other_onion_state(self) -> None:
|
||||
self.assertFalse(selection.parse("web-app-a@compose+clearnet").tor)
|
||||
|
||||
def test_a_role_id_ending_in_tor_is_not_read_as_an_axis(self) -> None:
|
||||
self.assertEqual(selection.parse("svc-net-tor"), selection.Pin("svc-net-tor"))
|
||||
|
||||
def test_an_unknown_mode_aborts(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.parse("web-app-a@swrm")
|
||||
|
||||
def test_an_unknown_onion_state_aborts(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.parse("web-app-a+onion")
|
||||
|
||||
def test_an_unparsable_token_aborts(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.parse("web-app-a#zwei")
|
||||
|
||||
|
||||
class TestParseLabel(unittest.TestCase):
|
||||
def _label(self, mode: str, tor: str) -> str:
|
||||
return f"{to_emoji(mode)}{to_emoji(tor)}web-app-a#2"
|
||||
|
||||
def test_the_glyphs_carry_mode_and_onion(self) -> None:
|
||||
self.assertEqual(
|
||||
selection.parse(self._label("swarm", "tor")),
|
||||
selection.Pin("web-app-a", (2,), "swarm", True),
|
||||
)
|
||||
|
||||
def test_the_clearnet_glyph_pins_the_clearnet_state(self) -> None:
|
||||
self.assertFalse(selection.parse(self._label("compose", "clearnet")).tor)
|
||||
|
||||
def test_the_priority_star_is_not_part_of_the_name(self) -> None:
|
||||
pasted = f"{self._label('compose', 'tor')} {to_emoji('priority')}"
|
||||
self.assertEqual(selection.parse(pasted).app, "web-app-a")
|
||||
|
||||
def test_a_host_label_pins_the_mode_only(self) -> None:
|
||||
pasted = f"{to_emoji('host')}{to_emoji('test_host')}web-app-a"
|
||||
self.assertEqual(
|
||||
selection.parse(pasted), selection.Pin("web-app-a", (), "host")
|
||||
)
|
||||
|
||||
def test_the_distro_and_filesystem_glyphs_are_read_too(self) -> None:
|
||||
pasted = (
|
||||
f"{to_emoji('swarm')}{to_emoji('tor')}"
|
||||
f"{to_emoji('fedora')}{to_emoji('btrfs')}web-app-a#2"
|
||||
)
|
||||
self.assertEqual(
|
||||
selection.parse(pasted),
|
||||
selection.Pin("web-app-a", (2,), "swarm", True, "fedora", "btrfs"),
|
||||
)
|
||||
|
||||
def test_a_title_pasted_with_its_caller_path_aborts_rather_than_guessing(
|
||||
self,
|
||||
) -> None:
|
||||
"""``/`` opens the filesystem axis, so a pasted caller path must abort
|
||||
loudly instead of splitting the token somewhere inside the path."""
|
||||
pasted = "🎶 Orchestrate CI / test-deploy-chunk-1 / " + self._label(
|
||||
"swarm", "tor"
|
||||
)
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.parse(pasted)
|
||||
|
||||
def test_the_two_spellings_may_not_contradict_each_other(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.parse(f"{to_emoji('swarm')}web-app-a@compose")
|
||||
|
||||
|
||||
class TestNames(unittest.TestCase):
|
||||
def test_the_query_filters_on_role_ids_only(self) -> None:
|
||||
pins = selection.parse_list("web-app-a#0@swarm web-app-b+tor")
|
||||
self.assertEqual(selection.names(pins), "web-app-a web-app-b")
|
||||
|
||||
def test_a_role_pinned_twice_is_named_once(self) -> None:
|
||||
pins = selection.parse_list("web-app-a#0 web-app-a#1")
|
||||
self.assertEqual(selection.names(pins), "web-app-a")
|
||||
|
||||
|
||||
_ROWS = [_row("web-app-a", 0), _row("web-app-a", 1), _row("web-app-b", 0)]
|
||||
|
||||
|
||||
class TestApply(unittest.TestCase):
|
||||
def test_no_selection_keeps_every_row_untouched(self) -> None:
|
||||
self.assertEqual(selection.apply(_ROWS, []), _ROWS)
|
||||
|
||||
def test_a_pinned_variant_drops_the_other_variants(self) -> None:
|
||||
kept = selection.apply(_ROWS, selection.parse_list("web-app-a#1"))
|
||||
self.assertEqual([row["variant"] for row in kept], [1])
|
||||
|
||||
def test_the_pinned_axes_ride_along_on_the_row(self) -> None:
|
||||
kept = selection.apply(_ROWS, selection.parse_list("web-app-a#0@swarm+tor"))
|
||||
self.assertEqual(kept[0]["pin_mode"], "swarm")
|
||||
self.assertTrue(kept[0]["pin_tor"])
|
||||
|
||||
def test_an_unpinned_row_carries_open_axes(self) -> None:
|
||||
kept = selection.apply(_ROWS, selection.parse_list("web-app-b"))
|
||||
self.assertEqual([kept[0]["pin_mode"], kept[0]["pin_tor"]], [None, None])
|
||||
self.assertEqual(
|
||||
[kept[0]["pin_distro"], kept[0]["pin_filesystem"]], [None, None]
|
||||
)
|
||||
|
||||
def test_the_pinned_distro_and_filesystem_ride_along_too(self) -> None:
|
||||
kept = selection.apply(_ROWS, selection.parse_list("web-app-a#0%debian/zfs"))
|
||||
self.assertEqual(kept[0]["pin_distro"], "debian")
|
||||
self.assertEqual(kept[0]["pin_filesystem"], "zfs")
|
||||
|
||||
def test_one_variant_that_failed_on_two_distros_comes_back_twice(self) -> None:
|
||||
kept = selection.apply(
|
||||
_ROWS, selection.parse_list("web-app-a#0%debian web-app-a#0%fedora")
|
||||
)
|
||||
self.assertEqual([row["pin_distro"] for row in kept], ["debian", "fedora"])
|
||||
|
||||
def test_one_variant_that_failed_in_two_modes_comes_back_twice(self) -> None:
|
||||
kept = selection.apply(
|
||||
_ROWS,
|
||||
selection.parse_list("web-app-a#0@compose+tor web-app-a#0@swarm+tor"),
|
||||
)
|
||||
self.assertEqual(
|
||||
[(row["variant"], row["pin_mode"], row["pin_tor"]) for row in kept],
|
||||
[(0, "compose", True), (0, "swarm", True)],
|
||||
)
|
||||
|
||||
def test_the_same_narrowing_written_twice_stays_one_deploy(self) -> None:
|
||||
kept = selection.apply(
|
||||
_ROWS,
|
||||
selection.parse_list("web-app-a#0@swarm+tor web-app-a#0@swarm+tor"),
|
||||
)
|
||||
self.assertEqual(len(kept), 1)
|
||||
|
||||
def test_a_bare_role_loses_to_a_token_that_narrows_the_same_row(self) -> None:
|
||||
kept = selection.apply(
|
||||
_ROWS, selection.parse_list("web-app-a web-app-a#0@swarm+tor")
|
||||
)
|
||||
self.assertEqual(
|
||||
[(row["variant"], row["pin_mode"]) for row in kept],
|
||||
[(0, "swarm"), (1, None)],
|
||||
)
|
||||
|
||||
def test_two_pins_on_one_role_keep_their_own_axes(self) -> None:
|
||||
kept = selection.apply(
|
||||
_ROWS, selection.parse_list("web-app-a#0@swarm web-app-a#1@compose")
|
||||
)
|
||||
self.assertEqual([row["pin_mode"] for row in kept], ["swarm", "compose"])
|
||||
|
||||
def test_the_query_order_survives_the_selection(self) -> None:
|
||||
kept = selection.apply(_ROWS, selection.parse_list("web-app-b web-app-a"))
|
||||
self.assertEqual([row["name"] for row in kept], [row["name"] for row in _ROWS])
|
||||
|
||||
def test_a_pin_that_matches_nothing_aborts_instead_of_running_empty(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.apply(_ROWS, selection.parse_list("web-app-b#7"))
|
||||
|
||||
def test_a_bare_role_that_matches_nothing_is_tolerated(self) -> None:
|
||||
self.assertEqual(selection.apply(_ROWS, selection.parse_list("web-app-c")), [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user