mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 10:59:59 +00:00
- Declare `__all__` in every package `__init__.py` whose `PROJECT_ROOT` is imported by its modules (163 packages), so the constant counts as exported; list every public name where the package defines more. - Empty the 15 package `__init__.py` files whose `PROJECT_ROOT` no module imports. - Declare the public API of `utils.github.variant.axes` in `__all__`, re-exports included, instead of silencing them with `noqa: F401`; the list matches the one `feature/store-i18n` already carries. - Use `import unittest.mock as mock` in `test_applications.py`, as the rest of the suite does. - Spell `_NON_PRINTABLE` in `utils/yaml_bootstrap.py` as the equivalent BMP-only positive class; it matches exactly the same 2111 code points as the old negated class across U+0000..U+10FFFF. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
730 lines
27 KiB
Python
730 lines
27 KiB
Python
"""Focused unit tests for ``utils.cache.applications``.
|
|
|
|
Pins the public API (`get_application_defaults`, `get_variants`,
|
|
`get_merged_applications`) plus the cache invariants (per-roles_dir
|
|
keying, deep-copy on read), and the strict ansible-free import-time
|
|
contract that the GitHub Actions runner-host CLI path depends on.
|
|
|
|
The matrix-deploy variants loader is exhaustively covered in
|
|
test_variants.py; this file owns the module-API contract.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
import unittest.mock as mock
|
|
from pathlib import Path
|
|
|
|
from utils.cache import _reset_cache_for_tests
|
|
from utils.cache import applications as cache_apps
|
|
from utils.cache.carrier import (
|
|
APPLICATIONS_RENDERED_FACT,
|
|
merged_applications_cache_key,
|
|
)
|
|
from utils.cache.yaml import load_yaml_any
|
|
from utils.roles.mapping import (
|
|
ROLE_DIR_META_ADDONS,
|
|
ROLE_FILE_META_MAIN,
|
|
ROLE_FILE_META_MCP,
|
|
ROLE_FILE_META_SERVICES,
|
|
ROLE_FILE_META_USERS,
|
|
ROLE_FILE_META_VARIANTS,
|
|
ROLE_FILE_META_VOLUMES,
|
|
)
|
|
|
|
from . import PROJECT_ROOT
|
|
|
|
|
|
def _write(path: Path, content: str) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(textwrap.dedent(content), encoding="utf-8")
|
|
|
|
|
|
def _seed_minimal_roles(tmp: Path) -> Path:
|
|
"""Create a minimal `<tmp>/roles/web-app-foo/meta/...` tree
|
|
so applications/variants can resolve a real role.
|
|
"""
|
|
roles = tmp / "roles"
|
|
role = roles / "web-app-foo"
|
|
_write(
|
|
role / ROLE_FILE_META_SERVICES,
|
|
"""
|
|
foo:
|
|
image: foo
|
|
version: latest
|
|
""",
|
|
)
|
|
_write(
|
|
role / ROLE_FILE_META_USERS,
|
|
"""
|
|
administrator: {}
|
|
""",
|
|
)
|
|
return roles
|
|
|
|
|
|
class TestGetApplicationDefaults(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_returns_dict_per_role(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
defaults = cache_apps.get_application_defaults(roles_dir=roles)
|
|
self.assertIn("web-app-foo", defaults)
|
|
self.assertIn("services", defaults["web-app-foo"])
|
|
|
|
def test_caches_per_roles_dir(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
first = cache_apps.get_application_defaults(roles_dir=roles)
|
|
first["web-app-foo"]["mutated"] = True
|
|
second = cache_apps.get_application_defaults(roles_dir=roles)
|
|
self.assertNotIn("mutated", second["web-app-foo"])
|
|
|
|
def test_users_block_rewritten_to_lookup_jinja(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
defaults = cache_apps.get_application_defaults(roles_dir=roles)
|
|
users = defaults["web-app-foo"]["users"]
|
|
self.assertEqual(
|
|
users["administrator"],
|
|
"{{ lookup('users', 'administrator') }}",
|
|
)
|
|
|
|
|
|
class TestGetCanonicalVolumes(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_forces_rebuild_when_registry_empty(self):
|
|
cache_apps._CANONICAL_VOLUMES_BY_ROLE.clear()
|
|
self.assertIn(
|
|
"ollama_models",
|
|
cache_apps.get_canonical_volumes("svc-ai-ollama"),
|
|
)
|
|
|
|
def test_warm_defaults_cache_does_not_mask_empty_registry(self):
|
|
cache_apps.get_application_defaults()
|
|
cache_apps._CANONICAL_VOLUMES_BY_ROLE.clear()
|
|
self.assertIn(
|
|
"ollama_models",
|
|
cache_apps.get_canonical_volumes("svc-ai-ollama"),
|
|
)
|
|
|
|
|
|
class TestGetVariants(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_returns_single_variant_when_no_meta_variants_yml(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
variants = cache_apps.get_variants(roles_dir=roles)
|
|
self.assertEqual(len(variants["web-app-foo"]), 1)
|
|
|
|
def test_returns_multiple_variants_when_meta_variants_yml_lists_them(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
_write(
|
|
roles / "web-app-foo" / ROLE_FILE_META_VARIANTS,
|
|
"""
|
|
- {}
|
|
- services:
|
|
foo:
|
|
image: foo-alt
|
|
""",
|
|
)
|
|
variants = cache_apps.get_variants(roles_dir=roles)
|
|
self.assertEqual(len(variants["web-app-foo"]), 2)
|
|
self.assertEqual(
|
|
variants["web-app-foo"][0]["services"]["foo"]["image"],
|
|
"foo",
|
|
)
|
|
self.assertEqual(
|
|
variants["web-app-foo"][1]["services"]["foo"]["image"],
|
|
"foo-alt",
|
|
)
|
|
|
|
def test_caches_per_roles_dir(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
first = cache_apps.get_variants(roles_dir=roles)
|
|
first["web-app-foo"][0]["mutated"] = True
|
|
second = cache_apps.get_variants(roles_dir=roles)
|
|
self.assertNotIn("mutated", second["web-app-foo"][0])
|
|
|
|
|
|
class TestGetVariantOverridesOnly(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_returns_single_empty_override_when_no_meta_variants_yml(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
overrides = cache_apps.get_variant_overrides_only(roles_dir=roles)
|
|
self.assertEqual(overrides["web-app-foo"], [{}])
|
|
|
|
def test_returns_raw_overrides_without_base_merge(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
_write(
|
|
roles / "web-app-foo" / ROLE_FILE_META_VARIANTS,
|
|
"""
|
|
- {}
|
|
- services:
|
|
foo:
|
|
image: foo-alt
|
|
""",
|
|
)
|
|
overrides = cache_apps.get_variant_overrides_only(roles_dir=roles)
|
|
self.assertEqual(overrides["web-app-foo"][0], {})
|
|
self.assertEqual(
|
|
overrides["web-app-foo"][1],
|
|
{"services": {"foo": {"image": "foo-alt"}}},
|
|
)
|
|
|
|
def test_caches_per_roles_dir(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
first = cache_apps.get_variant_overrides_only(roles_dir=roles)
|
|
first["web-app-foo"][0]["mutated"] = True
|
|
second = cache_apps.get_variant_overrides_only(roles_dir=roles)
|
|
self.assertNotIn("mutated", second["web-app-foo"][0])
|
|
|
|
|
|
class TestBuildRoleBaseConfig(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_empty_meta_yields_empty_dict(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
role = Path(tmp) / "roles" / "web-app-empty"
|
|
(role / "meta").mkdir(parents=True)
|
|
self.assertEqual(
|
|
cache_apps._build_role_base_config(role, role.parent),
|
|
{},
|
|
)
|
|
|
|
|
|
class TestAddonsNormalization(unittest.TestCase):
|
|
"""Each `meta/addons/<id>.yml` file (root = the addon spec) is loaded into
|
|
`applications.<role>.addons.<id>` with its enable state normalised: the
|
|
loader is the single place that resolves `required`/`enabled` defaults so
|
|
every consumer reads one already-resolved view.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def _addon_file(self, roles: Path, addon_id: str, spec_yaml: str) -> None:
|
|
_write(
|
|
roles / "web-app-foo" / ROLE_DIR_META_ADDONS / f"{addon_id}.yml",
|
|
spec_yaml,
|
|
)
|
|
|
|
def _defaults_with_addon(self, tmp: Path, addon_id: str, spec_yaml: str) -> dict:
|
|
roles = _seed_minimal_roles(tmp)
|
|
self._addon_file(roles, addon_id, spec_yaml)
|
|
defaults = cache_apps.get_application_defaults(roles_dir=roles)
|
|
return defaults["web-app-foo"]["addons"]
|
|
|
|
def test_required_true_defaults_enabled_true(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
addons = self._defaults_with_addon(
|
|
Path(tmp),
|
|
"core_module",
|
|
"""
|
|
mechanism: module
|
|
source: upstream
|
|
required: true
|
|
""",
|
|
)
|
|
self.assertTrue(addons["core_module"]["enabled"])
|
|
self.assertTrue(addons["core_module"]["required"])
|
|
|
|
def test_optional_addon_defaults_enabled_false(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
addons = self._defaults_with_addon(
|
|
Path(tmp),
|
|
"opt",
|
|
"""
|
|
mechanism: plugin
|
|
source: upstream
|
|
""",
|
|
)
|
|
self.assertIs(addons["opt"]["enabled"], False)
|
|
self.assertIs(addons["opt"]["required"], False)
|
|
|
|
def test_explicit_enabled_jinja_is_preserved(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
expr = (
|
|
"{{ lookup('config', application_id, 'services.ldap.enabled') | bool }}"
|
|
)
|
|
addons = self._defaults_with_addon(
|
|
Path(tmp),
|
|
"ldapauth",
|
|
f"""
|
|
mechanism: addon
|
|
source: upstream
|
|
enabled: "{expr}"
|
|
bridges:
|
|
- ldap
|
|
""",
|
|
)
|
|
self.assertEqual(addons["ldapauth"]["enabled"], expr)
|
|
|
|
def test_explicit_enabled_false_on_optional_is_preserved(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
addons = self._defaults_with_addon(
|
|
Path(tmp),
|
|
"bridge_net",
|
|
"""
|
|
mechanism: bridge
|
|
source: upstream
|
|
required: false
|
|
enabled: false
|
|
""",
|
|
)
|
|
self.assertIs(addons["bridge_net"]["enabled"], False)
|
|
|
|
def test_normalization_does_not_corrupt_yaml_cache(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
self._addon_file(
|
|
roles,
|
|
"opt",
|
|
"""
|
|
mechanism: plugin
|
|
source: upstream
|
|
""",
|
|
)
|
|
first = cache_apps.get_application_defaults(roles_dir=roles)
|
|
first["web-app-foo"]["addons"]["opt"]["enabled"] = "mutated"
|
|
_reset_cache_for_tests()
|
|
second = cache_apps.get_application_defaults(roles_dir=roles)
|
|
self.assertIs(second["web-app-foo"]["addons"]["opt"]["enabled"], False)
|
|
|
|
|
|
class TestGetMergedApplicationsRespectsOverrides(unittest.TestCase):
|
|
"""Smoke test that `get_merged_applications` deep-merges runtime
|
|
`applications` overrides on top of the role defaults. The full
|
|
rendering pipeline (with templar) is covered in test_data.py and
|
|
integration suites; this file pins the contract from the
|
|
applications module's perspective.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_runtime_override_wins_over_default(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
merged = cache_apps.get_merged_applications(
|
|
variables={
|
|
"applications": {
|
|
"web-app-foo": {
|
|
"services": {"foo": {"image": "override"}},
|
|
}
|
|
}
|
|
},
|
|
roles_dir=roles,
|
|
templar=None,
|
|
)
|
|
self.assertEqual(
|
|
merged["web-app-foo"]["services"]["foo"]["image"],
|
|
"override",
|
|
)
|
|
|
|
|
|
class TestGetMergedApplicationsCarrierFact(unittest.TestCase):
|
|
"""The constructor renders once and parks the payload with its cache key in
|
|
`_INFINITO_APPLICATIONS_RENDERED`. A worker process whose in-process cache is
|
|
empty serves that payload when the key matches and renders only when it
|
|
does not.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
@staticmethod
|
|
def _variables() -> dict:
|
|
return {
|
|
"applications": {
|
|
"web-app-foo": {"services": {"foo": {"image": "override"}}}
|
|
}
|
|
}
|
|
|
|
def test_matching_key_serves_the_carried_payload_without_rendering(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
variables = self._variables()
|
|
key = merged_applications_cache_key(variables, roles_dir=roles)
|
|
carried = {"web-app-foo": {"carried": True}}
|
|
variables[APPLICATIONS_RENDERED_FACT] = {
|
|
"key": [key[0], list(key[1])],
|
|
"applications": carried,
|
|
}
|
|
merged = cache_apps.get_merged_applications(
|
|
variables=variables, roles_dir=roles, templar=None
|
|
)
|
|
self.assertIs(merged, carried)
|
|
|
|
def test_stale_key_falls_back_to_a_fresh_render(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
variables = self._variables()
|
|
key = merged_applications_cache_key(variables, roles_dir=roles)
|
|
variables[APPLICATIONS_RENDERED_FACT] = {
|
|
"key": [key[0], ["stale"] * len(key[1])],
|
|
"applications": {"web-app-foo": {"carried": True}},
|
|
}
|
|
merged = cache_apps.get_merged_applications(
|
|
variables=variables, roles_dir=roles, templar=None
|
|
)
|
|
self.assertNotIn("carried", merged["web-app-foo"])
|
|
self.assertEqual(
|
|
merged["web-app-foo"]["services"]["foo"]["image"], "override"
|
|
)
|
|
|
|
def test_a_changed_tokens_file_rejects_the_carrier(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
tokens = Path(tmp) / "tokens.yml"
|
|
tokens.write_text("{}\n", encoding="utf-8")
|
|
variables = self._variables()
|
|
variables["FILE_TOKENS"] = str(tokens)
|
|
key = merged_applications_cache_key(variables, roles_dir=roles)
|
|
variables[APPLICATIONS_RENDERED_FACT] = {
|
|
"key": [key[0], list(key[1])],
|
|
"applications": {"web-app-foo": {"carried": True}},
|
|
}
|
|
tokens.write_text("users: {}\n", encoding="utf-8")
|
|
merged = cache_apps.get_merged_applications(
|
|
variables=variables, roles_dir=roles, templar=None
|
|
)
|
|
self.assertNotIn("carried", merged["web-app-foo"])
|
|
|
|
def test_malformed_carrier_is_ignored(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
for carrier in ("not-a-mapping", {"key": "x", "applications": {}}, {}):
|
|
_reset_cache_for_tests()
|
|
variables = self._variables()
|
|
variables[APPLICATIONS_RENDERED_FACT] = carrier
|
|
merged = cache_apps.get_merged_applications(
|
|
variables=variables, roles_dir=roles, templar=None
|
|
)
|
|
self.assertEqual(
|
|
merged["web-app-foo"]["services"]["foo"]["image"], "override"
|
|
)
|
|
|
|
|
|
class TestNestedLookupsReuseTheUnrenderedView(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_nested_calls_during_one_render_build_the_defaults_once(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
_write(
|
|
roles / "web-app-foo" / ROLE_FILE_META_SERVICES,
|
|
"""
|
|
foo:
|
|
image: "pre-{{ nested }}"
|
|
version: "v-{{ nested }}"
|
|
""",
|
|
)
|
|
variables = {
|
|
"applications": {
|
|
"web-app-foo": {"volumes": {"data": {"type": "volume"}}}
|
|
}
|
|
}
|
|
nested_views = []
|
|
|
|
class _NestingTemplar:
|
|
def __init__(self) -> None:
|
|
self.available_variables: dict = {}
|
|
|
|
def template(self, value, **_kwargs):
|
|
nested_views.append(
|
|
cache_apps.get_merged_applications(
|
|
variables=self.available_variables,
|
|
roles_dir=roles,
|
|
templar=self,
|
|
)
|
|
)
|
|
return "rendered"
|
|
|
|
with mock.patch.object(
|
|
cache_apps,
|
|
"get_application_defaults",
|
|
wraps=cache_apps.get_application_defaults,
|
|
) as defaults:
|
|
rendered = cache_apps.get_merged_applications(
|
|
variables=variables, roles_dir=roles, templar=_NestingTemplar()
|
|
)
|
|
|
|
self.assertEqual(defaults.call_count, 1)
|
|
self.assertGreaterEqual(len(nested_views), 2)
|
|
self.assertTrue(all(view is nested_views[0] for view in nested_views))
|
|
self.assertIn("volumes", nested_views[0]["web-app-foo"])
|
|
self.assertEqual(
|
|
rendered["web-app-foo"]["services"]["foo"]["image"], "rendered"
|
|
)
|
|
|
|
|
|
class TestRecursiveMetaWalk(unittest.TestCase):
|
|
"""`_build_role_base_config` walks the whole `meta/` tree: every
|
|
`meta/**/*.yml` lands at the matching nested path under
|
|
`applications.<app>` (the live `meta/addons/` case plus any deeper
|
|
nesting), while `main.yml`/`variants.yml` stay out as build mechanism.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_nested_dirs_map_to_nested_paths(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
role = roles / "web-app-foo"
|
|
_write(role / "meta" / "extra" / "widget.yml", "color: blue\n")
|
|
_write(role / "meta" / "a" / "b" / "c.yml", "deep: true\n")
|
|
cfg = cache_apps.get_application_defaults(roles_dir=roles)["web-app-foo"]
|
|
self.assertEqual(cfg["extra"]["widget"], {"color": "blue"})
|
|
self.assertEqual(cfg["a"]["b"]["c"], {"deep": True})
|
|
|
|
def test_main_and_variants_yml_stay_out_of_payload(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
role = roles / "web-app-foo"
|
|
_write(role / ROLE_FILE_META_MAIN, "galaxy_info:\n description: x\n")
|
|
_write(role / ROLE_FILE_META_VARIANTS, "- services: {foo: {image: alt}}\n")
|
|
cfg = cache_apps.get_application_defaults(roles_dir=roles)["web-app-foo"]
|
|
self.assertNotIn("main", cfg)
|
|
self.assertNotIn("variants", cfg)
|
|
|
|
|
|
class TestVolumesFoldedRawIntoMerged(unittest.TestCase):
|
|
"""`get_merged_applications` re-attaches the raw `meta/volumes.yml` subtree
|
|
onto `applications.<app>.volumes` AFTER the templar render, so the merged
|
|
SPOT carries volumes with their Jinja `source:` strings unrendered.
|
|
"""
|
|
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_volumes_present_and_raw(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
_write(
|
|
roles / "web-app-foo" / ROLE_FILE_META_VOLUMES,
|
|
"""
|
|
data:
|
|
type: volume
|
|
source: "{{ lookup('container', 'web-app-foo', 'x') }}"
|
|
""",
|
|
)
|
|
merged = cache_apps.get_merged_applications(
|
|
variables={}, roles_dir=roles, templar=None
|
|
)
|
|
vols = merged["web-app-foo"]["volumes"]
|
|
self.assertEqual(
|
|
vols["data"]["source"],
|
|
"{{ lookup('container', 'web-app-foo', 'x') }}",
|
|
)
|
|
|
|
def test_volumes_absent_when_role_has_none(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_roles(Path(tmp))
|
|
merged = cache_apps.get_merged_applications(
|
|
variables={}, roles_dir=roles, templar=None
|
|
)
|
|
self.assertNotIn("volumes", merged["web-app-foo"])
|
|
|
|
|
|
class TestApplicationsImportableWithoutAnsible(unittest.TestCase):
|
|
"""The CI runner-host CLI path
|
|
(`cli.administration.deploy.development.init` -> `plan_dev_inventory_matrix` ->
|
|
`get_variants`) MUST stay ansible-free. CI run 24935979190 broke
|
|
because `_build_variants` instantiated `ApplicationGidLookup`,
|
|
pulling `ansible.plugins.lookup.LookupBase` at call time. This
|
|
test pins both the import-time AND call-time invariants by
|
|
spawning a fresh subprocess so namespace-package / sys.modules
|
|
state cannot leak across tests.
|
|
"""
|
|
|
|
def test_module_imports_and_get_variants_callable_without_ansible(self):
|
|
import subprocess
|
|
|
|
repo_root = PROJECT_ROOT
|
|
snippet = (
|
|
"import sys\n"
|
|
f"sys.path.insert(0, {str(repo_root)!r})\n"
|
|
"class _Block:\n"
|
|
" def find_spec(self, name, path=None, target=None):\n"
|
|
" if name == 'ansible' or name.startswith('ansible.'):\n"
|
|
" raise ImportError(f'blocked: {name}')\n"
|
|
" return None\n"
|
|
"sys.meta_path.insert(0, _Block())\n"
|
|
"from utils.cache.applications import get_variants\n"
|
|
"assert callable(get_variants)\n"
|
|
"from utils.cache.base import ROLES_DIR\n"
|
|
"v = get_variants(roles_dir=ROLES_DIR)\n"
|
|
"assert len(v) > 0\n"
|
|
"print('OK', len(v))\n"
|
|
)
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", snippet],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(repo_root),
|
|
timeout=60,
|
|
check=False,
|
|
)
|
|
self.assertEqual(
|
|
result.returncode,
|
|
0,
|
|
msg=f"stderr=\n{result.stderr}\nstdout=\n{result.stdout}",
|
|
)
|
|
self.assertIn("OK", result.stdout)
|
|
|
|
|
|
class TestDeriveMcpPresence(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
@staticmethod
|
|
def _seed_pair(tmp: Path) -> Path:
|
|
roles = tmp / "roles"
|
|
_write(
|
|
roles / "web-app-prov" / ROLE_FILE_META_SERVICES,
|
|
"""
|
|
prov:
|
|
image: prov
|
|
version: latest
|
|
cons:
|
|
mcp_consumer: true
|
|
""",
|
|
)
|
|
_write(
|
|
roles / "web-app-prov" / ROLE_FILE_META_MCP,
|
|
"""
|
|
direction: server
|
|
""",
|
|
)
|
|
_write(
|
|
roles / "web-app-cons" / ROLE_FILE_META_SERVICES,
|
|
"""
|
|
cons:
|
|
image: cons
|
|
version: latest
|
|
mcp_consumer: true
|
|
""",
|
|
)
|
|
_write(
|
|
roles / "web-app-cons" / ROLE_FILE_META_MCP,
|
|
"""
|
|
direction: client
|
|
""",
|
|
)
|
|
return roles
|
|
|
|
def test_provider_resolves_to_its_admitted_consumers(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = self._seed_pair(Path(tmp))
|
|
defaults = cache_apps.get_application_defaults(roles_dir=roles)
|
|
self.assertEqual(
|
|
defaults["web-app-prov"]["mcp"]["enabled"],
|
|
"{{ 'web-app-cons' in group_names }}",
|
|
)
|
|
self.assertEqual(
|
|
defaults["web-app-prov"]["mcp"]["shared"],
|
|
defaults["web-app-prov"]["mcp"]["enabled"],
|
|
)
|
|
|
|
def test_provider_that_excludes_its_only_consumer_resolves_to_false(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = self._seed_pair(Path(tmp))
|
|
_write(
|
|
roles / "web-app-prov" / ROLE_FILE_META_SERVICES,
|
|
"""
|
|
prov:
|
|
image: prov
|
|
version: latest
|
|
cons:
|
|
mcp_consumer: false
|
|
""",
|
|
)
|
|
defaults = cache_apps.get_application_defaults(roles_dir=roles)
|
|
self.assertIs(defaults["web-app-prov"]["mcp"]["enabled"], False)
|
|
|
|
def test_configured_value_wins_over_the_derivation(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = self._seed_pair(Path(tmp))
|
|
_write(
|
|
roles / "web-app-prov" / ROLE_FILE_META_MCP,
|
|
"""
|
|
direction: server
|
|
enabled: false
|
|
shared: false
|
|
""",
|
|
)
|
|
defaults = cache_apps.get_application_defaults(roles_dir=roles)
|
|
self.assertIs(defaults["web-app-prov"]["mcp"]["enabled"], False)
|
|
|
|
def test_every_variant_carries_the_derived_value(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = self._seed_pair(Path(tmp))
|
|
_write(
|
|
roles / "web-app-prov" / ROLE_FILE_META_VARIANTS,
|
|
"""
|
|
- {}
|
|
- services:
|
|
prov:
|
|
image: prov-alt
|
|
""",
|
|
)
|
|
variants = cache_apps.get_variants(roles_dir=roles)["web-app-prov"]
|
|
self.assertEqual(len(variants), 2)
|
|
for variant in variants:
|
|
self.assertEqual(
|
|
variant["mcp"]["enabled"],
|
|
"{{ 'web-app-cons' in group_names }}",
|
|
)
|
|
|
|
def test_derivation_leaves_the_yaml_cache_untouched(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = self._seed_pair(Path(tmp))
|
|
mcp_path = roles / "web-app-prov" / ROLE_FILE_META_MCP
|
|
cache_apps.get_application_defaults(roles_dir=roles)
|
|
raw = load_yaml_any(str(mcp_path), default_if_missing={})
|
|
self.assertNotIn("enabled", raw)
|
|
self.assertNotIn("shared", raw)
|
|
|
|
def test_variant_override_wins_over_the_derivation(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = self._seed_pair(Path(tmp))
|
|
_write(
|
|
roles / "web-app-prov" / ROLE_FILE_META_VARIANTS,
|
|
"""
|
|
- {}
|
|
- mcp:
|
|
enabled: false
|
|
shared: false
|
|
""",
|
|
)
|
|
variants = cache_apps.get_variants(roles_dir=roles)["web-app-prov"]
|
|
self.assertEqual(
|
|
variants[0]["mcp"]["enabled"],
|
|
"{{ 'web-app-cons' in group_names }}",
|
|
)
|
|
self.assertIs(variants[1]["mcp"]["enabled"], False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|