mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-08 11:46:35 +00:00
kevinveenbirkenbachandClaude Opus 4.7
a8a117c86e
New lint guard
==============
tests/lint/repository/test_no_direct_yaml_calls.py AST-walks the
production tree (utils/, cli/, plugins/, filter_plugins/,
lookup_plugins/, roles/, scripts/, tasks/) and flags any direct
`yaml.safe_load` / `yaml.safe_dump` / `yaml.load` / `yaml.dump` /
`yaml.safe_load_all` / `yaml.dump_all` call. Aliased imports are
tracked: `import yaml as Y` ⇒ `Y.safe_load(...)` is flagged;
`from yaml import safe_load` ⇒ bare `safe_load(...)` is flagged.
Per-line opt-out: `# noqa: direct-yaml` on the call's physical line
(or any line spanned by a multi-line call). No central exemption
list — every legitimate exception lives next to its call with a
human-readable rationale in the surrounding comment.
Cache module additions
======================
* `dump_yaml_str(data)` — symmetric to `safe_dump(data)` with no file
path. Used for stdout output and filter return values where the
cache cannot help but the API needs to stay consistent.
* `load_yaml_str(text)` — symmetric to `safe_load(text)`. Used when
YAML comes from an HTTP body, a mutated buffer, or a mock-patched
source whose content does not correspond to a stable on-disk file.
Migrated production call sites (33 → 0 direct-yaml violations)
==============================================================
* utils/{entity_name_utils, valid_deploy_id, invokable, service_registry,
domains/list, applications/{config, in_group_deps},
docker/image/discovery, meta/{port_bands, scan},
roles/{dependency_resolver, meta_lookup, resource_validation}}
* cli/{build/{graph, inventory/full/command}, create/inventory/{host_vars,
mirror_overrides}, fix/vars_main_files, meta/{applications/in_group_deps,
callorder}, mirror/resolver, validate/inventory}
* plugins/{filter/{compose_volumes, get_all_application_ids, get_role,
invokable_paths, role_path_by_app_id}, lookup/applications_current_play}
* roles/{svc-db-postgres/filter_plugins/split_postgres_connections,
web-app-bigbluebutton/filter_plugins/compose_mods,
web-app-dashboard/lookup_plugins/docker_cards}
Behaviour adjustments
=====================
* utils/roles/dependency_resolver — `safe_load_all` (multi-doc) →
single-doc `load_yaml_any` wrapped in a list. Ansible task files are
single-doc by convention; the downstream multi-doc loop stays intact
for forward compatibility.
* roles/sys-svc-compose-ca/files/compose_ca.py — KEEPS direct
yaml.safe_load / yaml.safe_dump (annotated with noqa). The script is
copied to the target host via `ansible.builtin.copy` and the target
has no `utils/` package on PYTHONPATH.
noqa-annotated direct-yaml call sites (with rationale)
======================================================
* utils/cache/yaml.py — IS the cache; would recurse.
* utils/handler/yaml.py — vault-aware Loader subclass that wraps
\!vault scalars in VaultScalar; cache-by-path collapses the wrapping.
* cli/encrypt/inventory/__main__.py — vault-aware Dumper subclass that
emits \!vault literal blocks; the cached dump_yaml writes plain YAML
which is the wrong output here.
* roles/sys-svc-compose-ca/files/compose_ca.py — runtime-deployed.
* tasks/utils/migrate_meta_layout.py — once-off migration script.
Validation
==========
* Unit tests: 1573 passed.
* Lint: 56 passed (incl. new test_no_direct_yaml_calls).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
"""
|
|
Utility for validating deployment application IDs against defined roles and inventory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from utils.cache.yaml import load_yaml_any
|
|
|
|
from plugins.filter.get_all_application_ids import get_all_application_ids
|
|
|
|
|
|
class ValidDeployId:
|
|
def __init__(self) -> None:
|
|
"""
|
|
Always resolve roles/ from the repository root, independent of CWD.
|
|
"""
|
|
repo_root = Path(__file__).resolve().parents[1]
|
|
roles_dir = repo_root / "roles"
|
|
|
|
if not roles_dir.is_dir():
|
|
raise RuntimeError(
|
|
f"roles directory not found at expected location: {roles_dir}"
|
|
)
|
|
|
|
self.roles_dir = roles_dir
|
|
self.valid_ids = set(get_all_application_ids(str(roles_dir)))
|
|
|
|
def validate(
|
|
self, inventory_path: str, ids: list[str]
|
|
) -> dict[str, dict[str, bool]]:
|
|
"""
|
|
Validate a list of application IDs against both role definitions and inventory.
|
|
|
|
Returns:
|
|
{
|
|
"app1": {"in_roles": False, "in_inventory": True},
|
|
"app2": {"in_roles": True, "in_inventory": False},
|
|
}
|
|
"""
|
|
invalid: dict[str, dict[str, bool]] = {}
|
|
|
|
for app_id in ids:
|
|
in_roles = app_id in self.valid_ids
|
|
in_inventory = self._exists_in_inventory(inventory_path, app_id)
|
|
|
|
if not (in_roles and in_inventory):
|
|
invalid[app_id] = {
|
|
"in_roles": in_roles,
|
|
"in_inventory": in_inventory,
|
|
}
|
|
|
|
return invalid
|
|
|
|
def _exists_in_inventory(self, inventory_path: str, app_id: str) -> bool:
|
|
_, ext = os.path.splitext(inventory_path)
|
|
if ext in (".yml", ".yaml"):
|
|
return self._search_yaml_keys(inventory_path, app_id)
|
|
return self._search_ini_sections(inventory_path, app_id)
|
|
|
|
def _search_ini_sections(self, inventory_path: str, app_id: str) -> bool:
|
|
with open(inventory_path, "r", encoding="utf-8") as f:
|
|
current_section = None
|
|
for raw in f:
|
|
line = raw.strip()
|
|
if not line or line.startswith(("#", ";")):
|
|
continue
|
|
|
|
if line.startswith("[") and line.endswith("]"):
|
|
current_section = line[1:-1].strip()
|
|
if current_section == app_id:
|
|
return True
|
|
continue
|
|
|
|
if current_section:
|
|
for part in line.replace(",", " ").split():
|
|
if part.strip() == app_id:
|
|
return True
|
|
|
|
return False
|
|
|
|
def _search_yaml_keys(self, inventory_path: str, app_id: str) -> bool:
|
|
data = load_yaml_any(inventory_path, default_if_missing={})
|
|
return self._find_key(data, app_id)
|
|
|
|
def _find_key(self, node, key: str) -> bool:
|
|
if isinstance(node, dict):
|
|
for k, v in node.items():
|
|
if k == key and isinstance(v, (dict, list)):
|
|
return True
|
|
if self._find_key(v, key):
|
|
return True
|
|
elif isinstance(node, list):
|
|
for item in node:
|
|
if self._find_key(item, key):
|
|
return True
|
|
return False
|