cleveragents-core/robot/helper_plans_table_schema_alignment.py

"""Robot Framework helper for v3_plans schema alignment tests (issue #921).

Verifies effective_profile_snapshot, root_plan_id self-reference, and
automation_profile NOT NULL default.

Usage:
    python robot/helper_plans_table_schema_alignment.py snapshot-roundtrip
    python robot/helper_plans_table_schema_alignment.py root-self-ref
    python robot/helper_plans_table_schema_alignment.py child-root-id
"""

from __future__ import annotations

import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from sqlalchemy import create_engine  # noqa: E402
from sqlalchemy.orm import sessionmaker  # noqa: E402

from cleveragents.domain.models.core.action import (  # noqa: E402
    Action,
    ActionState,
)
from cleveragents.domain.models.core.plan import (  # noqa: E402
    AutomationProfileProvenance,
    AutomationProfileRef,
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
)
from cleveragents.infrastructure.database.models import Base  # noqa: E402
from cleveragents.infrastructure.database.repositories import (  # noqa: E402
    ActionRepository,
    LifecyclePlanRepository,
)

_NOW = datetime(2026, 3, 19, tzinfo=UTC)
_ACTION_NAME = "local/schema-test"


def _setup() -> tuple[Any, Any, Any]:
    """Create in-memory DB and return (session, plan_repo, action_repo)."""
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    sm = sessionmaker(bind=engine)
    session = sm()
    factory = lambda: session  # noqa: E731
    plan_repo = LifecyclePlanRepository(session_factory=factory)
    action_repo = ActionRepository(session_factory=factory)

    # Create prerequisite action for FK constraint
    ns = NamespacedName.parse(_ACTION_NAME)
    action = Action(
        namespaced_name=ns,
        description="Schema alignment test action",
        definition_of_done="Done",
        strategy_actor="local/s",
        execution_actor="local/e",
        state=ActionState.AVAILABLE,
        created_at=_NOW,
        updated_at=_NOW,
    )
    action_repo.create(action)
    session.commit()

    return session, plan_repo, action_repo


def _make_plan(
    plan_id: str,
    *,
    parent_plan_id: str | None = None,
    root_plan_id: str | None = None,
    effective_profile_snapshot: str = "{}",
    automation_profile: AutomationProfileRef | None = None,
) -> Plan:
    return Plan(
        identity=PlanIdentity(
            plan_id=plan_id,
            parent_plan_id=parent_plan_id,
            root_plan_id=root_plan_id,
            attempt=1,
        ),
        namespaced_name=NamespacedName(namespace="local", name="schema-test"),
        action_name=_ACTION_NAME,
        description="Schema alignment test plan",
        definition_of_done="All assertions pass",
        phase=PlanPhase.ACTION,
        processing_state=ProcessingState.QUEUED,
        automation_profile=automation_profile,
        effective_profile_snapshot=effective_profile_snapshot,
        strategy_actor="local/s",
        execution_actor="local/e",
        timestamps=PlanTimestamps(created_at=_NOW, updated_at=_NOW),
        created_by="test-schema",
        tags=[],
        reusable=True,
        read_only=False,
    )


def _cmd_snapshot_roundtrip() -> int:
    """Verify effective_profile_snapshot roundtrips through persistence."""
    session, plan_repo, _ = _setup()
    snapshot = '{"profile":"balanced","thresholds":{}}'
    plan = _make_plan(
        "01HV000000000000000000SR01",
        effective_profile_snapshot=snapshot,
    )
    plan_repo.create(plan)
    session.commit()

    retrieved = plan_repo.get("01HV000000000000000000SR01")
    if retrieved is None:
        print("FAIL: plan not retrieved")
        return 1
    if retrieved.effective_profile_snapshot != snapshot:
        print(
            f"FAIL: snapshot mismatch: {retrieved.effective_profile_snapshot!r} "
            f"!= {snapshot!r}"
        )
        return 1

    print("schema-snapshot-ok")
    return 0


def _cmd_root_self_ref() -> int:
    """Verify root plan self-references its own plan_id."""
    session, plan_repo, _ = _setup()
    plan_id = "01HV000000000000000000SR02"
    plan = _make_plan(plan_id, root_plan_id=None)
    plan_repo.create(plan)
    session.commit()

    retrieved = plan_repo.get(plan_id)
    if retrieved is None:
        print("FAIL: plan not retrieved")
        return 1
    if retrieved.identity.root_plan_id != plan_id:
        print(
            f"FAIL: root_plan_id={retrieved.identity.root_plan_id!r}, "
            f"expected={plan_id!r}"
        )
        return 1

    print("schema-root-self-ref-ok")
    return 0


def _cmd_child_root_id() -> int:
    """Verify child plan carries parent's root_plan_id."""
    session, plan_repo, _ = _setup()
    root_id = "01HV000000000000000000SR03"
    child_id = "01HV000000000000000000SR04"

    root_plan = _make_plan(root_id, root_plan_id=None)
    plan_repo.create(root_plan)
    session.commit()

    child_plan = _make_plan(child_id, parent_plan_id=root_id, root_plan_id=root_id)
    plan_repo.create(child_plan)
    session.commit()

    retrieved = plan_repo.get(child_id)
    if retrieved is None:
        print("FAIL: child plan not retrieved")
        return 1
    if retrieved.identity.root_plan_id != root_id:
        print(
            f"FAIL: root_plan_id={retrieved.identity.root_plan_id!r}, "
            f"expected={root_id!r}"
        )
        return 1
    if retrieved.identity.parent_plan_id != root_id:
        print(
            f"FAIL: parent_plan_id={retrieved.identity.parent_plan_id!r}, "
            f"expected={root_id!r}"
        )
        return 1

    print("schema-child-root-ok")
    return 0


def _cmd_grandchild_root_id() -> int:
    """Verify grandchild plan carries root ancestor's plan_id (3-level)."""
    session, plan_repo, _ = _setup()
    root_id = "01HV000000000000000000SR05"
    child_id = "01HV000000000000000000SR06"
    gc_id = "01HV000000000000000000SR07"

    root_plan = _make_plan(root_id, root_plan_id=None)
    plan_repo.create(root_plan)
    session.commit()

    child_plan = _make_plan(child_id, parent_plan_id=root_id, root_plan_id=root_id)
    plan_repo.create(child_plan)
    session.commit()

    gc_plan = _make_plan(gc_id, parent_plan_id=child_id, root_plan_id=root_id)
    plan_repo.create(gc_plan)
    session.commit()

    retrieved = plan_repo.get(gc_id)
    if retrieved is None:
        print("FAIL: grandchild plan not retrieved")
        return 1
    if retrieved.identity.root_plan_id != root_id:
        print(
            f"FAIL: root_plan_id={retrieved.identity.root_plan_id!r}, "
            f"expected={root_id!r}"
        )
        return 1
    if retrieved.identity.parent_plan_id != child_id:
        print(
            f"FAIL: parent_plan_id={retrieved.identity.parent_plan_id!r}, "
            f"expected={child_id!r}"
        )
        return 1

    print("schema-grandchild-root-ok")
    return 0


def _cmd_automation_profile_default() -> int:
    """Verify automation_profile=None roundtrips as None via 'balanced' default."""
    session, plan_repo, _ = _setup()
    plan_id = "01HV000000000000000000SR08"
    plan = _make_plan(plan_id)  # automation_profile defaults to None
    plan_repo.create(plan)
    session.commit()

    retrieved = plan_repo.get(plan_id)
    if retrieved is None:
        print("FAIL: plan not retrieved")
        return 1
    if retrieved.automation_profile is not None:
        print(
            f"FAIL: expected automation_profile=None, "
            f"got {retrieved.automation_profile!r}"
        )
        return 1

    # Verify raw column value is 'balanced'
    from sqlalchemy import text

    row = session.execute(
        text("SELECT automation_profile FROM v3_plans WHERE plan_id = :pid"),
        {"pid": plan_id},
    ).fetchone()
    if row is None or row[0] != "balanced":
        print(f"FAIL: raw column value={row[0] if row else 'N/A'}, expected='balanced'")
        return 1

    print("schema-automation-profile-default-ok")
    return 0


def _cmd_null_root_rejected() -> int:
    """Verify NULL root_plan_id is rejected by the NOT NULL constraint."""
    from sqlalchemy import text
    from sqlalchemy.exc import IntegrityError as SAIntegrityError

    session, _, _ = _setup()
    now_iso = _NOW.isoformat()
    try:
        session.execute(
            text(
                "INSERT INTO v3_plans "
                "(plan_id, root_plan_id, action_name, namespaced_name, "
                "namespace, phase, processing_state, attempt, description, "
                "tags_json, effective_profile_snapshot, automation_profile, "
                "reusable, read_only, created_at, updated_at) "
                "VALUES (:pid, NULL, :act, :ns, 'local', 'action', 'queued', "
                "1, 'test', '[]', '{}', 'balanced', 1, 0, :now, :now)"
            ),
            {
                "pid": "01HV000000000000000000SR09",
                "act": "local/schema-test",
                "ns": "local/schema-test",
                "now": now_iso,
            },
        )
        session.commit()
        print(
            "FAIL: expected IntegrityError for NULL root_plan_id, but insert succeeded"
        )
        return 1
    except SAIntegrityError:
        session.rollback()
        print("schema-null-root-rejected-ok")
        return 0


def _cmd_automation_profile_ref_roundtrip() -> int:
    """Verify AutomationProfileRef roundtrips through persistence."""
    session, plan_repo, _ = _setup()
    plan_id = "01HV000000000000000000SR10"
    ref = AutomationProfileRef(
        profile_name="supervised",
        provenance=AutomationProfileProvenance.PLAN,
    )
    plan = _make_plan(plan_id, automation_profile=ref)
    plan_repo.create(plan)
    session.commit()

    retrieved = plan_repo.get(plan_id)
    if retrieved is None:
        print("FAIL: plan not retrieved")
        return 1
    if retrieved.automation_profile is None:
        print("FAIL: automation_profile is None after round-trip")
        return 1
    if retrieved.automation_profile.profile_name != "supervised":
        print(
            f"FAIL: profile_name={retrieved.automation_profile.profile_name!r}, "
            f"expected='supervised'"
        )
        return 1
    if retrieved.automation_profile.provenance != AutomationProfileProvenance.PLAN:
        print(
            f"FAIL: provenance={retrieved.automation_profile.provenance!r}, "
            f"expected='plan'"
        )
        return 1

    print("schema-automation-profile-ref-roundtrip-ok")
    return 0


def _cmd_corrupted_snapshot_fallback() -> int:
    """Verify corrupted effective_profile_snapshot falls back to '{}'."""
    from sqlalchemy import text as sa_text

    session, plan_repo, _ = _setup()
    plan_id = "01HV000000000000000000SR11"
    plan = _make_plan(
        plan_id,
        effective_profile_snapshot='{"profile":"balanced"}',
    )
    plan_repo.create(plan)
    session.commit()

    # Corrupt the column directly
    session.execute(
        sa_text(
            "UPDATE v3_plans SET effective_profile_snapshot = :val WHERE plan_id = :pid"
        ),
        {"val": "not-valid-json{", "pid": plan_id},
    )
    session.commit()
    session.expire_all()

    retrieved = plan_repo.get(plan_id)
    if retrieved is None:
        print("FAIL: plan not retrieved after corruption")
        return 1
    if retrieved.effective_profile_snapshot != "{}":
        print(
            f"FAIL: expected fallback '{{}}', "
            f"got {retrieved.effective_profile_snapshot!r}"
        )
        return 1

    print("schema-corrupted-snapshot-fallback-ok")
    return 0


_COMMANDS: dict[str, Any] = {
    "snapshot-roundtrip": _cmd_snapshot_roundtrip,
    "root-self-ref": _cmd_root_self_ref,
    "child-root-id": _cmd_child_root_id,
    "grandchild-root-id": _cmd_grandchild_root_id,
    "automation-profile-default": _cmd_automation_profile_default,
    "automation-profile-ref-roundtrip": _cmd_automation_profile_ref_roundtrip,
    "null-root-rejected": _cmd_null_root_rejected,
    "corrupted-snapshot-fallback": _cmd_corrupted_snapshot_fallback,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} {{{','.join(_COMMANDS)}}}", file=sys.stderr)
        sys.exit(2)
    sys.exit(_COMMANDS[sys.argv[1]]())