FabledSteward/tests/integration/test_ansible_inventory.py

"""Integration tests: ansible_targets / ansible_groups tables + fetch_scope_targets.

Requires a live Postgres DB (STEWARD_DATABASE_URL).
Run with:
  pytest tests/integration/test_ansible_inventory.py -m integration -v
"""
from __future__ import annotations
import os
import uuid

import pytest
from sqlalchemy import select, text

pytestmark = pytest.mark.integration

_NEEDS_DB = pytest.mark.skipif(
    not os.environ.get("STEWARD_DATABASE_URL"),
    reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)",
)


def _make_app():
    # create_app is synchronous and runs its own internal asyncio.run for
    # settings/migrations — call it directly (matching the other integration
    # tests); wrapping it in asyncio.run nests event loops and fails.
    from steward.app import create_app
    return create_app(testing=False)


@_NEEDS_DB
def test_ansible_targets_table_exists():
    """Migration 0015 created the ansible_targets table."""
    app = _make_app()

    async def _check():
        async with app.db_sessionmaker() as db:
            result = await db.execute(
                text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_name = 'ansible_targets'"
                )
            )
            return {row[0] for row in result.fetchall()}

    import asyncio
    columns = asyncio.run(_check())
    assert "id" in columns
    assert "name" in columns
    assert "address" in columns
    assert "ansible_vars" in columns
    assert "host_id" in columns


@_NEEDS_DB
def test_ansible_groups_table_exists():
    """Migration 0016 created ansible_groups and ansible_target_groups."""
    app = _make_app()

    async def _check():
        async with app.db_sessionmaker() as db:
            groups_result = await db.execute(
                text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_name = 'ansible_groups'"
                )
            )
            groups_cols = {row[0] for row in groups_result.fetchall()}

            join_result = await db.execute(
                text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_name = 'ansible_target_groups'"
                )
            )
            join_cols = {row[0] for row in join_result.fetchall()}
        return groups_cols, join_cols

    import asyncio
    groups_cols, join_cols = asyncio.run(_check())
    assert "id" in groups_cols
    assert "name" in groups_cols
    assert "ansible_vars" in groups_cols
    assert "target_id" in join_cols
    assert "group_id" in join_cols


@_NEEDS_DB
def test_ansible_run_scope_column_exists():
    """Migration 0017 added inventory_scope to ansible_runs."""
    app = _make_app()

    async def _check():
        async with app.db_sessionmaker() as db:
            result = await db.execute(
                text(
                    "SELECT column_name FROM information_schema.columns "
                    "WHERE table_name = 'ansible_runs'"
                )
            )
            return {row[0] for row in result.fetchall()}

    import asyncio
    columns = asyncio.run(_check())
    assert "inventory_scope" in columns
    assert "inventory_path" in columns


@_NEEDS_DB
def test_create_target_and_group_and_fetch():
    """Can create a target+group, assign membership, and fetch via fetch_scope_targets."""
    from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup
    from steward.ansible.inventory_gen import fetch_scope_targets, generate_inventory

    app = _make_app()
    target_id = str(uuid.uuid4())
    group_id = str(uuid.uuid4())
    target_name = f"test-target-{target_id[:8]}"
    group_name = f"test-group-{group_id[:8]}"

    async def _run():
        # Create
        async with app.db_sessionmaker() as db:
            async with db.begin():
                grp = AnsibleGroup(
                    id=group_id,
                    name=group_name,
                    ansible_vars={"env": "test"},
                )
                tgt = AnsibleTarget(
                    id=target_id,
                    name=target_name,
                    address="10.0.99.1",
                    ansible_vars={"ansible_user": "ubuntu"},
                )
                tgt.groups.append(grp)
                db.add(grp)
                db.add(tgt)

        # Fetch all — target must be in the list
        async with app.db_sessionmaker() as db:
            all_targets = await fetch_scope_targets(db, "steward:all")
        assert any(t.id == target_id for t in all_targets)

        # Fetch by group
        async with app.db_sessionmaker() as db:
            group_targets = await fetch_scope_targets(db, f"steward:group:{group_id}")
        assert any(t.id == target_id for t in group_targets)

        # Fetch single target
        async with app.db_sessionmaker() as db:
            single = await fetch_scope_targets(db, f"steward:target:{target_id}")
        assert len(single) == 1
        assert single[0].id == target_id

        # Generate inventory and verify var merging
        async with app.db_sessionmaker() as db:
            targets = await fetch_scope_targets(db, f"steward:group:{group_id}")
        inv = generate_inventory(targets)
        assert target_name in inv["all"]["hosts"]
        assert target_name in inv[group_name]["hosts"]
        assert inv["_meta"]["hostvars"][target_name]["ansible_host"] == "10.0.99.1"
        assert inv["_meta"]["hostvars"][target_name]["ansible_user"] == "ubuntu"
        assert inv["_meta"]["hostvars"][target_name]["env"] == "test"

        # Cleanup
        async with app.db_sessionmaker() as db:
            async with db.begin():
                t = (await db.execute(
                    select(AnsibleTarget).where(AnsibleTarget.id == target_id)
                )).scalar_one_or_none()
                if t:
                    await db.delete(t)
                g = (await db.execute(
                    select(AnsibleGroup).where(AnsibleGroup.id == group_id)
                )).scalar_one_or_none()
                if g:
                    await db.delete(g)

    import asyncio
    asyncio.run(_run())