mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-13 06:06:38 +00:00
feat(ci): replace the tor axis with a network axis of node modes
Every deploy row now takes a node network mode, clearnet, tor or multi, instead of an onion yes or no. The workflow input `network` (auto|clearnet|tor|multi) replaces `tor`, and INFINITO_NETWORK replaces INFINITO_TOR. utils/github/variant/network.py decides which modes a row can take: clearnet only without a Tor bond or when the variant pins Tor off, no multi for a single_mode role, no clearnet for the Tor provider, whose rows are therefore no longer dropped when the rotation lands on clearnet. Priority rows cover every mode and network combination, regular rows rotate the network on sweep // 2, and clearnet rows carry disable=tor. Selection tokens pin the axis with +clearnet, +tor or +multi, and multi rows wear the new glyph. A row hands its mode to the deploy as the variable `network`, which provisioning writes into host_vars as NETWORK_MODE. Without it the node keeps deriving its mode from the deployed providers. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
6f8dcc40cc
commit
6314938133
47 files changed
+1194
-678
No files matched your search
@@ -51,32 +51,39 @@ class HeadlessSegmentTests(unittest.TestCase):
|
||||
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) || '' }}"
|
||||
NETWORK = (
|
||||
"${{ inputs.network != 'auto' && format('{0} ',"
|
||||
" inputs.network == 'clearnet' && '🌐'"
|
||||
" || inputs.network == 'tor' && '🧅' || '🌈') || '' }}"
|
||||
)
|
||||
|
||||
TPL = (
|
||||
"🕹️ "
|
||||
+ NETWORK
|
||||
+ "${{ inputs.distros != '' && format('🐧{0} ', inputs.distros) || '' }}"
|
||||
)
|
||||
|
||||
def test_it_is_the_network_segment_the_declared_run_name_renders(self) -> None:
|
||||
self.assertIn(self.NETWORK, run_name.template())
|
||||
|
||||
def test_it_yields_no_segment(self) -> None:
|
||||
self.assertNotIn("tor", run_name.heads(self.TPL))
|
||||
self.assertNotIn("tor", run_name.markers(self.TPL))
|
||||
self.assertNotIn("network", run_name.heads(self.TPL))
|
||||
self.assertNotIn("network", 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)
|
||||
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),
|
||||
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))
|
||||
self.assertNotIn("network", run_name.values_from_title("🕹️ 🧅 🐧arch", self.TPL))
|
||||
|
||||
|
||||
class OpeningTests(unittest.TestCase):
|
||||
|
||||
@@ -3,18 +3,41 @@ from __future__ import annotations
|
||||
import unittest
|
||||
import unittest.mock as mock
|
||||
|
||||
from utils.github.variant import axes, pools, tor
|
||||
from utils.github.variant import axes, network, pools
|
||||
from utils.roles.display import display_names
|
||||
from utils.symbol_glossary import to_emoji
|
||||
|
||||
_REACTIVE = "{{ 'svc-net-tor' in group_names }}"
|
||||
|
||||
_VARIANTS = {
|
||||
"web-app-a": [
|
||||
{"services": {"tor": {"enabled": True}}},
|
||||
{"services": {"tor": {"enabled": False}}},
|
||||
],
|
||||
"web-app-b": [{"services": {}}],
|
||||
"web-app-b": [{"services": {"tor": {"enabled": _REACTIVE}}}],
|
||||
"web-app-c": [{"services": {}}],
|
||||
"web-app-single": [
|
||||
{
|
||||
"services": {"tor": {"enabled": _REACTIVE}},
|
||||
"networks": {"reachability": {"single_mode": True}},
|
||||
}
|
||||
],
|
||||
"web-app-clear": [
|
||||
{
|
||||
"services": {"tor": {"enabled": _REACTIVE}},
|
||||
"networks": {"reachability": {"modes": ["clearnet"]}},
|
||||
}
|
||||
],
|
||||
"web-app-dark": [
|
||||
{
|
||||
"services": {"tor": {"enabled": _REACTIVE}},
|
||||
"networks": {"reachability": {"modes": ["tor"]}},
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
_ALL = ("clearnet", "tor", "multi")
|
||||
|
||||
|
||||
def _row(name: str, variant: int, modes: tuple[str, ...], **extra) -> dict:
|
||||
return {"name": name, "variant": variant, "modes": modes, **extra}
|
||||
@@ -28,14 +51,18 @@ def _assign(rows, **kwargs) -> list[dict[str, str]]:
|
||||
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)
|
||||
class TestResolveNetworkInput(unittest.TestCase):
|
||||
def test_known_inputs_pass_through(self) -> None:
|
||||
for value in network.NETWORK_INPUTS:
|
||||
self.assertEqual(network.resolve_network_input(value), value)
|
||||
|
||||
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")
|
||||
self.assertEqual(network.resolve_network_input("nonsense"), "auto")
|
||||
self.assertEqual(network.resolve_network_input(""), "auto")
|
||||
|
||||
def test_the_retired_tor_axis_values_are_not_network_modes(self) -> None:
|
||||
for value in ("enforced", "exclusive", "disabled"):
|
||||
self.assertEqual(network.resolve_network_input(value), "auto")
|
||||
|
||||
|
||||
class TestResolveSweep(unittest.TestCase):
|
||||
@@ -49,13 +76,41 @@ class TestResolveSweep(unittest.TestCase):
|
||||
|
||||
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))
|
||||
self.assertFalse(network.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))
|
||||
self.assertTrue(network.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))
|
||||
def test_a_reactive_gate_counts_as_capable(self) -> None:
|
||||
self.assertTrue(network.tor_capable("web-app-b", 0, _VARIANTS))
|
||||
|
||||
def test_a_role_without_a_tor_bond_is_incapable(self) -> None:
|
||||
self.assertFalse(network.tor_capable("web-app-c", 0, _VARIANTS))
|
||||
|
||||
|
||||
class TestRowStates(unittest.TestCase):
|
||||
def test_a_capable_row_takes_every_network_mode(self) -> None:
|
||||
self.assertEqual(network.row_states("web-app-b", 0, _VARIANTS), _ALL)
|
||||
|
||||
def test_an_incapable_row_only_takes_clearnet(self) -> None:
|
||||
self.assertEqual(network.row_states("web-app-a", 1, _VARIANTS), ("clearnet",))
|
||||
self.assertEqual(network.row_states("web-app-c", 0, _VARIANTS), ("clearnet",))
|
||||
|
||||
def test_a_single_mode_role_never_takes_multi(self) -> None:
|
||||
self.assertEqual(
|
||||
network.row_states("web-app-single", 0, _VARIANTS), ("clearnet", "tor")
|
||||
)
|
||||
|
||||
def test_restricted_modes_narrow_the_node_modes(self) -> None:
|
||||
self.assertEqual(
|
||||
network.row_states("web-app-clear", 0, _VARIANTS), ("clearnet",)
|
||||
)
|
||||
self.assertEqual(network.row_states("web-app-dark", 0, _VARIANTS), ("tor",))
|
||||
|
||||
def test_the_provider_never_takes_clearnet(self) -> None:
|
||||
provider = network.tor_provider()
|
||||
self.assertIsNotNone(provider)
|
||||
self.assertEqual(network.row_states(provider), ("tor", "multi"))
|
||||
|
||||
|
||||
class TestPickMode(unittest.TestCase):
|
||||
@@ -79,11 +134,25 @@ class TestPickMode(unittest.TestCase):
|
||||
|
||||
|
||||
class TestAxesDecouple(unittest.TestCase):
|
||||
def test_a_row_walks_all_four_combinations_in_four_sweeps(self) -> None:
|
||||
offered = ("compose", "swarm")
|
||||
def test_a_row_walks_all_six_combinations_in_six_sweeps(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose", "swarm"))
|
||||
seen = {
|
||||
(axes.pick_mode(offered, 0, sweep), axes.wants_tor(0, sweep))
|
||||
(entry["mode"], entry["network"])
|
||||
for sweep in range(6)
|
||||
for entry in _assign(
|
||||
[row], sweep=sweep, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
}
|
||||
self.assertEqual(len(seen), 6)
|
||||
|
||||
def test_a_two_state_row_walks_all_four_combinations_in_four_sweeps(self) -> None:
|
||||
row = _row("web-app-single", 0, ("compose", "swarm"))
|
||||
seen = {
|
||||
(entry["mode"], entry["network"])
|
||||
for sweep in range(4)
|
||||
for entry in _assign(
|
||||
[row], sweep=sweep, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
}
|
||||
self.assertEqual(len(seen), 4)
|
||||
|
||||
@@ -92,13 +161,18 @@ class TestGlyphBinding(unittest.TestCase):
|
||||
def test_the_local_glyph_is_the_house_symbol(self) -> None:
|
||||
self.assertEqual("🏠", axes.LOCAL_GLYPH)
|
||||
|
||||
def test_the_multi_mode_wears_the_rainbow(self) -> None:
|
||||
self.assertEqual("🌈", to_emoji("multi"))
|
||||
|
||||
|
||||
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)
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
for entry in entries:
|
||||
with self.subTest(entry["label"]):
|
||||
self.assertEqual(
|
||||
@@ -108,106 +182,144 @@ class TestArtifactSlug(unittest.TestCase):
|
||||
Failure(
|
||||
entry["mode"],
|
||||
entry["variant"],
|
||||
entry["tor"] == "true",
|
||||
entry["network"],
|
||||
entry["distro"],
|
||||
entry["filesystem"],
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
def test_the_onion_state_keeps_two_runs_of_one_variant_apart(self) -> None:
|
||||
def test_the_network_mode_keeps_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)
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual(len(entries), 3)
|
||||
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),
|
||||
axes.artifact_slug("host", "sys-front-proxy", "", "clearnet"),
|
||||
"host-sys-front-proxy",
|
||||
)
|
||||
|
||||
|
||||
class TestTorStates(unittest.TestCase):
|
||||
def test_a_capable_tor_mode_covers_both_states_under_auto(self) -> None:
|
||||
def test_only_a_network_beyond_clearnet_adds_a_shard(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.tor_states("compose", capable=True, tor_mode="auto"), [True, False]
|
||||
axes.artifact_slug("compose", "web-app-b", "0", "multi"),
|
||||
"compose-web-app-b-0-multi",
|
||||
)
|
||||
self.assertEqual(
|
||||
axes.artifact_slug("compose", "web-app-b", "0", "tor"),
|
||||
"compose-web-app-b-0-tor",
|
||||
)
|
||||
|
||||
|
||||
class TestNetworkStates(unittest.TestCase):
|
||||
def test_a_capable_row_covers_every_mode_under_auto(self) -> None:
|
||||
self.assertEqual(
|
||||
network.network_states("compose", states=_ALL, network_input="auto"),
|
||||
list(_ALL),
|
||||
)
|
||||
|
||||
def test_an_incapable_row_only_runs_clearnet(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.tor_states("compose", capable=False, tor_mode="auto"), [False]
|
||||
network.network_states(
|
||||
"compose", states=("clearnet",), network_input="auto"
|
||||
),
|
||||
["clearnet"],
|
||||
)
|
||||
|
||||
def test_host_carries_no_onion_axis(self) -> None:
|
||||
def test_host_carries_no_network_axis(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.tor_states("host", capable=True, tor_mode="auto"), [False]
|
||||
network.network_states("host", states=_ALL, network_input="auto"),
|
||||
["clearnet"],
|
||||
)
|
||||
|
||||
def test_an_explicit_narrowing_wins_over_full_coverage(self) -> None:
|
||||
def test_a_named_input_narrows_to_that_mode(self) -> None:
|
||||
for value in _ALL:
|
||||
with self.subTest(value):
|
||||
self.assertEqual(
|
||||
network.network_states("compose", states=_ALL, network_input=value),
|
||||
[value],
|
||||
)
|
||||
|
||||
def test_a_named_input_the_row_cannot_take_drops_it(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]
|
||||
network.network_states(
|
||||
"compose", states=("clearnet",), network_input="multi"
|
||||
),
|
||||
[],
|
||||
)
|
||||
|
||||
def test_exclusive_drops_an_incapable_row_entirely(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.tor_states("compose", capable=False, tor_mode="exclusive"), []
|
||||
)
|
||||
def test_only_clearnet_disables_the_tor_provider(self) -> None:
|
||||
self.assertEqual(network.disabled_services("clearnet"), "tor")
|
||||
self.assertEqual(network.disabled_services("tor"), "")
|
||||
self.assertEqual(network.disabled_services("multi"), "")
|
||||
|
||||
|
||||
class TestCombinations(unittest.TestCase):
|
||||
def test_two_modes_on_the_onion_axis_yield_four_runs(self) -> None:
|
||||
def test_two_modes_on_the_network_axis_yield_six_runs(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.combinations(("compose", "swarm"), capable=True, tor_mode="auto"),
|
||||
network.combinations(
|
||||
("compose", "swarm"), states=_ALL, network_input="auto"
|
||||
),
|
||||
[
|
||||
("compose", True),
|
||||
("compose", False),
|
||||
("swarm", True),
|
||||
("swarm", False),
|
||||
("compose", "clearnet"),
|
||||
("compose", "tor"),
|
||||
("compose", "multi"),
|
||||
("swarm", "clearnet"),
|
||||
("swarm", "tor"),
|
||||
("swarm", "multi"),
|
||||
],
|
||||
)
|
||||
|
||||
def test_a_stackless_role_yields_compose_pair_plus_one_host(self) -> None:
|
||||
def test_a_stackless_role_yields_compose_triple_plus_one_host(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.combinations(("compose", "host"), capable=True, tor_mode="auto"),
|
||||
[("compose", True), ("compose", False), ("host", False)],
|
||||
network.combinations(
|
||||
("compose", "host"), states=_ALL, network_input="auto"
|
||||
),
|
||||
[
|
||||
("compose", "clearnet"),
|
||||
("compose", "tor"),
|
||||
("compose", "multi"),
|
||||
("host", "clearnet"),
|
||||
],
|
||||
)
|
||||
|
||||
def test_an_incapable_variant_halves_the_cross_product(self) -> None:
|
||||
def test_an_incapable_variant_keeps_one_run_per_mode(self) -> None:
|
||||
self.assertEqual(
|
||||
axes.combinations(("compose", "swarm"), capable=False, tor_mode="auto"),
|
||||
[("compose", False), ("swarm", False)],
|
||||
network.combinations(
|
||||
("compose", "swarm"), states=("clearnet",), network_input="auto"
|
||||
),
|
||||
[("compose", "clearnet"), ("swarm", "clearnet")],
|
||||
)
|
||||
|
||||
|
||||
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)
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual(
|
||||
{(e["mode"], e["tor"]) for e in entries},
|
||||
{
|
||||
("compose", "true"),
|
||||
("compose", "false"),
|
||||
("swarm", "true"),
|
||||
("swarm", "false"),
|
||||
},
|
||||
{(e["mode"], e["network"]) for e in entries},
|
||||
{(mode, state) for mode in ("compose", "swarm") for state in _ALL},
|
||||
)
|
||||
|
||||
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)
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="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"])
|
||||
(e["mode"], e["network"])
|
||||
for e in _assign(
|
||||
rows, sweep=sweep, tor_mode="auto", variants_per_app=_VARIANTS
|
||||
rows, sweep=sweep, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
)
|
||||
for sweep in range(4)
|
||||
@@ -216,13 +328,24 @@ class TestPriorityCoverage(unittest.TestCase):
|
||||
|
||||
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)
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="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:
|
||||
def test_an_incapable_priority_variant_skips_its_tor_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"])
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual([e["network"] for e in entries], ["clearnet", "clearnet"])
|
||||
|
||||
def test_a_single_mode_priority_row_never_runs_multi(self) -> None:
|
||||
rows = [_row("web-app-single", 0, ("compose",), priority=True)]
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual([e["network"] for e in entries], ["clearnet", "tor"])
|
||||
|
||||
|
||||
class TestAssign(unittest.TestCase):
|
||||
@@ -231,71 +354,102 @@ class TestAssign(unittest.TestCase):
|
||||
_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)
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="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
|
||||
]
|
||||
entry = _assign(
|
||||
rows, sweep=0, network_input="clearnet", variants_per_app=_VARIANTS
|
||||
)[0]
|
||||
self.assertTrue(entry["label"].startswith(to_emoji("compose")))
|
||||
|
||||
def test_a_host_row_carries_no_onion_glyph(self) -> None:
|
||||
def test_a_host_row_carries_no_network_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"])
|
||||
entry = _assign(
|
||||
rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS
|
||||
)[0]
|
||||
self.assertEqual(entry["network"], "clearnet")
|
||||
for state in _ALL:
|
||||
self.assertNotIn(to_emoji(state), 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]
|
||||
entry = _assign(
|
||||
rows, sweep=0, network_input="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:
|
||||
def test_a_named_input_keeps_only_the_rows_that_can_take_it(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
|
||||
rows, sweep=0, network_input="tor", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual(
|
||||
[(e["variant"], e["network"]) for e in entries], [("0", "tor")]
|
||||
)
|
||||
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_a_named_input_never_rotates_a_row_onto_a_mode_that_cannot_take_it(
|
||||
self,
|
||||
) -> None:
|
||||
row = _row("web-app-b", 0, ("compose", "host"))
|
||||
for sweep in range(4):
|
||||
with self.subTest(sweep=sweep):
|
||||
entries = _assign(
|
||||
[row], sweep=sweep, network_input="tor", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual(
|
||||
[(e["mode"], e["network"]) for e in entries], [("compose", "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
|
||||
def test_a_row_no_mode_can_serve_under_the_input_is_dropped(self) -> None:
|
||||
rows = [_row("web-app-a", 1, ("compose", "host"))]
|
||||
self.assertEqual(
|
||||
_assign(rows, sweep=0, network_input="multi", variants_per_app=_VARIANTS),
|
||||
[],
|
||||
)
|
||||
self.assertEqual([e["variant"] for e in entries], ["0"])
|
||||
|
||||
def test_the_clearnet_input_serves_every_row_without_tor(self) -> None:
|
||||
rows = [_row("web-app-a", 0, ("compose",)), _row("web-app-a", 1, ("compose",))]
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="clearnet", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual([e["network"] for e in entries], ["clearnet", "clearnet"])
|
||||
self.assertEqual([e["disable"] for e in entries], ["tor", "tor"])
|
||||
|
||||
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]
|
||||
entry = _assign(
|
||||
rows, sweep=0, network_input="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(
|
||||
def test_tor_and_multi_rows_keep_the_provider(self) -> None:
|
||||
rows = [_row("web-app-b", 0, ("compose",), priority=True)]
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual(
|
||||
{e["network"]: e["disable"] for e in entries},
|
||||
{"clearnet": "tor", "tor": "", "multi": ""},
|
||||
)
|
||||
|
||||
def test_a_host_row_carries_the_local_glyph_where_network_rows_carry_theirs(
|
||||
self,
|
||||
) -> None:
|
||||
entries = _assign(
|
||||
[_row("web-app-a", 0, ("host",)), _row("web-app-a", 0, ("compose",))],
|
||||
[
|
||||
_row("web-app-a", 0, ("host",)),
|
||||
_row("web-app-a", 0, ("compose",), pin_network="tor"),
|
||||
],
|
||||
sweep=0,
|
||||
tor_mode="enforced",
|
||||
network_input="auto",
|
||||
variants_per_app=_VARIANTS,
|
||||
)
|
||||
host, compose = entries
|
||||
@@ -306,50 +460,50 @@ class TestAssign(unittest.TestCase):
|
||||
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()
|
||||
provider = network.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):
|
||||
for value in network.NETWORK_INPUTS:
|
||||
with self.subTest(sweep=sweep, network_input=value):
|
||||
entries = _assign(
|
||||
[_row(provider, 0, ("compose", "swarm"), priority=True)],
|
||||
sweep=sweep,
|
||||
tor_mode=tor_mode,
|
||||
network_input=value,
|
||||
)
|
||||
self.assertEqual(
|
||||
[e["disable"] for e in entries], [""] * len(entries)
|
||||
)
|
||||
self.assertNotIn("false", [e["tor"] for e in entries])
|
||||
self.assertNotIn("clearnet", [e["network"] 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)
|
||||
return _assign([row], sweep=0, network_input="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_a_pinned_network_mode_replaces_the_rotation(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose",), pin_network="multi")
|
||||
self.assertEqual([e["network"] for e in self._entries(row)], ["multi"])
|
||||
|
||||
def test_an_open_axis_still_rotates(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose", "swarm"), pin_tor=False)
|
||||
row = _row("web-app-b", 0, ("compose", "swarm"), pin_network="clearnet")
|
||||
picks = {
|
||||
_assign([row], sweep=sweep, tor_mode="auto", variants_per_app=_VARIANTS)[0][
|
||||
"mode"
|
||||
]
|
||||
_assign(
|
||||
[row], sweep=sweep, network_input="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)
|
||||
def test_pinning_tor_keeps_the_rotation_off_host(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose", "host"), pin_network="tor")
|
||||
for sweep in range(4):
|
||||
with self.subTest(sweep=sweep):
|
||||
entries = _assign(
|
||||
[row], sweep=sweep, tor_mode="auto", variants_per_app=_VARIANTS
|
||||
[row], sweep=sweep, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual([e["mode"] for e in entries], ["compose"])
|
||||
|
||||
@@ -359,8 +513,8 @@ class TestPinnedAxes(unittest.TestCase):
|
||||
)
|
||||
entries = self._entries(row)
|
||||
self.assertEqual(
|
||||
{(e["mode"], e["tor"]) for e in entries},
|
||||
{("compose", "true"), ("compose", "false")},
|
||||
{(e["mode"], e["network"]) for e in entries},
|
||||
{("compose", state) for state in _ALL},
|
||||
)
|
||||
|
||||
def test_a_fully_pinned_priority_row_runs_exactly_once(self) -> None:
|
||||
@@ -370,7 +524,7 @@ class TestPinnedAxes(unittest.TestCase):
|
||||
("compose", "swarm"),
|
||||
priority=True,
|
||||
pin_mode="swarm",
|
||||
pin_tor=False,
|
||||
pin_network="clearnet",
|
||||
)
|
||||
self.assertEqual(len(self._entries(row)), 1)
|
||||
|
||||
@@ -379,15 +533,22 @@ class TestPinnedAxes(unittest.TestCase):
|
||||
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)
|
||||
def test_an_impossible_network_mode_aborts_the_matrix(self) -> None:
|
||||
row = _row("web-app-a", 1, ("compose",), pin_network="tor")
|
||||
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)
|
||||
def test_multi_on_a_single_mode_role_aborts_the_matrix(self) -> None:
|
||||
row = _row("web-app-single", 0, ("compose",), pin_network="multi")
|
||||
with self.assertRaises(SystemExit):
|
||||
_assign([row], sweep=0, tor_mode="disabled", variants_per_app=_VARIANTS)
|
||||
self._entries(row)
|
||||
|
||||
def test_a_pin_fighting_the_runs_network_axis_aborts(self) -> None:
|
||||
row = _row("web-app-b", 0, ("compose",), pin_network="tor")
|
||||
with self.assertRaises(SystemExit):
|
||||
_assign(
|
||||
[row], sweep=0, network_input="clearnet", variants_per_app=_VARIANTS
|
||||
)
|
||||
|
||||
|
||||
class TestResolvePool(unittest.TestCase):
|
||||
@@ -418,7 +579,7 @@ class TestDistroAndFilesystemAxes(unittest.TestCase):
|
||||
entries = _assign(
|
||||
self._rows(len(axes.DISTROS)),
|
||||
sweep=0,
|
||||
tor_mode="auto",
|
||||
network_input="auto",
|
||||
variants_per_app=_VARIANTS,
|
||||
)
|
||||
self.assertEqual([e["distro"] for e in entries], list(axes.DISTROS))
|
||||
@@ -426,7 +587,10 @@ class TestDistroAndFilesystemAxes(unittest.TestCase):
|
||||
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
|
||||
self._rows(1),
|
||||
sweep=sweep,
|
||||
network_input="auto",
|
||||
variants_per_app=_VARIANTS,
|
||||
)[0]["distro"]
|
||||
for sweep in range(len(axes.DISTROS))
|
||||
]
|
||||
@@ -436,7 +600,7 @@ class TestDistroAndFilesystemAxes(unittest.TestCase):
|
||||
entries = _assign(
|
||||
self._rows(4),
|
||||
sweep=0,
|
||||
tor_mode="auto",
|
||||
network_input="auto",
|
||||
distros=("debian",),
|
||||
filesystems=("btrfs",),
|
||||
variants_per_app=_VARIANTS,
|
||||
@@ -446,17 +610,26 @@ class TestDistroAndFilesystemAxes(unittest.TestCase):
|
||||
|
||||
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))
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual(
|
||||
len({e["distro"] for e in entries}),
|
||||
min(len(entries), len(axes.DISTROS)),
|
||||
)
|
||||
|
||||
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)
|
||||
entries = _assign(
|
||||
rows, sweep=3, network_input="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)
|
||||
entries = _assign(
|
||||
rows, sweep=1, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
self.assertEqual(entries[0]["filesystem"], "ext4")
|
||||
|
||||
def test_collapsing_two_tokens_keeps_the_stronger_filesystem_claim(self) -> None:
|
||||
@@ -470,7 +643,7 @@ class TestDistroAndFilesystemAxes(unittest.TestCase):
|
||||
entries = _assign(
|
||||
rows,
|
||||
sweep=0,
|
||||
tor_mode="disabled",
|
||||
network_input="clearnet",
|
||||
distros=("debian",),
|
||||
filesystems=("zfs",),
|
||||
variants_per_app=_VARIANTS,
|
||||
@@ -484,16 +657,16 @@ class TestDistroAndFilesystemAxes(unittest.TestCase):
|
||||
_assign(
|
||||
rows,
|
||||
sweep=0,
|
||||
tor_mode="auto",
|
||||
network_input="auto",
|
||||
distros=("debian",),
|
||||
variants_per_app=_VARIANTS,
|
||||
)
|
||||
|
||||
def test_the_glyphs_follow_the_onion_slot_in_the_label(self) -> None:
|
||||
def test_the_glyphs_follow_the_network_slot_in_the_label(self) -> None:
|
||||
entry = _assign(
|
||||
[_row("web-app-b", 0, ("compose",))],
|
||||
sweep=0,
|
||||
tor_mode="disabled",
|
||||
network_input="clearnet",
|
||||
distros=("debian",),
|
||||
filesystems=("zfs",),
|
||||
variants_per_app=_VARIANTS,
|
||||
@@ -515,7 +688,7 @@ class TestSortKey(unittest.TestCase):
|
||||
_row("web-app-a", 1, ("compose",)),
|
||||
_row("web-app-a", 0, ("compose",)),
|
||||
]
|
||||
return _assign(rows, sweep=0, tor_mode="auto", variants_per_app=_VARIANTS)
|
||||
return _assign(rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS)
|
||||
|
||||
def test_rows_sort_by_name_then_variant(self) -> None:
|
||||
ordered = sorted(self._entries(), key=axes.sort_key)
|
||||
@@ -526,24 +699,24 @@ class TestSortKey(unittest.TestCase):
|
||||
|
||||
def test_the_mode_sorts_in_deploy_order_not_alphabetically(self) -> None:
|
||||
entries = [
|
||||
{"apps": "web-app-a", "variant": "0", "mode": mode, "tor": "false"}
|
||||
{"apps": "web-app-a", "variant": "0", "mode": mode, "network": "clearnet"}
|
||||
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:
|
||||
def test_clearnet_sorts_ahead_of_tor_ahead_of_multi(self) -> None:
|
||||
entries = [
|
||||
{"apps": "web-app-a", "variant": "0", "mode": "compose", "tor": tor}
|
||||
for tor in ("true", "false")
|
||||
{"apps": "web-app-a", "variant": "0", "mode": "compose", "network": state}
|
||||
for state in ("multi", "tor", "clearnet")
|
||||
]
|
||||
ordered = sorted(entries, key=axes.sort_key)
|
||||
self.assertEqual([e["tor"] for e in ordered], ["false", "true"])
|
||||
self.assertEqual([e["network"] for e in ordered], list(_ALL))
|
||||
|
||||
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"},
|
||||
{"apps": "web-app-a", "variant": "0", "mode": "compose", "network": "tor"},
|
||||
{"apps": "web-app-a", "variant": "", "mode": "compose", "network": "tor"},
|
||||
]
|
||||
ordered = sorted(entries, key=axes.sort_key)
|
||||
self.assertEqual([e["variant"] for e in ordered], ["", "0"])
|
||||
@@ -552,7 +725,7 @@ class TestSortKey(unittest.TestCase):
|
||||
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)[
|
||||
return _assign(rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS)[
|
||||
0
|
||||
]["label"]
|
||||
|
||||
@@ -566,15 +739,17 @@ class TestParseLabel(unittest.TestCase):
|
||||
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:
|
||||
def test_the_network_mode_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})
|
||||
entries = _assign(
|
||||
rows, sweep=0, network_input="auto", variants_per_app=_VARIANTS
|
||||
)
|
||||
parsed = {axes.parse_label(e["label"]).network for e in entries}
|
||||
self.assertEqual(parsed, set(_ALL))
|
||||
|
||||
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)
|
||||
self.assertEqual(axes.parse_label(title).network, "clearnet")
|
||||
|
||||
def test_the_priority_star_does_not_bleed_into_the_name(self) -> None:
|
||||
title = self._title("compose", "web-app-a", "0", priority=True)
|
||||
@@ -591,13 +766,16 @@ class TestParseLabel(unittest.TestCase):
|
||||
entry = _assign(
|
||||
[_row("web-app-a", 0, ("swarm",))],
|
||||
sweep=0,
|
||||
tor_mode="enforced",
|
||||
network_input="multi",
|
||||
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(
|
||||
(label.network, label.distro, label.filesystem),
|
||||
("multi", "centos", "btrfs"),
|
||||
)
|
||||
self.assertEqual(display_names().decode(label.name), "web-app-a")
|
||||
|
||||
def test_a_non_deploy_job_yields_nothing(self) -> None:
|
||||
@@ -610,9 +788,9 @@ class TestEnvironmentReads(unittest.TestCase):
|
||||
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")
|
||||
def test_the_network_input_comes_from_the_environment(self) -> None:
|
||||
with mock.patch.dict("os.environ", {"INFINITO_NETWORK": "multi"}):
|
||||
self.assertEqual(network.resolve_network_input(), "multi")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -19,7 +19,7 @@ class TestParseAscii(unittest.TestCase):
|
||||
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"),
|
||||
selection.Pin("web-app-a", (0, 2), "swarm", "tor", "debian", "zfs"),
|
||||
)
|
||||
|
||||
def test_the_distro_and_filesystem_are_optional_like_the_rest(self) -> None:
|
||||
@@ -44,8 +44,12 @@ class TestParseAscii(unittest.TestCase):
|
||||
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_every_network_word_pins_its_network_mode(self) -> None:
|
||||
for word in ("clearnet", "tor", "multi"):
|
||||
with self.subTest(word=word):
|
||||
self.assertEqual(
|
||||
selection.parse(f"web-app-a@compose+{word}").network, word
|
||||
)
|
||||
|
||||
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"))
|
||||
@@ -54,7 +58,7 @@ class TestParseAscii(unittest.TestCase):
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.parse("web-app-a@swrm")
|
||||
|
||||
def test_an_unknown_onion_state_aborts(self) -> None:
|
||||
def test_an_unknown_network_mode_aborts(self) -> None:
|
||||
with self.assertRaises(SystemExit):
|
||||
selection.parse("web-app-a+onion")
|
||||
|
||||
@@ -64,17 +68,21 @@ class TestParseAscii(unittest.TestCase):
|
||||
|
||||
|
||||
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 _label(self, mode: str, network: str) -> str:
|
||||
return f"{to_emoji(mode)}{to_emoji(network)}web-app-a#2"
|
||||
|
||||
def test_the_glyphs_carry_mode_and_onion(self) -> None:
|
||||
def test_the_glyphs_carry_mode_and_network_mode(self) -> None:
|
||||
self.assertEqual(
|
||||
selection.parse(self._label("swarm", "tor")),
|
||||
selection.Pin("web-app-a", (2,), "swarm", True),
|
||||
selection.Pin("web-app-a", (2,), "swarm", "tor"),
|
||||
)
|
||||
|
||||
def test_the_clearnet_glyph_pins_the_clearnet_state(self) -> None:
|
||||
self.assertFalse(selection.parse(self._label("compose", "clearnet")).tor)
|
||||
def test_every_network_glyph_pins_its_network_mode(self) -> None:
|
||||
for word in ("clearnet", "tor", "multi"):
|
||||
with self.subTest(word=word):
|
||||
self.assertEqual(
|
||||
selection.parse(self._label("compose", word)).network, word
|
||||
)
|
||||
|
||||
def test_the_priority_star_is_not_part_of_the_name(self) -> None:
|
||||
pasted = f"{self._label('compose', 'tor')} {to_emoji('priority')}"
|
||||
@@ -93,7 +101,7 @@ class TestParseLabel(unittest.TestCase):
|
||||
)
|
||||
self.assertEqual(
|
||||
selection.parse(pasted),
|
||||
selection.Pin("web-app-a", (2,), "swarm", True, "fedora", "btrfs"),
|
||||
selection.Pin("web-app-a", (2,), "swarm", "tor", "fedora", "btrfs"),
|
||||
)
|
||||
|
||||
def test_a_title_pasted_with_its_caller_path_aborts_rather_than_guessing(
|
||||
@@ -136,11 +144,11 @@ class TestApply(unittest.TestCase):
|
||||
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"])
|
||||
self.assertEqual(kept[0]["pin_network"], "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_mode"], kept[0]["pin_network"]], [None, None])
|
||||
self.assertEqual(
|
||||
[kept[0]["pin_distro"], kept[0]["pin_filesystem"]], [None, None]
|
||||
)
|
||||
@@ -162,10 +170,19 @@ class TestApply(unittest.TestCase):
|
||||
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)],
|
||||
[(row["variant"], row["pin_mode"], row["pin_network"]) for row in kept],
|
||||
[(0, "compose", "tor"), (0, "swarm", "tor")],
|
||||
)
|
||||
|
||||
def test_one_variant_that_failed_in_two_network_modes_comes_back_twice(
|
||||
self,
|
||||
) -> None:
|
||||
kept = selection.apply(
|
||||
_ROWS,
|
||||
selection.parse_list("web-app-a#0@swarm+tor web-app-a#0@swarm+multi"),
|
||||
)
|
||||
self.assertEqual([row["pin_network"] for row in kept], ["tor", "multi"])
|
||||
|
||||
def test_the_same_narrowing_written_twice_stays_one_deploy(self) -> None:
|
||||
kept = selection.apply(
|
||||
_ROWS,
|
||||
|
||||
Reference in new issue
Block a user