resolutionflow/backend/tests/test_script_builder_inline.py

"""Integration tests for inline pilot_inline script_builder_session behavior.

Covers:
- Idempotent get-or-create for (user, ai_session_id) on origin='pilot_inline'
- Authorization: ai_session_id must belong to current user
- list_sessions + count_user_sessions default-scope to 'standalone'
"""
from __future__ import annotations

import pytest
from httpx import AsyncClient
from sqlalchemy import select, func
from uuid import uuid4

from app.models.ai_session import AISession
from app.models.script_builder_session import ScriptBuilderSession


async def _make_pilot_session(test_db, user) -> str:
    """Helper: create a minimal pilot session owned by `user`.

    Matches the existing pattern used by test_fix_outcome_endpoint.py.
    `user` is the dict returned by the test_user fixture:
        {"email": ..., "password": ..., "user_data": {"id": ..., "account_id": ..., ...}}
    """
    user_id = user["user_data"]["id"]
    account_id = user["user_data"]["account_id"]
    session = AISession(
        id=uuid4(), user_id=user_id, account_id=account_id,
        session_type="tshoot", intake_type="psa_ticket",
        intake_content={}, title="QA",
        status="active", confidence_tier="exploring", confidence_score=0.0,
    )
    test_db.add(session)
    await test_db.commit()
    return str(session.id)


@pytest.mark.asyncio
async def test_inline_create_is_idempotent(
    client: AsyncClient, test_user, auth_headers, test_db
):
    """Second create with same (user, ai_session_id) returns the existing row."""
    ai_session_id = await _make_pilot_session(test_db, test_user)

    r1 = await client.post(
        "/api/v1/scripts/builder/sessions",
        json={"language": "powershell", "origin": "pilot_inline",
              "ai_session_id": ai_session_id},
        headers=auth_headers,
    )
    assert r1.status_code in (200, 201), r1.text
    first_id = r1.json()["id"]

    r2 = await client.post(
        "/api/v1/scripts/builder/sessions",
        json={"language": "powershell", "origin": "pilot_inline",
              "ai_session_id": ai_session_id},
        headers=auth_headers,
    )
    assert r2.status_code in (200, 201)
    assert r2.json()["id"] == first_id

    # DB confirms only one row
    row_count = await test_db.scalar(
        select(func.count()).select_from(ScriptBuilderSession).where(
            ScriptBuilderSession.user_id == test_user["user_data"]["id"],
            ScriptBuilderSession.origin == "pilot_inline",
        )
    )
    assert row_count == 1


@pytest.mark.asyncio
async def test_inline_requires_ai_session_id(
    client: AsyncClient, auth_headers
):
    """origin='pilot_inline' without ai_session_id is rejected."""
    r = await client.post(
        "/api/v1/scripts/builder/sessions",
        json={"language": "powershell", "origin": "pilot_inline"},
        headers=auth_headers,
    )
    assert r.status_code == 400
    assert "ai_session_id" in r.text.lower()


@pytest.mark.asyncio
async def test_inline_ai_session_must_belong_to_caller(
    client: AsyncClient, test_user, auth_headers, test_db
):
    """ai_session_id pointing at another user's session is rejected."""
    # Create pilot session owned by a DIFFERENT user
    from app.models.user import User
    from app.models.account import Account
    other_account = Account(id=uuid4(), name="other", display_code="OTH-0001")
    test_db.add(other_account)
    await test_db.flush()
    other_user = User(
        id=uuid4(), email="other@example.com",
        password_hash="x", name="Other", role="engineer",
        is_super_admin=False, is_team_admin=False, is_active=True,
        is_service_account=False, must_change_password=False,
        account_id=other_account.id, account_role="engineer",
    )
    test_db.add(other_user)
    await test_db.flush()
    # Build user dict in the same shape as the test_user fixture
    other_user_dict = {
        "user_data": {"id": str(other_user.id), "account_id": str(other_account.id)}
    }
    other_session_id = await _make_pilot_session(test_db, other_user_dict)

    r = await client.post(
        "/api/v1/scripts/builder/sessions",
        json={"language": "powershell", "origin": "pilot_inline",
              "ai_session_id": other_session_id},
        headers=auth_headers,
    )
    assert r.status_code in (403, 404), r.text


@pytest.mark.asyncio
async def test_list_sessions_excludes_inline(
    client: AsyncClient, test_user, auth_headers, test_db
):
    """GET /scripts/builder/sessions returns only standalone rows."""
    ai_session_id = await _make_pilot_session(test_db, test_user)

    # Create one inline session
    await client.post(
        "/api/v1/scripts/builder/sessions",
        json={"language": "powershell", "origin": "pilot_inline",
              "ai_session_id": ai_session_id},
        headers=auth_headers,
    )
    # Create one standalone session
    await client.post(
        "/api/v1/scripts/builder/sessions",
        json={"language": "powershell"},
        headers=auth_headers,
    )

    r = await client.get("/api/v1/scripts/builder/sessions", headers=auth_headers)
    assert r.status_code == 200
    body = r.json()
    # Depending on response shape, this may be a list or {"sessions": [...]}.
    items = body if isinstance(body, list) else body.get("sessions", body.get("items", []))
    # Response schema does not surface `origin`; len==1 is the only meaningful guard:
    # inline row would push this to 2.
    assert len(items) == 1


@pytest.mark.asyncio
async def test_inline_sessions_do_not_count_against_cap(
    client: AsyncClient, test_user, auth_headers, test_db
):
    """Creating 5 pilot_inline sessions does not block a subsequent standalone."""
    # Create 5 distinct pilot sessions and attach inline builder sessions to each
    for _ in range(5):
        ai_session_id = await _make_pilot_session(test_db, test_user)
        r = await client.post(
            "/api/v1/scripts/builder/sessions",
            json={"language": "powershell", "origin": "pilot_inline",
                  "ai_session_id": ai_session_id},
            headers=auth_headers,
        )
        assert r.status_code in (200, 201), r.text

    # A standalone create should still succeed — inline sessions don't count
    r = await client.post(
        "/api/v1/scripts/builder/sessions",
        json={"language": "powershell"},
        headers=auth_headers,
    )
    assert r.status_code in (200, 201), r.text