From ea47169049129df7635d2f023e892a9d6af480fe Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 5 Jun 2026 18:46:22 -0400 Subject: [PATCH] feat(ansible): generate_inventory() + fetch_scope_targets() + unit tests --- steward/ansible/inventory_gen.py | 85 +++++++++++++++++++++++++++++ tests/core/test_inventory_gen.py | 94 ++++++++++++++++++++++++++++++++ 2 files changed, 179 insertions(+) create mode 100644 steward/ansible/inventory_gen.py create mode 100644 tests/core/test_inventory_gen.py diff --git a/steward/ansible/inventory_gen.py b/steward/ansible/inventory_gen.py new file mode 100644 index 0000000..09b230c --- /dev/null +++ b/steward/ansible/inventory_gen.py @@ -0,0 +1,85 @@ +"""Ansible --list inventory generation from DB records.""" +from __future__ import annotations +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from steward.models.ansible_inventory import AnsibleTarget + + +def generate_inventory(targets: list) -> dict: + """Build an Ansible --list JSON inventory from a list of AnsibleTarget objects. + + Accepts any objects with .name, .address, .ansible_vars (dict), .groups (list + of objects with .name and .ansible_vars). Uses duck typing so unit tests can + pass SimpleNamespace objects. + + Var precedence (lowest → highest, matching Ansible default): + group vars (sorted alphabetically by group name) → target ansible_vars → ansible_host + """ + inv: dict = {"all": {"hosts": []}, "_meta": {"hostvars": {}}} + + for target in targets: + inv["all"]["hosts"].append(target.name) + + # Merge vars: group vars alphabetically, then host vars, then force ansible_host. + merged: dict = {} + for group in sorted(target.groups, key=lambda g: g.name): + merged.update(group.ansible_vars or {}) + merged.update(target.ansible_vars or {}) + merged["ansible_host"] = target.address + + inv["_meta"]["hostvars"][target.name] = merged + + for group in target.groups: + if group.name not in inv: + inv[group.name] = { + "hosts": [], + "vars": dict(group.ansible_vars or {}), + } + inv[group.name]["hosts"].append(target.name) + + return inv + + +async def fetch_scope_targets(db, scope: str) -> list: + """Query AnsibleTarget objects for a given inventory scope string. + + Scopes: + steward:all — every target + steward:group: — targets in that group + steward:target: — single target by ID + + Returns a list with groups pre-loaded (selectinload). Returns [] for + repo:* scopes (caller handles those via legacy path). + """ + from sqlalchemy import select + from sqlalchemy.orm import selectinload + from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup + + if scope == "steward:all": + stmt = ( + select(AnsibleTarget) + .options(selectinload(AnsibleTarget.groups)) + .order_by(AnsibleTarget.name) + ) + elif scope.startswith("steward:group:"): + group_id = scope[len("steward:group:"):] + stmt = ( + select(AnsibleTarget) + .join(AnsibleTarget.groups) + .where(AnsibleGroup.id == group_id) + .options(selectinload(AnsibleTarget.groups)) + .order_by(AnsibleTarget.name) + ) + elif scope.startswith("steward:target:"): + target_id = scope[len("steward:target:"):] + stmt = ( + select(AnsibleTarget) + .where(AnsibleTarget.id == target_id) + .options(selectinload(AnsibleTarget.groups)) + ) + else: + return [] + + result = await db.execute(stmt) + return list(result.scalars().all()) diff --git a/tests/core/test_inventory_gen.py b/tests/core/test_inventory_gen.py new file mode 100644 index 0000000..f67f1cb --- /dev/null +++ b/tests/core/test_inventory_gen.py @@ -0,0 +1,94 @@ +"""Unit tests for generate_inventory(). + +Uses SimpleNamespace duck-typed objects so no DB or SQLAlchemy is required. +""" +from types import SimpleNamespace +import pytest +from steward.ansible.inventory_gen import generate_inventory + + +def _group(name, ansible_vars=None): + g = SimpleNamespace() + g.name = name + g.ansible_vars = ansible_vars or {} + return g + + +def _target(name, address, ansible_vars=None, groups=None): + t = SimpleNamespace() + t.name = name + t.address = address + t.ansible_vars = ansible_vars or {} + t.groups = groups or [] + return t + + +def test_empty_targets(): + result = generate_inventory([]) + assert result == {"all": {"hosts": []}, "_meta": {"hostvars": {}}} + + +def test_single_target_no_groups(): + target = _target("webserver", "192.168.1.10", {"ansible_user": "ubuntu"}) + result = generate_inventory([target]) + assert result["all"]["hosts"] == ["webserver"] + assert result["_meta"]["hostvars"]["webserver"]["ansible_host"] == "192.168.1.10" + assert result["_meta"]["hostvars"]["webserver"]["ansible_user"] == "ubuntu" + assert set(result.keys()) == {"all", "_meta"} + + +def test_ansible_host_overrides_group_var(): + """ansible_host is always target.address, even if a group sets it to something else.""" + grp = _group("servers", {"ansible_host": "10.0.0.1"}) + target = _target("host1", "192.168.1.50", groups=[grp]) + result = generate_inventory([target]) + assert result["_meta"]["hostvars"]["host1"]["ansible_host"] == "192.168.1.50" + + +def test_group_vars_applied(): + grp = _group("webservers", {"http_port": 80, "ansible_user": "deploy"}) + target = _target("web1", "192.168.1.10", groups=[grp]) + result = generate_inventory([target]) + hostvars = result["_meta"]["hostvars"]["web1"] + assert hostvars["http_port"] == 80 + assert hostvars["ansible_user"] == "deploy" + assert result["webservers"]["hosts"] == ["web1"] + assert result["webservers"]["vars"]["http_port"] == 80 + + +def test_host_vars_override_group_vars(): + """Per-target ansible_vars beat group vars for the same key.""" + grp = _group("g1", {"ansible_user": "group_user"}) + target = _target("t1", "1.2.3.4", {"ansible_user": "host_user"}, groups=[grp]) + result = generate_inventory([target]) + assert result["_meta"]["hostvars"]["t1"]["ansible_user"] == "host_user" + + +def test_group_var_precedence_is_alphabetical(): + """When multiple groups set the same key, alphabetically-last group wins (Ansible default).""" + g_a = _group("aaa", {"timeout": 10}) + g_b = _group("bbb", {"timeout": 20}) + target = _target("t1", "1.1.1.1", groups=[g_b, g_a]) # deliberately reversed order + result = generate_inventory([target]) + # Groups sorted alphabetically: aaa then bbb → bbb wins + assert result["_meta"]["hostvars"]["t1"]["timeout"] == 20 + + +def test_multiple_targets_in_group(): + grp = _group("db", {"db_port": 5432}) + t1 = _target("db1", "10.0.0.1", groups=[grp]) + t2 = _target("db2", "10.0.0.2", groups=[grp]) + result = generate_inventory([t1, t2]) + assert set(result["db"]["hosts"]) == {"db1", "db2"} + assert result["all"]["hosts"] == ["db1", "db2"] + + +def test_target_in_multiple_groups(): + g1 = _group("web", {"role": "web"}) + g2 = _group("prod", {"env": "production"}) + target = _target("web-prod-01", "192.168.1.1", groups=[g1, g2]) + result = generate_inventory([target]) + assert "web-prod-01" in result["web"]["hosts"] + assert "web-prod-01" in result["prod"]["hosts"] + assert result["_meta"]["hostvars"]["web-prod-01"]["role"] == "web" + assert result["_meta"]["hostvars"]["web-prod-01"]["env"] == "production"