feat(ansible): generate_inventory() + fetch_scope_targets() + unit tests

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:<uuid>     — targets in that group
+      steward:target:<uuid>    — 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())

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"