cleveragents-core/robot/helper_invariant_model.py

"""Helper script for Robot Framework invariant model smoke tests."""

from __future__ import annotations

import sys
import uuid
from datetime import UTC, datetime
from pathlib import Path

# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import sessionmaker

from cleveragents.infrastructure.database.models import Base, InvariantModel


def _make_db() -> tuple[object, object]:
    """Create an in-memory SQLite DB and return (engine, session)."""
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    sm = sessionmaker(bind=engine)
    session = sm()
    return engine, session


def _make_invariant(description: str, is_active: bool = True) -> InvariantModel:
    return InvariantModel(
        id=str(uuid.uuid4()),
        description=description,
        created_at=datetime.now(tz=UTC).isoformat(),
        is_active=is_active,
    )


def _test_create_invariant() -> None:
    """Create an Invariant with all required fields and verify persistence."""
    _, session = _make_db()
    inv = _make_invariant("All plans must have a goal")
    session.add(inv)
    session.commit()

    retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
    assert retrieved is not None, "Invariant not found after persist"
    assert retrieved.description == "All plans must have a goal"
    print("invariant-create-ok")


def _test_default_is_active() -> None:
    """Verify that is_active defaults to True."""
    _, session = _make_db()
    inv = _make_invariant("Default active invariant")
    session.add(inv)
    session.commit()

    retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
    assert retrieved is not None
    assert retrieved.is_active is True or retrieved.is_active == 1, (
        f"Expected is_active=True, got {retrieved.is_active!r}"
    )
    print("invariant-default-active-ok")


def _test_created_at() -> None:
    """Verify that created_at is populated on insert."""
    _, session = _make_db()
    inv = _make_invariant("Timestamped invariant")
    session.add(inv)
    session.commit()

    retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
    assert retrieved is not None
    assert retrieved.created_at, "created_at should not be empty"
    assert len(str(retrieved.created_at)) > 0
    print("invariant-created-at-ok")


def _test_uuid_id() -> None:
    """Verify that the Invariant id is a valid UUID string."""
    _, session = _make_db()
    inv = _make_invariant("UUID invariant")
    session.add(inv)
    session.commit()

    retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
    assert retrieved is not None
    try:
        uuid.UUID(str(retrieved.id))
    except ValueError as exc:
        raise AssertionError(f"id '{retrieved.id}' is not a valid UUID") from exc
    print("invariant-uuid-ok")


def _test_query_active() -> None:
    """Query Invariants filtered by is_active=True."""
    _, session = _make_db()
    for i in range(3):
        session.add(_make_invariant(f"Active {i}", is_active=True))
    for i in range(2):
        session.add(_make_invariant(f"Inactive {i}", is_active=False))
    session.commit()

    results = session.query(InvariantModel).filter_by(is_active=True).all()
    assert len(results) == 3, f"Expected 3 active, got {len(results)}"
    print("invariant-query-active-ok")


def _test_query_inactive() -> None:
    """Query Invariants filtered by is_active=False."""
    _, session = _make_db()
    for i in range(3):
        session.add(_make_invariant(f"Active {i}", is_active=True))
    for i in range(2):
        session.add(_make_invariant(f"Inactive {i}", is_active=False))
    session.commit()

    results = session.query(InvariantModel).filter_by(is_active=False).all()
    assert len(results) == 2, f"Expected 2 inactive, got {len(results)}"
    print("invariant-query-inactive-ok")


def _test_deactivate() -> None:
    """Set is_active to False on an existing Invariant."""
    _, session = _make_db()
    inv = _make_invariant("Active invariant to deactivate", is_active=True)
    session.add(inv)
    session.commit()

    retrieved = session.query(InvariantModel).filter_by(id=inv.id).first()
    assert retrieved is not None
    retrieved.is_active = False
    session.commit()

    updated = session.query(InvariantModel).filter_by(id=inv.id).first()
    assert updated is not None
    assert updated.is_active is False or updated.is_active == 0, (
        f"Expected is_active=False, got {updated.is_active!r}"
    )
    print("invariant-deactivate-ok")


def _test_table_exists() -> None:
    """Verify the invariants table exists after schema creation."""
    engine, _ = _make_db()
    inspector = inspect(engine)
    tables = inspector.get_table_names()
    assert "invariants" in tables, f"Table 'invariants' not found. Available: {tables}"
    print("invariant-table-ok")


def _test_index_exists() -> None:
    """Verify the index on is_active exists after schema creation."""
    engine, _ = _make_db()
    inspector = inspect(engine)
    indexes = inspector.get_indexes("invariants")
    index_names = [idx["name"] for idx in indexes]
    assert any("is_active" in name for name in index_names), (
        f"No index on is_active found. Indexes: {index_names}"
    )
    print("invariant-index-ok")


_TESTS = {
    "create_invariant": _test_create_invariant,
    "default_is_active": _test_default_is_active,
    "created_at": _test_created_at,
    "uuid_id": _test_uuid_id,
    "query_active": _test_query_active,
    "query_inactive": _test_query_inactive,
    "deactivate": _test_deactivate,
    "table_exists": _test_table_exists,
    "index_exists": _test_index_exists,
}

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <test_name>")
        print(f"Available tests: {', '.join(sorted(_TESTS))}")
        sys.exit(1)

    test_name = sys.argv[1]
    if test_name not in _TESTS:
        print(f"Unknown test: {test_name}")
        print(f"Available: {', '.join(sorted(_TESTS))}")
        sys.exit(1)

    _TESTS[test_name]()