mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-10 04:36:40 +00:00
Merge origin/main into feature/svc-net-wireguard
Brings in 225 commits, including the merged web-app-openbao work and the
LMStudio loopback fix.
Seven conflicts:
- The legacy svc-net-wireguard-core/plain/firewalled roles this branch deleted
in e706170a0 were modified upstream. The deletion stands, which is the whole
premise of the branch.
- The MCP audit takes upstream's list plus svc-net-firewall, with the three
legacy wireguard entries collapsed into the single svc-net-wireguard: 23.
- web-app-openbao/README.md takes main's reviewed version wholesale.
- matrix.py takes both sides: upstream's app_id argument to _reset_credentials
and this branch's _write_mesh call after it.
The merge pushed matrix.py to 511 lines against the 500 cap, so the step
runner moved to utils/tests/swarm/run.py, which is self-contained and leaves
matrix.py at 427. Its test moved with it.
This commit is contained in:
437 files changed
+12903
-1724
No files matched your search
+24
-2
@@ -111,13 +111,35 @@ class TestStableVariablesSignature(unittest.TestCase):
|
||||
def test_empty_variables_collapses_to_canonical_tuple(self):
|
||||
self.assertEqual(
|
||||
base._stable_variables_signature(None),
|
||||
("0", "0", "", ""),
|
||||
("0", "0", "", "", ""),
|
||||
)
|
||||
self.assertEqual(
|
||||
base._stable_variables_signature({}),
|
||||
("0", "0", "", ""),
|
||||
("0", "0", "", "", ""),
|
||||
)
|
||||
|
||||
def test_two_hosts_of_one_play_do_not_share_a_key(self):
|
||||
shared = {"applications": {"web-app-pretix": {}}, "DOMAIN_PRIMARY": "x.test"}
|
||||
member = base._stable_variables_signature(
|
||||
{**shared, "group_names": ["web-app-pretix", "svc-db-postgres"]}
|
||||
)
|
||||
outsider = base._stable_variables_signature(
|
||||
{**shared, "group_names": ["web-app-pretix"]}
|
||||
)
|
||||
|
||||
self.assertNotEqual(
|
||||
member,
|
||||
outsider,
|
||||
"a service flag may read group_names, so one host's payload must "
|
||||
"never answer for a host whose groups differ",
|
||||
)
|
||||
|
||||
def test_group_order_does_not_split_the_key(self):
|
||||
first = base._stable_variables_signature({"group_names": ["a", "b"]})
|
||||
second = base._stable_variables_signature({"group_names": ["b", "a"]})
|
||||
|
||||
self.assertEqual(first, second)
|
||||
|
||||
def test_includes_domain_primary_and_email_domain_strings(self):
|
||||
sig = base._stable_variables_signature(
|
||||
{
|
||||
|
||||
+2
-2
@@ -129,8 +129,8 @@ class TestStableVariablesSignature(unittest.TestCase):
|
||||
_reset_cache_for_tests()
|
||||
|
||||
def test_empty_variables_returns_sentinel(self):
|
||||
self.assertEqual(_stable_variables_signature(None), ("0", "0", "", ""))
|
||||
self.assertEqual(_stable_variables_signature({}), ("0", "0", "", ""))
|
||||
self.assertEqual(_stable_variables_signature(None), ("0", "0", "", "", ""))
|
||||
self.assertEqual(_stable_variables_signature({}), ("0", "0", "", "", ""))
|
||||
|
||||
def test_captures_key_subset(self):
|
||||
sig = _stable_variables_signature(
|
||||
|
||||
+66
@@ -337,5 +337,71 @@ class TestOnionDomainInjection(unittest.TestCase):
|
||||
self.assertEqual(out["web-app-x"], ["infinito.test", self.ONION])
|
||||
|
||||
|
||||
class TestOnionFlagsAreRendered(unittest.TestCase):
|
||||
"""`services.tor.*` reaches this map as Jinja in 99 of the 105 roles that
|
||||
declare it, and as its own source text whenever the applications render is
|
||||
still in flight. Reading that text as a boolean answers False and drops the
|
||||
onion domains for every app, which is what froze a clearnet asset URL into
|
||||
the applications payload of run 34922994823.
|
||||
"""
|
||||
|
||||
ONION = TestOnionDomainInjection.ONION
|
||||
JINJA_ON = "{{ 'svc-net-tor' in group_names }}"
|
||||
|
||||
_apps = TestOnionDomainInjection._apps
|
||||
|
||||
def _templar(self, **variables):
|
||||
from ansible.parsing.dataloader import DataLoader
|
||||
from ansible.template import Templar
|
||||
|
||||
templar = Templar(loader=DataLoader(), variables=dict(variables))
|
||||
templar.available_variables = dict(variables)
|
||||
return templar
|
||||
|
||||
def _inject(self, apps, **variables):
|
||||
return cache_domains._inject_onion_domains(
|
||||
{"web-app-x": ["x.infinito.test"]},
|
||||
apps,
|
||||
"infinito.test",
|
||||
self.ONION,
|
||||
templar=self._templar(**variables),
|
||||
variables=dict(variables),
|
||||
)
|
||||
|
||||
def test_a_templated_enabled_flag_still_injects_the_onion(self):
|
||||
out = self._inject(
|
||||
self._apps(enabled=self.JINJA_ON),
|
||||
group_names=["svc-net-tor", "web-app-x"],
|
||||
)
|
||||
|
||||
self.assertEqual(out["web-app-x"], ["x.infinito.test", f"x.{self.ONION}"])
|
||||
|
||||
def test_a_templated_flag_resolving_false_leaves_the_clearnet_domain(self):
|
||||
out = self._inject(self._apps(enabled=self.JINJA_ON), group_names=["web-app-x"])
|
||||
|
||||
self.assertEqual(out["web-app-x"], ["x.infinito.test"])
|
||||
|
||||
def test_templated_exclusive_replaces_the_clearnet_domain(self):
|
||||
out = self._inject(
|
||||
self._apps(enabled=self.JINJA_ON, exclusive=self.JINJA_ON),
|
||||
group_names=["svc-net-tor", "web-app-x"],
|
||||
)
|
||||
|
||||
self.assertEqual(out["web-app-x"], [f"x.{self.ONION}"])
|
||||
|
||||
def test_a_flag_that_cannot_be_rendered_is_an_error(self):
|
||||
"""Without a templar the text cannot resolve, and answering False
|
||||
would drop the onion domains without saying so."""
|
||||
with self.assertRaisesRegex(ValueError, "services.tor.enabled"):
|
||||
cache_domains._inject_onion_domains(
|
||||
{"web-app-x": ["x.infinito.test"]},
|
||||
self._apps(enabled=self.JINJA_ON),
|
||||
"infinito.test",
|
||||
self.ONION,
|
||||
templar=None,
|
||||
variables={"group_names": ["svc-net-tor"]},
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+28
@@ -164,5 +164,33 @@ class TestReset(_ProjectRootFixture, unittest.TestCase):
|
||||
self.assertEqual(read_text(str(self.root / "x.txt")), "v1-changed")
|
||||
|
||||
|
||||
class TestWalkIsKeyedOnTheRoot(_ProjectRootFixture, unittest.TestCase):
|
||||
def test_a_second_root_is_walked_rather_than_answered_from_the_first(
|
||||
self,
|
||||
) -> None:
|
||||
_touch(self.root / "first.txt", "a")
|
||||
self.assertEqual(
|
||||
{str(self.root / "first.txt")},
|
||||
set(iter_project_files()),
|
||||
)
|
||||
|
||||
with TemporaryDirectory() as other:
|
||||
other_root = Path(other)
|
||||
_touch(other_root / "second.txt", "b")
|
||||
with patch.object(files_module, "PROJECT_ROOT", other_root):
|
||||
self.assertEqual(
|
||||
{str(other_root / "second.txt")},
|
||||
set(iter_project_files()),
|
||||
"the walk answered for the previous root, so a test that "
|
||||
"points PROJECT_ROOT at a fixture silently sweeps the real "
|
||||
"repository instead",
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
{str(self.root / "first.txt")},
|
||||
set(iter_project_files()),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
unittest.main()
|
||||
@@ -0,0 +1,99 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from utils.docker.image.discovery import docker_hub_source, image_source
|
||||
from utils.docker.image.pin import (
|
||||
DIGEST,
|
||||
REF,
|
||||
RULE_BY_CLASS,
|
||||
SEMVER,
|
||||
is_digest,
|
||||
pin_class,
|
||||
pull_reference,
|
||||
reference_separator,
|
||||
)
|
||||
|
||||
DIGEST_VALUE = "sha256:d5591131b1d898cd02fc7a17adb8ea026352d67809bc9289707778e9e03a76a4"
|
||||
|
||||
|
||||
class TestPinClass(unittest.TestCase):
|
||||
def test_a_semver_tag_is_a_semver_pin(self):
|
||||
for value in ("3.0.1", "16", "17-3.5", "5.4.5-php8.3-apache", "v1.12.27"):
|
||||
self.assertEqual(pin_class(value), SEMVER, value)
|
||||
|
||||
def test_a_content_address_is_a_digest_pin(self):
|
||||
self.assertEqual(pin_class(DIGEST_VALUE), DIGEST)
|
||||
|
||||
def test_a_moving_name_is_neither(self):
|
||||
for value in ("latest", "main", "stable", "buildx-stable-1"):
|
||||
self.assertEqual(pin_class(value), REF, value)
|
||||
|
||||
def test_surrounding_whitespace_does_not_change_the_class(self):
|
||||
self.assertEqual(pin_class(f" {DIGEST_VALUE} "), DIGEST)
|
||||
self.assertEqual(pin_class(" 3.0.1 "), SEMVER)
|
||||
|
||||
def test_a_digest_declares_itself_apart_from_a_tag(self):
|
||||
self.assertNotEqual(
|
||||
RULE_BY_CLASS[DIGEST],
|
||||
RULE_BY_CLASS[REF],
|
||||
"a digest is chosen for a reason a non-semver tag never carries, so "
|
||||
"the two cannot share one marker",
|
||||
)
|
||||
self.assertEqual(RULE_BY_CLASS[DIGEST], "docker-digest")
|
||||
self.assertEqual(RULE_BY_CLASS[REF], "docker-version")
|
||||
|
||||
def test_a_semver_pin_needs_no_declaration(self):
|
||||
self.assertNotIn(SEMVER, RULE_BY_CLASS)
|
||||
|
||||
|
||||
class TestPullReference(unittest.TestCase):
|
||||
def test_a_digest_is_joined_with_an_at_sign(self):
|
||||
self.assertEqual(reference_separator(DIGEST_VALUE), "@")
|
||||
self.assertEqual(
|
||||
pull_reference("ghcr.io/owner/app", DIGEST_VALUE),
|
||||
f"ghcr.io/owner/app@{DIGEST_VALUE}",
|
||||
)
|
||||
|
||||
def test_a_tag_is_joined_with_a_colon(self):
|
||||
self.assertEqual(reference_separator("3.0.1"), ":")
|
||||
self.assertEqual(
|
||||
pull_reference("ghcr.io/owner/app", "3.0.1"), "ghcr.io/owner/app:3.0.1"
|
||||
)
|
||||
|
||||
def test_is_digest_rejects_a_tag_that_merely_mentions_sha256(self):
|
||||
self.assertFalse(is_digest("v2-sha256:abc"))
|
||||
|
||||
|
||||
class TestSourceKeepsTheDigest(unittest.TestCase):
|
||||
"""A pull ref that silently drops the digest pins nothing at all."""
|
||||
|
||||
def test_image_source_keeps_a_digest(self):
|
||||
self.assertEqual(
|
||||
image_source("ghcr.io/owner/app", DIGEST_VALUE),
|
||||
f"ghcr.io/owner/app@{DIGEST_VALUE}",
|
||||
)
|
||||
|
||||
def test_image_source_still_tags_a_semver(self):
|
||||
self.assertEqual(
|
||||
image_source("ghcr.io/owner/app", "3.0.1"), "ghcr.io/owner/app:3.0.1"
|
||||
)
|
||||
|
||||
def test_docker_hub_source_keeps_a_digest(self):
|
||||
self.assertEqual(
|
||||
docker_hub_source("postgres", DIGEST_VALUE),
|
||||
f"docker.io/library/postgres@{DIGEST_VALUE}",
|
||||
)
|
||||
|
||||
def test_a_digest_never_yields_a_tag_separator(self):
|
||||
source = image_source("quay.io/owner/app", DIGEST_VALUE)
|
||||
|
||||
self.assertNotIn(
|
||||
f":{DIGEST_VALUE}",
|
||||
source,
|
||||
"a colon here would make the registry read the digest as a tag name",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -199,6 +199,11 @@ class TestApply(unittest.TestCase):
|
||||
def test_a_bare_role_that_matches_nothing_is_tolerated(self) -> None:
|
||||
self.assertEqual(selection.apply(_ROWS, selection.parse_list("web-app-c")), [])
|
||||
|
||||
def test_a_pin_on_an_undiscovered_role_is_tolerated_like_a_bare_one(self) -> None:
|
||||
self.assertEqual(
|
||||
selection.apply(_ROWS, selection.parse_list("web-app-c#0,2")), []
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,260 @@
|
||||
import json
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from utils import PROJECT_ROOT
|
||||
from utils.cache.files import read_text
|
||||
from utils.cache.yaml import load_yaml_any
|
||||
from utils.install.collections import declared_pins, unsatisfied
|
||||
|
||||
_DOCKERFILE = PROJECT_ROOT / "Dockerfile"
|
||||
_COPY = re.compile(r"^COPY\s+(?P<source>\S+)\s+\$\{INFINITO_SRC_DIR\}/(?P<target>\S+)$")
|
||||
|
||||
|
||||
def _materialise_image_slice(destination: Path) -> None:
|
||||
"""Reproduce what the image holds when it first runs the collection check.
|
||||
|
||||
The Dockerfile copies a handful of paths, runs ``scripts/install/ansible.sh``
|
||||
against them, and only afterwards copies the rest of the tree. Anything
|
||||
``utils.install.collections`` imports beyond that slice is absent in the
|
||||
layer that executes it, which no import inside a full checkout can show.
|
||||
The list is read from the Dockerfile rather than restated here, so a COPY
|
||||
added or dropped moves this probe with it.
|
||||
|
||||
The caller must run the slice under ``python -S``. This repository installs
|
||||
``utils`` into site-packages, so without it the interpreter resolves every
|
||||
submodule from the checkout however the slice is built, and the probe
|
||||
passes on a module that cannot import inside the image.
|
||||
|
||||
Args:
|
||||
destination: an empty directory to build the slice in.
|
||||
"""
|
||||
for line in read_text(str(_DOCKERFILE)).splitlines():
|
||||
if line.startswith("COPY . "):
|
||||
return
|
||||
match = _COPY.match(line.strip())
|
||||
if not match:
|
||||
continue
|
||||
source = PROJECT_ROOT / match.group("source")
|
||||
target = destination / match.group("target")
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
if source.is_dir():
|
||||
shutil.copytree(source, target, dirs_exist_ok=True)
|
||||
else:
|
||||
shutil.copy2(source, target)
|
||||
|
||||
|
||||
_REQUIREMENTS = (
|
||||
PROJECT_ROOT / "requirements" / "requirements.galaxy.yml",
|
||||
PROJECT_ROOT / "requirements" / "requirements.git.yml",
|
||||
)
|
||||
|
||||
_BOOTSTRAP_PROBE = """
|
||||
import sys
|
||||
|
||||
|
||||
class _NoPyYAML:
|
||||
def find_spec(self, fullname, path=None, target=None):
|
||||
if fullname == "yaml" or fullname.startswith("yaml."):
|
||||
raise ImportError("PyYAML is absent during the lint bootstrap")
|
||||
return None
|
||||
|
||||
|
||||
sys.meta_path.insert(0, _NoPyYAML())
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from utils.install.lint.ansible import collections as lint_collections
|
||||
from utils.install.collections import declared_pins
|
||||
|
||||
for fqcn, version in declared_pins(Path("requirements/requirements.galaxy.yml")):
|
||||
print(fqcn, version)
|
||||
"""
|
||||
|
||||
REQUIREMENTS = """---
|
||||
collections:
|
||||
- name: community.general
|
||||
version: 13.4.0
|
||||
- name: hetzner.hcloud
|
||||
version: 7.0.1
|
||||
"""
|
||||
|
||||
|
||||
def _write_requirements(root: Path, body: str = REQUIREMENTS) -> Path:
|
||||
path = root / "requirements.galaxy.yml"
|
||||
path.write_text(body, encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def _install(collections_dir: Path, fqcn: str, version: str) -> None:
|
||||
namespace, _, name = fqcn.partition(".")
|
||||
target = collections_dir / "ansible_collections" / namespace / name
|
||||
target.mkdir(parents=True)
|
||||
(target / "MANIFEST.json").write_text(
|
||||
json.dumps({"collection_info": {"version": version}}), encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
class TestAgainstPyYAML(unittest.TestCase):
|
||||
def test_the_hand_parser_agrees_with_pyyaml(self):
|
||||
for path in _REQUIREMENTS:
|
||||
with self.subTest(requirements=path.name):
|
||||
declared = load_yaml_any(str(path), default_if_missing={}) or {}
|
||||
expected = [
|
||||
(str(entry.get("name")), entry.get("version"))
|
||||
for entry in declared.get("collections") or []
|
||||
if isinstance(entry, dict) and entry.get("name")
|
||||
]
|
||||
expected = [
|
||||
(name, None if version is None else str(version))
|
||||
for name, version in expected
|
||||
]
|
||||
|
||||
self.assertEqual(declared_pins(path), expected)
|
||||
|
||||
def test_the_parser_runs_with_pyyaml_unimportable(self):
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", _BOOTSTRAP_PROBE],
|
||||
cwd=str(PROJECT_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertIn("community.general", result.stdout)
|
||||
|
||||
def test_the_checker_runs_on_the_slice_the_dockerfile_copies(self):
|
||||
tmp = TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
slice_root = Path(tmp.name)
|
||||
_materialise_image_slice(slice_root)
|
||||
requirements = slice_root / "requirements" / "requirements.galaxy.yml"
|
||||
self.assertTrue(
|
||||
requirements.is_file(),
|
||||
"the Dockerfile stopped copying requirements/ before the ansible "
|
||||
"install layer; this probe needs it to have something to check",
|
||||
)
|
||||
|
||||
result = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-S",
|
||||
"-m",
|
||||
"utils.install.collections",
|
||||
str(requirements),
|
||||
str(slice_root / "collections"),
|
||||
],
|
||||
cwd=str(slice_root),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
self.assertNotIn(
|
||||
"ModuleNotFoundError",
|
||||
result.stderr,
|
||||
"utils.install.collections reached for a module the image layer "
|
||||
"that runs it does not hold; keep it on the standard library",
|
||||
)
|
||||
self.assertEqual(
|
||||
result.returncode,
|
||||
1,
|
||||
f"an empty collections dir satisfies nothing, so the checker owes "
|
||||
f"exit 1 and a list; stderr: {result.stderr}",
|
||||
)
|
||||
self.assertIn("community.general", result.stdout)
|
||||
|
||||
|
||||
class TestDeclaredPins(unittest.TestCase):
|
||||
def test_a_quoted_version_is_read_without_its_quotes(self):
|
||||
body = (
|
||||
"---\ncollections:\n"
|
||||
' - name: double.quoted\n version: "13.4.0"\n'
|
||||
" - name: single.quoted\n version: '1.0.0'\n"
|
||||
" - name: commented\n version: 2.2.2 # pinned\n"
|
||||
)
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
requirements = _write_requirements(root, body)
|
||||
|
||||
self.assertEqual(
|
||||
declared_pins(requirements),
|
||||
[
|
||||
("double.quoted", "13.4.0"),
|
||||
("single.quoted", "1.0.0"),
|
||||
("commented", "2.2.2"),
|
||||
],
|
||||
)
|
||||
|
||||
def test_an_entry_without_a_version_stays_unpinned(self):
|
||||
body = "---\ncollections:\n - name: bare.entry\n"
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
|
||||
self.assertEqual(
|
||||
declared_pins(_write_requirements(root, body)), [("bare.entry", None)]
|
||||
)
|
||||
|
||||
def test_a_missing_file_declares_nothing(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
self.assertEqual(declared_pins(Path(tmp) / "absent.yml"), [])
|
||||
|
||||
|
||||
class TestUnsatisfied(unittest.TestCase):
|
||||
def test_every_pin_installed_at_its_version_is_satisfied(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
requirements = _write_requirements(root)
|
||||
_install(root, "community.general", "13.4.0")
|
||||
_install(root, "hetzner.hcloud", "7.0.1")
|
||||
|
||||
self.assertEqual(unsatisfied(requirements, root), [])
|
||||
|
||||
def test_absent_collection_is_reported(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
requirements = _write_requirements(root)
|
||||
_install(root, "community.general", "13.4.0")
|
||||
|
||||
self.assertEqual(unsatisfied(requirements, root), ["hetzner.hcloud:7.0.1"])
|
||||
|
||||
def test_wrong_version_is_reported_although_the_directory_exists(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
requirements = _write_requirements(root)
|
||||
_install(root, "community.general", "13.4.0")
|
||||
_install(root, "hetzner.hcloud", "6.9.0")
|
||||
|
||||
self.assertEqual(unsatisfied(requirements, root), ["hetzner.hcloud:7.0.1"])
|
||||
|
||||
def test_unpinned_entry_can_never_be_confirmed(self):
|
||||
body = "---\ncollections:\n - name: community.general\n"
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
requirements = _write_requirements(root, body)
|
||||
_install(root, "community.general", "13.4.0")
|
||||
|
||||
self.assertEqual(unsatisfied(requirements, root), ["community.general"])
|
||||
|
||||
def test_unreadable_manifest_is_reported(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
requirements = _write_requirements(root)
|
||||
_install(root, "community.general", "13.4.0")
|
||||
_install(root, "hetzner.hcloud", "7.0.1")
|
||||
manifest = (
|
||||
root / "ansible_collections" / "hetzner" / "hcloud" / "MANIFEST.json"
|
||||
)
|
||||
manifest.write_text("{not json", encoding="utf-8")
|
||||
|
||||
self.assertEqual(unsatisfied(requirements, root), ["hetzner.hcloud:7.0.1"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,13 +1,16 @@
|
||||
import base64
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import ClassVar
|
||||
from unittest import TestCase, main, mock
|
||||
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
|
||||
|
||||
from utils.cache.files import PROJECT_ROOT
|
||||
from utils.cache.yaml import load_yaml_any
|
||||
from utils.handler.vault import VaultScalar
|
||||
from utils.manager.inventory import InventoryManager
|
||||
from utils.manager.inventory import InventoryManager, validate_supplied_value
|
||||
from utils.manager.value_generator import ValueGenerator
|
||||
from utils.roles.mapping import (
|
||||
ROLE_FILE_META_SECRETS,
|
||||
@@ -760,5 +763,104 @@ class TestInventoryManagerVariant(TestCase):
|
||||
self.assertEqual(cfg, {})
|
||||
|
||||
|
||||
class TestSuppliedValueValidation(TestCase):
|
||||
"""A schema that declares a shape must reject a value contradicting it.
|
||||
|
||||
Without this the declaration is decoration: the 192 `validation:` blocks
|
||||
already in the role schemas are enforced nowhere, and a `regex` would have
|
||||
joined them.
|
||||
"""
|
||||
|
||||
SCHEMA: ClassVar[dict[str, str]] = {
|
||||
"type": "string",
|
||||
"regex": "sk-[A-Za-z0-9]{4,}",
|
||||
}
|
||||
|
||||
def test_a_matching_value_passes(self):
|
||||
validate_supplied_value("credentials.api_key", "sk-abcd1234", self.SCHEMA)
|
||||
|
||||
def test_an_empty_value_is_not_configured_and_passes(self):
|
||||
validate_supplied_value("credentials.api_key", "", self.SCHEMA)
|
||||
validate_supplied_value("credentials.api_key", None, self.SCHEMA)
|
||||
|
||||
def test_a_value_failing_the_pattern_aborts(self):
|
||||
with self.assertRaises(SystemExit):
|
||||
validate_supplied_value("credentials.api_key", "nope", self.SCHEMA)
|
||||
|
||||
def test_a_partial_match_is_not_enough(self):
|
||||
with self.assertRaises(SystemExit):
|
||||
validate_supplied_value("credentials.api_key", "sk-ab!!", self.SCHEMA)
|
||||
|
||||
def test_a_value_of_the_wrong_type_aborts(self):
|
||||
with self.assertRaises(SystemExit):
|
||||
validate_supplied_value("credentials.port", 8080, {"type": "string"})
|
||||
|
||||
def test_an_unknown_declared_type_aborts(self):
|
||||
with self.assertRaises(SystemExit):
|
||||
validate_supplied_value("credentials.x", "v", {"type": "uuid"})
|
||||
|
||||
def test_a_schema_declaring_neither_accepts_anything(self):
|
||||
validate_supplied_value("credentials.x", "anything", {"description": "free"})
|
||||
|
||||
def test_an_unrendered_template_is_not_a_value(self):
|
||||
for value in ("{{ API.openai.api_key }}", "{% if x %}a{% endif %}"):
|
||||
with self.subTest(value=value):
|
||||
validate_supplied_value("credentials.api_key", value, self.SCHEMA)
|
||||
|
||||
|
||||
class TestProviderKeyPattern(TestCase):
|
||||
"""The shipped provider-key pattern must outlive a vendor's format change.
|
||||
|
||||
OpenAI moved from `sk-` to `sk-proj-` without notice. A schema that pinned
|
||||
the prefix would reject a valid key and abort inventory creation, so the
|
||||
pattern checks what a paste can get wrong instead.
|
||||
"""
|
||||
|
||||
ACCEPTED: ClassVar[tuple[str, ...]] = (
|
||||
"sk-proj-AbCdEfGh1234567890xyz",
|
||||
"sk-ant-api03-AbCdEfGh1234567890",
|
||||
"sk-or-v1-AbCdEfGh1234567890abc",
|
||||
"sk-AbCdEfGh1234567890legacyKey",
|
||||
"some-future-format-2099-abcdef",
|
||||
)
|
||||
REJECTED: ClassVar[tuple[str, ...]] = (
|
||||
"sk-proj-AbCdEfGh1234567890xyz\n",
|
||||
"sk-short",
|
||||
"<your-key-here>",
|
||||
"sk-proj-Ab Cd 1234567890 xyz",
|
||||
)
|
||||
|
||||
def _credentials(self) -> dict:
|
||||
path = PROJECT_ROOT / "roles/svc-ai-litellm" / ROLE_FILE_META_SECRETS
|
||||
return load_yaml_any(str(path))["credentials"]
|
||||
|
||||
def _schema(self) -> dict:
|
||||
return self._credentials()["openai_api_key"]
|
||||
|
||||
def test_every_declared_default_survives_its_own_schema(self):
|
||||
"""A default is written verbatim and read back as a supplied value.
|
||||
|
||||
Run 34883491022 died in every job because the provider keys' own
|
||||
`{{ API.<vendor>.api_key }}` reached the pattern on the second read.
|
||||
"""
|
||||
for key, schema in self._credentials().items():
|
||||
if "default" not in schema:
|
||||
continue
|
||||
with self.subTest(key=key):
|
||||
validate_supplied_value(f"credentials.{key}", schema["default"], schema)
|
||||
|
||||
def test_every_known_provider_form_is_accepted(self):
|
||||
schema = self._schema()
|
||||
for value in self.ACCEPTED:
|
||||
with self.subTest(value=value):
|
||||
validate_supplied_value("credentials.openai_api_key", value, schema)
|
||||
|
||||
def test_a_broken_paste_is_rejected(self):
|
||||
schema = self._schema()
|
||||
for value in self.REJECTED:
|
||||
with self.subTest(value=value), self.assertRaises(SystemExit):
|
||||
validate_supplied_value("credentials.openai_api_key", value, schema)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -4,12 +4,15 @@ import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from utils.cache.files import PROJECT_ROOT, read_text
|
||||
from utils.cache.yaml import dump_yaml_str
|
||||
from utils.roles.applications.services.registry import (
|
||||
GROUP_GATED_CONSTANTS,
|
||||
ServiceRegistryError,
|
||||
build_service_registry_from_applications,
|
||||
detect_service_bucket,
|
||||
detect_service_channel,
|
||||
is_explicit_truth,
|
||||
ordered_primary_service_entries,
|
||||
resolve_service_dependency_roles_from_config,
|
||||
)
|
||||
@@ -335,5 +338,40 @@ class TestServiceRegistryOrdering(unittest.TestCase):
|
||||
ordered_primary_service_entries(registry, roles_dir)
|
||||
|
||||
|
||||
class TestExplicitTruth(unittest.TestCase):
|
||||
"""A flag may resolve a dependency through a group-gated constant.
|
||||
|
||||
The substring rule reads the role id out of the flag itself, so a SPOT
|
||||
constant hides it. Roles referencing one keep their provider only while
|
||||
the constant is registered.
|
||||
"""
|
||||
|
||||
def test_a_literal_true_resolves(self):
|
||||
self.assertTrue(is_explicit_truth(True))
|
||||
|
||||
def test_the_in_group_names_form_resolves(self):
|
||||
self.assertTrue(is_explicit_truth("{{ 'web-app-keycloak' in group_names }}"))
|
||||
|
||||
def test_every_registered_constant_resolves(self):
|
||||
for constant in GROUP_GATED_CONSTANTS:
|
||||
with self.subTest(constant=constant):
|
||||
self.assertTrue(is_explicit_truth("{{ " + constant + " }}"))
|
||||
|
||||
def test_an_unregistered_jinja_expression_does_not_resolve(self):
|
||||
self.assertFalse(
|
||||
is_explicit_truth("{{ lookup('config', 'web-app-gitea', 'mcp.enabled') }}")
|
||||
)
|
||||
|
||||
def test_a_literal_false_does_not_resolve(self):
|
||||
self.assertFalse(is_explicit_truth(False))
|
||||
|
||||
def test_every_registered_constant_is_defined_in_group_vars(self):
|
||||
group_vars = PROJECT_ROOT / "group_vars" / "all"
|
||||
defined = "".join(read_text(str(path)) for path in group_vars.glob("*.yml"))
|
||||
for constant in GROUP_GATED_CONSTANTS:
|
||||
with self.subTest(constant=constant):
|
||||
self.assertIn(f"{constant}:", defined)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Unit tests for `utils.roles.applications.variants`' provider overrides.
|
||||
|
||||
The service registry is stubbed: what matters is which role a claim is
|
||||
attributed to, not how a service key resolves to its provider.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from utils.roles.applications.variants import (
|
||||
NestedOverrideConflictError,
|
||||
collect_provider_overrides,
|
||||
)
|
||||
|
||||
_REGISTRY = {
|
||||
"openwebui": {"role": "web-app-openwebui"},
|
||||
"gitea": {"role": "web-app-gitea"},
|
||||
}
|
||||
|
||||
_PATCH = "utils.roles.applications.variants.build_service_registry_from_roles_dir"
|
||||
|
||||
|
||||
class TestCollectProviderOverrides(unittest.TestCase):
|
||||
def _collect(self, payloads, registry=None):
|
||||
with patch(_PATCH, autospec=True) as build:
|
||||
build.return_value = _REGISTRY if registry is None else registry
|
||||
return collect_provider_overrides(payloads, roles_dir="/roles")
|
||||
|
||||
def test_a_named_topic_reaches_its_provider(self) -> None:
|
||||
payloads = {
|
||||
"web-app-moodle": {"services": {"openwebui": {"services": None}}},
|
||||
}
|
||||
self.assertEqual(
|
||||
{"web-app-openwebui": {"services": None}}, self._collect(payloads)
|
||||
)
|
||||
|
||||
def test_several_topics_of_one_provider_are_collected_together(self) -> None:
|
||||
payloads = {
|
||||
"web-app-moodle": {
|
||||
"services": {
|
||||
"openwebui": {"services": None, "addons": {"whisper": False}},
|
||||
},
|
||||
},
|
||||
}
|
||||
self.assertEqual(
|
||||
{"web-app-openwebui": {"services": None, "addons": {"whisper": False}}},
|
||||
self._collect(payloads),
|
||||
)
|
||||
|
||||
def test_a_roles_own_entry_is_not_an_override(self) -> None:
|
||||
payloads = {"web-app-gitea": {"services": {"gitea": {"users": {"a": 1}}}}}
|
||||
self.assertEqual({}, self._collect(payloads))
|
||||
|
||||
def test_an_unregistered_service_key_is_skipped(self) -> None:
|
||||
payloads = {"web-app-moodle": {"services": {"nowhere": {"services": None}}}}
|
||||
self.assertEqual({}, self._collect(payloads))
|
||||
|
||||
def test_two_roles_agreeing_is_not_a_conflict(self) -> None:
|
||||
payloads = {
|
||||
"web-app-moodle": {"services": {"openwebui": {"services": None}}},
|
||||
"web-app-gitea": {"services": {"openwebui": {"services": None}}},
|
||||
}
|
||||
self.assertEqual(
|
||||
{"web-app-openwebui": {"services": None}}, self._collect(payloads)
|
||||
)
|
||||
|
||||
def test_two_roles_demanding_different_values_raises(self) -> None:
|
||||
payloads = {
|
||||
"web-app-moodle": {"services": {"openwebui": {"services": None}}},
|
||||
"web-app-gitea": {"services": {"openwebui": {"services": {"a": 1}}}},
|
||||
}
|
||||
with self.assertRaises(NestedOverrideConflictError):
|
||||
self._collect(payloads)
|
||||
|
||||
def test_a_round_without_overrides_never_builds_the_registry(self) -> None:
|
||||
payloads = {"web-app-moodle": {"services": {"openwebui": {"enabled": True}}}}
|
||||
with patch(_PATCH, autospec=True) as build:
|
||||
build.return_value = _REGISTRY
|
||||
self.assertEqual(
|
||||
{}, collect_provider_overrides(payloads, roles_dir="/roles")
|
||||
)
|
||||
build.assert_not_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,51 @@
|
||||
import unittest
|
||||
|
||||
from utils.api import declared_api, provider_enabled, resolve_api
|
||||
|
||||
|
||||
class TestResolveApi(unittest.TestCase):
|
||||
def test_a_partial_override_keeps_the_declared_providers(self) -> None:
|
||||
declared = declared_api()
|
||||
self.assertIn("github", declared)
|
||||
|
||||
resolved = resolve_api({"API": {"openai": {"api_key": "sk-test"}}})
|
||||
|
||||
self.assertEqual(set(resolved), set(declared))
|
||||
self.assertEqual(resolved["openai"]["api_key"], "sk-test")
|
||||
self.assertEqual(resolved["github"]["client_id"], "")
|
||||
|
||||
def test_an_override_wins_per_key(self) -> None:
|
||||
resolved = resolve_api({"API": {"github": {"client_id": "ID42"}}})
|
||||
|
||||
self.assertEqual(resolved["github"]["client_id"], "ID42")
|
||||
self.assertEqual(resolved["github"]["client_secret"], "")
|
||||
|
||||
def test_a_missing_api_falls_back_to_the_declaration(self) -> None:
|
||||
self.assertEqual(resolve_api({}), declared_api())
|
||||
|
||||
def test_a_non_mapping_api_is_an_error(self) -> None:
|
||||
with self.assertRaises(TypeError):
|
||||
resolve_api({"API": "nope"})
|
||||
|
||||
|
||||
class TestProviderEnabled(unittest.TestCase):
|
||||
def test_every_credential_must_carry_a_value(self) -> None:
|
||||
api = {"github": {"client_id": "ID42", "client_secret": "SEC42"}}
|
||||
|
||||
self.assertTrue(provider_enabled(api, "github"))
|
||||
|
||||
def test_one_empty_credential_disables_the_provider(self) -> None:
|
||||
api = {"github": {"client_id": "ID42", "client_secret": " "}}
|
||||
|
||||
self.assertFalse(provider_enabled(api, "github"))
|
||||
|
||||
def test_a_provider_without_credentials_is_disabled(self) -> None:
|
||||
self.assertFalse(provider_enabled({"github": {}}, "github"))
|
||||
|
||||
def test_an_undeclared_provider_is_an_error(self) -> None:
|
||||
with self.assertRaises(KeyError):
|
||||
provider_enabled({}, "github")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+3
-3
@@ -3,7 +3,7 @@ import io
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from utils.tests.swarm import matrix
|
||||
from utils.tests.swarm import run
|
||||
|
||||
|
||||
class TestRunReporting(unittest.TestCase):
|
||||
@@ -18,10 +18,10 @@ class TestRunReporting(unittest.TestCase):
|
||||
def _run_capturing(self, rc: int) -> tuple[int, str]:
|
||||
buffer = io.StringIO()
|
||||
with (
|
||||
mock.patch.object(matrix, "_run_watched", return_value=rc) as watched,
|
||||
mock.patch.object(run, "run_watched", return_value=rc) as watched,
|
||||
contextlib.redirect_stdout(buffer),
|
||||
):
|
||||
returned = matrix._run(["true"], env={"A": "b"}, label="demo phase")
|
||||
returned = run.run_step(["true"], env={"A": "b"}, label="demo phase")
|
||||
watched.assert_called_once_with(["true"], env={"A": "b"})
|
||||
return returned, buffer.getvalue()
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
"""Immutable tags the updater could not order before.
|
||||
|
||||
Two shapes reach the repository that name a fixed release yet failed the semver
|
||||
gate, so the bump job skipped them and nothing could tell whether a newer one
|
||||
existed: a vendor patch counter such as ``2.4.0p32`` and a five-component
|
||||
version such as ``26.04.2.1.1``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from utils.update.base import (
|
||||
is_semver,
|
||||
latest_semver,
|
||||
version_depth,
|
||||
version_flavor,
|
||||
version_key,
|
||||
)
|
||||
|
||||
|
||||
def _latest(current: str, tags: list[str]) -> str | None:
|
||||
return latest_semver(tags, version_depth(current), version_flavor(current))
|
||||
|
||||
|
||||
class TestPatchCounterTags(unittest.TestCase):
|
||||
def test_a_patch_counter_is_orderable(self):
|
||||
self.assertTrue(is_semver("2.4.0p32"))
|
||||
|
||||
def test_it_bumps_within_its_own_counter(self):
|
||||
self.assertEqual(_latest("2.4.0p32", ["2.4.0p32", "2.4.0p33"]), "2.4.0p33")
|
||||
|
||||
def test_the_counter_orders_numerically_not_lexically(self):
|
||||
self.assertEqual(_latest("2.4.0p9", ["2.4.0p9", "2.4.0p10"]), "2.4.0p10")
|
||||
|
||||
def test_a_later_release_wins_over_a_later_counter(self):
|
||||
self.assertEqual(_latest("2.4.0p32", ["2.4.0p99", "2.4.1p1"]), "2.4.1p1")
|
||||
|
||||
def test_it_never_crosses_into_a_four_component_version(self):
|
||||
self.assertEqual(
|
||||
_latest("2.4.0p32", ["2.4.0p32", "2.4.0.99"]),
|
||||
"2.4.0p32",
|
||||
"2.4.0.99 is a different tag family and bumping into it would "
|
||||
"silently change what the role deploys",
|
||||
)
|
||||
|
||||
def test_the_counter_letter_keeps_families_apart(self):
|
||||
self.assertEqual(version_flavor("2.4.0p32"), "p")
|
||||
self.assertEqual(version_flavor("2.4.0.32"), "")
|
||||
self.assertNotEqual(version_depth("2.4.0p32"), version_depth("2.4.0.32"))
|
||||
|
||||
def test_a_plain_release_does_not_bump_into_a_counter(self):
|
||||
self.assertEqual(_latest("2.4.0", ["2.4.0", "2.4.0p1"]), "2.4.0")
|
||||
|
||||
|
||||
class TestFiveComponentTags(unittest.TestCase):
|
||||
def test_five_components_are_orderable(self):
|
||||
self.assertTrue(is_semver("26.04.2.1.1"))
|
||||
self.assertEqual(version_depth("26.04.2.1.1"), 5)
|
||||
|
||||
def test_it_bumps_within_its_own_depth(self):
|
||||
self.assertEqual(
|
||||
_latest("26.04.2.1.1", ["26.04.2.1.1", "26.04.2.1.2"]), "26.04.2.1.2"
|
||||
)
|
||||
|
||||
def test_it_never_crosses_into_a_shorter_version(self):
|
||||
self.assertEqual(
|
||||
_latest("26.04.2.1.1", ["26.04.2.1.1", "27.0.0.0"]), "26.04.2.1.1"
|
||||
)
|
||||
|
||||
|
||||
class TestNothingElseChanged(unittest.TestCase):
|
||||
"""The shapes the updater already handled must keep their family key."""
|
||||
|
||||
def test_known_shapes_keep_their_classification(self):
|
||||
cases = {
|
||||
"1.2.3": (True, 3, "", (1, 2, 3, 0)),
|
||||
"v1.12.27": (True, 3, "", (1, 12, 27, 0)),
|
||||
"16": (True, 1, "", (16, 0, 0, 0)),
|
||||
"5.4.5-php8.3-apache": (True, 3, "-php8.3-apache", (5, 4, 5, 0)),
|
||||
}
|
||||
|
||||
for tag, (semver, depth, flavor, key) in cases.items():
|
||||
self.assertEqual(is_semver(tag), semver, tag)
|
||||
self.assertEqual(version_depth(tag), depth, tag)
|
||||
self.assertEqual(version_flavor(tag), flavor, tag)
|
||||
self.assertEqual(version_key(tag), key, tag)
|
||||
|
||||
def test_moving_names_are_still_refused(self):
|
||||
for tag in ("latest", "stable", "lts", "alpine", "main"):
|
||||
self.assertFalse(is_semver(tag), tag)
|
||||
|
||||
def test_a_flavored_tag_still_refuses_a_different_runtime(self):
|
||||
self.assertEqual(
|
||||
_latest("5.4.5-php8.3-apache", ["5.4.6-php8.4-apache", "5.4.6"]),
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
class TestChannelPrefixedTags(unittest.TestCase):
|
||||
"""A release line written before the number is a family, not a version."""
|
||||
|
||||
def test_a_channel_prefixed_tag_is_orderable(self):
|
||||
self.assertTrue(is_semver("main-v1.77.7-stable"))
|
||||
self.assertEqual(version_depth("main-v1.77.7-stable"), 3)
|
||||
self.assertEqual(version_key("main-v1.77.7-stable"), (1, 77, 7, 0))
|
||||
|
||||
def test_it_bumps_inside_its_own_channel(self):
|
||||
tags = ["main-v1.77.7-stable", "main-v1.83.14-stable"]
|
||||
self.assertEqual(_latest("main-v1.77.7-stable", tags), "main-v1.83.14-stable")
|
||||
|
||||
def test_it_never_crosses_into_a_sibling_channel(self):
|
||||
tags = ["main-v1.83.14-nightly", "main-v1.83.14-stable"]
|
||||
self.assertEqual(_latest("main-v1.77.7-stable", tags), "main-v1.83.14-stable")
|
||||
self.assertEqual(_latest("main-v1.77.7-nightly", tags), "main-v1.83.14-nightly")
|
||||
|
||||
def test_a_bare_number_is_a_different_family_than_a_channelled_one(self):
|
||||
self.assertNotEqual(version_flavor("stable-9646"), version_flavor("9646"))
|
||||
self.assertEqual(
|
||||
_latest("stable-9646", ["stable-9646", "9999"]),
|
||||
"stable-9646",
|
||||
"a bare build number leaves the stable channel and would change "
|
||||
"what the role deploys",
|
||||
)
|
||||
|
||||
def test_it_bumps_inside_the_stable_channel(self):
|
||||
self.assertEqual(
|
||||
_latest("stable-9646", ["stable-9646", "stable-9999"]), "stable-9999"
|
||||
)
|
||||
|
||||
def test_a_channel_without_a_number_stays_refused(self):
|
||||
for tag in ("act-latest", "bookworm-slim", "main-latest", "buildx-stable-1"):
|
||||
self.assertFalse(is_semver(tag), tag)
|
||||
|
||||
def test_a_date_stamped_tag_orders_as_one_number(self):
|
||||
self.assertEqual(version_depth("20260917"), 1)
|
||||
self.assertEqual(
|
||||
_latest("20260702", ["20260702", "20260706", "20260917"]), "20260917"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,114 @@
|
||||
import json
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from utils.update.collection_source import (
|
||||
RATIO,
|
||||
WINDOW,
|
||||
append_sample,
|
||||
decide,
|
||||
load_history,
|
||||
)
|
||||
|
||||
|
||||
def _samples(galaxy, git, count=WINDOW):
|
||||
return [{"galaxy": galaxy, "git": git} for _ in range(count)]
|
||||
|
||||
|
||||
class TestAppendSample(unittest.TestCase):
|
||||
def test_the_window_never_grows_past_its_size(self):
|
||||
history = _samples(10.0, 10.0, WINDOW)
|
||||
|
||||
grown = append_sample(history, {"galaxy": 99.0, "git": 1.0})
|
||||
|
||||
self.assertEqual(len(grown), WINDOW)
|
||||
self.assertEqual(grown[-1], {"galaxy": 99.0, "git": 1.0})
|
||||
|
||||
def test_the_oldest_sample_is_the_one_dropped(self):
|
||||
history = [{"galaxy": float(i), "git": 1.0} for i in range(WINDOW)]
|
||||
|
||||
grown = append_sample(history, {"galaxy": 99.0, "git": 1.0})
|
||||
|
||||
self.assertEqual(grown[0]["galaxy"], 1.0)
|
||||
|
||||
|
||||
class TestDecide(unittest.TestCase):
|
||||
def test_below_a_full_window_the_default_is_left_alone(self):
|
||||
history = _samples(300.0, 10.0, WINDOW - 1)
|
||||
|
||||
verdict = decide(history, "galaxy")
|
||||
|
||||
self.assertFalse(verdict.changed)
|
||||
self.assertEqual(verdict.samples, WINDOW - 1)
|
||||
|
||||
def test_exactly_the_ratio_swaps_the_preferred_source(self):
|
||||
history = _samples(30.0, 10.0)
|
||||
|
||||
verdict = decide(history, "galaxy")
|
||||
|
||||
self.assertTrue(verdict.changed)
|
||||
self.assertEqual(verdict.proposed, "git")
|
||||
|
||||
def test_just_under_the_ratio_keeps_the_preferred_source(self):
|
||||
history = _samples(10.0 * RATIO - 0.1, 10.0)
|
||||
|
||||
self.assertFalse(decide(history, "galaxy").changed)
|
||||
|
||||
def test_a_failed_install_ranks_worse_than_any_duration(self):
|
||||
history = _samples(None, 10.0)
|
||||
|
||||
verdict = decide(history, "galaxy")
|
||||
|
||||
self.assertTrue(verdict.changed)
|
||||
self.assertEqual(verdict.proposed, "git")
|
||||
self.assertEqual(verdict.medians["galaxy"], float("inf"))
|
||||
|
||||
def test_the_faster_source_is_never_swapped_away_from(self):
|
||||
history = _samples(10.0, 300.0)
|
||||
|
||||
self.assertFalse(decide(history, "galaxy").changed)
|
||||
|
||||
def test_both_sources_failing_changes_nothing(self):
|
||||
history = _samples(None, None)
|
||||
|
||||
self.assertFalse(decide(history, "galaxy").changed)
|
||||
|
||||
def test_a_single_outlier_does_not_move_the_median(self):
|
||||
history = _samples(10.0, 10.0, WINDOW - 1)
|
||||
history = append_sample(history, {"galaxy": 9000.0, "git": 10.0})
|
||||
|
||||
self.assertFalse(decide(history, "galaxy").changed)
|
||||
|
||||
def test_the_verdict_is_taken_from_the_preferred_side(self):
|
||||
history = _samples(10.0, 30.0)
|
||||
|
||||
verdict = decide(history, "git")
|
||||
|
||||
self.assertTrue(verdict.changed)
|
||||
self.assertEqual(verdict.proposed, "galaxy")
|
||||
|
||||
|
||||
class TestLoadHistory(unittest.TestCase):
|
||||
def test_a_missing_file_reads_as_an_empty_history(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
self.assertEqual(load_history(Path(tmp) / "absent.json"), [])
|
||||
|
||||
def test_a_corrupt_file_reads_as_an_empty_history(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
path = Path(tmp) / "history.json"
|
||||
path.write_text("{not json", encoding="utf-8")
|
||||
|
||||
self.assertEqual(load_history(path), [])
|
||||
|
||||
def test_a_written_history_round_trips(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
path = Path(tmp) / "history.json"
|
||||
history = _samples(1.0, 2.0, 3)
|
||||
path.write_text(json.dumps(history), encoding="utf-8")
|
||||
|
||||
self.assertEqual(load_history(path), history)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,112 @@
|
||||
"""Several images of one role at one version move together or not at all.
|
||||
|
||||
Upstream does not push a release group atomically. Measured on Docker Hub:
|
||||
``jitsi/prosody``, ``jitsi/jicofo`` and ``jitsi/jvb`` all publish
|
||||
``stable-9457`` while ``jitsi/web`` does not, so a bump run landing in that
|
||||
window would raise three components and leave the fourth behind.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from utils.roles.mapping import ROLE_FILE_META_SERVICES
|
||||
from utils.update.docker import (
|
||||
DockerImageVersionEntry,
|
||||
DockerImageVersionUpdate,
|
||||
_held_to_release_groups,
|
||||
)
|
||||
|
||||
JITSI = ("web", "prosody", "jicofo", "jvb")
|
||||
|
||||
|
||||
def _entry(role: str, service: str, image: str, version: str):
|
||||
return DockerImageVersionEntry(
|
||||
role=role,
|
||||
service=service,
|
||||
image=image,
|
||||
version=version,
|
||||
config_path=Path(ROLE_FILE_META_SERVICES),
|
||||
)
|
||||
|
||||
|
||||
def _jitsi_updates(version: str, latest: str):
|
||||
return [
|
||||
DockerImageVersionUpdate(
|
||||
entry=_entry("web-app-jitsi", name, f"jitsi/{name}", version),
|
||||
latest=latest,
|
||||
)
|
||||
for name in JITSI
|
||||
]
|
||||
|
||||
|
||||
class TestReleaseGroupsMoveTogether(unittest.TestCase):
|
||||
def test_a_partial_publish_holds_the_whole_group_back(self):
|
||||
tags = {
|
||||
"jitsi/web": ["stable-9400", "stable-9646"],
|
||||
"jitsi/prosody": ["stable-9400", "stable-9646", "stable-9700"],
|
||||
"jitsi/jicofo": ["stable-9400", "stable-9646", "stable-9700"],
|
||||
"jitsi/jvb": ["stable-9400", "stable-9646", "stable-9700"],
|
||||
}
|
||||
updates = _jitsi_updates("stable-9400", "stable-9700")
|
||||
|
||||
held = _held_to_release_groups(updates, tags)
|
||||
|
||||
self.assertEqual({update.latest for update in held}, {"stable-9646"})
|
||||
self.assertEqual(len(held), 4)
|
||||
|
||||
def test_a_group_all_publish_moves_to_the_head(self):
|
||||
tags = {f"jitsi/{name}": ["stable-9400", "stable-9700"] for name in JITSI}
|
||||
updates = _jitsi_updates("stable-9400", "stable-9700")
|
||||
|
||||
held = _held_to_release_groups(updates, tags)
|
||||
|
||||
self.assertEqual({update.latest for update in held}, {"stable-9700"})
|
||||
|
||||
def test_a_group_with_nothing_in_common_is_dropped(self):
|
||||
tags = {
|
||||
"jitsi/web": ["stable-9400", "stable-9701"],
|
||||
"jitsi/prosody": ["stable-9400", "stable-9702"],
|
||||
"jitsi/jicofo": ["stable-9400", "stable-9703"],
|
||||
"jitsi/jvb": ["stable-9400", "stable-9704"],
|
||||
}
|
||||
updates = _jitsi_updates("stable-9400", "stable-9704")
|
||||
|
||||
self.assertEqual(_held_to_release_groups(updates, tags), [])
|
||||
|
||||
def test_a_lone_entry_is_untouched(self):
|
||||
update = DockerImageVersionUpdate(
|
||||
entry=_entry("web-app-fider", "fider", "getfider/fider", "v0.35.0"),
|
||||
latest="v0.36.0",
|
||||
)
|
||||
|
||||
held = _held_to_release_groups([update], {"getfider/fider": ["v0.36.0"]})
|
||||
|
||||
self.assertEqual(held, [update])
|
||||
|
||||
def test_two_roles_on_one_version_are_not_one_group(self):
|
||||
tags = {
|
||||
"jitsi/web": ["stable-9400", "stable-9700"],
|
||||
"other/thing": ["stable-9400"],
|
||||
}
|
||||
updates = [
|
||||
DockerImageVersionUpdate(
|
||||
entry=_entry("web-app-jitsi", "web", "jitsi/web", "stable-9400"),
|
||||
latest="stable-9700",
|
||||
),
|
||||
DockerImageVersionUpdate(
|
||||
entry=_entry("web-app-other", "thing", "other/thing", "stable-9400"),
|
||||
latest="stable-9400",
|
||||
),
|
||||
]
|
||||
|
||||
held = _held_to_release_groups(updates, tags)
|
||||
|
||||
self.assertEqual(
|
||||
[update.latest for update in held], ["stable-9700", "stable-9400"]
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,127 @@
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
from unittest.mock import patch
|
||||
|
||||
from utils.update.galaxy import apply_updates, collect_pins, find_outdated_updates
|
||||
|
||||
GALAXY = """---
|
||||
collections:
|
||||
- name: community.general
|
||||
version: 13.2.0
|
||||
- name: kewlfft.aur
|
||||
version: 0.13.0
|
||||
"""
|
||||
|
||||
GIT = """---
|
||||
collections:
|
||||
- name: community.general
|
||||
source: https://github.com/ansible-collections/community.general.git
|
||||
type: git
|
||||
version: 13.2.0
|
||||
|
||||
- name: kewlfft.aur
|
||||
source: https://github.com/kewlfft/ansible-aur.git
|
||||
type: git
|
||||
version: master
|
||||
"""
|
||||
|
||||
|
||||
def _read(root: Path, name: str) -> str:
|
||||
path = root / "requirements" / name
|
||||
return path.read_text() # nocheck: cache-read -- rewritten by this test
|
||||
|
||||
|
||||
def _repo(root: Path) -> Path:
|
||||
(root / "requirements").mkdir()
|
||||
(root / "requirements" / "requirements.galaxy.yml").write_text(
|
||||
GALAXY, encoding="utf-8"
|
||||
)
|
||||
(root / "requirements" / "requirements.git.yml").write_text(GIT, encoding="utf-8")
|
||||
return root
|
||||
|
||||
|
||||
class TestCollectPins(unittest.TestCase):
|
||||
def test_only_collections_pinned_to_a_semver_in_both_files_are_collected(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = _repo(Path(tmp))
|
||||
|
||||
pins = collect_pins(root)
|
||||
|
||||
self.assertEqual([pin.fqcn for pin in pins], ["community.general"])
|
||||
|
||||
def test_a_version_the_two_files_disagree_on_is_not_collected(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = _repo(Path(tmp))
|
||||
(root / "requirements" / "requirements.galaxy.yml").write_text(
|
||||
GALAXY.replace("13.2.0", "13.4.0"), encoding="utf-8"
|
||||
)
|
||||
|
||||
self.assertEqual(collect_pins(root), [])
|
||||
|
||||
|
||||
class TestFindOutdatedUpdates(unittest.TestCase):
|
||||
def test_a_higher_tag_of_the_same_depth_is_an_update(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = _repo(Path(tmp))
|
||||
with patch(
|
||||
"utils.update.galaxy.git_ls_remote_tags",
|
||||
return_value=["13.1.0", "13.2.0", "13.4.0"],
|
||||
):
|
||||
updates = find_outdated_updates(root)
|
||||
|
||||
self.assertEqual([u.latest for u in updates], ["13.4.0"])
|
||||
|
||||
def test_the_newest_tag_already_pinned_is_no_update(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = _repo(Path(tmp))
|
||||
with patch(
|
||||
"utils.update.galaxy.git_ls_remote_tags",
|
||||
return_value=["13.1.0", "13.2.0"],
|
||||
):
|
||||
self.assertEqual(find_outdated_updates(root), [])
|
||||
|
||||
def test_a_tag_of_a_different_depth_is_not_a_candidate(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = _repo(Path(tmp))
|
||||
with patch(
|
||||
"utils.update.galaxy.git_ls_remote_tags",
|
||||
return_value=["13.2.0", "14"],
|
||||
):
|
||||
self.assertEqual(find_outdated_updates(root), [])
|
||||
|
||||
|
||||
class TestApplyUpdates(unittest.TestCase):
|
||||
def test_both_requirement_files_are_bumped_together(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = _repo(Path(tmp))
|
||||
with patch(
|
||||
"utils.update.galaxy.git_ls_remote_tags",
|
||||
return_value=["13.4.0"],
|
||||
):
|
||||
updates = find_outdated_updates(root)
|
||||
changed = apply_updates(root, updates)
|
||||
|
||||
self.assertEqual(len(changed), 2)
|
||||
galaxy_text = _read(root, "requirements.galaxy.yml")
|
||||
git_text = _read(root, "requirements.git.yml")
|
||||
self.assertIn("version: 13.4.0", galaxy_text)
|
||||
self.assertIn("version: 13.4.0", git_text)
|
||||
self.assertNotIn("13.2.0", galaxy_text)
|
||||
self.assertNotIn("13.2.0", git_text)
|
||||
|
||||
def test_an_entry_outside_the_update_keeps_its_version(self):
|
||||
with TemporaryDirectory() as tmp:
|
||||
root = _repo(Path(tmp))
|
||||
with patch(
|
||||
"utils.update.galaxy.git_ls_remote_tags",
|
||||
return_value=["13.4.0"],
|
||||
):
|
||||
apply_updates(root, find_outdated_updates(root))
|
||||
|
||||
self.assertIn("version: master", _read(root, "requirements.git.yml"))
|
||||
self.assertIn("version: 0.13.0", _read(root, "requirements.galaxy.yml"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -5,10 +5,12 @@ import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from utils.cache.files import read_text
|
||||
from utils.roles.mapping import ROLE_DIR_META_ADDONS, ROLE_FILE_META_SERVICES
|
||||
from utils.update.repository import (
|
||||
RepositoryRefEntry,
|
||||
RepositoryRefUpdate,
|
||||
apply_updates,
|
||||
collect_entries,
|
||||
suppressed_ref_lines,
|
||||
update_config_refs,
|
||||
walk_repo_ref_pairs,
|
||||
@@ -216,5 +218,46 @@ class TestApplyUpdates(unittest.TestCase):
|
||||
self.assertEqual(read_text(str(config_b)), " ref: v2.1.0\n")
|
||||
|
||||
|
||||
class TestCollectEntriesCoversAddons(unittest.TestCase):
|
||||
def _role(self, root: Path) -> Path:
|
||||
role = root / "roles" / "web-app-example"
|
||||
(role / ROLE_DIR_META_ADDONS).mkdir(parents=True)
|
||||
return role
|
||||
|
||||
def test_addon_ref_is_collected_beside_the_service_ref(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
root = Path(tmpdir)
|
||||
role = self._role(root)
|
||||
(role / ROLE_FILE_META_SERVICES).write_text(
|
||||
"app:\n repository: https://example.test/app.git\n ref: v1.0.0\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
(role / ROLE_DIR_META_ADDONS / "plug.yml").write_text(
|
||||
"config:\n repository: https://example.test/plug.git\n ref: v2.0.0\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
by_ref = {entry.ref: entry for entry in collect_entries(root)}
|
||||
|
||||
self.assertEqual(set(by_ref), {"v1.0.0", "v2.0.0"})
|
||||
addon = by_ref["v2.0.0"]
|
||||
self.assertEqual(addon.entity_path[:2], ("addons", "plug"))
|
||||
self.assertEqual(addon.config_path.name, "plug.yml")
|
||||
|
||||
def test_addon_ref_honours_its_own_suppression_marker(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
root = Path(tmpdir)
|
||||
role = self._role(root)
|
||||
(role / ROLE_DIR_META_ADDONS / "plug.yml").write_text(
|
||||
"config:\n"
|
||||
" repository: https://example.test/plug.git\n"
|
||||
" # nocheck: repository-version\n"
|
||||
" ref: v2.0.0\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
self.assertEqual(collect_entries(root), [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user